MCP服务器发送事件服务

Name: MCP服务器发送事件服务
Author: Amplify360

作者：Amplify3602 星标更新：2025-08-22

安装使用来源：spider==github.com

项目介绍

MCP SSE 服务器 - 参考实现

这是一个使用服务器发送事件（SSE）传输的最小MCP（模型上下文协议）服务器实现，演示了电子邮件发送功能。与许多使用标准输入输出（stdio）传输的MCP服务器示例不同，这是一个远程SSE服务器，可以通过Docker部署到任何托管平台。这提供了一个干净且易于理解的参考，用于构建可部署的MCP服务器。

注意： 此实现优先考虑清晰度和易理解性，而不是生产就绪的功能。它不包括全面的生产方面，如健壮的错误处理、监控、安全加固、速率限制或可扩展性考虑，这些对于企业部署是必需的。

功能

SSE传输：使用服务器发送事件进行远程MCP服务器部署
Docker就绪：容器化以便轻松部署到任何云平台
简单邮件发送：通过Postmark SMTP发送电子邮件
MCP集成：将电子邮件功能作为MCP工具公开
简洁架构：最小依赖项和明确的关注点分离
可配置日志记录：控制台日志记录，可选的日志级别覆盖
基于环境的配置：使用.env文件进行设置

快速开始

1. 环境设置

复制.env.example到.env并更新您的配置：

cp .env.example .env

然后编辑.env并填入您的值：

# 必需
MCP_SERVER_AUTH_KEY=your-mcp-auth-key
POSTMARK_API_KEY=your-postmark-api-key
SENDER_EMAIL=your-sender@example.com

# 可选
LOG_LEVEL=INFO
ENVIRONMENT=development

# Docker/部署特定（可选）
FILE_LOGGING=true

2. 安装

首先，安装uv（参见Astral的安装指南）：

# 创建虚拟环境并安装依赖项
uv venv

# 激活虚拟环境
# 在macOS/Linux上：
source .venv/bin/activate
# 在Windows上：
# .venv\Scripts\activate

uv sync  # 将依赖项安装到虚拟环境中

3. 运行服务器

uv run python mcp_server.py

部署

Docker部署

# 构建并运行
docker build -f deployment/Dockerfile -t mcp-sse-server .
docker run -d --name mcp-sse-server -p 8080:8080 --env-file .env mcp-sse-server

Azure容器应用部署

快速部署

cd deployment/bicep
chmod +x deploy.sh
./deploy.sh

自定义部署选项

部署脚本现在默认从您的.env文件中读取BASE_NAME和REGION_CODE。对于一次性部署，您可以覆盖这些值：

# 使用来自环境变量的自定义名称
export BASE_NAME=myclient-mcp REGION_CODE=eastus
./deploy.sh

# 或者通过命令行覆盖（优先于.env）
./deploy.sh --base-name myclient-mcp --environment prod --region-code eastus

# 更新代码（无基础设施更改）
./deploy.sh --update

Azure资源创建

部署会根据标准命名约定创建以下Azure资源：

资源	名称模式	示例
资源组	rg-{service}-{env}-{region}	rg-mcp-sse-dev-weu
容器应用	ca-{service}-{env}-{region}	ca-mcp-sse-dev-weu
容器应用环境	cae-{service}-{env}-{region}	cae-mcp-sse-dev-weu
日志分析工作区	log-{service}-{env}-{region}	log-mcp-sse-dev-weu
容器注册表	cr{service}{env}{region}	crmcpssedevweu

配置：

CPU：0.5 vCPU，内存：1GB
固定缩放：1个副本（最小=1，最大=1）
启用HTTPS入口

先决条件

已安装并配置Azure CLI
已安装并运行Docker
包含所需变量的.env文件

实时部署

服务当前部署在：

URL：https://ca-mcp-sse-development-weu.mangosea-a4cea9ef.westeurope.azurecontainerapps.io
环境：Development（西欧）
资源组：rg-mcp-sse-dev-weu

Azure管理

门户访问：

导航至Azure门户
搜索资源组：rg-mcp-sse-dev-weu
查找容器应用：ca-mcp-sse-development-weu

CLI命令：

# 获取容器应用详情
az containerapp show --name ca-mcp-sse-development-weu --resource-group rg-mcp-sse-dev-weu

