Skip to main content
模型上下文协议 (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库。此库提供了一种简单的方式来定义工具并将它们作为服务器运行。
pip install 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)

其他资源