MCP 系列 3：使用 LangChain Agent 调用 MCP 服务

上一篇已经用 FastMCP 写好了 Python MCP Server，也验证了 Tool、Resource 和 Prompt。这一篇把 MCP 能力接入 LangChain，让本地 Qwen3 Agent 根据用户问题选择和调用 MCP Tool。

这条链路中有三个容易混淆的角色：

• MCP Server 定义并执行能力。
• MCP Client 负责连接、发现和调用 Server。
• Agent 把 MCP Tool 交给模型，并组织工具调用循环。

MCP Server 不负责生成最终自然语言回答。工具返回结果后，仍由 Qwen3 组织答案。

1. 完整调用链

用户问“18 加 24 等于多少”时，执行顺序如下：

1. MultiServerMCPClient 从 MCP Server 获取 Tool。
2. Adapter 把 MCP Tool 转换为 LangChain Tool。
3. Agent 将 Tool Schema 和用户问题发送给 Qwen3。
4. Qwen3 返回包含 tool_calls 的 AIMessage。
5. Agent 调用转换后的 LangChain Tool，Adapter 再通过 MCP Client 调用 Server 中的 add。
6. MCP Server 返回 42.0。
7. Adapter 转换 MCP 结果，Agent 将结果记录为与请求对应的 ToolMessage。
8. Qwen3 生成最终中文回答。

2. 安装 MCP Adapter

继续使用 .venv_mcp，新增：

langchain-mcp-adapters==0.2.2

环境中同时固定前面已经使用的 LangChain 版本：

langchain==1.2.13
langchain-core==1.2.22
langchain-openai==1.1.12
langgraph==1.1.3
openai==2.30.0

安装和检查：

cd source/_posts/llm_learning
source .venv_mcp/bin/activate

python -m pip install \
  -r mcp/p02_fastmcp_server/requirements-lock.txt
python -m pip check

3. 启动 FastMCP 和 Qwen3

第一个终端启动 FastMCP：

source .venv_mcp/bin/activate

fastmcp run mcp/p02_fastmcp_server/server.py \
  --transport http \
  --host 127.0.0.1 \
  --port 18100

第二个终端启动 Qwen3：

source .venv_tool_server/bin/activate

"$VIRTUAL_ENV/bin/python" -m mlx_lm server \
  --model model \
  --host 127.0.0.1 \
  --port 18080 \
  --prompt-cache-size 0 \
  --chat-template-args '{"enable_thinking": false}'

检查模型服务：

curl --noproxy '*' http://127.0.0.1:18080/v1/models

文章不保留响应中的动态创建时间，只要 HTTP 状态为 200 且模型列表非空即可。

4. 获取 Tools、Resources 和 Prompts

MultiServerMCPClient 不只可以加载 Tool，也能读取 Resource 和 Prompt。

01_list_mcp_capabilities.py：

"""使用 LangChain MCP Adapter 获取 Tool、Resource 和 Prompt。"""

import asyncio

from langchain_mcp_adapters.client import MultiServerMCPClient


async def main() -> None:
    """连接 FastMCP HTTP Server 并打印三种能力。"""

    client = MultiServerMCPClient(
        {
            "learning": {
                "transport": "http",
                "url": "http://127.0.0.1:18100/mcp",
            }
        }
    )

    tools = await client.get_tools(server_name="learning")
    resources = await client.get_resources(
        "learning",
        uris=["guide://mcp"],
    )
    prompt_messages = await client.get_prompt(
        "learning",
        "explain_topic",
        arguments={"topic": "MCP Tool"},
    )

    print("LangChain Tools:", [tool.name for tool in tools])
    print("Resource:", resources[0].as_string())
    print("Prompt message type:", type(prompt_messages[0]).__name__)
    print("Prompt:", prompt_messages[0].content)


if __name__ == "__main__":
    asyncio.run(main())

运行结果：

LangChain Tools: ['add', 'get_city_info']
Resource: MCP Server 可以通过 Tools、Resources 和 Prompts 向客户端提供能力。
Prompt message type: HumanMessage
Prompt: 请面向初学者，用三句话解释 MCP Tool。

可以看到：

• MCP Tool 被转换成 LangChain Tool。
• MCP Resource 被转换成 LangChain Blob，文本内容可以使用 as_string() 读取。
• MCP Prompt 被转换成 LangChain Message。

这些能力不会自动进入 Agent。代码必须明确决定加载什么、怎样使用。

5. 让 Qwen3 调用 HTTP MCP Tool

02_agent_with_http_mcp.py：

"""把 Streamable HTTP MCP 工具交给本地 Qwen3 Agent。"""

import asyncio

import httpx
from langchain.agents import create_agent
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_openai import ChatOpenAI


