PydanticAI代理利用MCP服务器

模型上下文协议（MCP）是Anthropic引入的一项倡议，旨在定义一种标准化的方式，使AI代理能够连接到外部上下文，例如提供信息的资源或执行操作的工具。其基本想法是一个MCP服务器提供一组工具或资源，一个MCP客户端可以连接到一个或多个服务器以访问其功能并将其提供给AI代理。

这旨在消除提供上下文或工具的服务与需要访问它们的AI框架或模型之间的耦合。了解更多请访问这里！

PydanticAI 是一种用Python构建生产级应用的AI代理框架，旨在让生成式AI的应用开发更加轻松愉快。

PydanticAI是一个比笨重的替代品如LangChain更直观和灵活的框架来开发AI代理。然而，它目前还不支持MCP，因此要创建一个能够利用通过协议提供的所有酷炫功能的AI代理并不那么简单。我决定借此机会深入研究MCP并找出如何“从头开始”集成它！（提示：它并不美观）。

1、准备MCP服务器

我希望专注于将现有的MCP服务器实现与PydanticAI代理连接起来，而不是自己创建自己的服务器（因为这就是协议的目的！）。幸运的是，有一个MCP服务器参考实现的仓库，提供了访问Slack、Google Drive、Github以及各种数据库等服务的基本MCP服务器。为了保持简单和无聊，我决定使用文件系统服务器来帮助创建一个能够协助软件开发的代理，因为它能够读取和编辑文件。

它是作为一个简单的Node包/工具实现的，可以在本地运行并通过STDIO进行通信：

npx -y @modelcontextprotocol/server-filesystem "/path/to/working/dir"

经过一些测试后，我发现文件搜索和目录树工具存在一些问题，在代码库中使用时变得难以处理，因为它们在node_modules或Python虚拟环境中被内容淹没，或者IDE通常忽略的其他文件。因此，我做了一个MCP服务器实现的副本并添加了一些自定义增强功能，使其相关工具尊重任何.gitignore规则，这样它们的输出会更有用。（我还提交了一个PR来修复search_files工具的行为）。

2、向PydanticAI代理提供工具

下一个挑战是如何将MCP服务器提供的功能注册为PydanticAI代理的工具。使用MCP Python SDK启动MCP服务器、连接到客户端并查看可用工具看起来像这样：

from mcp import ClientSession, StdioServerParameters  
from mcp.client.stdio import stdio_client  

# 定义服务器配置  
server_params = StdioServerParameters(  
      command="npx",  
      args=[  
          "tsx",  
          "server/index.ts",  
          "path/to/working/directory",  
      ],  
  )  

# 启动服务器并连接客户端  
async with stdio_client(server_params) as (read, write):  
    async with ClientSession(read, write) as session:  
        # 初始化连接  
        await session.initialize()  
          
        # 获取可用工具的详细信息  
        tools_result = await session.list_tools()  
        tools = tools_result.tools

这里，tools是一个Tool Pydantic模型列表，它定义了每个可用工具的名称、描述和输入模式，基本上与大多数LLM API要求工具规范提供的格式相同。然而，不幸的是，PydanticAI目前还不支持直接提供函数工具的输入模式；它只能通过检查函数签名来构造它。

因此，为了绕过这个限制，我需要构建一种能力，将MCP工具规范转换为动态函数定义，只是为了能让PydanticAI再次将其转换回JSON工具规范。总体上，它的工作方式如下：

async def get_tools(session: ClientSession) -> list[Tool[AgentDeps]]:  
    """  
    从MCP会话获取所有工具并将其转换为Pydantic AI工具。  
    """  
    tools_result = await session.list_tools()  
    return [pydantic_tool_from_mcp_tool(session, tool) for tool in tools_result.tools]  
  
  
def pydantic_tool_from_mcp_tool(session: ClientSession, tool: MCPTool) -> Tool[AgentDeps]:  
    """  
    将MCP工具转换为Pydantic AI工具。  
    """  
    tool_function = create_function_from_schema(session=session, name=tool.name, schema=tool.inputSchema)  
    return Tool(name=tool.name, description=tool.description, function=tool_function, takes_ctx=True)

对于create_function_from_schema()的实现，关键部分是我们需要为每个工具定义一个实际请求MCP服务器的函数，该函数将执行实际操作并返回响应。这看起来像这样：

