Skip to article frontmatterSkip to article content

MCP Server

本节，我们将在 LangGraph 中接入 MCP Server。在接入 MCP Server 之前，必须得有 MCP Server。也就是说，我们得开发一个 MCP Server。这可是我的老本行。在《新瓶装旧酒：纸牌魔术 MCP》一文中，我已经总结出一套高效的开发方法了。创建 MCP Server 的完整代码，我放在仓库的 mcp_server 路径下，如有兴趣可往一观。

一、开发 MCP 服务

1）天气 MCP

以 get_weather_mcp 为例，我们要把这个 MCP 写成一个 Python 包。当然仅供本地使用，如果你想传到 PyPI 上当然可以，但那就是另外的流程了，敬请参考我的博客 《PyPI 打包小记》。

为了让它被识别为 Python 包，我们要在项目下，新建一个 __init__.py 文件。然后把主逻辑写在 server.py 中，接着在 __main__.py 中使用 from . import server 引入它。最后用 streamable-http 的方式部署它：

def http():
    """streamable-http entry point for the package."""
    asyncio.run(server.mcp.run(transport="http",
                               host=host,
                               port=port,
                               path="/mcp"))

写到这里就齐活了。这里使用 __main__.py 是有小巧思的，这样我们可以将这个包作为模块直接在命令行使用。什么意思呢？就是我们用 python -m [包名] 就等于直接运行了 __main__.py 这个特殊文件。由于我们先前在该特殊文件中启动了 http() 函数，这样就能快捷方便地把 MCP Server 启动起来了！对于我们的 get_weather_mcp，启动命令如下：

python -m get_weather_mcp

2）算数 MCP

这还需要赘述吗？开发流程照抄上面的步骤。

真的是超级模版化。__init__.py 和 __main__.py 几乎完全相同。

唯一需要改动的是 __main__.py。需要把端口 port 改成新号码，一般来说加 1 就行。这里我们把 8000 改成 8001，其他不变：

# -*- coding: utf-8 -*-
import asyncio
import os

from . import server


host = os.getenv('HOST', '127.0.0.1')
port = int(os.getenv('PORT', 8001))


def stdio():
    """Stdio entry point for the package."""
    asyncio.run(server.mcp.run(transport="stdio"))


def http():
    """streamable-http entry point for the package."""
    asyncio.run(server.mcp.run(transport="http",
                               host=host,
                               port=port,
                               path="/mcp"))


if __name__ == "__main__":
    http()

二、使用 supervisord 管理 MCP 服务

supervisord 是一个 进程管理工具。你告诉它有哪些 MCP 要跑，它会守护你的 MCP 宝宝。当 MCP 挂掉的时候，supervisord 能够自动拉起 MCP。这些内容在我的博客 《后台管理工具介绍》 中有做简略的介绍（但更多是关于 systemd 和 pm2 的）。

首先，我们打开项目的 mcp_server 路径，在这里创建一个配置文件 mcp_supervisor.conf，来给 supervisord 使用。我的配置如下：

[unix_http_server]
file=/tmp/supervisor.sock

[supervisord]
logfile=/tmp/supervisord.log
logfile_maxbytes=50MB
logfile_backups=10
loglevel=info
pidfile=/tmp/supervisord.pid
nodaemon=false
minfds=1024
minprocs=200

[rpcinterface:supervisor]
supervisor.rpcinterface_factory = supervisor.rpcinterface:make_main_rpcinterface

[supervisorctl]
serverurl=unix:///tmp/supervisor.sock

[program:math_mcp]
command=python -m mcp_server.math_mcp
directory=..
autostart=true
autorestart=true
startsecs=5
stopwaitsecs=10
stdout_logfile=/tmp/math_mcp.log
stderr_logfile=/tmp/math_mcp_err.log

[program:weather_mcp]
command=python -m mcp_server.get_weather_mcp
directory=..
autostart=true
autorestart=true
startsecs=5
stopwaitsecs=10
stdout_logfile=/tmp/weather_mcp.log
stderr_logfile=/tmp/weather_mcp_err.log

