这里是慕雪的小助手
慕雪小助手的总结
DeepSeek & LongCat

欢迎阅读慕雪撰写的AI Agent专栏,本专栏目录如下

  1. 【MCP】详细了解MCP协议:和function call的区别何在?如何使用MCP?
  2. 【AI】AI对26届及今后计算机校招的影响
  3. 【Agent.01】AI Agent智能体开发专题引言
  4. 【Agent.02】市面上常见的大模型有哪些?
  5. 【Agent.03】带你学会写一个基础的Prompt
  6. 【Agent.04】AI时代的hello world:调用OpenAI接口,与大模型交互
  7. 【Agent.05】OpenAI接口Function Calling工具调用详解
  8. 【Agent.06】使用openai sdk实现多轮对话
  9. 【Agent.07】什么是Agent?从Chat到ReAct的AI进化之路
  10. 【Agent.08】LangChain的第一个Demo:从零开始构建Agent
  11. 【Agent.09】LangChain里面使用MCP工具

本专栏的所有代码都可以在https://gitee.com/musnows/agent-blog找到。

在上一篇文章中,我们学习了LangChain的基础用法和如何创建简单的Agent。本文将说明如何在LangChain中集成MCP(Model Context Protocol)工具。

1. 什么是MCP?

MCP的介绍请移步MCP协议博客,本文不多介绍这个协议的原理。

简而言之,MCP就是在各个Agent SDK所支持的Function Calling上抽象了一层中间层,让这些Function有了一个外部统一调用和处理的协议,从而让我们的各类Tools只需要使用MCP协议编写出一个server,就可以在各个支持MCP的client里面无缝使用,不再需要针对目标工具、SDK做单独的二次工具开发,大大节省了工具兼容的开发时间。

当然,如果你的工具是Python函数,不涉及到任何兼容逻辑,也不打算切换你的Agent框架的时候,也可以不使用MCP。MCP只是提供了方便,并不代表我们的工具能力就一定要用MCP协议来编写。

在上一个Demo里面,我们使用了LangChain成功调用了直接在python里面实现的tools,本文将介绍如何在LangChain中集成MCP工具,顺带介绍如何使用fastmcp库编写MCP的server。

2. 创建MCP服务器

2.1. 环境准备

首先需要安装MCP相关的依赖,第一个是用于编写MCP服务器的库,第二个是LangChain里面的MCP客户端实现。

bash
1
pip install fastmcp langchain-mcp-adapters

如果你使用的是uv,则使用uv add添加这些依赖。

2.2. 创建数学计算服务器

本次Demo使用一个比较简单的数学服务器作为演示。

创建一个名为01.math_server.py的文件:

python
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
from mcp.server.fastmcp import FastMCP

# 创建MCP服务器实例,"Math"是服务器名称
mcp = FastMCP("Math")

@mcp.tool()
def add(a: int, b: int) -> int:
"""Add two numbers"""
return a + b

@mcp.tool()
def multiply(a: int, b: int) -> int:
"""Multiply two numbers"""
return a * b

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

就是这么短短的几行代码,我们就已经写好了一个MCP服务器了。MCP服务器的启动方式分为两种,一个是stdio(字节流),另外一个是sse(远程)。关于sse协议,在本站也有博客详解。这里为了方便,使用了stdio协议。

同理,FastMCP框架也会使用python函数的lint和docstring来作为工具的desc,方便Agent理解工具的作用。

如果你使用的是Anthopic官方提供的MCP Python SDK,则还有另外一套更加全面的MCP服务端编写方式,支持自定义的字段更多,但编写也更加复杂。如果没有特殊需要,直接使用FastMCP就可以了,直接在原本的函数上加上一个@mcp.tool()装饰器,就能把普通的函数变成MCP Server的Tools,非常方便。

2.3. 测试MCP服务器

我们可以直接运行这个服务器来测试:

bash
1
uv run mcp/01.math_server.py

服务器启动后会等待MCP客户端的连接,终端里面啥都不会输出。只要你观测到终端没有报错,阻塞运行了,那就说明server已经正常启动了。

3. 在LangChain中使用MCP工具

现在让我们在LangChain中集成我们刚创建的MCP工具。

3.1. 完整集成代码

python
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
import asyncio
import os
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from dotenv import load_dotenv

# 加载环境变量
load_dotenv(override=True)


async def main():
# 连接到 MCP 服务器
client = MultiServerMCPClient({
"math": {
"transport": "stdio", # 使用stdio传输协议
"command": "uv", # 启动命令
"args": ["--directory", "mcp", "run", "01.math_server.py"],
}
})

# 配置模型
model = init_chat_model(
model=os.getenv("OPENAI_MODEL"),
model_provider="openai",
api_key=os.getenv("OPENAI_API_KEY"),
base_url=os.getenv("OPENAI_BASE_URL")
)

