execute_sql
Execute SQL queries against ClickHouse databases to retrieve, analyze, or modify data through a secure interface.
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 primary handler for the 'execute_sql' tool. Extracts the SQL query from arguments, performs a dangerous query check, executes the query using ClickHouseClient, formats the result as JSON, and returns it as TextContent or an error message.
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 input schema for the 'execute_sql' tool in the list_tools method, specifying 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 '_handle_execute_sql' method as the handler for the 'execute_sql' tool within the call_tool method's tool_handlers dictionary.
tool_handlers = { "execute_sql": self._handle_execute_sql } - Helper method in ClickHouseClient class that executes the SQL query on the ClickHouse database, converts results to list of dictionaries, and handles errors.
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 - Utility function that checks the SQL query for dangerous keywords like INSERT, UPDATE, etc., to prevent potential SQL injection or destructive operations.
def dangerous_check(query: str) -> tuple[bool, str]: """ Perform a security check on the query to prevent SQL injection attacks. This function checks for the presence of potentially dangerous keywords and patterns that could be used to inject malicious code. Args: query (str): The SQL query to be checked. Returns: tuple[bool, str]: A tuple containing a boolean indicating whether the query is dangerous and a string with the detected dangerous pattern, if any. """ # List of dangerous keywords and patterns dangerous_keywords = [ "insert", "update", "delete", "drop", "truncate", "alter" ] # Check for dangerous keywords for keyword in dangerous_keywords: if re.search(rf"\b{re.escape(keyword)}\b", query, re.IGNORECASE): logger.warning(f"Dangerous keyword '{keyword}' detected in query: {query}") return True, f"Query contains dangerous keyword '{keyword}'" return False, "Query is safe"