# 查看日志
az containerapp logs show --name ca-mcp-sse-development-weu --resource-group rg-mcp-sse-dev-weu

# 重启应用
az containerapp restart --name ca-mcp-sse-development-weu --resource-group rg-mcp-sse-dev-weu

故障排除

常见问题：

缺少环境变量：确保.env文件存在并包含所有必需的变量
Azure CLI问题：验证登录状态，使用az account show
容器故障：检查Docker守护进程是否正在运行
运行时问题：查看Azure日志分析中的容器日志

健康检查：

curl https://your-container-app-url.azurecontainerapps.io/health

本地开发使用ngrok

对于带有Web客户端的本地开发，您可以使用ngrok来暴露您的本地服务器：

安装ngrok：https://ngrok.com/download
启动您的本地MCP服务器：
```
uv run python mcp_server.py
```
在另一个终端中暴露服务器：
```
ngrok http 8080
```
在您的Web客户端中使用提供的HTTPS URL（例如，https://abc123.ngrok.io）
记得在请求时包含您的X-API-Key头

使用AI Buddy测试

您可以在AI Buddy中使用ngrok创建一个安全隧道来测试您的MCP SSE服务器：

启动您的本地服务器：
```
uv run python mcp_server.py
```
创建一个ngrok隧道：
```
ngrok http 8080
```
配置AI Buddy MCP连接器：
- 打开AI Buddy并创建一个新的MCP连接器
- 将服务器URL设置为您ngrok的公共URL，并加上/sse端点
- 示例：https://abc123.ngrok.io/sse
- 将X-API-Key头值设置为与.env中的MCP_SERVER_AUTH_KEY匹配
测试连接：
- AI Buddy现在应该能够连接到您的本地MCP服务器
- 请求专家揭示它拥有的工具调用
- 您应该看到通过ngrok和应用服务器（在控制台输出）的请求
- 专家应在其工具列表中包括电子邮件发送工具
- 您可以请求它发送一封测试邮件

项目结构

mcp-sse-server/
├── mcp_server.py           # 主入口点和核心应用程序逻辑
├── src/                    # 主源代码
│   ├── __init__.py         # 包标记
│   ├── config.py           # 配置管理
│   ├── mcp_tools.py        # MCP服务器和工具注册
│   ├── utils/              # 工具模块
│   │   ├── __init__.py     # 包标记
│   │   └── email.py        # 电子邮件工具（从email_utils.py移动）
│   └── actions/            # MCP动作实现
│       ├── __init__.py     # 包标记
│       └── send_email.py   # 发送电子邮件动作
├── tests/                  # 测试文件
│   ├── test_config.py      # 配置测试
│   ├── test_email_utils.py # 电子邮件工具测试
│   ├── test_mcp_tools.py   # MCP工具测试
│   └── test_*.py           # 其他测试文件
├── deployment/             # 部署文件
│   ├── Dockerfile          # 容器配置
│   └── bicep/              # Azure Bicep模板和脚本
│       ├── deploy.sh       # 自动部署脚本
│       └── main.bicep      # Azure资源定义
├── logs/                   # 运行时日志（自动创建）
├── pyproject.toml          # 依赖项和项目配置
└── README.md              # 本文件

配置

环境变量

必需：

MCP_SERVER_AUTH_KEY：MCP请求的身份验证密钥
POSTMARK_API_KEY：您的Postmark API密钥，用于发送电子邮件
SENDER_EMAIL：要从中发送的电子邮件地址

可选：

LOG_LEVEL：日志级别（默认：INFO）
ENVIRONMENT：环境名称（默认：development）
FILE_LOGGING：启用文件日志记录（用于Docker容器）

开发

动作系统 - 添加新的MCP工具

服务器使用透明的动作基础架构，其中每个MCP工具都作为一个单独的动作模块实现。依赖项在函数签名中显式声明，使得系统易于理解和扩展。

目录结构

src/actions/
├── __init__.py          # 包标记
├── send_email.py        # 发送电子邮件功能
└── status.py            # 服务器状态功能（无依赖项）

工作原理

系统采用依赖项注册表方法：

中央注册表：所有服务器依赖项都在src/mcp_tools.py中声明：

