跳到内容

客户端

PydanticAI 可以作为 MCP 客户端，连接到 MCP 服务器以使用它们的工具。

安装 (Install)

您需要安装 pydantic-ai，或者pydantic-ai-slim 以及 mcp 可选组

pip install "pydantic-ai-slim[mcp]"

uv add "pydantic-ai-slim[mcp]"

注意

MCP 集成需要 Python 3.10 或更高版本。

用法 (Usage)

PydanticAI 提供了两种连接到 MCP 服务器的方式

下面显示了两个示例； mcp-run-python 在两个示例中都用作 MCP 服务器。

SSE 客户端 (SSE Client)

MCPServerHTTP 通过 HTTP 使用 HTTP + 服务器发送事件传输 连接到服务器。

注意

MCPServerHTTP 需要在调用 agent.run_mcp_servers() 之前，MCP 服务器正在运行并接受 HTTP 连接。 运行服务器不由 PydanticAI 管理。

之所以使用名称 “HTTP”，是因为此实现将在未来进行调整，以使用目前正在开发中的新的 Streamable HTTP。

在创建 SSE 客户端之前，我们需要运行服务器（文档 这里）

终端 (运行 sse 服务器)
npx @pydantic/mcp-run-python sse
mcp_sse_client.py
from pydantic_ai import Agent
from pydantic_ai.mcp import MCPServerHTTP

server = MCPServerHTTP(url='https://127.0.0.1:3001/sse')  # (1)!
agent = Agent('openai:gpt-4o', mcp_servers=[server])  # (2)!


async def main():
    async with agent.run_mcp_servers():  # (3)!
        result = await agent.run('How many days between 2000-01-01 and 2025-03-18?')
    print(result.data)
    #> There are 9,208 days between January 1, 2000, and March 18, 2025.
  1. 使用用于连接的 URL 定义 MCP 服务器。
  2. 创建一个附加了 MCP 服务器的代理。
  3. 创建一个客户端会话以连接到服务器。

（此示例是完整的，可以使用 Python 3.10+ “按原样” 运行 — 您需要添加 asyncio.run(main()) 来运行 main）

这里发生了什么？

  • 模型正在接收提示 “2000-01-01 和 2025-03-18 之间有多少天？”
  • 模型决定 “哦，我有这个 run_python_code 工具，这将是回答这个问题的好方法”，并编写一些 python 代码来计算答案。
  • 模型返回一个工具调用
  • PydanticAI 使用 SSE 传输将工具调用发送到 MCP 服务器
  • 再次调用模型，并传入运行代码的返回值
  • 模型返回最终答案

您可以通过添加三行代码来使用 logfire 工具来可视化这一点，甚至可以看到运行的代码，从而检测示例

mcp_sse_client_logfire.py
import logfire

logfire.configure()
logfire.instrument_pydantic_ai()

将显示如下

Logfire run python code

MCP "stdio" 服务器 (Server)

MCP 提供的另一种传输是 stdio 传输，其中服务器作为子进程运行，并通过 stdin 和 stdout 与客户端通信。 在这种情况下，您将使用 MCPServerStdio 类。

注意

当使用 MCPServerStdio 服务器时，agent.run_mcp_servers() 上下文管理器负责启动和停止服务器。

mcp_stdio_client.py
from pydantic_ai import Agent
from pydantic_ai.mcp import MCPServerStdio

server = MCPServerStdio('npx', ['-y', '@pydantic/mcp-run-python', 'stdio'])
agent = Agent('openai:gpt-4o', mcp_servers=[server])


async def main():
    async with agent.run_mcp_servers():
        result = await agent.run('How many days between 2000-01-01 and 2025-03-18?')
    print(result.data)
    #> There are 9,208 days between January 1, 2000, and March 18, 2025.