在本指南中,你将学习如何构建一个 PydanticAI 智能体,使其 能够通过 ClickHouse 的 MCP 服务器 与 ClickHouse 的 SQL playground 交互。
前置条件
- 你的系统中需要安装 Python。
- 你的系统中需要安装
pip。 - 你需要一个 Anthropic API key,或其他 LLM 提供商的 API key。
你可以在 Python REPL 中或通过脚本执行以下步骤。
安装库
运行以下命令安装所需库:
pip install -q --upgrade pip
pip install -q "pydantic-ai-slim[mcp]"
pip install -q "pydantic-ai-slim[anthropic]" # 如果使用其他 LLM 提供商,请替换为相应的包配置凭据
接下来,你需要提供 Anthropic API key:
import os, getpass
os.environ["ANTHROPIC_API_KEY"] = getpass.getpass("Enter Anthropic API Key:")输入 Anthropic API Key: ········接下来,定义连接到 ClickHouse SQL playground 所需的凭据:
env = {
"CLICKHOUSE_HOST": "sql-clickhouse.clickhouse.com",
"CLICKHOUSE_PORT": "8443",
"CLICKHOUSE_USER": "demo",
"CLICKHOUSE_PASSWORD": "",
"CLICKHOUSE_SECURE": "true"
}初始化 MCP 服务器和 PydanticAI 智能体
现在,将 ClickHouse MCP 服务器配置为指向 ClickHouse SQL playground:
from pydantic_ai import Agent
from pydantic_ai.mcp import MCPServerStdio
from pydantic_ai.messages import ToolCallPart, ToolReturnPart
server = MCPServerStdio(
'uv',
args=[
'run',
'--with', 'mcp-clickhouse',
'--python', '3.13',
'mcp-clickhouse'
],
env=env
)
agent = Agent('anthropic:claude-sonnet-4-0', mcp_servers=[server])向智能体提问
最后,你可以向智能体提出一个问题:
async with agent.run_mcp_servers():
result = await agent.run("Who's done the most PRs for ClickHouse?")
print(result.output)你会得到类似下面的响应:
根据 ClickHouse GitHub 仓库中的数据,按创建的拉取请求数量统计,排名靠前的贡献者如下:
**按已发起 PR 数量统计的 ClickHouse 顶级贡献者:**
1. **alexey-milovidov** - 已发起 3,370 个 PR
2. **azat** - 已发起 1,905 个 PR
3. **rschu1ze** - 已发起 979 个 PR
4. **alesapin** - 已发起 947 个 PR
5. **tavplubix** - 已发起 896 个 PR
6. **kssenii** - 已发起 871 个 PR
7. **Avogar** - 已发起 805 个 PR
8. **KochetovNicolai** - 已发起 700 个 PR
9. **Algunenano** - 已发起 658 个 PR
10. **kitaisreal** - 已发起 630 个 PR
**Alexey Milovidov** 显然是迄今为止最活跃的贡献者,已发起超过 3,370 个拉取请求,远高于其他任何贡献者。考虑到 Alexey Milovidov 是 ClickHouse 的创始人和核心开发者之一,这并不令人意外。
数据还显示,alexey-milovidov 在管理 PR 方面也非常活跃。除了创建自己的 PR 之外,他还有 12,818 个“closed”事件(很可能是在审查并关闭其他贡献者的 PR)。
值得注意的是,我过滤掉了处理自动化流程的各类机器人/bot 账户,重点关注人工贡献者,以便更准确地回答谁为 ClickHouse 贡献了最多 PR。