def create_function_from_schema(session: ClientSession, name: str, schema: Dict[str, Any]) -> types.FunctionType:  
    """  
    根据JSON模式创建具有签名的函数。这是必要的，因为PydanticAI目前还不支持直接提供工具JSON模式。  
    """  
    # 从工具模式创建参数列表  
    parameters = convert_schema_to_params(schema)  
  
    # 创建签名  
    sig = inspect.Signature(parameters=parameters)  
  
    # 创建函数体  
    async def function_body(ctx: RunContext[AgentDeps], **kwargs) -> str:  
        # 使用提供的参数调用MCP工具  
        result = await session.call_tool(name, arguments=kwargs)  
  
        # 假设响应始终是TextContent  
        if isinstance(result.content[0], TextContent):  
            return result.content[0].text  
        else:  
            raise ValueError("Expected TextContent, got ", type(result.content[0]))  
  
    # 使用正确的签名创建函数  
    dynamic_function = types.FunctionType(  
        function_body.__code__,  
        function_body.__globals__,  
        name=name,  
        argdefs=function_body.__defaults__,  
        closure=function_body.__closure__,  
    )  
  
    # 添加签名和注解  
    dynamic_function.__signature__ = sig  # type: ignore  
    dynamic_function.__annotations__ = {param.name: param.annotation for param in parameters}  
  
    return dynamic_function

对于那些勇敢或自虐到想要看看convert_schema_to_params()实际涉及的内容的人，请看这里。感谢Claude/Cursor帮我做了这个脏活……我只能祈祷PydanticAI团队尽快合并这个PR，这样其他人就不用再接触这个怪物了。

3、组装在一起

现在我们已经有了标准的PydanticAI 工具列表，可以直接提供给代理正常使用，例如：

from pydantic_ai import Agent  
from rich.console import Console  
from rich.prompt import Prompt  
...  
  
def run():    
    # 配置模型和代理依赖项  
    ...  
    # 初始化并连接MCP服务器，构建工具  
    ...  
  
    agent = Agent(  
        model=model,  
        deps_type=type(deps),  
        system_prompt=SYSTEM_PROMPT,  
        tools=tools,  
    )  
  
    message_history: list[ModelMessage] = []  
    while True:  
        prompt = Prompt.ask("[cyan]>[/cyan] ").strip()  
  
        if not prompt:  
            continue  
  
        # 处理s特殊命令  
        if prompt.lower() in EXIT_COMMANDS:  
            break  
  
        # 通过代理处理正常输入  
        result = await agent.run(prompt, deps=deps, message_history=message_history)  
        response = result.data  
  
        console.print(f"[bold green]代理:[/bold green] {response}")  
        message_history = result.all_messages()

要查看我的完整实现，请访问这里的存储库 。亲自尝试使用文件系统代理看看它能做什么！以下是一个示例：

欢迎来到MCP演示CLI！输入/quit退出。  
  
[DEBUG] 启动MCP服务器，工作目录为：/Users/finn.andersen/projects/mcp_demo  
[DEBUG] 安全的MCP文件系统服务器运行在stdio上  
[DEBUG] 允许的目录：[ '/Users/finn.andersen/projects/mcp_demo' ]  
  
> : 创建一个包含项目目录树结构的文件，以分层格式显示每个项目的深度，使用类似“tree”工具的结果格式。排除隐藏文件和文件夹  
  
[DEBUG] 调用工具directory_tree，参数为：{'path': '.'}  
[DEBUG] 调用工具write_file，参数为：{'path': 'directory_tree.txt', 'content': '...'}  
  
代理：目录树已成功写入名为`directory_tree.txt`的文件中。如果你还需要其他帮助，请随时告诉我！  
> :

它成功创建了一个名为directory_tree.txt的文件，内容如下：

Makefile  
README.md  
client  
    __init__.py  
    mcp_agent  
        __init__.py  
        agent.py  
        cli.py  
        deps.py  
        llm.py  
        run.py  
        tools.py  
        util  
            __init__.py  
            schema_to_params.py  
pyproject.toml  
requirements-dev.txt  
requirements.txt  
server  
    Dockerfile  
    README.md  
    index.ts  
    package-lock.json  
    package.json  
    test.ts  
    tsconfig.json  
uv.lock

4、结束语

希望这能帮助你理解MCP的工作原理以及如何将其用于集成AI代理的新功能。去构建一些酷炫的东西吧，利用所有免费提供的MCP服务器！

原文链接：How to use MCP tools with a PydanticAI Agent

汇智网翻译整理，转载请标明出处