MCP Server Notifications

带真实服务器通知的MCP服务器-安装指南

本指南演示了如何创建发送以下内容的MCP服务器 实际服务器通知 （不仅仅是工具响应）使用低级MCP SDK和具有会话访问的FastMCP。

🔍 关键发现

研究表明 FastMCP抽象会话访问 需要真正的服务器通知。要发送真实的服务器通知，您需要：

1. 使用低级 mcp.server.lowlevel.Server 用于完全会话控制
2. 直接在FastMCP中访问会话 通过 ctx.session.send_progress_notification()

📋 需求

• Python 3.10或更高版本
• mcp 包版本1.8.0+（包括流式HTTP支持）
• uv 包管理器（推荐）

🚀 快速开始

1.安装依赖项

# Install uv if you don't have it
curl -LsSf https://astral.sh/uv/install.sh | sh

# Create a new project
uv init mcp-notifications-demo
cd mcp-notifications-demo

# Install MCP with CLI support
uv add "mcp[cli]>=1.8.0" pydantic

2.选择你的方法

选项A：带会话访问的FastMCP（推荐）

将“带有真实服务器通知的FastMCP”代码另存为 fastmcp_server.py

选项B：低级服务器（高级）

将“具有真实服务器通知的低级MCP服务器”代码另存为 lowlevel_server.py

3.运行服务器

# FastMCP approach (Streamable HTTP)
python fastmcp_server.py

# Low-level approach (STDIO)
python lowlevel_server.py stdio

4.与客户进行测试

# Save the client code as notification_client.py

# Run notification demo
python notification_client.py

# Run interactive testing
python notification_client.py --interactive

🔑 研究的关键见解

是什么让真正的服务器通知工作

1. 会话访问：您需要直接访问MCP会话对象
2. 进度通知：使用 session.send_progress_notification() 用于服务器发起的消息
3. 流式HTTP：允许对SSE进行动态连接升级以进行流式传输

关键代码模式

# In FastMCP, access session through context
async def send_server_notification(ctx: Context, message: str):
    await ctx.session.send_progress_notification(
        progress_token=f"notification_{int(time.time())}",
        progress=1.0,
        total=1.0,
        message=f"[SERVER] {message}",
        related_request_id=ctx.request_id
    )

传输协议详细信息

流式HTTP 使能够：

• 单端点(/mcp)用于所有通信
• 向SSE进行流媒体动态连接升级
• 实时服务器通知
• 具有可选会话ID的会话管理

🛠️ 包括什么

服务器功能

真实服务器通知：

• send_notification() -发送具有优先级的通知
• start_background_task() -启动具有定期更新的任务
• long_running_operation() -带有进度通知的操作
• 自动会话跟踪和广播

通知类型：

• 即时通知 -作为对工具调用的响应发送
• 进度通知 -在长时间运行的操作中发送
• 背景通知 -从后台任务发送
• 广播通知 -发送到所有连接的会话

客户端功能

通知处理：

• 增强的进度通知处理程序
• 实时通知显示
• 进度跟踪和总结
• 交互式测试命令

📡 服务器通知示例

1.简单通知

await send_server_notification(ctx, "Hello from server!", "info")

2.进度更新

await ctx.session.send_progress_notification(
    progress_token="task_123",
    progress=3,
    total=10,
    message="Processing step 3 of 10"
)

3.背景广播

for session_id, session_data in active_sessions.items():
    await session_data["session"].send_progress_notification(
        progress_token="broadcast",
        progress=1.0,
        total=1.0,
        message="Server maintenance starting in 5 minutes"
    )

🧪 测试场景

演示序列

1. 简单通知 -基本服务器消息
2. 广播通知 -向所有会话发送消息
3. 长时间运行 -进度更新
4. 后台任务 -定期通知
5. 服务器状态 -当前状态信息

交互式命令

# In interactive mode:
notify Hello world
broadcast Important announcement
task BackgroundDemo 10
operation 5
status
summary

🔧 配置选项

FastMCP配置

# Stateful server (maintains sessions)
mcp = FastMCP("NotificationServer", stateless_http=False)

# Stateless server (for serverless deployments)  
mcp = FastMCP("NotificationServer", stateless_http=True)

# Run with Streamable HTTP
mcp.run(transport="streamable-http", host="localhost", port=8000)

低级服务器配置

# Create server with notification support
server = Server("NotificationServer")

# Run with different transports
# STDIO (local process)
python server.py stdio

# SSE (requires HTTP server setup)
python server.py sse

# Streamable HTTP (modern approach)
python server.py streamable-http

📚 体系结构比较

FastMCP与低级服务器

推荐

• 使用FastMCP 对于大多数应用程序（使用ctx.session访问更容易）
• 使用低级服务器 当您需要对通知进行最大控制时

🐛 故障排除

常见问题

未出现通知：

• 确保您正在使用 ctx.session.send_progress_notification() 不仅仅是返回值
• 检查服务器是否使用Streamable HTTP传输运行
• 验证客户端是否正确处理进度通知

会话跟踪问题：

• 使用有状态的FastMCP(stateless_http=False)
• 在请求处理程序中实现适当的会话管理
• 在全球范围内跟踪会议以进行广播

运输问题：

• 流式HTTP需要MCP SDK 1.8.0+
• SSE需要正确的HTTP服务器设置
• STDIO仅适用于本地进程通信

调试提示

# Add logging to see notifications being sent
print(f"📡 Sending notification: {message}")

# Check active sessions
print(f"Active sessions: {len(active_sessions)}")

# Monitor client notification handlers
print(f"📊 Notification #{self.notification_count} received")

🚀 后续步骤

1. 扩展通知：添加自定义通知类型和处理程序
2. 永久存储：将通知历史记录存储在数据库中
3. 认证：添加安全会话管理
4. 扩展：使用负载均衡器和多个实例进行部署
5. 监控：添加指标和健康检查

📖 额外资源

• MCP规范
• -进度通知修复
• MCP Python SDK示例

______________________________________________________________________

备注：此实现演示了真实的服务器通知，而不仅仅是工具调用响应。服务器通过Streamable HTTP协议中的SSE流主动向连接的客户端推送通知。