ClickHouse MCP Server

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)