# 获取 MCP 服务器提供的工具
tools = await client.get_tools()

# 创建 Agent
agent = create_agent(model=model, tools=tools)

# 调用 Agent
result = await agent.ainvoke(
{"messages": [{
"role": "user",
"content": "3 加 5 等于多少?然后乘以 12?"
}]}
)

print(result["messages"][-1].content)


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

3.2. 代码解析

下面对这段代码比较关键的部分单独解释:

第一步:创建MCP客户端

python
1
2
3
4
5
6
7
client = MultiServerMCPClient({
"math": {
"transport": "stdio",
"command": "uv",
"args": ["--directory", "mcp", "run", "01.math_server.py"],
}
})

这里我们创建了一个多服务器MCP客户端,它可以同时连接多个MCP服务器。配置说明:

  • “math”:服务器的针对Agent暴露的名称,可以自定义。注意在prompt中也需要使用对应的名字,方便Agent确认需要使用的工具是谁。
  • transport:传输协议,可选为"stdio"或"streamable_http"(对应sse)
  • command:启动MCP服务器的命令(只有stdio模式下需要)
  • args:启动命令的参数列表(只有stdio模式下需要)
  • url:MCP服务器的链接地址(只有streamable_http模式下需要)

MCP其中一个优势就在于,MCP客户端的链接配置基本上是所有支持MCP的客户端工具通用的,只要把这个json拷贝出去,就可以给其他工具(如claude-code)使用了。

这里需要注意两点:

  • 如果配置的stdio的工具,一定要确定配置的命令能够正常执行。可以先执行一下这个命令确认是否能无报错正常启动MCP服务器。
  • 如果配置的是sse(streamable_http)的工具,要确认这个远程的http服务器地址能够正常在客户端的主机上访问到。

另外,部分Agent工具在MCP服务器无法链接的时候是不会报错提示的(比如claude-code)一定要确认好Agent工具或SDK真的链接上了你提供的MCP工具,否则在缺少工具的时候Agent只会胡乱编造数据!

第二步:获取工具

python
1
tools = await client.get_tools()

这行代码会启动MCP服务器进程,并获取服务器提供的所有工具,tools是一个包含所有可用工具的列表。

同时,这些工具里面也会包含FastMCP库从docstring里面解析出来的简介、参数、返回值,这些信息都会提供给Agent进行Function Calling。

plaintext
1
[StructuredTool(name='add', description='Add two numbers', args_schema={'properties': {'a': {'title': 'A', 'type': 'integer'}, 'b': {'title': 'B', 'type': 'integer'}}, 'required': ['a', 'b'], 'title': 'addArguments', 'type': 'object'}, response_format='content_and_artifact', coroutine=<function convert_mcp_tool_to_langchain_tool.<locals>.call_tool at 0x110448e00>), StructuredTool(name='multiply', description='Multiply two numbers', args_schema={'properties': {'a': {'title': 'A', 'type': 'integer'}, 'b': {'title': 'B', 'type': 'integer'}}, 'required': ['a', 'b'], 'title': 'multiplyArguments', 'type': 'object'}, response_format='content_and_artifact', coroutine=<function convert_mcp_tool_to_langchain_tool.<locals>.call_tool at 0x11244e980>)]

第三步:创建Agent

python
1
agent = create_agent(model=model, tools=tools)

和之前的LangChain示例一样,我们将模型和工具传递给create_agent函数。不同的是,这里的工具来自MCP服务器。

第四步:异步调用

python
1
2
3
4
5
6
result = await agent.ainvoke(
{"messages": [{
"role": "user",
"content": "3 加 5 等于多少?然后乘以 12?"
}]}
)

由于MCP客户端使用异步方式与服务器通信,我们需要使用ainvoke方法而不是invoke方法。

3.3. 运行结果

运行这个程序,你会看到类似这样的输出:

plaintext
1
3 加 5 等于 8,然后 8 乘以 12 等于 96。

Agent成功地:

  1. 理解了用户的数学问题
  2. 调用了add工具计算 3 + 5 = 8
  3. 调用了multiply工具计算 8 × 12 = 96
  4. 给出了完整的答案

从日志中可以看到,日志里面打印出来了CallToolRequest,也就是我们的MCP客户端去请求调用了MCP服务端的服务,这些信息是Agent根据工具调用结果返回的,而不是根据自己的理解“算”出来的。

image.png

4. The end

本文简单介绍了如何在LangChain里面使用MCP工具。

在实际项目中,你可以:

  • 将现有的Python功能函数封装成MCP工具,方便不同的Agent使用
  • 使用第三方提供的MCP工具,有非常多开源的牛逼MCP工具,比如serena
  • 构建自己的工具生态系统,比如对接公司内网系统OpenAPI的各类MCP工具

本文到这里就结束啦!