基于SSE传输的MCP服务器实现

基于SSE传输的MCP服务器实现

MCP协议支持多种传输机制，其中包括stdio、Server-Sent Events（SSE） 和Streamable HTTP。

但目前MCP SDK只提供了stdio和SSE两种传输方式的开发库，而暂时还没有提供基于流式HTTP传输的开发工具。因此在进行MCP开发过程中，目前只考虑实现stdio和SSE传输方式。

在使用MCP Python SDK开发MCP服务器时，只需要在此处进行设置，即可让MCP服务器开启SSE模式。

mcp.run(transport='sse')

本文以创建一个查询天气的MCP服务器为例进行开发。

1. 环境准备

本文使用Ubuntu22.04作为开发环境。

1.1 nodejs安装

在Ubuntu上安装Node.js可以通过多种方法来实现，以下是三种常用的方法：

• 使用Ubuntu存储库

• 通过NodeSource PPA

• 使用nvm (Node Version Manager)

在Ubuntu 22.04，apt 默认的nodejs 版本是 v12，而最新的 nodejs都已经 v20+了，，因此使用第一种方式安装的nodejs版本较老，后续执行npm命令执行会报错（已测试）。

使用第二种方式进行nodejs的安装。

# 安装Node.js 20.x版本
sudo apt update
sudo apt install curl
curl -fsSL https://deb.nodesource.com/setup_20.x | sudo bash -
sudo apt install -y nodejs
node -v
npm -v

1.2 uv安装

MCP开发要借助uv进行虚拟环境创建和依赖管理。uv 是一个Python依赖管理工具，类似于 pip 和 conda，但它更快、更高效，并且可以更好地管理Python虚拟环境和依赖项。它的核心目标是替代pip、venv 和pip-tools，提供更好的性能和更低的管理开销。

uv 的特点：

• 速度更快：相比 pip，uv 采用Rust 编写，性能更优。
• 支持PEP 582：无需 virtualenv，可以直接使用 __pypackages__ 进行管理。
• 兼容pip：支持 requirements.txt 和 pyproject.toml 依赖管理。
• 替代 venv：提供 uv venv 进行虚拟环境管理，比 venv 更轻量。
• 跨平台：支持 Windows、macOS 和 Linux。

使用pip安装uv：

pip install uv

1.3 OpenWeather秘钥获取

获取天气的接口，使用OpenWeather提供的接口，官方地址：http://openweathermap.org/

1.4 pypi 秘钥获取

将开发测试完的依赖库，需要pypi的秘钥，官方地址：https://pypi.org/

该过程需要VPN，具体步骤可参考：https://blog.csdn.net/qq_53545309/article/details/140788076

2. 代码编写

2.1 初始化环境

• 创建项目主目录

  cd /root/autodl-tmp/mcp
  mkdir mcp-sse-test
  cd mcp-sse-test
  

• 创建基础项目结构

  uv init mcp-get-weather-scorpios
  cd mcp-get-weather-scorpios
  
  # 创建虚拟环境
  uv venv
  
  # 激活虚拟环境
  source .venv/bin/activate
  
  uv add mcp httpx
  

创建完的目录结构如下：

各文件解释如下：

2.2 代码编写

删除主目录下的main.py文件，并创建代码文件夹：

mkdir -p ./src/mcp_get_weather
cd ./src/mcp_get_weather

下面创建服务器核心代码， 在src/mcp_get_weather中创建三个代码文件：
其中server.py主要负责进行天气查询，代码如下：

import json
import httpx
import argparse  
from typing import Any
from mcp.server.fastmcp import FastMCP

# 初始化 MCP 服务器
mcp = FastMCP("WeatherServer")

# OpenWeather API 配置
OPENWEATHER_API_BASE = "https://api.openweathermap.org/data/2.5/weather"
API_KEY = None  
USER_AGENT = "weather-app/1.0"

async def fetch_weather(city: str) -> dict[str, Any] | None:
    """
    从 OpenWeather API 获取天气信息。
    """
    if API_KEY is None:
        return {"error": "API_KEY 未设置，请提供有效的 OpenWeather API Key。"}

    params = {
        "q": city,
        "appid": API_KEY,
        "units": "metric",
        "lang": "zh_cn"
    }
    headers = {"User-Agent": USER_AGENT}

    async with httpx.AsyncClient() as client:
        try:
            response = await client.get(OPENWEATHER_API_BASE, params=params, headers=headers, timeout=30.0)
            response.raise_for_status()
            return response.json()
        except httpx.HTTPStatusError as e:
            return {"error": f"HTTP 错误: {e.response.status_code}"}
        except Exception as e:
            return {"error": f"请求失败: {str(e)}"}

