ssh_cancel_async_task
Stop a running asynchronous SSH task to manage server operations and control resource usage within the MCP SSH Orchestrator environment.
Instructions
Cancel a running async task.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| task_id | No |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| result | Yes |
Implementation Reference
- src/mcp_ssh/mcp_server.py:1622-1662 (handler)The main handler function for the 'ssh_cancel_async_task' MCP tool. It validates the task_id input, invokes ASYNC_TASKS.cancel_task(), logs the event, and returns a JSON response indicating success or failure.
@mcp.tool() def ssh_cancel_async_task(task_id: str = "", ctx: Context | None = None) -> ToolResult: """Cancel a running async task.""" try: # Input validation valid, error_msg = _validate_task_id(task_id) if not valid: return f"Error: {error_msg}" task_id = task_id.strip() success = ASYNC_TASKS.cancel_task(task_id) response = { "task_id": task_id, "cancelled": bool(success), "message": ( "Cancellation signaled" if success else "Task not found or not cancellable" ), } _ctx_log( ctx, "info", "ssh_cancel_async_task", {"task_id": task_id, "cancelled": bool(success)}, ) return response except Exception as e: error_str = str(e) log_json( {"level": "error", "msg": "cancel_async_exception", "error": error_str} ) _ctx_log( ctx, "debug", "ssh_cancel_async_task_error", {"task_id": task_id.strip(), "error": sanitize_error(error_str)}, ) return f"Cancel error: {sanitize_error(error_str)}" - Core cancellation logic in AsyncTaskManager.cancel_task(). Sets the task's cancel event, updates status to 'cancelled', sends a notification, and returns True if the task was cancellable.
def cancel_task(self, task_id: str) -> bool: """Cancel a running task.""" with self._lock: task_info = self._tasks.get(task_id) if task_info and task_info["status"] in ["pending", "running"]: task_info["cancel"].set() task_info["status"] = "cancelled" # Send cancellation notification self._send_notification( "cancelled", task_id, { "reason": "user_requested", "max_seconds": int( task_info.get("limits", {}).get("max_seconds", 60) ), }, ) return True return False - src/mcp_ssh/mcp_server.py:1622-1622 (registration)The @mcp.tool() decorator registers the function as an MCP tool named 'ssh_cancel_async_task'.
@mcp.tool()