一、FastMCP简介

FastMCP是一个基于Python的高级框架，专为构建MCP服务器而设计。它极大简化了MCP服务器的开发流程，让开发者能够以最小的代码量创建功能强大的MCP服务器。

FastMCP的主要特点包括：

简洁的API：通过装饰器模式，简化MCP服务器的创建
丰富的功能：支持工具（Tools）、资源（Resources）、提示模板（Prompts）等MCP核心元素
多种传输方式：支持stdio和SSE等不同传输协议
类型安全：利用Python的类型提示，自动生成MCP协议所需的模式定义
内置图像处理：支持图像数据的自动格式转换和处理。

二、MCP中SSE和STDIO的区别

MCP服务端当前支持两种与客户端的数据通信方式：标准输入输出(stdio)和基于HTTP的服务器推送事件(SSE)。这两种方式有着明显的区别和各自的适用场景。

3.1 标准输入输出(STDIO)
原理：

STDIO是一种用于本地通信的传输方式。在这种模式下，MCP客户端会将服务器程序作为子进程启动，双方通过标准输入和标准输出进行数据交换。具体而言：

客户端通过标准输入（stdin）向服务器发送请求
服务器通过标准输出（stdout）返回响应
服务器可以通过标准错误（stderr）输出日志和错误信息
特点：

低延迟：本地进程间通信，无网络开销
简单直接：不需要网络配置和端口管理
安全性高：通信限制在本地进程内，无需考虑网络安全
适合单会话：一个客户端对应一个服务器实例
适用场景：

客户端和服务器在同一台机器上运行的场景
需要本地访问文件系统或设备的工具
对延迟敏感的应用
开发测试阶段
3.2 服务器推送事件(SSE)
原理：
SSE（Server-Sent Events）是一种基于HTTP的单向通信技术，允许服务器向客户端推送数据。在MCP中，SSE实现了：
客户端通过HTTP POST请求发送数据到服务器
服务器通过持久的HTTP连接向客户端推送事件和数据
通信基于标准的HTTP协议，便于在网络环境中部署

特点：
网络传输：基于HTTP协议，可跨网络通信
多客户端支持：单个服务器实例可同时服务多个客户端
易于部署：可部署在云服务、容器或微服务架构中
扩展性好：适合分布式环境

适用场景：
客户端和服务器位于不同物理位置
需要支持多客户端连接
需要远程访问的服务
生产环境和企业级应用
3.3 SSE与STDIO的详细比较

三、FastMCP实现SSE方式

3.1 基本示例

下面是使用FastMCP实现SSE通信方式的示例，创建一个简单的天气服务：

# weather_sse.py
from fastmcp import FastMCP
import random

# 创建MCP服务器实例，指定端口
mcp = FastMCP("Weather Service", port=8000)

# 模拟的天气数据
weather_data = {
    "大连": {"temp": range(10, 25), "conditions": ["晴", "多云", "小雨"]},
    "北京": {"temp": range(5, 20), "conditions": ["cloudy", "rainy", "foggy"]},
    "齐齐哈尔": {"temp": range(15, 30), "conditions": ["sunny", "cloudy", "humid"]},
    "呼伦贝尔": {"temp": range(20, 35), "conditions": ["sunny", "clear", "hot"]},
}


@mcp.tool()
def get_weather(city: str) -> dict:
    """获取指定城市的当前天气"""
    if city not in weather_data:
        return {"error": f"无法找到城市 {city} 的天气数据"}

    data = weather_data[city]
    temp = random.choice(list(data["temp"]))
    condition = random.choice(data["conditions"])

    return {
        "city": city,
        "temperature": temp,
        "condition": condition,
        "unit": "celsius"
    }


@mcp.resource("weather://cities")
def get_available_cities() -> list:
    """获取所有可用的城市列表"""
    return list(weather_data.keys())


@mcp.resource("weather://forecast/{city}")
def get_forecast(city: str) -> dict:
    """获取指定城市的天气预报资源"""
    if city not in weather_data:
        return {"error": f"无法找到城市 {city} 的天气预报"}

    forecast = []
    for i in range(5):  # 5天预报
        data = weather_data[city]
        temp = random.choice(list(data["temp"]))
        condition = random.choice(data["conditions"])
        forecast.append({
            "day": i + 1,
            "temperature": temp,
            "condition": condition
        })

    return {
        "city": city,
        "forecast": forecast,
        "unit": "celsius"
    }


if __name__ == "__main__":
    # 使用SSE传输方式启动服务器
    mcp.run(transport="sse")

运行SSE模式的MCP服务器：
python weather_sse.py

在Pycharm中启动的SSE服务

服务器将在指定端口（本例中为8000）启动，并监听HTTP连接。您可以通过浏览器访问：
http://localhost:8000/sse

可以使用Cherry Studio来测试SSE的server。
配置MCP server：

问答测试：

另外一种方式，可以使用支持SSE传输的MCP客户端，或者使用如下Python代码创建一个简单的客户端：

# sse_client.py
import asyncio
from mcp import ClientSession
from mcp.client.sse import sse_client


async def main():
    # 连接到SSE服务器
    async with sse_client(url="http://localhost:8000/sse") as streams:
        async with ClientSession(*streams) as session:
            # 初始化会话
            await session.initialize()

            # 列出可用工具
            tools_response = await session.list_tools()
            print("Available tools:")
            for tool in tools_response.tools:
                print(f" - {tool.name}: {tool.description}")

            # 列出可用资源
            resources_response = await session.list_resources()
            print("\nAvailable resources:")
            for resource in resources_response.resources:
                print(f" - {resource.uri}: {resource.description}")

            # 调用天气工具
            print("\nCalling get_weather tool for 大连...")
            weather_response = await session.call_tool("get_weather", {"city": "大连"})
            print(weather_response.content[0].text)

            # 读取资源
            print("\nReading weather://cities resource...")
            cities_response = await session.read_resource("weather://cities")
            print(cities_response[0].content)

            # 读取带参数的资源
            print("\nReading weather forecast for Tokyo...")
            forecast_response = await session.read_resource("weather://forecast/大连")
            print(forecast_response[0].content)


if __name__ == "__main__":
    asyncio.run(main())