让大模型应用使用 MCP 工具

大模型本身不会调用 MCP——它只会"说要调用哪个工具"。真正的工作在宿主代码里：把 MCP 服务器上的工具翻译成模型认识的 tools 数组，跑一遍 function calling 循环，再把执行结果回填给模型。本章给出打通"LLM ↔ MCP server"的最小架构与关键代码。

整体思路

整条链路共四步，适配层是核心：

1. 连接 MCP server，用 list_tools 读出工具（名称、描述、inputSchema）；
2. 把工具转换成 OpenAI 兼容的 tools 数组喂给模型（各家模型的 tools 格式大同小异，这里以 OpenAI 为例）；
3. 模型返回 tool_calls 时，逐条映射回 MCP 的 call_tool 真正执行；
4. 把执行结果以 tool 角色回填进消息列表，再交给模型，直到模型不再要工具、给出最终回答。

LLM 应用(编排循环)
   │ tools 数组 / tool_calls
   ▼
适配层：schema 转换 + MCP 执行回填   ← 本章要写的薄封装
   │ list_tools / call_tool
   ▼
MCP client ──stdio / Streamable HTTP──► MCP server(工具/资源)

第一步：把 MCP 工具转成模型认识的 tools

MCP 工具的参数声明 inputSchema 本身就是 JSON Schema，与 OpenAI function calling 的 parameters 字段兼容，做一层浅转换即可：

def to_openai_tools(mcp_tools):
    """把 MCP 的 Tool 列表转成 OpenAI 兼容的 tools 数组"""
    return [
        {
            "type": "function",
            "function": {
                "name": t.name,
                "description": t.description or "",
                "parameters": t.inputSchema,   # JSON Schema，基本可直接透传
            },
        }
        for t in mcp_tools
    ]

个别模型对 JSON Schema 支持有限（如不允许 required 缺失或复杂嵌套），转换时如报参数错误，就按该模型文档把 schema 收敛成更简单的子集。

第二步：function calling 主循环

参考 agent 实践的 run_with_tools 思路：循环里模型说调哪个工具，我们就经 MCP session 执行并回填：

import json
from openai import OpenAI  # 需 pip install openai，并配置 API Key

async def run_with_tools(session, messages, model="你的模型名"):
    """messages 里放历史对话；session 是已 initialize 的 MCP ClientSession"""
    tools = to_openai_tools((await session.list_tools()).tools)
    client = OpenAI()

    while True:
        reply = client.chat.completions.create(
            model=model, messages=messages, tools=tools,
        )
        msg = reply.choices[0].message
        messages.append(msg)
        if not msg.tool_calls:                 # 模型不再要工具 → 收敛
            return msg.content
        for call in msg.tool_calls:            # 逐个执行模型要的工具
            name = call.function.name
            args = json.loads(call.function.arguments)
            result = await session.call_tool(name, args)     # 交给 MCP 执行
            text = "".join(b.text for b in result.content if hasattr(b, "text"))
            messages.append({                  # 以 tool 角色回填执行结果
                "role": "tool",
                "tool_call_id": call.id,
                "content": text,
            })

关键点：assistant 消息必须连同 tool_calls 一起放回消息列表；每个 tool_calls 都要有一条同 id 的 tool 消息回填，否则模型无法继续推理。执行出错时把错误文本也回填（而不是抛异常中断），让模型自己修正参数。

拼起来的最小打通

# demo.py：需与上面的 to_openai_tools、run_with_tools 放在同一文件再运行
# （还需你的 API Key 与可用的 MCP server，model 名替换为你的模型）
import asyncio
import sys
from mcp import ClientSession
from mcp.client.stdio import StdioServerParameters, stdio_client

async def main():
    # command 填可执行程序路径：本地 python 用 sys.executable 最稳妥
    params = StdioServerParameters(command=sys.executable, args=["server_echo.py"])
    async with stdio_client(params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            answer = await run_with_tools(
                session,
                [{"role": "user", "content": "计算 31 加 11 等于多少？"}],
            )
            print(answer)   # 输出：42（由模型给出，可能带解释）

python demo.py
# 输出：42（模型的实际回答可能附带解释，结果以模型输出为准）

上面演示串起了完整闭环：模型读到 add 工具 → 请求调用 → 你的循环经 MCP 执行 → 结果回填 → 模型给出最终答案。

有没有更省事的封装

MCP SDK 自身也在提供更高层的客户端封装（例如对 stdio / HTTP 服务器的一站式托管类），具体类名与用法随版本演进，以官方 SDK 文档为准。除此之外，成熟 agent 框架大多内置了 MCP 支持；若你只用 OpenAI 直连，本章 30 行左右的薄封装就是最透明的方案。 小结：接入大模型应用 = 把 MCP 工具 schema 转成 tools 数组 + 跑 function calling 循环 + 把 call_tool 结果回填给模型；无现成高层集成时自建薄封装即可，SDK 的高层客户端以官方文档为准。