DEPENDENCIES: dict[str, object] = {
    "postmark_api_key": api_key,
    "sender_email": from_email,
    # 在此处添加新依赖项 ↓
    # "weather_api_key": os.getenv("WEATHER_API_KEY"),
}

签名注入：只有出现在函数签名中的依赖项才会被注入——没有隐藏行为。

添加新动作（小于60秒）

步骤1：编写动作

创建src/actions/my_feature.py：

"""
我的特性动作实现。
"""

import logging
from typing import Any

logger = logging.getLogger(__name__)


async def my_feature_action(
    user_param1: str,
    user_param2: int,
    postmark_api_key: str,    # 如果需要则注入
    sender_email: str,        # 如果需要则注入
) -> Any:
    """
    描述此动作的作用。
    
    参数：
        user_param1：用户提供的参数
        user_param2：另一个用户提供的参数
        postmark_api_key：Postmark API密钥（注入）
        sender_email：发件人电子邮件（注入）
        
    返回：
        动作的结果
    """
    logger.info("我的特性动作被调用")
    
    # 您的实现在此处
    result = f"处理了 {user_param1} 值为 {user_param2}"
    
    logger.info("我的特性动作完成")
    return result

步骤2：添加新依赖项（如果需要）

如果您的动作需要额外的服务（如天气API密钥），请将其添加到src/mcp_tools.py中的DEPENDENCIES注册表中：

DEPENDENCIES: dict[str, object] = {
    "postmark_api_key": api_key,
    "sender_email": from_email,
    "weather_api_key": os.getenv("WEATHER_API_KEY"),  # ← 添加此行
}

步骤3：重启服务器

就这样！动作将自动注册为my_feature_tool。

动作示例

简单动作（无依赖项）：

async def status_action() -> dict:
    """获取服务器状态 - 不需要外部依赖项。"""
    return {"status": "ok", "version": "1.0.0"}

仅使用用户参数的动作：

async def greet_user_action(name: str, greeting: str = "Hello") -> str:
    """问候用户 - 不需要服务器依赖项。"""
    return f"{greeting}, {name}!"

使用服务器依赖项的动作：

async def send_notification_action(
    message: str,
    recipient: str,
    postmark_api_key: str,  # 因为在DEPENDENCIES中所以注入
    sender_email: str,      # 因为在DEPENDENCIES中所以注入
) -> str:
    """使用服务器依赖项发送通知电子邮件。"""
    # 使用postmark_api_key和sender_email
    return f"已发送 '{message}' 给 {recipient}"

具有自定义依赖项的动作：

async def fetch_weather_action(
    city: str,
    weather_api_key: str,   # 必须先添加到DEPENDENCIES中
) -> dict:
    """使用外部API获取天气数据。"""
    # 使用weather_api_key调用外部服务
    return {"city": city, "temperature": "22°C"}

函数要求

命名约定：

函数名称必须以_action结尾（例如，send_email_action）
注册的MCP工具将通过替换_action为_tool命名

参数：

用户参数：对MCP客户端公开，必须有文档
依赖项参数：必须与DEPENDENCIES注册表中的名称匹配
类型提示：所有参数都需要类型提示
不允许**kwargs：依赖项作为显式的命名参数传递

返回值：

可以返回任何可序列化的类型（字符串、字典、列表等）
返回值将发送回MCP客户端

异步函数：

必须是一个使用async def的异步函数
可以使用await进行I/O操作

自动发现过程

当服务器启动时：

register_tools()函数填充DEPENDENCIES注册表
它扫描src/actions/包中的Python模块
它查找以_action结尾的异步函数
对于每个动作，它检查函数签名
它创建一个只注入动作请求的依赖项的包装器
它将包装器注册为MCP工具

测试

# 运行所有测试
uv run python -m pytest tests/ -v

# 特别测试动作注册
uv run python -m pytest tests/test_mcp_tools.py::TestRegisterTools -v

# 测试单个动作
uv run python -m pytest tests/test_send_email_action.py -v

依赖项

httpx：HTTP客户端
mcp[cli]：模型上下文协议实现
starlette：用于Web服务器的ASGI框架
uvicorn：ASGI服务器
python-dotenv：环境变量加载
pydantic-settings：配置管理

许可证

本项目根据MIT许可证授权。