mcp-clickhouse-long-running

Instructions

Implementation Reference

• mcp_clickhouse/mcp_server.py:156-189 (handler)

  The primary handler function for the 'run_select_query' MCP tool. It manages query execution asynchronously with timeout handling, error catching, and structured responses.
  @mcp.tool() 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: result = future.result(timeout=SELECT_QUERY_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 {SELECT_QUERY_TIMEOUT_SECS} seconds: {query}" ) future.cancel() # Return a properly structured response for timeout errors return { "status": "error", "message": f"Query timed out after {SELECT_QUERY_TIMEOUT_SECS} seconds", } except Exception as e: logger.error(f"Unexpected error in run_select_query: {str(e)}") # Catch all other exceptions and return them in a structured format # to prevent MCP serialization failures return {"status": "error", "message": f"Unexpected error: {str(e)}"}
• mcp_clickhouse/mcp_server.py:135-154 (helper)

  Supporting helper function that performs the actual ClickHouse query execution, processes results into a list of dictionaries, and returns structured errors if any occur.
  def execute_query(query: str): client = create_clickhouse_client() try: read_only = get_readonly_setting(client) res = client.query(query, settings={"readonly": read_only}) 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 a structured dictionary rather than a string to ensure proper serialization # by the MCP protocol. String responses for errors can cause BrokenResourceError. return {"error": str(err)}
• mcp_clickhouse/mcp_server.py:191-209 (helper)

  Helper function to create and test a ClickHouse client connection using configuration from mcp_env.py, used by query execution.
  def create_clickhouse_client(): client_config = get_config().get_client_config() logger.info( f"Creating ClickHouse client connection to {client_config['host']}:{client_config['port']} " f"as {client_config['username']} " f"(secure={client_config['secure']}, verify={client_config['verify']}, " f"connect_timeout={client_config['connect_timeout']}s, " f"send_receive_timeout={client_config['send_receive_timeout']}s)" ) try: client = clickhouse_connect.get_client(**client_config) # Test the connection version = client.server_version logger.info(f"Successfully connected to ClickHouse server version {version}") return client except Exception as e: logger.error(f"Failed to connect to ClickHouse: {str(e)}") raise
• mcp_clickhouse/__init__.py:1-13 (registration)

  Package __init__ exports the run_select_query tool function for import in tests and other modules.
  from .mcp_server import ( create_clickhouse_client, list_databases, list_tables, run_select_query, ) __all__ = [ "list_databases", "list_tables", "run_select_query", "create_clickhouse_client", ]