LangChain 调用 MCP 实战：从 @tool 到标准化工具协议

Host (你的 LLM 应用)
  └─ Client (MCP SDK，管理连接)
       └─ Server (FastMCP，提供能力)

# mcp_math_server.py
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 langchain_mcp_adapters.client import MultiServerMCPClient

client = MultiServerMCPClient({
    "math": {
        "command": "python",
        "args": ["mcp_math_server.py"],
        "transport": "stdio",  # 本地进程通信
    }
})

tools = await client.get_tools() #返回LangChain BaseTool列表

client = MultiServerMCPClient({
    "math": {
        "command": "python",
        "args": ["mcp_math_server.py"],
        "transport": "stdio",
    },
    "weather": {
        "command": "python",
        "args": ["mcp_weather_server.py"],
        "transport": "stdio",
    }
})

tools = await client.get_tools()
# 返回 [add, multiply, get_weather] — 来自两个服务器的工具无缝合并

from langchain.agents import create_agent

agent = create_agent(llm, tools)
response = await agent.ainvoke({
    "messages": "北京天气如何？帮我算一下 (3+5)*12"
})

-> 调用工具: get_weather({'city': '北京'})
-> 调用工具: add({'a': 3, 'b': 5})        # 3+5=8
-> 调用工具: multiply({'a': 8, 'b': 12})   # 8*12=96

# Tool：LLM 执行查询
@mcp.tool()
async def get_weather(city: str) -> str:
    """Get current weather for a city"""
    return WEATHER_DATA.get(city, f"{city}：暂无数据")

# Resource：LLM 获取只读上下文
@mcp.resource("weather://cities")
def get_supported_cities() -> str:
    """List supported cities"""
    return "北京、上海、广州、深圳"

# Prompt：可复用的分析模板
@mcp.prompt()
def weather_report(city: str) -> list[dict]:
    """Weather analysis prompt"""
    return [{"role": "user", "content": f"分析{city}天气并给出出行建议"}]

# 读取 Resource — 获取服务器提供的数据
resources = await client.get_resources("weather")
# -> weather://cities: "北京、上海、广州、深圳"

# 加载 Prompt — 获取预定义的模板消息
messages = await client.get_prompt("weather", "weather_report",arguments={"city": "北京"})
# -> [HumanMessage("分析北京天气并给出出行建议")]

@mcp.tool()
def get_weather(city: str) -> str:
    """Get current weather for a city""" #说清：做什么、输入什么、返回什么

@mcp.tool()
def weather(data):     # 参数名模糊、无类型、描述为空
    """handle weather"""

# 手动方式 — 繁琐
response = llm_with_tools.invoke(query)
if response.tool_calls:
    for tc in response.tool_calls:
        result = tool.invoke(tc["args"])
        # ... 拼消息、再调 LLM ...

agent = create_agent(llm, tools)
response = await agent.ainvoke({"messages": query})

from langchain_mcp_adapters.tools import to_fastmcp

@tool
def search_database(query: str) -> str:
    """Search database"""
    ...

server = FastMCP("Database", tools=[to_fastmcp(search_database)])
server.run(transport="stdio")   # 现在 Claude Desktop 也能调用你的工具

# 正确：用 __file__ 计算绝对路径
SERVER_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "mcp_math_server.py")

# 错误
result = tool.ainvoke({"a": 1, "b": 2})    # 返回 coroutine，不是结果

# 正确
result = await tool.ainvoke({"a": 1, "b": 2})

pip install langchain-mcp-adapters mcp langgraph