跳转至

使用python开发mcp server之sse

前面文章我们使用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 类型

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
# 初始化项目
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 文件，写入以下代码

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
# -*- 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