run_select_query
Execute SQL SELECT queries on ClickHouse MCP Server to retrieve and analyze data. Simplify database interactions by directly querying structured information.
Input Schema
| Name | Required | Description | Default |
|---|---|---|---|
| query | Yes |
Input Schema (JSON Schema)
{
"properties": {
"query": {
"title": "Query",
"type": "string"
}
},
"required": [
"query"
],
"title": "run_select_queryArguments",
"type": "object"
}
Implementation Reference
- mcp_clickhouse/mcp_server.py:84-102 (handler)The main handler function for the 'run_select_query' MCP tool. It executes the provided SELECT query on a ClickHouse database using a readonly setting, formats the result rows into a list of dictionaries with column names as keys, logs the operation, and returns an error message if the query fails.@mcp.tool() def run_select_query(query: str): logger.info(f"Executing SELECT query: {query}") client = create_clickhouse_client() try: res = client.query(query, settings={"readonly": 1}) column_names = res.column_names rows = [] for row in res.result_rows: row_dict = {} for i, col_name in enumerate(column_names): row_dict[col_name] = row[i] rows.append(row_dict) logger.info(f"Query returned {len(rows)} rows") return rows except Exception as err: logger.error(f"Error executing query: {err}") return f"error running query: {err}"
- mcp_clickhouse/mcp_server.py:104-115 (helper)Helper function used by 'run_select_query' (and other tools) to create a ClickHouse client connection based on environment variables.def create_clickhouse_client(): host = os.getenv("CLICKHOUSE_HOST") port = os.getenv("CLICKHOUSE_PORT") username = os.getenv("CLICKHOUSE_USER") logger.info(f"Creating ClickHouse client connection to {host}:{port} as {username}") return clickhouse_connect.get_client( host=host, port=port, username=username, password=os.getenv("CLICKHOUSE_PASSWORD"), )
- mcp_clickhouse/mcp_server.py:25-25 (registration)Initialization of the FastMCP server instance where tools like 'run_select_query' are registered via decorators.mcp = FastMCP(MCP_SERVER_NAME, dependencies=deps)