def format_weather(data: dict[str, Any] | str) -> str:
    """
    将天气数据格式化为易读文本。
    """
    if isinstance(data, str):
        try:
            data = json.loads(data)
        except Exception as e:
            return f"无法解析天气数据: {e}"

    if "error" in data:
        return f"⚠️ {data['error']}"

    city = data.get("name", "未知")
    country = data.get("sys", {}).get("country", "未知")
    temp = data.get("main", {}).get("temp", "N/A")
    humidity = data.get("main", {}).get("humidity", "N/A")
    wind_speed = data.get("wind", {}).get("speed", "N/A")
    weather_list = data.get("weather", [{}])
    description = weather_list[0].get("description", "未知")

    return (
        f"🌍 {city}, {country}\n"
        f"🌡 温度: {temp}°C\n"
        f"💧 湿度: {humidity}%\n"
        f"🌬 风速: {wind_speed} m/s\n"
        f"🌤 天气: {description}\n"
    )

@mcp.tool()
async def query_weather(city: str) -> str:
    """
    输入指定城市的英文名称，返回今日天气查询结果。
    """
    data = await fetch_weather(city)
    return format_weather(data)

def main():
    parser = argparse.ArgumentParser(description="Weather Server")
    parser.add_argument("--api_key", type=str, required=True, help="你的 OpenWeather API Key")
    args = parser.parse_args()
    global API_KEY
    API_KEY = args.api_key
    mcp.run(transport='sse')

if __name__ == "__main__":
    main()

采用了fastmcp进行创建，而在mcp.run中设置了使用sse方式进行传输。

mcp.run(transport='sse')

在__init__.py中写入

from .server import main

在__main__.py中写入：

from mcp_get_weather import main

main()

同时回到主目录，修改项目配置文件pyproject.toml：

[build-system]
requires = ["setuptools>=61.0", "wheel"]
build-backend = "setuptools.build_meta"

[project]
name = "mcp-get-weather-scorpios"
version = "0.1.0"
description = "输入OpenWeather-API-KEY，获取天气信息。"
readme = "README.md"
requires-python = ">=3.10"
dependencies = [
    "httpx>=0.28.1",
    "mcp>=1.11.0",
]

[project.scripts]
mcp-get-weather = "mcp_get_weather:main"

[tool.setuptools]
package-dir = {"" = "src"}

[tool.setuptools.packages.find]
where = ["src"]

3. 代码测试

对编写完的SSE MCP服务进行测试，使用MCP提供的Inspector进行服务器性能测试，需要先开启MCP服务器，然后再开启Inspector：

在stdio模式下是开启Inspector时同步开启MCP Server

• 开启SSE MCP服务器：

# 回到项目主目录
# cd /root/autodl-tmp/mcp/mcp-sse-test/mcp-get-weather-scorpios

uv run ./src/mcp_get_weather/server.py --api_key YOUR_KEY

• 开启Inspector

# 回到项目主目录
# cd /root/autodl-tmp/mcp/mcp-sse-test/mcp-get-weather-scorpios

npx -y @modelcontextprotocol/inspector uv run ./src/mcp_get_weather/server.py --api_key YOUR_KEY

• 进行测试

打开Inspector，并选择SSE模式，选择默认运行地址：http://localhost:8000/sse，再配置下代理网关和秘钥（使用AutoDL服务器需要），然后点击connect，输入地名进行测试：：

• 选择SSE和填写地址
• 配置相关参数
• 选择Tools、tools-list
• 输入地址进行调试

测试完成后，即可上线发布。

4. 打包上传

# 回到项目主目录
# cd /root/autodl-tmp/mcp/mcp-sse-test/mcp-get-weather-scorpios

pip install build
uv pip install build twine
python -m build
python -m twine upload dist/*

查看发布的库：https://pypi.org/search/?q=mcp-get-weather-scorpios

5. 下载验证

• 本地安装：

pip install mcp-get-weather-scorpios -i https://pypi.org/simple/

• 开启服务：

  uv run mcp-get-weather --api_key 36d………………b
  

然后即可打开浏览器输入http://localhost:8000/sse测试连接情况

注意，这里在服务器上开启服务然后本地连接也可以，和stdio不同，SSE模式下的MCP服务器并不需要本地运行。

6. CherryStudio 验证

使用Cherry studio连接SSE模式下的MCP服务器，只需要输入服务器地址即可：