ClickHouse自然语言查询代理

git clone <repository-url>
cd clickhouse-agent

pip install anthropic python-dotenv mcp clickhouse-connect

ANTHROPIC_API_KEY=your_api_key_here

python mcp_client.py

Query: What tables are available in the database?

Query: Show me the first 10 rows from the orders table

Query: Find all orders where the shipping city is different from the billing city

Query: Create a table called customer_data from the customers.csv file

Query: What are the top 5 products by total order amount?

clickhouse_client = clickhouse_connect.get_client(
    host='your_host',
    port=your_port,
    username='your_username',
    password='your_password'  # if required
)

SERVER_SCRIPT_PATH = "clickhouse_mcp_server.py"

clickhouse-agent/
├── clickhouse_mcp_server.py   # MCP server with ClickHouse tools
├── mcp_client.py               # AI-powered MCP client
├── .env                        # Environment variables (API keys)
├── README.md                   # This file
└── csv_files/                  # Sample CSV files (if any)

┌─────────────────┐
│   User Query    │
│  (Natural Lang) │
└────────┬────────┘
         │
         ▼
┌─────────────────┐
│   MCP Client    │
│  (mcp_client)   │
└────────┬────────┘
         │
         ▼
┌─────────────────┐
│   Claude API    │
│  (Anthropic)    │
└────────┬────────┘
         │
         ▼
┌─────────────────┐
│   MCP Server    │
│  (clickhouse)   │
└────────┬────────┘
         │
         ▼
┌─────────────────┐
│   ClickHouse    │
│    Database     │
└─────────────────┘

@mcp.tool()
async def your_new_tool(param1: str, param2: int) -> dict[str, Any]:
    """Description of what your tool does"""
    try:
        # Your implementation
        return {
            "status": "success",
            "data": result
        }
    except Exception as e:
        return {
            "status": "error",
            "message": f"Error: {e}"
        }