run_select_query
Execute SELECT queries to retrieve data from ClickHouse databases, enabling structured data extraction and analysis.
Instructions
Run a SELECT query in a ClickHouse database
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| query | Yes |
Implementation Reference
- mcp_clickhouse/mcp_server.py:398-425 (handler)The main handler function for the 'run_select_query' tool. It executes the provided SELECT query using a thread pool executor with timeout handling, logs the query execution, and returns the result as a dictionary containing columns and rows, or an error structure.def run_select_query(query: str): """Run a SELECT query in a ClickHouse database""" logger.info(f"Executing SELECT query: {query}") try: future = QUERY_EXECUTOR.submit(execute_query, query) try: timeout_secs = get_mcp_config().query_timeout result = future.result(timeout=timeout_secs) # Check if we received an error structure from execute_query if isinstance(result, dict) and "error" in result: logger.warning(f"Query failed: {result['error']}") # MCP requires structured responses; string error messages can cause # serialization issues leading to BrokenResourceError return { "status": "error", "message": f"Query failed: {result['error']}", } return result except concurrent.futures.TimeoutError: logger.warning(f"Query timed out after {timeout_secs} seconds: {query}") future.cancel() raise ToolError(f"Query timed out after {timeout_secs} seconds") except ToolError: raise except Exception as e: logger.error(f"Unexpected error in run_select_query: {str(e)}") raise RuntimeError(f"Unexpected error during query execution: {str(e)}")
- mcp_clickhouse/mcp_server.py:562-566 (registration)Registration of the 'run_select_query' tool (along with other ClickHouse tools) to the MCP server, conditional on CLICKHOUSE_ENABLED being true.if os.getenv("CLICKHOUSE_ENABLED", "true").lower() == "true": mcp.add_tool(Tool.from_function(list_databases)) mcp.add_tool(Tool.from_function(list_tables)) mcp.add_tool(Tool.from_function(run_select_query)) logger.info("ClickHouse tools registered")
- mcp_clickhouse/mcp_server.py:386-396 (helper)Helper function called by run_select_query to actually execute the query on the ClickHouse client, handling readonly settings and returning structured results.def execute_query(query: str): client = create_clickhouse_client() try: read_only = get_readonly_setting(client) res = client.query(query, settings={"readonly": read_only}) logger.info(f"Query returned {len(res.result_rows)} rows") return {"columns": res.column_names, "rows": res.result_rows} except Exception as err: logger.error(f"Error executing query: {err}") raise ToolError(f"Query execution failed: {str(err)}")