# MCP Server

本节，我们将在 LangGraph 中接入 MCP Server。在接入 MCP Server 之前，必须开发 MCP Server。这可是我的老本行。在[《新瓶装旧酒：纸牌魔术 MCP》](https://luochang212.github.io/posts/card_magic_mcp/)一文中，我已经总结出一套高效的开发方法了。下面我将使用这种方法创建 MCP Servers，再将它们接入 LangGraph。

> **Note**
> 创建 MCP Servers 的完整代码放在 [GitHub 仓库](https://github.com/luochang212/dive-into-langgraph/tree/main/mcp_server) 了，如有兴趣可往一观＼(\`Δ’)／

## 一、开发 MCP 服务

### 1）天气 MCP

以 `get_weather_mcp` 为例，我们要把这个 MCP 写成一个 Python 包。当然仅供本地使用，如果你想传到 PyPI 上当然可以，但那就是另外的流程了，请参考我的博客 [《PyPI 打包小记》](https://luochang212.github.io/posts/pypi_packaging/)。

为了让它被识别为 Python 包，我们要在项目下，新建一个 `__init__.py` 文件。然后把主逻辑写在 `server.py` 中。

```python
# server.py

# -*- coding: utf-8 -*-
from fastmcp import FastMCP


mcp = FastMCP("get_weather_mcp")


@mcp.tool
def get_weather(city: str) -> str:
    """Get weather for a given city."""
    return f"It's always sunny in {city}!"


if __name__ == "__main__":
    mcp.run()
```

接着在 `__main__.py` 中使用 `from . import server` 引入它。最后用 streamable-http 的方式部署：

```python
# __main__.py

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

from . import server


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


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()
```

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

```bash
python -m get_weather_mcp
```

### 2）算数 MCP

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

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

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

```python
# -*- 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](https://github.com/Supervisor/supervisor) 是一个进程管理工具。你告诉它有哪些 MCP 要跑，它会守护你的 MCP 宝宝。当 MCP 挂掉的时候，supervisord 能够自动拉起 MCP。这些内容在我的博客 [《后台管理工具介绍》](https://luochang212.github.io/posts/process_manager/) 中有做简略的介绍（但更多是关于 `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](https://www.trae.ai/) 帮我写的。下面是关于常用命令的说明！

### 1）安装 `supervisord`

```bash
pip install supervisor
```

### 2）启动 `supervisord`

```bash
supervisord -c ./mcp_supervisor.conf
```

### 3）关闭 `supervisord`

```bash
pkill -f supervisord
```

### 4）检查端口状态

```bash
lsof -i :8000
lsof -i :8001
```

## 三、在 LangGraph 中使用 MCP

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

```
pip install langchain-mcp-adapters
```

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

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

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

### 1）启动 MCP 服务

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

启动 `get_weather_mcp`：

```bash
python -m mcp_server.get_weather_mcp 
```

测试 MCP Server 是否成功启动：

```python
# !lsof -i :8000
```

### 2）接入 MCP 服务

使用 `MultiServerMCPClient` 接入 MCP Server。

```python
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` 文件中，这种调用方法会失败。

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

```
'福州的天气总是晴朗明媚！如果您计划前往福州，可以期待阳光充足的天气。不过，建议您出行前还是查看一下最新的天气预报，以确保做好相应的准备。'
```

```python
# 调用算数 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`，改动部分如下：

```python
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())

```