[group:mcp_servers]
programs=math_mcp,weather_mcp

至此，math_mcp、weather_mcp 的配置就完成了。这种东西没必要自己写，我是让 TRAE 帮我写的。下面是关于常用命令的说明！

1）安装 supervisord

pip install supervisor

2）启动 supervisord

supervisord -c ./mcp_supervisor.conf

3）关闭 supervisord

pkill -f supervisord

4）检查端口状态

lsof -i :8000
lsof -i :8001

三、在 LangGraph 中使用 MCP

在使用之前，需要安装配适该功能的 Python 包：

pip install langchain-mcp-adapters

我也是服了开发团队，依我看 LangChain、LangGraph 不如合成一个包。还要我们去功能在哪个包里，真费劲！而且各种功能也被拆得稀碎，看看我到目前为止都安装多少包了：

langchain[openai]
langchain-mcp-adapters
langgraph
langgraph-cli[inmem]
langgraph-supervisor
langgraph-checkpoint-sqlite

若非 LangGraph 1.0 更新了不少好功能，我是打心眼里看不上这个开源项目。衷心祝愿后起之秀 AgentScope 吸收 LangGraph 1.0 的长处并超越它。当然在此之前，我们得承认 LangGraph 的地位。它虽不完美，但依然是最强大的那个。

1）启动 MCP 服务

我们只启动天气 MCP。算数 MCP 稍后我们将以 stdio 的方式调用，无需单独启动服务。

启动 get_weather_mcp：

python -m mcp_server.get_weather_mcp

测试 MCP Server 是否成功启动：

# !lsof -i :8000

2）接入 MCP 服务

使用 MultiServerMCPClient 接入 MCP Server.

import os

from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_mcp_adapters.client import MultiServerMCPClient  
from langchain.agents import create_agent

# 加载模型配置
_ = load_dotenv()

# 加载模型
llm = ChatOpenAI(
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    base_url=os.getenv("DASHSCOPE_BASE_URL"),
    model="qwen3-coder-plus",
    temperature=0.7,
)

async def mcp_agent():
    # 我们用两种方式启动 MCP Server：stdio 和 streamable_http
    client = MultiServerMCPClient(  
        {
            "math": {
                "command": "python",
                "args": [os.path.abspath("./mcp_server/math_mcp/server.py")],
                "transport": "stdio",
            },
            "weather": {
                "url": "http://localhost:8000/mcp",
                "transport": "streamable_http",
            }
        }
    )
    
    tools = await client.get_tools()
    agent = create_agent(
        llm,
        tools=tools,
    )

    return agent

async def use_mcp(messages):
    agent = await mcp_agent()
    response = await agent.ainvoke(messages)
    return response

在 Jupyter Notebook 中，使用 response = await use_mcp(messages) 命令调用函数。但是在 .py 文件中，这种调用方法会失败。

# 调用天气 MCP
messages = {"messages": [{"role": "user", "content": "福州天气怎么样？"}]}
response = await use_mcp(messages)
response["messages"][-1].content
'福州的天气总是晴朗明媚！如果您计划前往福州，可以期待阳光充足的天气。不过，建议您出行前还是查看一下最新的天气预报，以确保做好相应的准备。'
# 调用算数 MCP，由于是 stdio，启动会慢一点
messages = {"messages": [{"role": "user", "content": "计算 (3 + 5) * 12"}]}
response = await use_mcp(messages)
response["messages"][-1].content
'(3 + 5) * 12 的结果是 96。'

在 .py 文件中，应该使用 asyncio，改动部分如下：

import asyncio

async def main():
    # 调用天气 MCP
    messages = {"messages": [{"role": "user", "content": "福州天气怎么样？"}]}
    response = await use_mcp(messages)
    print(response["messages"][-1].content)

if __name__ == "__main__":
    asyncio.run(main())
首页
上下文工程
首页
监督者模式