LangChain+MCP（模型上下文协议）实现案例

基于Langchain框架构建含MCP功能的大模型应用简单案例。

WH_214

1084人浏览 · 2025-05-20 23:02:12

WH_214 · 2025-05-20 23:02:12 发布

LangChain+MCP（模型上下文协议）实现案例

概述：通过Langchain选择LLM构建Agent，配置Agent的MCP client（包括其可调用的MCP server信息），案例则开发一些简单的MCP server，提供给Agent调用，从而使其输出更满意的结果。

1.环境准备

（1）Python版本 >= 3.10

（2）依赖包安装：

langchain-mcp-adapters
langgraph
langchain-openai
python-dotenv

pip install langchain-mcp-adapters langgraph langchain-openai python-dotenv

（3）.env 环境变量配置

OPENAI_API_KEY=xxx
# ChatOpenAI()构建openai自己的model只需指定model的名称即可

DEEPSEEK_API_BASE=https://api.deepseek.com
DEEPSEEK_API_KEY=sk-xxx

ZHIPUAI_API_BASE=https://open.bigmodel.cn/api/paas/v4/
ZHIPUAI_API_KEY=xxx

2.创建MCP server

案例构建两个MCP server，包括一个math_server和weather_server，模拟数学运算和天气服务。

# math_server.py
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("Math")

@mcp.tool()
def add(a: int, b: int) -> int:
    """对两个整数相加"""
    return a + b

@mcp.tool()
def multiple(a: int, b: int) -> int:
    """对两个整数相乘"""
    return a * b

if  __name__ == "__main__":
    mcp.run(transport="stdio")  # 使用标准输入输出

# weather_server.py
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("Weather")

@mcp.tool()
async def get_weather(location: str) -> str:
    """获取位置的天气。"""
    return f"{location}当前天气晴朗，温度 25°C"

if __name__ == "__main__":
    mcp.run(transport="sse")  # 使用 SSE 传输，端口 8000

3.启动MCP server

开启两个终端，使用python启动MCP server，注意在对应py文件目录下面。

python math_server.py

本地的标准输入/输出（Stdio）传输server启动后无任何输出
在这里插入图片描述

python weather_server.py

基于HTTP的服务器发送事件（SSE）传输server启动后，通过uvicorn启动http服务如下
在这里插入图片描述

4.客户端集成（LangGraph + MCP）

构建Agent，配置MCP client，连接MCP server，测试Agent输出内容效果。

from langchain_mcp_adapters.client import MultiServerMCPClient
from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI
import asyncio
import os
from dotenv import load_dotenv

load_dotenv() # 加载.env文件环境变量

# 创建Agent的model
model = ChatOpenAI(
    openai_api_base=os.getenv("ZHIPUAI_API_BASE"),  # 智谱LLM调用base url
    openai_api_key=os.getenv("ZHIPUAI_API_KEY"),  # 环境变量API_KEY
    model_name="GLM-4-FlashX"  # 模型名称
)

# 定义异步任务运行 agent，并输入两个测试问题，打印其输出内容，测试MCP server被使用情况
async def run_client():  
    async with MultiServerMCPClient(
            {
                "math": {
                    "command": "python",
                    "args": ["./math_server.py"],  # 确保定位的MCP server 源py文件路径正确
                    "transport": "stdio",
                },
                "weather": {
                    # 确保您在 8000 端口启动天气服务器
                    "url": "http://localhost:8000/sse",
                    "transport": "sse",
                }
            }
    ) as client:
        # 构建agent,传入model和配置好的MCP client获取的tools
        agent = create_react_agent(model, client.get_tools())

        # 测试数学工具
        math_response = await agent.ainvoke({"messages": "请问(3 + 5) x 12=多少?"})
        print("Math Response:", math_response["messages"][-1].content)
        # 测试天气工具
        weather_response = await agent.ainvoke({"messages": "请问北京今天天气怎么样?"})
        print("Weather Response:", weather_response["messages"][-1].content)


# 运行异步函数
if __name__ == "__main__":
    # 使用asyncio包异步运行client测试任务
    asyncio.run(run_client())

5.运行与验证

启动两个服务器
- 确保在两个终端运行了 math_server.py 和 weather_server.py，保证成功启动MCP server。

运行客户端

# 运行client.py文件
multi_server_MCP_client.py

预期输出结果：
在这里插入图片描述

6.使用FastApi提供接口调用

尝试通过FastApi接口对结合了MCP的Agent进行调用，但是一直有问题。

示例代码如下，实际创建model、MCPClient、Agent的代码都应该封装，且调用应放在业务层，这里为了快速演示直接使用。

@router.get("/agent/test/weather/{location}")
async def get_location_weather(location: str):
    try:
        model = ChatOpenAI(
            openai_api_base=os.getenv("ZHIPUAI_API_BASE"),  # 智谱LLM调用base url
            openai_api_key=os.getenv("ZHIPUAI_API_KEY"),  # 环境变量API_KEY
            model_name="GLM-4-FlashX"  # 模型名称
        )
        async with MultiServerMCPClient(
                {
                    "math": {
                        "command": "python",
                        "args": ["MCP/math_server.py"],
                        "transport": "stdio",
                    },
                    "weather": {
                        # 确保您在 8000 端口启动天气服务器
                        "url": "http://localhost:8000/sse",
                        "transport": "sse",
                    }
                }
        ) as client:
            agent = create_react_agent(model, client.get_tools())
            math_response = await agent.ainvoke({"messages": f"请问{location}今天天气如何?"})
            response_content = math_response["messages"][-1].content

            return {
                "code": 200,
                "response": response_content,
                "msg": f"获取{location}位置天气成功"
            }

    except Exception as e:
        print(f"服务端异常: {str(e)}")
        return {
            "code": 400,
            "response": None,
            "msg": f"获取{location}位置天气失败"
        }