基于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

创建完的目录结构如下：
在这里插入图片描述
各文件解释如下：

文件/文件夹作用

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

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",
]