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

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)}"

src/mcp_ssh/tools/utilities.py:533-554 (helper)

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()
```

MCP SSH Orchestrator