Support parameterized views in list tables; optimize row counts via system schema; update README with test steps and lint fixes.

Author: mstump

README.md

@@ -1,4 +1,5 @@
 # ClickHouse MCP Server
+
 [![PyPI - Version](https://img.shields.io/pypi/v/mcp-clickhouse)](https://pypi.org/project/mcp-clickhouse)
 
 An MCP server for ClickHouse.
@@ -10,22 +11,22 @@ An MCP server for ClickHouse.
 ### Tools
 
 * `run_select_query`
-  - Execute SQL queries on your ClickHouse cluster.
-  - Input: `sql` (string): The SQL query to execute.
-  - All ClickHouse queries are run with `readonly = 1` to ensure they are safe.
+  * Execute SQL queries on your ClickHouse cluster.
+  * Input: `sql` (string): The SQL query to execute.
+  * All ClickHouse queries are run with `readonly = 1` to ensure they are safe.
 
 * `list_databases`
-  - List all databases on your ClickHouse cluster.
+  * List all databases on your ClickHouse cluster.
 
 * `list_tables`
-  - List all tables in a database.
-  - Input: `database` (string): The name of the database.
+  * List all tables in a database.
+  * Input: `database` (string): The name of the database.
 
 ## Configuration
 
 1. Open the Claude Desktop configuration file located at:
-   - On macOS: `~/Library/Application Support/Claude/claude_desktop_config.json`
-   - On Windows: `%APPDATA%/Claude/claude_desktop_config.json`
+   * On macOS: `~/Library/Application Support/Claude/claude_desktop_config.json`
+   * On Windows: `%APPDATA%/Claude/claude_desktop_config.json`
 
 2. Add the following:
 
@@ -89,7 +90,6 @@ Or, if you'd like to try it out with the [ClickHouse SQL Playground](https://sql
 }
 ```
 
-
 3. Locate the command entry for `uv` and replace it with the absolute path to the `uv` executable. This ensures that the correct version of `uv` is used when starting the server. On a mac, you can find this path using `which uv`.
 
 4. Restart Claude Desktop to apply the changes.
@@ -102,7 +102,7 @@ Or, if you'd like to try it out with the [ClickHouse SQL Playground](https://sql
 
 *Note: The use of the `default` user in this context is intended solely for local development purposes.*
 
-```
+```bash
 CLICKHOUSE_HOST=localhost
 CLICKHOUSE_PORT=8123
 CLICKHOUSE_USER=default
@@ -118,36 +118,39 @@ CLICKHOUSE_PASSWORD=clickhouse
 The following environment variables are used to configure the ClickHouse connection:
 
 #### Required Variables
+
 * `CLICKHOUSE_HOST`: The hostname of your ClickHouse server
 * `CLICKHOUSE_USER`: The username for authentication
 * `CLICKHOUSE_PASSWORD`: The password for authentication
 
-> [!CAUTION]  
+> [!CAUTION]
 > It is important to treat your MCP database user as you would any external client connecting to your database, granting only the minimum necessary privileges required for its operation. The use of default or administrative users should be strictly avoided at all times.
 
 #### Optional Variables
+
 * `CLICKHOUSE_PORT`: The port number of your ClickHouse server
-  - Default: `8443` if HTTPS is enabled, `8123` if disabled
-  - Usually doesn't need to be set unless using a non-standard port
+  * Default: `8443` if HTTPS is enabled, `8123` if disabled
+  * Usually doesn't need to be set unless using a non-standard port
 * `CLICKHOUSE_SECURE`: Enable/disable HTTPS connection
-  - Default: `"true"`
-  - Set to `"false"` for non-secure connections
+  * Default: `"true"`
+  * Set to `"false"` for non-secure connections
 * `CLICKHOUSE_VERIFY`: Enable/disable SSL certificate verification
-  - Default: `"true"`
-  - Set to `"false"` to disable certificate verification (not recommended for production)
+  * Default: `"true"`
+  * Set to `"false"` to disable certificate verification (not recommended for production)
 * `CLICKHOUSE_CONNECT_TIMEOUT`: Connection timeout in seconds
-  - Default: `"30"`
-  - Increase this value if you experience connection timeouts
+  * Default: `"30"`
+  * Increase this value if you experience connection timeouts
 * `CLICKHOUSE_SEND_RECEIVE_TIMEOUT`: Send/receive timeout in seconds
-  - Default: `"300"`
-  - Increase this value for long-running queries
+  * Default: `"300"`
+  * Increase this value for long-running queries
 * `CLICKHOUSE_DATABASE`: Default database to use
-  - Default: None (uses server default)
-  - Set this to automatically connect to a specific database
+  * Default: None (uses server default)
+  * Set this to automatically connect to a specific database
 
 #### Example Configurations
 
 For local development with Docker:
+
 ```env
 # Required variables
 CLICKHOUSE_HOST=localhost
@@ -160,6 +163,7 @@ CLICKHOUSE_VERIFY=false
 ```
 
 For ClickHouse Cloud:
+
 ```env
 # Required variables
 CLICKHOUSE_HOST=your-instance.clickhouse.cloud
@@ -172,6 +176,7 @@ CLICKHOUSE_PASSWORD=your-password
 ```
 
 For ClickHouse SQL Playground:
+
 ```env
 CLICKHOUSE_HOST=sql-clickhouse.clickhouse.com
 CLICKHOUSE_USER=demo
@@ -204,6 +209,17 @@ You can set these variables in your environment, in a `.env` file, or in the Cla
   }
 }
 ```
+
+### Running tests
+
+```bash
+uv sync --all-extras --dev # install dev dependencies
+uv run ruff check . # run linting
+
+docker compose up -d test_services # start ClickHouse
+uv run pytest tests
+```
+
 ## YouTube Overview
 
 [![YouTube](http://i.ytimg.com/vi/y9biAm_Fkqw/hqdefault.jpg)](https://www.youtube.com/watch?v=y9biAm_Fkqw)

mcp_clickhouse/mcp_server.py

@@ -1,15 +1,50 @@
 import logging
-from typing import Sequence
+import json
+from typing import Optional, List, Any
 import concurrent.futures
 import atexit
 
 import clickhouse_connect
-from clickhouse_connect.driver.binding import quote_identifier, format_query_value
+from clickhouse_connect.driver.binding import format_query_value
 from dotenv import load_dotenv
 from mcp.server.fastmcp import FastMCP
+from dataclasses import dataclass, field, asdict, is_dataclass
 
 from mcp_clickhouse.mcp_env import get_config
 
+
+@dataclass
+class Column:
+    database: str
+    table: str
+    name: str
+    column_type: str
+    default_kind: Optional[str]
+    default_expression: Optional[str]
+    comment: Optional[str]
+
+
+@dataclass
+class Table:
+    database: str
+    name: str
+    engine: str
+    create_table_query: str
+    dependencies_database: str
+    dependencies_table: str
+    engine_full: str
+    sorting_key: str
+    primary_key: str
+    total_rows: int
+    total_bytes: int
+    total_bytes_uncompressed: int
+    parts: int
+    active_parts: int
+    total_marks: int
+    comment: Optional[str] = None
+    columns: List[Column] = field(default_factory=list)
+
+
 MCP_SERVER_NAME = "mcp-clickhouse"
 
 # Configure logging
@@ -34,6 +69,24 @@
 mcp = FastMCP(MCP_SERVER_NAME, dependencies=deps)
 
 
+def result_to_table(query_columns, result) -> List[Table]:
+    return [Table(**dict(zip(query_columns, row))) for row in result]
+
+
+def result_to_column(query_columns, result) -> List[Column]:
+    return [Column(**dict(zip(query_columns, row))) for row in result]
+
+
+def to_json(obj: Any) -> str:
+    if is_dataclass(obj):
+        return json.dumps(asdict(obj), default=to_json)
+    elif isinstance(obj, list):
+        return [to_json(item) for item in obj]
+    elif isinstance(obj, dict):
+        return {key: to_json(value) for key, value in obj.items()}
+    return obj
+
+
 @mcp.tool()
 def list_databases():
     """List available ClickHouse databases"""
@@ -45,85 +98,38 @@ def list_databases():
 
 
 @mcp.tool()
-def list_tables(database: str, like: str = None):
+def list_tables(
+    database: str, like: Optional[str] = None, not_like: Optional[str] = None
+):
     """List available ClickHouse tables in a database, including schema, comment,
     row count, and column count."""
     logger.info(f"Listing tables in database '{database}'")
     client = create_clickhouse_client()
-    query = f"SHOW TABLES FROM {quote_identifier(database)}"
+    query = f"SELECT database, name, engine, create_table_query, dependencies_database, dependencies_table, engine_full, sorting_key, primary_key, total_rows, total_bytes, total_bytes_uncompressed, parts, active_parts, total_marks, comment FROM system.tables WHERE database = {format_query_value(database)}"
     if like:
-        query += f" LIKE {format_query_value(like)}"
-    result = client.command(query)
+        query += f" AND name LIKE {format_query_value(like)}"
 
-    # Get all table comments in one query
-    table_comments_query = (
-        f"SELECT name, comment FROM system.tables WHERE database = {format_query_value(database)}"
-    )
-    table_comments_result = client.query(table_comments_query)
-    table_comments = {row[0]: row[1] for row in table_comments_result.result_rows}
-
-    # Get all column comments in one query
-    column_comments_query = f"SELECT table, name, comment FROM system.columns WHERE database = {format_query_value(database)}"
-    column_comments_result = client.query(column_comments_query)
-    column_comments = {}
-    for row in column_comments_result.result_rows:
-        table, col_name, comment = row
-        if table not in column_comments:
-            column_comments[table] = {}
-        column_comments[table][col_name] = comment
-
-    def get_table_info(table):
-        logger.info(f"Getting schema info for table {database}.{table}")
-        schema_query = f"DESCRIBE TABLE {quote_identifier(database)}.{quote_identifier(table)}"
-        schema_result = client.query(schema_query)
-
-        columns = []
-        column_names = schema_result.column_names
-        for row in schema_result.result_rows:
-            column_dict = {}
-            for i, col_name in enumerate(column_names):
-                column_dict[col_name] = row[i]
-            # Add comment from our pre-fetched comments
-            if table in column_comments and column_dict["name"] in column_comments[table]:
-                column_dict["comment"] = column_comments[table][column_dict["name"]]
-            else:
-                column_dict["comment"] = None
-            columns.append(column_dict)
-
-        # Get row count and column count from the table
-        row_count_query = (
-            f"SELECT count() FROM {quote_identifier(database)}.{quote_identifier(table)}"
-        )
-        row_count_result = client.query(row_count_query)
-        row_count = row_count_result.result_rows[0][0] if row_count_result.result_rows else 0
-        column_count = len(columns)
-
-        create_table_query = f"SHOW CREATE TABLE {database}.`{table}`"
-        create_table_result = client.command(create_table_query)
-
-        return {
-            "database": database,
-            "name": table,
-            "comment": table_comments.get(table),
-            "columns": columns,
-            "create_table_query": create_table_result,
-            "row_count": row_count,
-            "column_count": column_count,
-        }
-
-    tables = []
-    if isinstance(result, str):
-        # Single table result
-        for table in (t.strip() for t in result.split()):
-            if table:
-                tables.append(get_table_info(table))
-    elif isinstance(result, Sequence):
-        # Multiple table results
-        for table in result:
-            tables.append(get_table_info(table))
+    if not_like:
+        query += f" AND name NOT LIKE {format_query_value(not_like)}"
+
+    result = client.query(query)
+
+    # Deserialize result as Table dataclass instances
+    tables = result_to_table(result.column_names, result.result_rows)
+
+    for table in tables:
+        column_data_query = f"SELECT database, table, name, type AS column_type, default_kind, default_expression, comment FROM system.columns WHERE database = {format_query_value(database)} AND table = {format_query_value(table.name)}"
+        column_data_query_result = client.query(column_data_query)
+        table.columns = [
+            c
+            for c in result_to_column(
+                column_data_query_result.column_names,
+                column_data_query_result.result_rows,
+            )
+        ]
 
     logger.info(f"Found {len(tables)} tables")
-    return tables
+    return [asdict(table) for table in tables]
 
 
 def execute_query(query: str):
@@ -160,10 +166,15 @@ def run_select_query(query: str):
                 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 {
+                    "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}")
+            logger.warning(
+                f"Query timed out after {SELECT_QUERY_TIMEOUT_SECS} seconds: {query}"
+            )
             future.cancel()
             # Return a properly structured response for timeout errors
             return {