langgraph实战mcp

发表于 更新于 分类于 阅读次数：

环境配置

1
pip install langchain-mcp-adapters

使用langgraph调用mcp

要点主要是利用MultiServerMCPClient构建服务，获取tool

利用预设的create_react_agent构建ReAct架构的智能体并调用工具

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
import asyncio # 需要导入 asyncio 来运行异步函数
# 从langchain_mcp_adapters.client模块导入MultiServerMCPClient类
# 从langgraph.prebuilt模块导入create_react_agent函数
from langchain_mcp_adapters.client import MultiServerMCPClient
from langgraph.prebuilt import create_react_agent

# 导入 LLM 相关库
from langchain_openai import ChatOpenAI

# 将主要逻辑封装在一个异步函数中
async def main():
    # 创建MultiServerMCPClient实例，配置两个不同的服务
    client = MultiServerMCPClient(
        {
            "math": {  # 数学计算服务
                "command": "python",  # 使用python命令启动
                # 替换为你的math_server.py文件的绝对路径
                "args": ["/workspace/langgraph-mcp/math_server.py"],
                "transport": "stdio",  # 使用标准输入输出传输
            },
            "weather": {  # 天气服务
                # 确保你的天气服务器在8000端口运行
                # *** 确保这个 URL 是正确的，并且服务器正在运行 ***
                "url": "http://localhost:8000/mcp",
                "transport": "streamable_http",  # 使用可流式HTTP传输
            }
        }
    )

    tools = []
    try:
        # 在异步函数内部正确使用 await
        tools = await client.get_tools()
        print(f"成功获取到 {len(tools)} 个MCP工具。")
        for tool_item in tools:
            print(f"  - {tool_item.name}: {tool_item.description}")
    except Exception as e:
        print(f"获取MCP工具失败: {e}")
        print("请确保MCP服务URL有效且可访问，或者您已正确配置了认证信息。")
        # 在函数内部，如果出错可以选择返回或继续处理
        # return # 这里可以 return，但会结束 main 函数

    if not tools:
        print("没有获取到工具，无法创建代理。")
        return

    # 创建ReAct代理
    llm = ChatOpenAI(
        model="qwen3-235b-a22b-thinking-2507",
        api_key="sk-a8ef27c47ea84224ac6eed6d4bba1bab",
        base_url="https://dashscope.aliyuncs.com/compatible-mode/v1" # 修正了末尾多余的空格
    )
    agent = create_react_agent(llm, tools)
    

    # 异步调用代理来解决数学问题
    # 确保在异步函数内部使用 await
    math_response = await agent.ainvoke(
        {"messages": [{"role": "user", "content": "what's (3 + 5) x 12?"}]}
    )
    print("\n--- 数学问题回答 ---")
    print(math_response["messages"][-1].content) # 打印最后一条消息（LLM的回答）

    # 异步调用代理来查询天气
    weather_response = await agent.ainvoke(
        {"messages": [{"role": "user", "content": "what is the weather in nyc?"}]}
    )
    print("\n--- 天气问题回答 ---")
    print(weather_response["messages"][-1].content) # 打印最后一条消息（LLM的回答）


# --- 这是脚本的入口点 ---
# 使用 asyncio.run() 来运行你的主异步函数
if __name__ == "__main__":
    asyncio.run(main())

参考资料

使用 MCP - LangChain 框架

框架流程

image-20250801164905578

✅ 三个角色（系统组件）

角色 作用
Client 前端或用户界面，发起请求
Auth Provider 认证服务（如 OAuth、JWT 提供者），负责登录和签发 token
LangGraph Backend 应用的后端服务，处理业务逻辑
Secret Store 存放用户敏感信息（如 token、密钥等）
MCP Server 后端工具服务，提供具体的工具或资源接口

✅ 流程详解（12步）

🔐 阶段一：用户登录 & 获取 Token（1~6）

  1. 用户登录
    Client 提交用户名和密码给 Auth Provider。

  2. 返回 Token
    Auth Provider 验证成功后，返回一个访问令牌（token）。

  3. 携带 Token 请求
    Client 将 token 附加在请求头中，发给 LangGraph Backend。

  4. 验证 Token
    LangGraph Backend 使用 @auth.authenticate 中间件验证 token 是否有效。

  5. 获取用户信息
    验证通过后，LangGraph Backend 从 Auth Provider 拉取用户详细信息。

  6. 确认有效性
    后端确认用户信息无误，流程继续。

🔑 阶段二：获取用户权限 Token（6a~6b）

6a. 拉取用户权限 Token
LangGraph Backend 从 Secret Store 获取该用户对应的权限 token（可能是 MCP 所需的访问凭证）。

6b. 返回权限 Token
Secret Store 返回该 token。

🛠️ 阶段三：调用工具 & 返回结果（7~12）

  1. 权限控制检查
    LangGraph Backend 使用 @auth.on.* 权限控制逻辑，确认用户是否有权调用该工具。

  2. 构建 MCP Client
    后端用用户的权限 token 构建一个 MCP 客户端。

  3. 调用 MCP 工具
    MCP Client 发起请求，调用某个具体工具，携带 token（通常放在请求头中）。

  4. MCP 验证并执行
    MCP Server 验证 token 是否有效，确认无误后执行工具逻辑。

  5. 工具返回结果
    MCP Server 返回工具执行结果或资源数据。

  6. 返回给前端
    LangGraph Backend 将结果返回给 Client，完成整个链路。