cancel_query

Instructions

Implementation Reference

• sql_query_mcp/async_queries.py:182-230 (handler)

  The core handler method on AsyncQueryService that cancels a running query: sets job.status=CANCELLED, invokes the stored cancel_callback (if any), logs audit, and returns the updated job info.

  def cancel_query(self, query_id: str) -> Dict[str, object]:
      started = time.perf_counter()
      try:
          cancel_callback = None
          with self._lock:
              job = self._get_job_locked(query_id)
              if job.status == RUNNING:
                  job.status = CANCELLED
                  job.updated_at = time.time()
                  cancel_callback = job.cancel_callback
                  job.cancel_callback = None
              result: Dict[str, object] = {
                  "query_id": job.query_id,
                  "connection_id": job.connection_id,
                  "engine": job.engine,
                  "status": job.status,
              }
          cancel_error = None
          if cancel_callback is not None:
              try:
                  cancel_callback()
              except Exception as exc:
                  cancel_error = sanitize_error_message(str(exc))
          self._audit.log(
              tool="cancel_query",
              connection_id=job.connection_id,
              success=True,
              duration_ms=_elapsed_ms(started),
              sql_summary=job.sql_summary,
              extra={
                  "engine": job.engine,
                  "status": job.status,
                  "cancel_error": cancel_error,
              },
          )
          return result
      except Exception as exc:
          sanitized = sanitize_error_message(str(exc))
          self._audit.log(
              tool="cancel_query",
              connection_id=None,
              success=False,
              duration_ms=_elapsed_ms(started),
              error=sanitized,
              extra={"query_id": query_id},
          )
          if isinstance(exc, QueryExecutionError):
              raise
          raise QueryExecutionError(sanitized) from exc

• sql_query_mcp/app.py:105-109 (registration)

  Registers 'cancel_query' as an MCP tool via @mcp.tool() decorator in the FastMCP app, delegating to AsyncQueryService.cancel_query.

  @mcp.tool()
  def cancel_query(query_id: str) -> dict:
      """Cancel a running asynchronous query."""
  
      return _run_tool(lambda: async_queries.cancel_query(query_id))

• sql_query_mcp/async_queries.py:379-388 (helper)

  Helper that builds the cancel callback for a job by checking adapter.cancel_query, cursor.cancel, conn.cancel, or conn.close in priority order.

  def _build_cancel_callback(adapter: Any, conn: Any, cursor: Any) -> Optional[Callable[[], None]]:
      if hasattr(adapter, "cancel_query"):
          return lambda: adapter.cancel_query(conn, cursor)
      if hasattr(cursor, "cancel"):
          return cursor.cancel
      if hasattr(conn, "cancel"):
          return conn.cancel
      if hasattr(conn, "close"):
          return conn.close
      return None

• sql_query_mcp/async_queries.py:39-57 (handler)

  Data class for async query job state, including 'cancel_callback' field used by cancel_query to store the driver-level cancellation function.

  @dataclass
  class _AsyncQueryJob:
      query_id: str
      connection_id: str
      engine: str
      sql_text: str
      sql_summary: str
      applied_limit: int
      status: str = RUNNING
      active: bool = True
      columns: List[str] = field(default_factory=list)
      rows: List[object] = field(default_factory=list)
      row_count: int = 0
      truncated: bool = False
      duration_ms: Optional[int] = None
      error: Optional[str] = None
      created_at: float = field(default_factory=time.time)
      updated_at: float = field(default_factory=time.time)
      cancel_callback: Optional[Callable[[], None]] = None