mcp-clickhouse

Instructions

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)}")