LangChain上手 MCP：从用别人工具到自己写工具

读完这篇文章，你将能：用别人的 MCP Server（比如百度地图）、自己写一个 MCP Server（比如计算器）、用 Agent 自动调度多个工具。

先看效果

用别人的：百度地图 MCP

十几行代码，让 AI 变成地图助手：

1import asyncio
2from langchain_mcp_adapters.client import MultiServerMCPClient
3from langchain.agents import create_agent
4from llm import llm
5
6async def main():
7    client = MultiServerMCPClient({
8        "baidu_map": {
9            "transport": "streamable_http",
10            "url": "https://mcp.map.baidu.com/mcp?ak=你的AK",
11        }
12    })
13    tools = await client.get_tools()
14    print("✅ 工具列表：", [t.name for t in tools])
15
16    agent = create_agent(
17        llm, tools,
18        system_prompt="你是地图助手，必须使用工具查询。",
19    )
20    result = await agent.ainvoke({
21        "messages": [("user", "北京今天天气怎么样？")]
22    })
23    print(result["messages"][-1].content)
24
25asyncio.run(main())
26

输出：

1✅ 工具列表：['map_geocode', 'map_reverse_geocode', 'map_weather', ...]
2北京今天晴朗，温度23°C，微风...
3

自己写的：计算器 MCP Server

1from mcp.server.fastmcp import FastMCP
2
3mcp = FastMCP("计算器")
4
5@mcp.tool()
6def add(a: float, b: float) -> float:
7    return a + b
8
9mcp.run()  # 就这样，一个 MCP Server 写好了
10

MCP 是什么

一句话：MCP 是工具和大模型之间的"USB 接口"。

以前你写 @tool def cal(...)，这个工具只能给 LangChain 用。写成 MCP Server 后，LangChain、Claude Desktop、ChatGPT、Go 写的 Agent——谁都能调。

1┌──────────┐                  ┌──────────┐
2│  MCP     │                  │  MCP     │
3│  Server  │ ←── 标准协议 ──→  │  Client  │
4│ (工具端)  │                  │ (模型端)  │
5└──────────┘                  └──────────┘
6

MCP Server：暴露工具（数据库、文件、API、你的计算器）
MCP Client：连到大模型，负责调用 Server 提供的工具
协议：JSON-RPC，就是 JSON 格式的"调哪个函数、传什么参数、返回什么结果"

📖 官方文档：LangChain MCP | MCP 协议 | FastMCP

3 分钟上手：用别人的 MCP Server

安装

1pip install langchain-mcp-adapters
2

百度地图 MCP 实战

去 lbsyun.baidu.com 申请 AK（应用类别选"服务端"，勾选 MCP 服务，要实名认证）
设置环境变量 export BAIDU_MAP_AK="你的AK"
跑代码：

1import asyncio
2from langchain_mcp_adapters.client import MultiServerMCPClient
3from langchain.agents import create_agent
4from llm import llm
5import os
6
7async def main():
8    client = MultiServerMCPClient({
9        "baidu_map": {
10            "transport": "streamable_http",
11            "url": f"https://mcp.map.baidu.com/mcp?ak={os.getenv('BAIDU_MAP_AK')}",
12        }
13    })
14    tools = await client.get_tools()
15    print("✅ 工具列表：", [t.name for t in tools])
16
17    # 先直接调用，不用 Agent —— 看看工具能返回什么
18    tool = {t.name: t for t in tools}
19    r = await tool["map_geocode"].ainvoke({"address": "北京市朝阳区国贸中心"})
20    print(f"📍 经纬度：{r}")
21
22    # 再用 Agent —— 让大模型自动判断调哪个工具
23    agent = create_agent(
24        llm, tools,
25        system_prompt="你是地图助手，当用户询问天气、地址、路线、地点等信息时，必须使用工具查询。",
26    )
27    result = await agent.ainvoke({
28        "messages": [("user", "北京今天天气怎么样？")]
29    })
30    print(f"\n🌤 {result['messages'][-1].content}")
31
32asyncio.run(main())
33

关键点： MCP 和 Agent 是解耦的。client.get_tools() 拿到工具后，可以直接调，也可以交给 Agent 自动调。

两种传输方式（必知）

MCP 支持两种通信方式，你都要知道：

1. stdio — 本地开发首选

Client 自动拉起 Server 作为子进程，不需要启动 HTTP 服务：

1client = MultiServerMCPClient({
2    "math": {
3        "transport": "stdio",
4        "command": "python",
5        "args": ["demo/mcp-server-stdio.py"],
6    }
7})
8tools = await client.get_tools()
9

