跳到主要内容

使用python开发mcp server之sse

· 阅读需 8 分钟

前面文章我们使用python 实现了简单的天气查询mcp server，并使用cherry studio 和 Cline 实现了mcp server 的调用，之前是使用stdio 类型的mcp，本章看一下如何开发sse类型的mcp 服务。

与stdio 的区别​

sse，Server-Sent Events 是基于http 长连接的服务，以往我们在开发大模型应用的时候，由于大模型生成token 不是一次性生成的，是一个token 一个token 生成的，为了避免调用端长时间等待，会一点点的输出给调用端。

在mcp 的开发过程中，也会用到sse，但是这里的sse 和大模型输出稍微有一点不同，sse 类型的mcp 会建立一个http 监听端口，客户端在调用mcp server 时，先和mcp 建立一个长连接，之后工具的输出都是通过这个长连接发送给客户端，由于sse是单向传输，客户端只能被动的接受服务端的数据，而不能在建立连接以后再次接收客户端的数据，所以客户端在调用mcp 工具时，需要向另外一个接口发送数据。所以开发sse 的mcp 会稍微复杂一些，但是官方python sdk 中FastMCP已经做了高级封装，开发者只需要关注工具的具体实现即可。

本地stdio 类型的mcp server，通过标准的输入输出来进行数据的交互，调用端需要提前安装好环境，如python 或者 node，当然如果mcp 是一个可执行的二进制文件也可以不用安装配置环境，只是目前大部分的mcp 使用python 或者 node 进行开发的。

sse mcp server​

本节依然使用uv来创建项目，但是这次不再创建package类型的项目，而是默认的app 类型

# 初始化项目
uv init gaodeweather --python 3.11

cd gaodeweather

# 同步虚拟环境 
uv sync 

# 激活虚拟环境
.venv\Scripts\activate  # source .venv/bin/activate # linux or mac

# 安装包
uv add mcp

创建完项目以后，由于是app 类型的项目，所以这里没有src 目录，只需要在项目根目录上开发即可，新建一个main.py 文件，写入以下代码

# -*- coding: utf-8 -*-
from mcp.server.fastmcp import FastMCP
from pydantic import Field
import httpx
import json
import os
import logging

logger = logging.getLogger("mcp")

settings = {
    "log_level": "DEBUG"
}

# 初始化mcp服务
mcp = FastMCP("hello-mcp-server", log_level="ERROR", settings=settings)

# 定义工具
@mcp.tool(name="高德天气查询助手", description="高德天气查询，输入城市名，返回该城市的天气情况，例如：北京")
async def query_logistics(city: str = Field(description="要查询天气的城市名称")) -> str:
    logger.info("收到查询天气请求，城市名：{}".format(city))
    api_key = "xxxx"
    api_domain = 'https://restapi.amap.com/v3'
    url = f"{api_domain}/config/district?keywords={city}"f"&subdistrict=0&extensions=base&key={api_key}"
    headers = {"Content-Type": "application/json; charset=utf-8"}
    async with httpx.AsyncClient(headers=headers) as client:
        response = await client.get(url)
        if response.status_code != 200:
            return "查询失败"
        
        city_info = response.json()
        if city_info["info"] != "OK":
            return "获取城市信息查询失败"
        CityCode = city_info['districts'][0]['adcode']
        weather_url = f"{api_domain}/weather/weatherInfo?city={CityCode}&extensions=all&key={api_key}"
        weatherInfo_response = await client.get(weather_url)
        if weatherInfo_response.status_code != 200:
            return "查询天气信息失败"
        weatherInfo = weatherInfo_response.json()
        if weatherInfo['info'] != "OK":
            return "查询天气信息失败"
        weatherInfo_data = weatherInfo_response.json()
        contents = []
        if len(weatherInfo_data.get('forecasts')) <= 0:
            return "没有获取到该城市的天气信息"
        for item in weatherInfo_data['forecasts'][0]['casts']:
            content = {
                'date': item.get('date'),
                'week': item.get('week'),
                'dayweather': item.get('dayweather'),
                'daytemp_float': item.get('daytemp_float'),
                'daywind': item.get('daywind'),
                'nightweather': item.get('nightweather'),
                'nighttemp_float': item.get('nighttemp_float')
            }
            contents.append(content)
        return json.dumps(contents, ensure_ascii=False)
    
def run():
    mcp.run(transport="sse")


if __name__ == "__main__":
   run()

