execute_sql
Run SQL queries on ClickHouse databases securely through an AI interface. Enables direct database interaction for data retrieval and analysis via controlled query execution.
Instructions
Execute a query against the ClickHouse database
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| query | Yes | The SQL query to be executed |
Implementation Reference
- The main handler function for the 'execute_sql' tool. It extracts the query from arguments, checks for dangerous SQL patterns, executes the query using ClickHouseClient.execute_query, formats the result as JSON, and returns it as TextContent. Handles errors appropriately.async def _handle_execute_sql(self, arguments: Dict[str, str]) -> List[TextContent]: """Handle execute_sql tool""" self.logger.debug("Handling execute_sql tool") # Get query query = arguments.get("query") if not query: self.logger.error("Query is required") return [] # Check query is_dangerous, pattern = dangerous_check(query) if is_dangerous: self.logger.error(f"Dangerous query detected: {pattern}") return [TextContent(value=f"Error: Dangerous query detected: {pattern}")] try: # Execute query result = self.client.execute_query(query) json_result = json.dumps(result, default=str, indent=2) return [ TextContent( type='text', text=json_result, mimeType='application/json' ) ] except Exception as e: self.logger.error(f"Failed to execute query: {e}") return [TextContent(type='text', text=f"Error executing query: {str(e)}")]
- Defines the tool schema returned by list_tools, specifying the inputSchema as an object with a required 'query' string property.Tool( name="execute_sql", description="Execute a query against the ClickHouse database", inputSchema={ "type": "object", "properties": { "query": { "type": "string", "description": "The SQL query to be executed" } }, "required": ["query"], } ) ]
- src/clickhouse_mcp_server/server.py:242-244 (registration)Registers the 'execute_sql' tool name to its handler function (_handle_execute_sql) in the tool_handlers dictionary used by call_tool.tool_handlers = { "execute_sql": self._handle_execute_sql }
- Helper method in ClickHouseClient that executes the SQL query on the ClickHouse server (with optional readonly mode), converts results to list of dictionaries, and returns them. Called by the tool handler.def execute_query(self, query: str, readonly: bool = True): """Execute a query against the ClickHouse database""" try: client = self.get_client() settings = {"readonly": 1} if readonly else {} res = client.query(query, settings=settings) # convert result to list of dicts rows = [] for row in res.result_rows: row_dict = {} for i, col_name in enumerate(res.column_names): row_dict[col_name] = row[i] rows.append(row_dict) self.logger.debug(f"Query executed successfully: {query}") return rows except Exception as e: self.logger.error(f"Failed to execute query: {e}") raise