模型上下文协议 (MCP) 是一个开放协议，用于标准化应用程序向大型语言模型 (LLM) 提供工具和上下文的方式。LangChain 代理可以使用 langchain-mcp-adapters 库使用 MCP 服务器上定义的工具。
安装 langchain-mcp-adapters 库以在 LangGraph 中使用 MCP 工具
pip install langchain-mcp-adapters
传输类型
MCP 支持不同的传输机制用于客户端-服务器通信
- stdio – 客户端将服务器作为子进程启动，并通过标准输入/输出进行通信。最适合本地工具和简单设置。
- 可流式 HTTP – 服务器作为独立进程运行，处理 HTTP 请求。支持远程连接和多个客户端。
- 服务器发送事件 (SSE) – 可流式 HTTP 的一种变体，针对实时流式通信进行了优化。
langchain-mcp-adapters 使代理能够使用在一个或多个 MCP 服务器上定义的工具。
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain.agents import create_agent
client = MultiServerMCPClient(
{
"math": {
"transport": "stdio", # Local subprocess communication
"command": "python",
# Absolute path to your math_server.py file
"args": ["/path/to/math_server.py"],
},
"weather": {
"transport": "streamable_http", # HTTP-based remote server
# Ensure you start your weather server on port 8000
"url": "https://:8000/mcp",
}
}
)
tools = await client.get_tools()
agent = create_agent(
"claude-sonnet-4-5-20250929",
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:
"""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")
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("Weather")
@mcp.tool()
async def get_weather(location: str) -> str:
"""Get weather for location."""
return "It's always sunny in New York"
if __name__ == "__main__":
mcp.run(transport="streamable-http")
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)
附加资源
