Trino MCP Server

Instructions

    Cancel a running query.
    
    Args:
        query_id: ID of the query to cancel.
        
    Returns:
        Dict[str, Any]: Result of the cancellation operation.
    

Implementation Reference

• src/trino_mcp/tools/__init__.py:78-112 (handler)

  MCP tool handler for cancel_query: calls TrinoClient.cancel_query and formats response.

  def cancel_query(query_id: str) -> Dict[str, Any]:
      """
      Cancel a running query.
      
      Args:
          query_id: ID of the query to cancel.
          
      Returns:
          Dict[str, Any]: Result of the cancellation operation.
      """
      logger.info(f"Cancelling query: {query_id}")
      
      try:
          success = client.cancel_query(query_id)
          
          if success:
              return {
                  "success": True,
                  "message": f"Query {query_id} cancelled successfully"
              }
          else:
              return {
                  "success": False,
                  "message": f"Failed to cancel query {query_id}"
              }
      
      except Exception as e:
          error_msg = str(e)
          logger.error(f"Query cancellation failed: {error_msg}")
          return {
              "success": False,
              "error": error_msg,
              "query_id": query_id
          }

• src/trino_mcp/server.py:84-85 (registration)

  Registration of Trino tools (including cancel_query) via register_trino_tools call during app lifespan.

  register_trino_resources(mcp, trino_client)
  register_trino_tools(mcp, trino_client)

• src/trino_mcp/trino_client.py:262-281 (helper)

  TrinoClient helper method implementing query cancellation via Trino system procedure.

  def cancel_query(self, query_id: str) -> bool:
      """
      Cancel a running query.
      
      Args:
          query_id: The ID of the query to cancel.
          
      Returns:
          bool: True if the query was successfully canceled, False otherwise.
      """
      self.ensure_connection()
      
      try:
          # Use system procedures to cancel the query
          self.execute_query(f"CALL system.runtime.kill_query(query_id => '{query_id}')")
          return True
      except Exception as e:
          logger.error(f"Failed to cancel query {query_id}: {e}")
          return False