async def main() -> None:
    """让 Qwen3 选择并调用 MCP 加法工具。"""

    client = MultiServerMCPClient(
        {
            "learning": {
                "transport": "http",
                "url": "http://127.0.0.1:18100/mcp",
            }
        }
    )
    tools = await client.get_tools()

    # trust_env=False 避免系统代理或 VPN 转发本机请求。
    with httpx.Client(trust_env=False) as http_client:
        model = ChatOpenAI(
            model="default_model",
            base_url="http://127.0.0.1:18080/v1",
            api_key="not-needed",
            temperature=0,
            max_tokens=128,
            http_client=http_client,
        )
        agent = create_agent(
            model=model,
            tools=tools,
            system_prompt="遇到算术问题必须使用 MCP 计算工具。",
        )
        result = await agent.ainvoke(
            {"messages": [{"role": "user", "content": "请计算 18 加 24。"}]}
        )

    for message in result["messages"]:
        if getattr(message, "tool_calls", None):
            call = message.tool_calls[0]
            print("MCP Tool:", call["name"])
            print("Arguments:", call["args"])
        if message.type == "tool":
            # MCP Adapter 可能返回内容块列表，只提取稳定的文本结果。
            tool_result = (
                message.content[0]["text"]
                if isinstance(message.content, list)
                else message.content
            )
            print("Tool result:", tool_result)

    print("Qwen3 answer:", result["messages"][-1].content)


if __name__ == "__main__":
    asyncio.run(main())

真实输出：

MCP Tool: add
Arguments: {'a': 18, 'b': 24}
Tool result: 42.0
Qwen3 answer: 18 加 24 的结果是 42。

结果说明 Qwen3 不是自己直接计算后返回，而是生成了 add 调用，MCP Server 执行函数后再由模型整理回答。

6. 一个 Agent 连接多个 MCP Server

实际应用通常不只连接一个 Server。下面再创建一个 stdio MCP Server，提供固定用户兴趣资料。

stdio_profile_server.py：

"""提供固定用户兴趣资料的 stdio MCP Server。"""

from fastmcp import FastMCP


mcp = FastMCP("profile-mcp-server")


@mcp.tool
def get_user_hobby(name: str) -> str:
    """查询固定用户的兴趣爱好。"""

    hobbies = {
        "小明": "周末去西湖散步",
        "小红": "阅读科普书籍",
    }
    return hobbies.get(name, f"没有找到 {name} 的兴趣资料。")


if __name__ == "__main__":
    mcp.run(transport="stdio", show_banner=False)

主 Agent 同时连接 HTTP 和 stdio：

MultiServerMCPClient 保存两套连接配置，并把两个 Server 的 Tool 转换成一组 LangChain Tool。设置 tool_name_prefix=True 后，前缀来自连接配置中的 Server 名称；它只解决工具重名问题，不会把两个 MCP Server 合并成同一个协议会话。

"""让一个 Agent 同时使用 HTTP MCP Server 和 stdio MCP Server。"""

import asyncio
import sys
from pathlib import Path

import httpx
from langchain.agents import create_agent
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_openai import ChatOpenAI


PROFILE_SERVER = Path(__file__).with_name("stdio_profile_server.py")


async def main() -> None:
    """验证一个 Agent 可以组合多个 MCP Server 的工具。"""

    client = MultiServerMCPClient(
        {
            "learning": {
                "transport": "http",
                "url": "http://127.0.0.1:18100/mcp",
            },
            "profile": {
                "transport": "stdio",
                "command": sys.executable,
                "args": [str(PROFILE_SERVER)],
            },
        },
        tool_name_prefix=True,
    )
    tools = await client.get_tools()
    print("All tools:", [tool.name for tool in tools])

    with httpx.Client(trust_env=False) as http_client:
        model = ChatOpenAI(
            model="default_model",
            base_url="http://127.0.0.1:18080/v1",
            api_key="not-needed",
            temperature=0,
            max_tokens=192,
            http_client=http_client,
        )
        agent = create_agent(
            model=model,
            tools=tools,
            system_prompt="必须使用 MCP 工具分别查询兴趣并完成计算。",
        )
        result = await agent.ainvoke(
            {
                "messages": [
                    {
                        "role": "user",
                        "content": "小明喜欢做什么？另外请计算 8 加 9。",
                    }
                ]
            }
        )

    used_tools = []
    for message in result["messages"]:
        for call in getattr(message, "tool_calls", []):
            used_tools.append(call["name"])
        if message.type == "tool":
            # 只打印 ToolMessage 中的文本，过滤动态内容块 ID。
            tool_result = (
                message.content[0]["text"]
                if isinstance(message.content, list)
                else message.content
            )
            print("Tool result:", tool_result)

    print("Used tools:", used_tools)
    print("Qwen3 answer:", result["messages"][-1].content)


if __name__ == "__main__":
    asyncio.run(main())

这里设置 tool_name_prefix=True，Adapter 会在工具名前增加 Server 名，降低多个 Server 出现同名工具的风险。

真实输出：

