work in progress

Author: ycedres

src/mcp_server_uyuni/server.py

@@ -568,7 +568,7 @@ async def check_all_systems_for_updates(ctx: Context) -> List[Dict[str, Any]]:
     return systems_with_updates
 
 @write_tool()
-async def schedule_apply_pending_updates_to_system(system_identifier: Union[str, int], ctx: Context, confirm: bool = False) -> str:
+async def schedule_apply_pending_updates_to_system(system_identifier: Union[str, int], ctx: Context, confirm: Union[bool, str] = False) -> str:
 
     """
     Checks for pending updates on a system, schedules all of them to be applied,
@@ -580,7 +580,11 @@ async def schedule_apply_pending_updates_to_system(system_identifier: Union[str,
 
     Args:
         system_identifier: The unique identifier of the system. It can be the system name (e.g. "buildhost") or the system ID (e.g. 1000010000).
-        confirm: False by default. Only set confirm to True if the user has explicetely confirmed. Ask the user for confirmation.
+        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:
         str: The action url if updates were successfully scheduled.
@@ -590,7 +594,10 @@ async def schedule_apply_pending_updates_to_system(system_identifier: Union[str,
     logger.info(log_string)
     await ctx.info(log_string)
 
-    if not confirm:
+    # Coerce string "false" to boolean False for models that send strings
+    is_confirmed = str(confirm).lower() == 'true'
+
+    if not is_confirmed:
         return f"CONFIRMATION REQUIRED: This will apply pending updates to the system {system_identifier}.  Do you confirm?"
 
     # 1. Use check_system_updates to get relevant errata
@@ -642,15 +649,19 @@ async def schedule_apply_pending_updates_to_system(system_identifier: Union[str,
             return ""
 
 @write_tool()
-async def schedule_apply_specific_update(system_identifier: Union[str, int], errata_id: int, ctx: Context, confirm: bool = False) -> str:
+async def schedule_apply_specific_update(system_identifier: Union[str, int], errata_id: Union[str, int], ctx: Context, confirm: Union[bool, str] = False) -> str:
 
     """
     Schedules a specific update (erratum) to be applied to a system.
 
     Args:
         system_identifier: The unique identifier of the system. It can be the system name (e.g. "buildhost") or the system ID (e.g. 1000010000).
-        errata_id: The unique identifier of the erratum (also referred to as update ID) to be applied.
-        confirm: False by default. Only set confirm to True if the user has explicetely confirmed. Ask the user for confirmation.
+        errata_id: The unique identifier of the erratum (also referred to as update ID) to be applied. It must be an integer.
+        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:
         str: The action URL if the update was successfully scheduled.