代码和之前的stdio 类型的mcp 基本上一致，调用查询的实现是一样的，主要有两个差异

  1. 在代码中直接写入高德接口的apikey, api_key = "xxxx"， 因为作为sse方式的mcp，用户是直接通过http 接口来访问服务的，所以可以不需要用户提供api_key，使用固定的key，且key 是放到服务端的，调用者是拿不到的。
  2. mcp.run 方法，transport 参数之前是 stdio , 这里改为了 sse

使用 uv run main.py 或者 python main.py 来启动sse server。

FastMCP 会启动本地8000端口，并且默认添加了 /sse 和 /messages/ 两个路由。

  • /sse 为客户端首次进行sse连接， 连接成功以后，后返回一个唯一的url， 之后客户端再发送工具调用时，就是向这个url 发送。
  • /messages/ ，客户端发送工具调用时的路由

此时可以使用浏览器打开 http://localhost:8000/sse

image.png

客户端第一次打开 /sse 时，mcp server 会和client 建立一个长连接，会返回一个消息入口点 /messages/?session_id=c89ba28190ab40adb2e947fc42e71f28, 之后客户端发送工具调用的http 请求，都是向这个url 发送。

Inspector 调试​

在之前调试 stdio mcp server时，用到了官方的 MCP Inspector ，这个Inspector 同样也可以调试sse 类型的mcp server。使用 mcp dev main.py 启动 Inspector， 注意这里的 main.py 只是为了填充参数用，实际是根本用不到，因为后面我们会在界面上选择sse方式连接。

image.png

打开 Inspector，Transport Type 选择 sse， url 填写 http://localhost:8000/sse, 点击Connect 按钮。 显示连接成功以后，点击导航栏的 Tools，之后再点击 List Tools

image.png

会显示 高德天气查询助手 工具， 点击这个工具名，即可在右侧进行输入输出调试了。

image.png

调试效果和之前 stdio 是一样的。

client 调用​

本文使用 Cherry studio 和 Corsur 来调用这个 sse mcp server。

cherry studio​

打开 cherry studio 的mcp 添加配置界面，类型选择 SSE，url 填写 http://localhost:8000/sse, SSE 类型的mcp server 配置起来要比stdio 类型要简单很多，只需要配个url 即可。

image.png

之后回到聊天界面，依然问一个需要查询天气的问题： ”我明天要去重庆出差，我应该穿什么衣服“

image.png

模型可以分析出要调用的工具以及需要的入参。

Cursor​

在Cursor 配置同样很简单 ,选择sse类型，url 填写 http://127.0.0.1:8000/sse

image.png

返回对话界面，对话模式选择 Agent

image.png

问同样的问题，可以看到，cursor 同样可以正常调用mcp 服务。

image.png

如何选择stdio 还是 sse？​

从sse 和 stdio 的代码来看，工具的实现都是样的，只是FastMCP对象运行时的transport参数不同，那么我们应该如何选择时调用sse 还是 stdio 呢？

以下是我个人的看法，仅供参考

  1. sse 类型的mcp server， 需要部署在云服务器上，如果有服务器资源的话建议优先使用sse方式，因为sse方式对于客户端来讲比较友好，客户端不用下载安装python 或者 node 环境，只要有网络就行
  2. 对于需要操作本地文件的mcp 服务，如读取本地文件内容，创建文件或目录，只能使用本地stdio 方式
  3. 目前对于sse 的鉴权，只能通过在header 中添加请求头的方式，所以复杂的鉴权不太适合sse，建议使用 stdio

如果有服务器资源，且不需要操作本地文件，优先考虑使用sse，如果要操作本地文件，也就是云端服务无法完成的操作只能选择使用stdio 方式。

Todo list​

我们已经使用 stdio 和 sse 方式开发了不同的mcp server，实现过程中，我们一直使用 FastMCP 这个高级的封装类，使用这个类的好处是开发起来比较简单，开发者只需要关注具体的工具实现即可，不好的地方在于，屏蔽的很多实现，如果有定制化的需求，如我们想要开发一个github的mcp 工具，对于调用者需要填写自己的github token，实现对自己拥有权限的仓库操作，这种目前来讲使用FastMCP 就不太适合了，好在官方的sdk 也提供了底层的实现类，可以让开发者自定义一些实现。之后的文章里我们再学习一下如何使用。

  1. 如何做鉴权？
  2. 如何结合 FastAPI 使用？
  3. 如何使用云部署，将自己写的mcp 部署到 mcp.so 或者 smithery.ai

另外多说一句，官方对于目前基于sse的实现，发现了不少问题，如无法做负载均衡等，并且已经提出了相应的调整，协议标准在变，后期这种sse 方式大概率不再使用了，具体的规范参考

https://spec.modelcontextprotocol.io/specification/2025-03-26/

image.png

wp