connect_database
Connect to a specified ClickHouse database for executing queries.
Instructions
Connect to a specific ClickHouse database
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| database | Yes |
Implementation Reference
- The main handler function for the 'connect_database' tool. It uses the MCP tool decorator (@mcp.tool()), takes a database name and context, executes a USE statement via QueryExecutor, and returns the result as JSON.
def connect_database(database: str, ctx: Context) -> str: """Connect to a specific ClickHouse database""" try: executor = _get_executor(ctx) result = executor.execute_single_query(f"USE {database}") return json.dumps(result, indent=2) except (ConnectionError, QueryError) as e: return str(e) - src/clickhouse_mcp_server/server.py:182-183 (registration)The tool is registered with the FastMCP server via the @mcp.tool() decorator on the connect_database function.
@mcp.tool() def connect_database(database: str, ctx: Context) -> str: - Helper function _get_executor creates the QueryExecutor from the MCP context's lifespan context (ClickHouseContext).
def _get_executor(ctx: Context) -> QueryExecutor: """Helper function to get QueryExecutor from context""" clickhouse_ctx = ctx.request_context.lifespan_context return QueryExecutor(clickhouse_ctx) - The execute_single_query method of QueryExecutor handles USE statements by switching the database context and executing the USE command on the ClickHouse client.
def execute_single_query(self, query: str) -> Dict[str, Any]: """Execute a single query and return results""" self.context.ensure_connected() try: # Handle USE statements if self._is_use_statement(query): db_name = query.strip().split()[-1].strip('`').strip() self.context.database = db_name self.context.client.execute(f'USE {db_name}') return {"message": f"Switched to database: {db_name}"} - The function signature defines the schema: takes a string 'database' parameter and returns a string.
def connect_database(database: str, ctx: Context) -> str: """Connect to a specific ClickHouse database"""