@@ -659,41 +670,52 @@ async def schedule_apply_specific_update(system_identifier: Union[str, int], err
     log_string = f"Attempting to apply specific update (errata ID: {errata_id}) to system ID: {system_identifier}"
     logger.info(log_string)
     await ctx.info(log_string)
+
+    # Coerce string "false" to boolean False for models that send strings
+    is_confirmed = str(confirm).lower() == 'true'
+
+    try:
+        # Coerce errata_id to an integer, as some models might pass it as a string.
+        errata_id_int = int(errata_id)
+    except (ValueError, TypeError):
+        return f"Invalid errata ID '{errata_id}'. The ID must be an integer."
+
+
     system_id = await _resolve_system_id(system_identifier)
     if not system_id:
         return "" # Helper function already logged the reason for failure.
 
     print(f"Attempting to apply specific update (errata ID: {errata_id}) to system: {system_identifier}")
 
-    if not confirm:
+    if not is_confirmed:
         return f"CONFIRMATION REQUIRED: This will apply specific update (errata ID: {errata_id}) to the system {system_identifier}. Do you confirm?"
 
     async with httpx.AsyncClient(verify=UYUNI_MCP_SSL_VERIFY) as client:
         # The API expects a list of errata IDs, even if it's just one.
-        payload = {"sid": int(system_id), "errataIds": [errata_id]}
+        payload = {"sid": int(system_id), "errataIds": [errata_id_int]}
         api_result = await _call_uyuni_api(
             client=client,
             method="POST",
             api_path="/rhn/manager/api/system/scheduleApplyErrata",
             json_body=payload,
-            error_context=f"scheduling specific update (errata ID: {errata_id}) for system {system_identifier}",
+            error_context=f"scheduling specific update (errata ID: {errata_id_int}) for system {system_identifier}",
             default_on_error=None # Helper returns None on error
         )
 
         if isinstance(api_result, list) and api_result and isinstance(api_result[0], int):
             action_id = api_result[0]
-            success_message = f"Update (errata ID: {errata_id}) successfully scheduled for system {system_identifier}. Action URL: {UYUNI_SERVER}/rhn/schedule/ActionDetails.do?aid={action_id}"
+            success_message = f"Update (errata ID: {errata_id_int}) successfully scheduled for system {system_identifier}. Action URL: {UYUNI_SERVER}/rhn/schedule/ActionDetails.do?aid={action_id}"
             print(success_message)
             return success_message
         # Some schedule APIs might return int directly in result (though scheduleApplyErrata usually returns a list)
         elif isinstance(api_result, int): # Defensive check
             action_id = api_result
-            success_message = f"Update (errata ID: {errata_id}) successfully scheduled. Action URL: {UYUNI_SERVER}/rhn/schedule/ActionDetails.do?aid={action_id}"
+            success_message = f"Update (errata ID: {errata_id_int}) successfully scheduled. Action URL: {UYUNI_SERVER}/rhn/schedule/ActionDetails.do?aid={action_id}"
             print(success_message)
             return success_message
         else:
             if api_result is not None: # Log if not None but also not expected format
-                print(f"Failed to schedule specific update (errata ID: {errata_id}) for system {system_identifier} or unexpected API result format. Result: {api_result}")
+                print(f"Failed to schedule specific update (errata ID: {errata_id_int}) for system {system_identifier} or unexpected API result format. Result: {api_result}")
             return ""
 
 @write_tool()
@@ -705,7 +727,7 @@ async def add_system(
     ssh_user: str = "root",
     proxy_id: int = None,
     salt_ssh: bool = False,
-    confirm: bool = False,
+    confirm: Union[bool, str] = False,
 ) -> str:
     """
     Adds a new system to be managed by Uyuni.
@@ -721,10 +743,11 @@ async def add_system(
         ssh_user: The user to connect with via SSH (default: 'root').
         proxy_id: The system ID of a Uyuni proxy to use (optional).
         salt_ssh: Manage the system with Salt SSH (default: False).
-        confirm: User confirmation is required to execute this action. Set to False
-                 by default. If False, the tool returns a confirmation message. The
-                 model must present this message to the user and, if they agree, call
-                 the tool again with this parameter set to True.
+        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 confirmation message if 'confirm' is False.
@@ -735,6 +758,10 @@ async def add_system(
     log_string = f"Attempting to add system ID: {host}"
     logger.info(log_string)
     await ctx.info(log_string)
+
+    # Coerce string "false" to boolean False for models that send strings
+    is_confirmed = str(confirm).lower() == 'true'
+
     if ctx.session.check_client_capability(types.ClientCapabilities(elicitation=types.ElicitationCapability())):
         # Check for activation key
         if not activation_key:
@@ -761,7 +788,7 @@ async def add_system(
             await ctx.info(message)
             return message
 
-    if not confirm:
+    if not is_confirmed:
         return f"CONFIRMATION REQUIRED: This will add system {host} with activation key {activation_key} to Uyuni. Do you confirm?"
 
     ssh_priv_key_raw = os.environ.get('UYUNI_SSH_PRIV_KEY')
@@ -817,7 +844,7 @@ async def add_system(
 
 
 @write_tool()
-async def remove_system(system_identifier: Union[str, int], ctx: Context, cleanup: bool = True, confirm: bool = False) -> str:
+async def remove_system(system_identifier: Union[str, int], ctx: Context, cleanup: bool = True, confirm: Union[bool, str] = False) -> str:
     """
     Removes/deletes a system from being managed by Uyuni.
 
@@ -827,8 +854,11 @@ async def remove_system(system_identifier: Union[str, int], ctx: Context, cleanu
         system_identifier: The unique identifier of the system to remove. It can be the system name (e.g. "buildhost") or the system ID (e.g. 1000010000).
         cleanup: If True (default), Uyuni will attempt to run cleanup scripts on the client before deletion.
                  If False, the system is deleted from Uyuni without attempting client-side cleanup.
-        confirm: User confirmation is required. If False, the tool returns a confirmation prompt. The
-                 model must ask the user and call the tool again with confirm=True if they agree.
+        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 confirmation message if 'confirm' is False.
@@ -837,6 +867,10 @@ async def remove_system(system_identifier: Union[str, int], ctx: Context, cleanu
     log_string = f"Attempting to remove system with id {system_identifier}"
     logger.info(log_string)
     await ctx.info(log_string)
+
+    # Coerce string "false" to boolean False for models that send strings
+    is_confirmed = str(confirm).lower() == 'true'
+
     system_id = await _resolve_system_id(system_identifier)
     if not system_id:
         return "" # Helper function already logged the reason for failure.
@@ -848,7 +882,7 @@ async def remove_system(system_identifier: Union[str, int], ctx: Context, cleanu
         logger.warning(message)
         return message
 
-    if not confirm:
+    if not is_confirmed:
         return (f"CONFIRMATION REQUIRED: This will permanently remove system {system_id} from Uyuni. "
                 f"Client-side cleanup is currently {'ENABLED' if cleanup else 'DISABLED'}. Do you confirm?")
 
@@ -974,7 +1008,7 @@ async def get_systems_needing_security_update_for_cve(cve_identifier: str, ctx:
     return list(affected_systems_map.values())
 
 @mcp.tool()
-async def get_systems_needing_reboot(ctx: Context) -> List[Dict[str, Any]]:
+async def get_systems_needing_reboot(ctx: Context) -> List[Dict[str, Any]]: # No change needed here
     """
     Fetches a list of systems from the Uyuni server that require a reboot.
 
@@ -1024,14 +1058,18 @@ async def get_systems_needing_reboot(ctx: Context) -> List[Dict[str, Any]]:
     return systems_needing_reboot_list
 
 @write_tool()
-async def schedule_system_reboot(system_identifier: Union[str, int], ctx:Context, confirm: bool = False) -> str:
+async def schedule_system_reboot(system_identifier: Union[str, int], ctx:Context, confirm: Union[bool, str] = False) -> str:
 
     """
     Schedules an immediate reboot for a specific system on the Uyuni server.
 
     Args:
         system_identifier: The unique identifier of the system. It can be the system name (e.g. "buildhost") or the system ID (e.g. 1000010000).
-        confirm: False by default. Only set confirm to True if the user has explicetely confirmed. Ask the user for confirmation.
+        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`.
 
     The reboot is scheduled to occur as soon as possible (effectively "now").
     Returns:
@@ -1042,11 +1080,15 @@ async def schedule_system_reboot(system_identifier: Union[str, int], ctx:Context
     log_string = f"Schedule system reboot for system {system_identifier}"
     logger.info(log_string)
     await ctx.info(log_string)
+
+    # Coerce string "false" to boolean False for models that send strings
+    is_confirmed = str(confirm).lower() == 'true'
+
     system_id = await _resolve_system_id(system_identifier)
     if not system_id:
         return "" # Helper function already logged the reason for failure.
 
-    if not confirm:
+    if not is_confirmed:
         return f"CONFIRMATION REQUIRED: This will reboot system {system_identifier}. Do you confirm?"
 
     schedule_reboot_path = '/rhn/manager/api/system/scheduleReboot'
@@ -1126,7 +1168,7 @@ async def list_all_scheduled_actions(ctx: Context) -> List[Dict[str, Any]]:
     return processed_actions_list
 
 @write_tool()
-async def cancel_action(action_id: int, ctx: Context, confirm: bool = False) -> str:
+async def cancel_action(action_id: int, ctx: Context, confirm: Union[bool, str] = False) -> str:
     """
     Cancels a specified action on the Uyuni server.
 
@@ -1135,7 +1177,11 @@ async def cancel_action(action_id: int, ctx: Context, confirm: bool = False) ->
 
     Args:
         action_id: The integer ID of the action to be canceled.
-        confirm: False by default. Only set confirm to True if the user has explicetely confirmed. Ask the user for confirmation.
+        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:
         str: A success message if the action was canceled,
@@ -1148,12 +1194,15 @@ async def cancel_action(action_id: int, ctx: Context, confirm: bool = False) ->
     logger.info(log_string)
     await ctx.info(log_string)
 
+    # Coerce string "false" to boolean False for models that send strings
+    is_confirmed = str(confirm).lower() == 'true'
+
     cancel_actions_path = '/rhn/manager/api/schedule/cancelActions'
 
     if not isinstance(action_id, int): # Basic type check, though FastMCP might handle this
         return "Invalid action ID provided. Must be an integer."
 
-    if not confirm:
+    if not is_confirmed:
         return f"CONFIRMATION REQUIRED: This will schedule action {action_id} to be canceled. Do you confirm?"
 
     async with httpx.AsyncClient(verify=UYUNI_MCP_SSL_VERIFY) as client:

test/test_cases_ops.json

@@ -5,7 +5,7 @@
   },
   {
     "id": "TC-OPS-001_confirm_request",
-    "prompt": "Add a new system at host {new_system_host} with activation key '{key_deblike}'",
+    "prompt": "Can you add a new system at host {new_system_host} with activation key '{key_deblike}'?",
     "expected_output": "Return a message requiring confirmation."
   },
   {

test/test_cases_ops_2.json

@@ -5,7 +5,7 @@
   },
   {
     "id": "TC-OPS-002_confirm_request",
-    "prompt": "Remove system {new_system_host}",
+    "prompt": "Can you remove system {new_system_host}?",
     "expected_output": "Return a message requiring confirmation."
   },
   {

test/test_cases_sch.json

@@ -6,7 +6,7 @@
   },
   {
     "id": "TC-SCH-002_confirm_request",
-    "prompt": "Cancel an action with a valid ID.",
+    "prompt": "Can you cancel an action with a valid ID?",
     "expected_output": "Returns a message the requires confirmation"
   },
   {