模型上下文协议（MCP） 是一种开放协议，用于标准化应用程序如何向大语言模型（LLM）提供工具和上下文。LangChain 智能体可通过 langchain-mcp-adapters 库使用定义在 MCP 服务器上的工具。
MCP
安装 langchain-mcp-adapters 库，以便在 LangGraph 中使用 MCP 工具：
pip install langchain-mcp-adapters
传输类型
MCP 支持多种客户端-服务器通信的传输机制：
- stdio：客户端将服务器作为子进程启动，并通过标准输入/输出进行通信。最适合本地工具和简单设置。
- 可流式 HTTP：服务器作为独立进程运行，处理 HTTP 请求。支持远程连接和多个客户端。
- 服务器发送事件（SSE）：一种针对实时流式通信优化的可流式 HTTP 变体。
使用 MCP 工具
langchain-mcp-adapters 使智能体能够使用一个或多个 MCP 服务器上定义的工具。
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain.agents import create_agent
client = MultiServerMCPClient(
{
"math": {
"transport": "stdio", # 本地子进程通信
"command": "python",
# 替换为你的 math_server.py 文件的绝对路径
"args": ["/path/to/math_server.py"],
},
"weather": {
"transport": "streamable_http", # 基于 HTTP 的远程服务器
# 请确保你的天气服务器在 8000 端口启动
"url": "http://localhost:8000/mcp",
}
}
)
tools = await client.get_tools()
agent = create_agent(
"anthropic:claude-3-7-sonnet-latest",
tools
)
math_response = await agent.ainvoke(
{"messages": [{"role": "user", "content": "what's (3 + 5) x 12?"}]}
)
weather_response = await agent.ainvoke(
{"messages": [{"role": "user", "content": "what is the weather in nyc?"}]}
)
MultiServerMCPClient 默认是无状态的。每次工具调用都会创建一个新的 MCP ClientSession，执行工具，然后清理资源。
自定义 MCP 服务器
要创建你自己的 MCP 服务器，可以使用 mcp 库。该库提供了一种简单的方式来定义工具 并将其作为服务器运行。
请使用以下参考实现来测试你的智能体与 MCP 工具服务器的交互。
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("Math")
@mcp.tool()
def add(a: int, b: int) -> int:
"""加两个数字"""
return a + b
@mcp.tool()
def multiply(a: int, b: int) -> int:
"""乘两个数字"""
return a * b
if __name__ == "__main__":
mcp.run(transport="stdio")
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("Weather")
@mcp.tool()
async def get_weather(location: str) -> str:
"""获取指定地点的天气。"""
return "纽约总是阳光明媚"
if __name__ == "__main__":
mcp.run(transport="streamable-http")
通过 MCP 暴露 LangChain 工具
你也可以使用 to_fastmcp 函数，通过 MCP 服务器暴露现有的 LangChain 工具。这允许你将 LangChain 工具提供给任何 MCP 客户端使用。
from langchain_core.tools import tool
from langchain_mcp_adapters.tools import to_fastmcp
from mcp.server.fastmcp import FastMCP
@tool
def add(a: int, b: int) -> int:
"""加两个数字"""
return a + b
@tool
def get_user_info(user_id: str) -> str:
"""获取用户信息"""
return f"用户 {user_id} 处于活跃状态"
# 将 LangChain 工具转换为 FastMCP 格式
fastmcp_tools = [to_fastmcp(tool) for tool in (add, get_user_info)]
# 使用转换后的工具创建服务器
mcp = FastMCP("LangChain 工具", tools=fastmcp_tools)
mcp.run(transport="stdio")
有状态工具使用
对于需要在工具调用之间保持上下文的有状态服务器，请使用 client.session() 创建持久的 ClientSession。
使用 MCP ClientSession 实现有状态工具调用
from langchain_mcp_adapters.tools import load_mcp_tools
client = MultiServerMCPClient({...})
async with client.session("math") as session:
tools = await load_mcp_tools(session)
更多资源
