Enable system group functionalities

src/mcp_server_uyuni/server.py

@@ -1547,6 +1547,260 @@ async def get_unscheduled_errata(system_id: int, ctx: Context) -> List[Dict[str,
             logger.error(msg)
             return msg
 
+@mcp.tool()
+async def list_system_groups(ctx: Context) -> List[Dict[str, str]]:
+    """
+    Fetches a list of system groups from the Uyuni server.
+
+    This tool retrieves all system groups visible to the user and returns a list containing for
+    each group the identifier, name, description and system count.
+
+    Returns:
+        A list of dictionaries, where each dictionary represents a system group with 'id', 'name',
+        'description' and 'system_count' fields. The 'system_count' refers to the number of systems
+        assigned to each group.
+
+        Returns an empty list if the API call fails, the response is not in the expected format,
+        or no groups are found.
+
+        Example:
+            [
+                {
+                    "id": "1",
+                    "name": "Default Group",
+                    "description": "Default group for all systems",
+                    "system_count": "10"
+                },
+                {
+                    "id": "2",
+                    "name": "Test Group",
+                    "description": "Group for testing purposes",
+                    "system_count": "5"
+                }
+            ]
+    """
+    list_groups_path = '/rhn/manager/api/systemgroup/listAllGroups'
+
+    async with httpx.AsyncClient(verify=CONFIG["UYUNI_MCP_SSL_VERIFY"]) as client:
+        api_result = await call_uyuni_api(
+            client=client,
+            method="GET",
+            api_path=list_groups_path,
+            error_context="listing system groups",
+            token=ctx.get_state('token')
+        )
+
+    filtered_groups = []
+    if isinstance(api_result, list):
+        for group_data in api_result:
+            if isinstance(group_data, dict):
+                filtered_groups.append({'id': str(group_data.get('id')), 'name': group_data.get('name'),
+                    'description': group_data.get('description'), 'system_count': str(group_data.get('system_count'))})
+            else:
+                msg = f"Unexpected item format in system group list: {group_data}")
+                logger.warning(msg)
+                await ctx.warning(msg)
+    return filtered_groups
+
+@write_tool()
+async def create_system_group(name: str, ctx: Context, description: str = "", confirm: Union[bool, str] = False) -> str:
+    """
+    Creates a new system group in Uyuni.
+
+    Args:
+        name: The name of the new system group.
+        description: An optional description for the system group.
+        confirm: User confirmation is required to execute this action. This parameter
+                 is `False` by default. To obtain the confirmation message that must
+                 be presented to the user, the model must first call the tool with
+                 `confirm=False`. If the user agrees, the model should call the tool
+                 a second time with `confirm=True`.
+
+    Returns:
+        A success message if the group was created, e.g., "Successfully created system group 'my-group'".
+        Returns an error message if the creation failed.
+
+        Example:
+            Successfully created system group 'my-group'.
+    """
+    log_string = f"Creating system group '{name}'"
+    logger.info(log_string)
+    await ctx.info(log_string)
+
+    is_confirmed = _to_bool(confirm)
+
+    if not is_confirmed:
+        return f"CONFIRMATION REQUIRED: This will create a new system group named '{name}' with description '{description}'. Do you confirm?"
+
+    create_group_path = '/rhn/manager/api/systemgroup/create'
+
+    async with httpx.AsyncClient(verify=CONFIG["UYUNI_MCP_SSL_VERIFY"]) as client:
+        api_result = await call_uyuni_api(
+            client=client,
+            method="POST",
+            api_path=create_group_path,
+            json_body={"name": name, "description": description},
+            error_context=f"creating system group '{name}'",
+            token=ctx.get_state('token')
+        )
+
+        if isinstance(api_result, dict) and 'id' in api_result:
+             # The API returns the created group object
+            msg = f"Successfully created system group '{name}'."
+            logger.info(msg)
+        elif api_result:
+             # If it returns something truthy that we didn't expect, but not None (which is error)
+             msg = f"Successfully created system group '{name}'. (API returned: {api_result})"
+             logger.warning(msg)
+        else:
+            msg = f"Failed to create system group '{name}'. Check server logs."
+            logger.error(msg)
+
+@mcp.tool()
+async def list_group_systems(group_name: str, ctx: Context) -> List[Dict[str, Any]]:
+    """
+    Lists the systems in a system group.
+
+    Args:
+        group_name: The name of the system group.
+
+    Returns:
+        A list of dictionaries, where each dictionary represents a system with 'system_id' and
+        'system_name' fields.
+
+        Returns an empty list if the API call fails or no systems are found.
+
+        Example:
+            [
+                {
+                    "system_id": "123456789",
+                    "system_name": "my-system"
+                },
+                {
+                    "system_id": "987654321",
+                    "system_name": "my-other-system"
+                }
+            ]
+    """
+    log_string = f"Listing systems in group '{group_name}'"
+    logger.info(log_string)
+    await ctx.info(log_string)
+
+    list_systems_path = '/rhn/manager/api/systemgroup/listSystemsMinimal'
+
+    async with httpx.AsyncClient(verify=CONFIG["UYUNI_MCP_SSL_VERIFY"]) as client:
+        api_result = await call_uyuni_api(
+            client=client,
+            method="GET",
+            api_path=list_systems_path,
+            params={"groupName": group_name},
+            error_context=f"listing systems in group '{group_name}'",
+            token=ctx.get_state('token')
+        )
+
+    filtered_systems = []
+    if isinstance(api_result, list):
+        for system in api_result:
+            if isinstance(system, dict):
+                filtered_systems.append({
+                    'system_id': system.get('id'),
+                    'system_name': system.get('name')
+                })
+            else:
+                msg = f"Unexpected item format in group systems list: {system}"
+                logger.warning(msg)
+                await ctx.warning(msg)
+    return filtered_systems
+
+@write_tool()
+async def add_systems_to_group(group_name: str, system_identifiers: List[Union[str, int]], ctx: Context, confirm: Union[bool, str] = False) -> str:
+    """
+    Adds systems to a system group.
+
+    Args:
+        group_name: The name of the system group.
+        system_identifiers: A list of system names or IDs to add to the group.
+        confirm: User confirmation is required to execute this action. This parameter
+                 is `False` by default. To obtain the confirmation message that must
+                 be presented to the user, the model must first call the tool with
+                 `confirm=False`. If the user agrees, the model should call the tool
+                 a second time with `confirm=True`.
+
+    Returns:
+        A success message if the systems were added.
+
+        Example:
+            Successfully added 1 systems to/from group 'test-group'.
+    """
+    return await _manage_group_systems(group_name, system_identifiers, True, ctx, confirm)
+
+@write_tool()
+async def remove_systems_from_group(group_name: str, system_identifiers: List[Union[str, int]], ctx: Context, confirm: Union[bool, str] = False) -> str:
+    """
+    Removes systems from a system group.
+
+    Args:
+        group_name: The name of the system group.
+        system_identifiers: A list of system names or IDs to remove from the group.
+        confirm: User confirmation is required to execute this action. This parameter
+                 is `False` by default. To obtain the confirmation message that must
+                 be presented to the user, the model must first call the tool with
+                 `confirm=False`. If the user agrees, the model should call the tool
+                 a second time with `confirm=True`.
+
+    Returns:
+        A success message if the systems were removed.
+
+        Example:
+            Successfully removed 1 systems to/from group 'test-group'.
+    """
+    return await _manage_group_systems(group_name, system_identifiers, False, ctx, confirm)
+
+async def _manage_group_systems(group_name: str, system_identifiers: List[Union[str, int]], add: bool, ctx: Context, confirm: Union[bool, str] = False) -> str:
+    """
+    Internal helper to add or remove systems from a group.
+    """
+    action_str = ("add", "to") if add else ("remove", "from")
+    log_string = f"Attempting to {action_str[0]} systems {system_identifiers} {action_str[1]} group '{group_name}'"
+    logger.info(log_string)
+    await ctx.info(log_string)
+
+    is_confirmed = _to_bool(confirm)
+
+    if not is_confirmed:
+        return f"CONFIRMATION REQUIRED: This will {action_str[0]} {len(system_identifiers)} systems {action_str[1]} group '{group_name}'. Do you confirm?"
+
+    # Resolve all system IDs
+    resolved_ids = []
+    for identifier in system_identifiers:
+        sid = await _resolve_system_id(identifier)
+        if sid:
+            resolved_ids.append(int(sid))
+        else:
+            print(f"Warning: Could not resolve system identifier '{identifier}'. Skipping.")
+
+    if not resolved_ids:
+        return "No valid system identifiers found. Aborting."
+
+    add_remove_path = '/rhn/manager/api/systemgroup/addOrRemoveSystems'
+
+    async with httpx.AsyncClient(verify=CONFIG["UYUNI_MCP_SSL_VERIFY"]) as client:
+        api_result = await call_uyuni_api(
+            client=client,
+            method="POST",
+            api_path=add_remove_path,
+            json_body={"systemGroupName": group_name, "serverIds": resolved_ids, "add": add},
+            error_context=f"attempting to {action_str[0]} systems {action_str[1]} group '{group_name}'",
+            token=ctx.get_state('token')
+        )
+
+        if api_result == 1:
+            past_tense_action = "added" if add else "removed"
+            return f"Successfully {past_tense_action} {len(resolved_ids)} systems to/from group '{group_name}'."
+        else:
+            msg = f"Failed to {action_str[0]} systems. Check server logs. (API Result: {api_result})"
+            logger.error(msg)
+
 def main_cli():
 
     logger.info("Running Uyuni MCP server.")

test/test_cases_sys.json

@@ -1,6 +1,5 @@
 [
   {
-
     "id": "TC-SYS-001",
     "prompt": "Get the name and system id of all systems in the uyuni server.",
     "expected_output": "The response must contain the following system and ID pairs: \n* {build_host_name}: {build_host_id} \n* {deblike_minion_name}: {deblike_minion_id} \n* {proxy_name}: {proxy_id} \n* {rhlike_minion_name}: {rhlike_minion_id} \n* {suse_minion_name}: {suse_minion_id} \n* {suse_ssh_minion_name}: {suse_ssh_minion_id}"
@@ -29,5 +28,15 @@
     "id": "TC-SYS-006",
     "prompt": "Get CPU details for system '{build_host_name}'.",
     "expected_output": "Returns a message with CPU attributes of model {build_host_cpu_model}."
+  },
+  {
+    "id": "TC-SYS-007",
+    "prompt": "Create a new system group named 'test-group' with description 'Test Group Description'.",
+    "expected_output": "CONFIRMATION REQUIRED: This will create a new system group named 'test-group' with description 'Test Group Description'. Do you confirm?"
+  },
+  {
+    "id": "TC-SYS-008",
+    "prompt": "Create a new system group named 'test-group' with description 'Test Group Description'. Confirm yes.",
+    "expected_output": "Successfully created system group 'test-group'."
   }
-]
+]