Model Context Protocol (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)
额外资源