这段配置翻译成人话就是一句命令行：python demo/mcp-server-stdio.py

Client 自动帮你执行这个命令，然后通过 stdin/stdout 和它对话。你不需要手动开终端去启动 Server。

2. streamable_http — 远程生产环境

Server 作为 HTTP 服务运行，可以跨机器调用：

1# 服务端
2mcp = FastMCP("服务", host="0.0.0.0", port=8080)
3mcp.run(transport="streamable-http")
4
5# 客户端
6client = MultiServerMCPClient({
7    "service": {
8        "transport": "streamable_http",
9        "url": "http://localhost:8080/mcp",
10    }
11})
12

这段配置翻译成人话：Client 向 http://localhost:8080/mcp 发 HTTP 请求，Server 在那等着响应。

和 stdio 的区别是：Server 是一个独立的 HTTP 服务，你需要先手动启动它，可以部署在任何机器上。

对比

stdio	streamable_http
需要先启动 Server	❌ 自动拉起	✅ 手动启动
跨机器	❌	✅
鉴权	不需要	支持 headers
场景	本地开发	生产环境

SSE 呢？

老版本 MCP 用 SSE（Server-Sent Events）做远程通信，但连接单向、实现复杂。2025 年 3 月官方推出 streamable_http 替代 SSE，新项目直接用 streamable_http，不用管 SSE。

自己写一个 MCP Server

最小版本 — stdio 模式

服务端：

1# demo/mcp-server-stdio.py
2from mcp.server.fastmcp import FastMCP
3
4mcp = FastMCP("计算器服务")
5
6@mcp.tool()
7def add(a: float, b: float) -> float:
8    return a + b
9
10@mcp.tool()
11def multiply(a: float, b: float) -> float:
12    return a * b
13
14if __name__ == "__main__":
15    mcp.run()  # 默认 stdio
16

客户端：

1# demo/mcp-client-stdio.py
2import asyncio
3from pathlib import Path
4from langchain_mcp_adapters.client import MultiServerMCPClient
5
6async def main():
7    client = MultiServerMCPClient({
8        "math": {
9            "transport": "stdio",
10            "command": "python",
11            "args": [str(Path(__file__).parent / "mcp-server-stdio.py")],
12        }
13    })
14    tools = await client.get_tools()
15    tool = {t.name: t for t in tools}
16    r = await tool["add"].ainvoke({"a": 100, "b": 50})
17    print(f"100 + 50 = {r}")
18
19asyncio.run(main())
20

多工具版 — streamable_http 模式

服务端：

1# demo/mcp-server-calculator.py
2from mcp.server.fastmcp import FastMCP
3
4mcp = FastMCP("计算器服务", host="0.0.0.0", port=8080)
5
6@mcp.tool()
7def add(a: float, b: float) -> float:
8    return a + b
9
10@mcp.tool()
11def multiply(a: float, b: float) -> float:
12    return a * b
13
14@mcp.tool()
15def divide(a: float, b: float) -> float:
16    if b == 0:
17        raise ValueError("除数不能为0")
18    return a / b
19
20if __name__ == "__main__":
21    mcp.run(transport="streamable-http")
22

启动：

1python3 demo/mcp-server-calculator.py
2# → 监听 http://localhost:8080/mcp
3

客户端：

1# demo/mcp-client-calculator.py
2client = MultiServerMCPClient({
3    "my_calculator": {
4        "transport": "streamable_http",
5        "url": "http://localhost:8080/mcp",
6    }
7})
8tools = await client.get_tools()
9
10# 直接调
11tool = {t.name: t for t in tools}
12r = await tool["add"].ainvoke({"a": 100, "b": 50})
13
14# Agent 调
15agent = create_agent(llm, tools, system_prompt="你是数学助手，计算必须用工具。")
16result = await agent.ainvoke({
17    "messages": [("user", "100乘以50再加20等于多少？")]
18})
19

完整流程

1┌────────────────────┐         ┌────────────────────┐
2│ mcp-server.py      │         │ mcp-client.py      │
3│                    │  HTTP   │                    │
4│ @mcp.tool() add    │◄───────►│ MultiServerMCPClient│
5│ @mcp.tool() multi  │  JSON   │ → get_tools()      │
6│ :8080/mcp          │  -RPC   │ → tool.ainvoke()   │
7│                    │         │ → create_agent()   │
8└────────────────────┘         └────────────────────┘
9

