cancel_query
Stop a running query in Trino by providing its query ID to manage query execution and optimize resource usage.
Instructions
Cancel a running query.
Args:
query_id: ID of the query to cancel.
Returns:
Dict[str, Any]: Result of the cancellation operation.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| query_id | Yes |
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) - 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