模型上下文协议 (MCP) 是一种开放协议,它标准化了应用程序向大型语言模型 (LLM) 提供工具和上下文的方式。LangChain 智能体可以使用 langchain-mcp-adapters 库在 MCP 服务器上定义的工具。
安装 langchain-mcp-adapters 库以在 LangGraph 中使用 MCP 工具:
pip install langchain-mcp-adapters
传输类型
MCP支持不同的客户端-服务器通信传输机制:
stdio:客户端以子进程的形式启动服务器,并通过标准输入/输出进行通信。最适合本地工具和简单设置。
可流式HTTP:服务器作为一个独立进程运行,处理HTTP请求。支持远程连接和多个客户端。
服务器发送事件(SSE):一种针对实时流式通信优化的可流式HTTP变体。
使用MCP工具
langchain-mcp-adapters 允许智能体使用跨一个或多个MCP服务器定义的工具。
Accessing multiple MCP servers
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" : "http://localhost:8000/mcp" ,
}
}
)
tools = await client.get_tools()
agent = create_agent(
"anthropic:claude-sonnet-4-5" ,
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工具服务器测试您的智能体。
Math server (stdio transport)
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" )
Weather server (streamable HTTP transport)
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。
Using MCP ClientSession for stateful tool usage
from langchain_mcp_adapters.tools import load_mcp_tools
client = MultiServerMCPClient({ ... })
async with client.session( "math" ) as session:
tools = await load_mcp_tools(session)
其他资源