基于SSE传输的MCP服务器实现_nodejs mcp sse client

基于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 updatesudo apt install curlcurl -fsSL https://deb.nodesource.com/setup_20.x | sudo bash -sudo apt install -y nodejsnode -vnpm -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/mcpmkdir mcp-sse-testcd mcp-sse-test

• 创建基础项目结构

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

创建完的目录结构如下：

各文件解释如下：

文件/文件夹 作用 .git/ Git 版本控制目录 .venv/ 虚拟环境 .gitignore Git 忽略规则 .python-version Python版本声明 main.py 主程序入口 pyproject.toml 项目配置文件 README.md 项目说明文档

2.2 代码编写

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

mkdir -p ./src/mcp_get_weathercd ./src/mcp_get_weather

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

import jsonimport httpximport argparse from typing import Anyfrom 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 mainmain()

同时回到主目录，修改项目配置文件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-scorpiosuv run ./src/mcp_get_weather/server.py --api_key YOUR_KEY

• 开启Inspector

# 回到项目主目录# cd /root/autodl-tmp/mcp/mcp-sse-test/mcp-get-weather-scorpiosnpx -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-scorpiospip install builduv pip install build twinepython -m buildpython -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服务器，只需要输入服务器地址即可：