All tools: ['learning_add', 'learning_get_city_info', 'profile_get_user_hobby']
Tool result: 周末去西湖散步
Tool result: 17.0
Used tools: ['profile_get_user_hobby', 'learning_add']
Qwen3 answer: 小明喜欢周末去西湖散步。另外，8加9的结果是17。

HTTP Server 已经独立运行，stdio Server 则由 Client 在需要时启动子进程。

7. 默认临时 Session 与显式 Session

MultiServerMCPClient 默认按无状态方式管理 Tool：get_tools() 返回的工具包装器在每次执行时都会创建新的 ClientSession，完成初始化和 tools/call 后再清理连接。get_resources()、get_prompt() 等便捷方法也会为本次读取建立并关闭 Session。

如果同一个有状态 Server 的多次操作必须共享协议会话，应显式打开 Session。

04_persistent_mcp_session.py：

"""使用显式 ClientSession 在同一个 MCP Session 中执行多次操作。"""

import asyncio

from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_mcp_adapters.prompts import load_mcp_prompt
from langchain_mcp_adapters.resources import load_mcp_resources
from langchain_mcp_adapters.tools import load_mcp_tools


async def main() -> None:
    """在一个显式 Session 中加载 Tool、Resource 和 Prompt。"""

    client = MultiServerMCPClient(
        {
            "learning": {
                "transport": "http",
                "url": "http://127.0.0.1:18100/mcp",
            }
        }
    )

    # 离开 async with 前，下面三次操作复用同一个 ClientSession。
    async with client.session("learning") as session:
        tools = await load_mcp_tools(session)
        resources = await load_mcp_resources(session, uris=["guide://mcp"])
        prompts = await load_mcp_prompt(
            session,
            "explain_topic",
            arguments={"topic": "持久 MCP Session"},
        )

        print("Same session tools:", [tool.name for tool in tools])
        print("Same session resource:", resources[0].as_string())
        print("Same session prompt:", prompts[0].content)


if __name__ == "__main__":
    asyncio.run(main())

运行结果：

Same session tools: ['add', 'get_city_info']
Same session resource: MCP Server 可以通过 Tools、Resources 和 Prompts 向客户端提供能力。
Same session prompt: 请面向初学者，用三句话解释 持久 MCP Session。

显式 Session 只在 async with 上下文中保持，退出后仍会关闭。它不是聊天记忆，不会自动保存 AgentState，也不是进程重启后仍能恢复的持久化存储。

8. MCP Tool 转换后发生了什么

client.get_tools() 返回的是 LangChain 可调用 Tool。Adapter 负责：

• 从 MCP Server 读取工具名称、说明和参数 Schema。
• 将 LangChain Tool 调用转换成 MCP tools/call。
• 将 MCP CallToolResult 转换成 LangChain Tool 可以返回的内容与 Artifact。
• Agent 的工具执行节点再把结果写成与 Tool Call ID 对应的 ToolMessage。

Adapter 不负责决定调用哪个 Tool。这个决策由 Qwen3 和 Agent 循环完成。

9. 常见问题

9.1 ModuleNotFoundError: langchain.agents

如果项目根目录也存在带 init.py 的 langchain/ 文件夹，它会遮蔽虚拟环境中的第三方包。检查实际导入路径：

python -c 'import langchain; print(langchain.__file__)'

输出应该指向 .venv_mcp/lib/python3.12/site-packages/langchain/。学习代码目录只用于分类时，不要在根级创建会把它变成同名普通包的 langchain/init.py。

9.2 连接了 Qwen3 端口而不是 MCP 端口

18080 是 OpenAI 兼容模型接口，18100/mcp 才是 FastMCP。两者协议完全不同。

9.3 VPN 影响本机 Qwen3 请求

为 ChatOpenAI 显式传入：

http_client=httpx.Client(trust_env=False)

这样本机请求不会读取系统代理环境变量。

9.4 两个 Server 存在同名 Tool

使用 tool_name_prefix=True，或者从设计上给工具使用清晰且不冲突的名称。

9.5 直接打印 ToolMessage 出现动态 ID

MCP Adapter 可能把结果表示为内容块列表。教学输出只提取 content[0][“text”]，不要把动态内容块 ID 和整段元数据写入日志或文章。

10. 总结

组件	本例中的职责	不负责的工作
FastMCP Server	定义并执行 add、城市资料等能力	不选择工具，不生成最终回答
MultiServerMCPClient	连接 Server，加载 Tool、Resource、Prompt	不进行模型推理
LangChain MCP Adapter	在 MCP 能力与 LangChain 对象间转换	不自动把所有能力放进 Agent
Qwen3	根据问题和 Schema 生成工具请求及最终回答	不直接执行 MCP Tool
Agent	组织模型、Tool 和 ToolMessage 循环	不替代 MCP 传输层
显式 Session	复用同一个 MCP 协议会话	不等于聊天记忆或 Checkpointer