diff --git a/src/mcp_server_uyuni/server.py b/src/mcp_server_uyuni/server.py index c244f8c..0259082 100644 --- a/src/mcp_server_uyuni/server.py +++ b/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.") diff --git a/test/test_cases_sys.json b/test/test_cases_sys.json index a383ae5..ea65d76 100644 --- a/test/test_cases_sys.json +++ b/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'." } -] +] \ No newline at end of file