MCP SSH Orchestrator

Instructions

Implementation Reference

• src/mcp_ssh/mcp_server.py:1353-1385 (handler)

  The handler function for the 'ssh_cancel' MCP tool. Validates the task_id input, signals cancellation via TASKS.cancel(), and returns a status dictionary.

  def ssh_cancel(task_id: str = "", ctx: Context | None = None) -> ToolResult:
      """Request cancellation for a running 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()
          ok = TASKS.cancel(task_id)
          response = {
              "task_id": task_id,
              "cancelled": bool(ok),
              "message": "Cancellation signaled" if ok else "Task not found",
          }
          _ctx_log(
              ctx,
              "info",
              "ssh_cancel",
              {"task_id": task_id, "cancelled": bool(ok)},
          )
          return response
      except Exception as e:
          error_str = str(e)
          log_json({"level": "error", "msg": "cancel_exception", "error": error_str})
          _ctx_log(
              ctx,
              "debug",
              "ssh_cancel_error",
              {"task_id": task_id.strip(), "error": sanitize_error(error_str)},
          )
          return f"Cancel error: {sanitize_error(error_str)}"

• src/mcp_ssh/mcp_server.py:1353-1353 (registration)

  The @mcp.tool() decorator registers the ssh_cancel function as an MCP tool.

  def ssh_cancel(task_id: str = "", ctx: Context | None = None) -> ToolResult:

• src/mcp_ssh/tools/utilities.py:147-154 (helper)

  The TaskManager.cancel() method, called by ssh_cancel handler to signal task cancellation by setting the threading.Event().

  def cancel(self, task_id: str) -> bool:
      """Signal cancellation for task id."""
      with self._lock:
          if task_id in self._tasks:
              self._tasks[task_id]["cancel"].set()
              return True
          return False

• src/mcp_ssh/tools/utilities.py:600-600 (helper)

  Instantiation of the TaskManager as TASKS, used by the ssh_cancel handler.

  TASKS = TaskManager()