LangChain搭建MCP服务端和客户端流程

发布时间：2026-03-18 20:11:50编辑：123阅读（324）

LangChain搭建MCP客户端流程，如何将MCP工具接入LangGraph中并创建智能体？





创建自定义MCP工具

作为大模型开发者，掌握MCP工具开发流程是基本功，这里先尝试自定义MCP工具，并接入LangGraph。




借助conda创建Python项目

conda create --name mcp_demo python=3.10

conda activate mcp_demo




安装依赖

pip install httpx dotenv langgraph langchain langchain_mcp_adapters lxml




编写MCP服务器

查询天气服务器 weather_server.py

代码如下：

import requests
from lxml import etree
from typing import Any
import time
from mcp.server.fastmcp import FastMCP

# 初始化mcp服务器
mcp = FastMCP('WeatherServer', log_level="ERROR")

@mcp.tool()
async def fetch_weather(city: str) -> dict[str, Any] | None:
    """
        查询即时天气函数
        :param city: 必要参数，字符串类型。用于表示查询天气的具体城市名称
        :return: 返回即时天气的结果，dict类型
        """
    today_time = time.strftime("%Y-%m-%d", time.localtime())
    url = f'https://www.ks121.com/history/?location={city}&startdate={today_time}&enddate={today_time}'
    headers = {
        "sec-ch-ua": '"Not(A:Brand";v="8", "Chromium";v="144", "Google Chrome";v="144"',
        "user-agent": "Mozilla/5.0(Windows NT 10.0;Win64; x64) AppleWebKit/537.36 (KHTML, 
        likeGecko) Chrome/ 144.0.0.0Safari / 537.36"}
    response = requests.get(url=url, headers=headers, timeout=10)
    html = etree.HTML(response.text)
    res = html.xpath("//div[@class='box']/div[2]/div[1]/div/table/tbody/tr")
    ret = {}
    for i in res:
        one = i.xpath("./td[1]/p[2]/text()")[0]
        two = i.xpath("./td[2]/p[2]/span/text()")[0]
        three = i.xpath("./td[3]/p/text()")[0]
        four = i.xpath("./td[4]/p/text()")[0]
        ret.setdefault('日期', one)
        ret.setdefault('气象', two)
        ret.setdefault('温度', three)
        ret.setdefault('风级', four)
    print(f"{today_time}:{city}天气信息:{ret}")
    return ret

if __name__ == '__main__':
    mcp.run(transport='stdio')




写入本地文档服务器write_server.py

用于将指定文本信息写入本地文件的外部函数


代码如下：

import os
from datetime import datetime
from mcp.server.fastmcp import FastMCP

# 初始化mcp服务器
mcp = FastMCP('WriteServer', log_level="ERROR")

USER_AGENT = 'write-app/1.0'

output_dir = './output'

# 确保目录存在
os.makedirs(output_dir, exist_ok=True)

@mcp.tool()
async def write_file(content:str) -> str:
    """
    将指定内容写入本地文件，并返回生成的文件名。
    :param content: 必要参数，字符串类型，用于表示需要写入文档的具体内容。
    :return: 写入结果与文件路径
    """
    # 创建带时间戳的文件名
    timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
    filename = f'note_{timestamp}.txt'
    filepath = os.path.join(output_dir, filename)
    try:
        with open(filepath, 'w', encoding='utf-8') as f:
            f.write(content)
        return f"成功写入文件:{filepath}"
    except Exception as e:
        return f"写入失败:{e}"

if __name__ == '__main__':
    mcp.run(transport='stdio')




创建LangGraph MCP客户端


使用LangGraph接入MCP工具，核心需要使用langchain_mcp_adapters库，该库可以将MCP工具信息进行解析，并让LangGraph顺利识别，识别后即可像任意其他工具一样接入LangGraph中并搭建智能体。

创建MCP配置文件


为了完整实现一个标准的MCP调用流程，即通过配置文件灵活说明MCP工具信息然后再进行调用，这里先创建一个servers_config.json文件，用于记录MCP工具信息：

{
  "mcpServers": {
    "weather": {
      "command": "python",
      "args": ["weather_server.py"],
      "transport": "stdio"
    },
    "write": {
      "command": "python",
      "args": ["write_server.py"],
      "transport": "stdio"
    }
  }
}




然后为当前Agent创建一个提示词模板agent_prompts.txt

内容如下：

你是一个智能体，具备以下两种能力:

1.查询天气：调用query_weather(city:str),返回指定城市的实时天气

2.写入文件：调用write_file(content:str),将文本内容写入本地文件并返回路径

当用户提出请求时，你需要理解意图并选择相应的工具。如果请求缺少必要信息，先与用户确认后再调用工具。返回结果时
以简洁、友好的方式回复。
如果用户提出的问题与你的功能无关，请礼貌告知无法处理。




创建client.py主函数文件

代码如下：


import asyncio
import json
from langchain_ollama import ChatOllama
from typing import Any,Dict
from langchain.agents import create_agent
from langchain_mcp_adapters.client import MultiServerMCPClient
from langgraph.checkpoint.memory import InMemorySaver

# 设置记忆存储
checkpointer = InMemorySaver()

# 读取提示词
with open("./agent_prompts.txt", mode='r', encoding='utf-8') as f:
    prompt = f.read()

# 设置对话配置
config = {
    "configurable":{
        "thread_id":"1"
    }
}

class Configuration:
    @staticmethod
    def load_servers(file_path: str = "servers_config.json") -> Dict[str, Any]:
        with open(file_path, mode='r', encoding='utf-8') as f:
            return json.load(f).get("mcpServers", {})

async def run_chat_loop():
    """启动 MCP-Agent 聊天循环"""
    cfg = Configuration()
    servers_cfg = Configuration.load_servers()
    # 连接多台MCP服务器
    mcp_client = MultiServerMCPClient(servers_cfg)
    _tools = await mcp_client.get_tools()
    print(f"已加载{len(_tools)}个MCP工具:{[t.name for t in _tools]}")

    # 初始化大模型
    model = ChatOllama(
    model="qwen3:8b",
    temperature=0,
    top_p=0.95,
    )

    # 构造 LangGraph Agent
    agent = create_agent(
        model=model,
        tools=_tools,
        system_prompt=prompt,
        checkpointer=checkpointer
    )
    print("\n MCP Agent 已启动，输入'quit'退出")
    while True:
        user_input = input("\n你：").strip()
        if user_input.lower() == "quit":
            break
        try:
            result = await agent.ainvoke(
                {"messages": [{"role":"user", "content":user_input}]},
                config
            )
            print(f"\nAI：{result['messages'][-1].content}")
        except Exception as e:
            print(f"\n 错误:{e}")

if __name__ == '__main__':
    asyncio.run(run_chat_loop())

运行client.py，执行结果如下：

已加载2个MCP工具:['query_weather', 'write_file']

 MCP Agent 已启动，输入'quit'退出

你：今天上海的天气，并把结果保存到本地文件

AI：今天上海的天气已成功保存到本地文件。您可以在以下路径查看详细信息：

`/output/note_20260317_224447.txt`

文件内容包含：

```

日期: 2026-03-17

气象: 小雨

温度: 10℃

～ 11℃

风级: 北风转西北风<3级转3-4级

```

目录结构

关键字：