对比之前在 LangChain 里 @tool def cal(...)：MCP 版多了一步 mcp.run()，但换来的是任何语言、任何框架都能调。

多 Server 同时接入 + 鉴权

一个 Agent 可以同时连多个 MCP Server：

1client = MultiServerMCPClient({
2    "math": {
3        "transport": "stdio",
4        "command": "python",
5        "args": ["/path/to/math_server.py"],
6    },
7    "baidu_map": {
8        "transport": "streamable_http",
9        "url": "https://mcp.map.baidu.com/mcp?ak=xxx",
10    }
11})
12tools = await client.get_tools()
13# tools = math 的 add/multiply + baidu_map 的 13 个地图工具
14agent = create_agent(llm, tools)
15

带 Token 鉴权：

企业场景下，Server 通常需要带 Token 鉴权。在写服务端的时候增加一个鉴权中间件，Client 调用时在 headers 里带上 Token 就好：

1client = MultiServerMCPClient({
2    "service": {
3        "transport": "streamable_http",
4        "url": "http://localhost:8000/mcp",
5        "headers": {"Authorization": "Bearer YOUR_TOKEN"},
6    }
7})
8

总结

MCP = 工具和模型的 USB 接口，谁的 Server 都能给谁的 Client 用
会用别人的：MultiServerMCPClient({...}) → get_tools() → 直接调或放 Agent
会写自己的：FastMCP + @mcp.tool() + mcp.run()
本地开发用 stdio，生产用 streamable_http
多个 Server 自由组合，Agent 自动判断调哪个

附录 A：MCP 是什么协议？JSON-RPC 详解

MCP 底层是 JSON-RPC（JSON Remote Procedure Call）。没有 HTTP Headers、没有 REST 语义——就是用 JSON 格式请求远程服务器执行一个函数，然后返回结果。

请求：

1{
2  "jsonrpc": "2.0",
3  "method": "tools/call",
4  "params": { "name": "add", "arguments": { "a": 100, "b": 50 } },
5  "id": 1
6}
7

响应：

1{
2  "jsonrpc": "2.0",
3  "result": { "content": [{ "type": "text", "text": "150" }] },
4  "id": 1
5}
6

为什么不是 REST？

REST	JSON-RPC
通信方式	URL 路径 + HTTP 动词	单一端点，方法名在 body
工具调用	POST /api/tools/cal	{"method": "tools/call", ...}
批量调用	多次请求	一次请求数组
实时推送	需要 WebSocket	同一连接支持 notification

工具调用天然是"调用函数"，和 JSON-RPC 的语义完全匹配。

三种消息类型

类型	说明	有 id？
Request	Client → Server，请求执行	✅
Response	Server → Client，返回结果	✅
Notification	单向通知，无需回复	❌

一句话：请求-响应是对话，通知是广播。

附录 B：MCP Server 功能分类

按"工具是干嘛的"分，常见 9 大类：

类别	场景	例子
文件系统	读写本地文件	FileSystem MCP
数据库	执行 SQL	PostgreSQL MCP
搜索网页	抓取、搜索	Brave Search MCP
云存储	管理云端文件	S3 MCP
版本控制	Git 操作	GitHub MCP
开发工具	终端、Lint	命令行 MCP
通信平台	发消息	Slack MCP
监控	查日志、指标	可观测性 MCP
其他	地图、支付、笔记	百度地图 MCP

任何能被 AI 调用的能力，都能做成一个 MCP Server。

附录 C：常见踩坑

Q1：TypeError: FastMCP.run() got an unexpected keyword argument 'host'

host/port 写在构造函数里，不是 run() 里：

1# ✅ 正确
2mcp = FastMCP("服务", host="0.0.0.0", port=8080)
3mcp.run(transport="streamable-http")
4
5# ❌ 错误
6mcp.run(transport="streamable-http", host="0.0.0.0", port=8080)
7

Q2：NotImplementedError: MultiServerMCPClient cannot be used as a context manager

0.1.0 版本去掉了 async with，直接创建就行：

1# ✅ 新版
2client = MultiServerMCPClient({...})
3tools = await client.get_tools()
4
5# ❌ 旧版
6async with MultiServerMCPClient({...}) as client:
7    ...
8

Q3：MCP 工具 Agent 调用报错

MCP 工具只支持异步。用 agent.ainvoke() 而不是 agent.invoke()。

Q4：stdio 模式 Server 找不到

用绝对路径，或者在客户端用 Path(__file__).parent 拼接。

《LangChain上手 MCP：从用别人工具到自己写工具》是转载文章，点击查看原文。