MCP SSE服务器

一个使用服务器发送事件(SSE)传输的最小化MCP(模型上下文协议)服务器实现,可作为构建可部署MCP服务器的清晰参考。

  • by Amplify360
  • Python
  • 本地部署
  • 更新于
在 GitHub 查看 amplify360/mcp-demo-2

使用说明

MCP SSE服务器-参考实现

使用服务器发送事件(SSE)传输的最小MCP(模型上下文协议)服务器实现。与许多使用stdio传输的MCP服务器示例不同,这是一个可以通过Docker部署到任何托管平台的远程SSE服务器。这为构建可部署的MCP服务器提供了一个干净、易于理解的参考。

注: 此实现将清晰度和易理解性置于生产就绪功能之上。它不包括企业部署所需的全面生产方面,如健壮的错误处理、监控、安全强化、速率限制或可扩展性考虑。

特性

  • SSE 运输:远程MCP服务器部署的服务器发送事件
  • Docker就绪:容器化,便于部署到任何云平台
  • MCP集成
  • 清洁建筑:相关性最小,关注点明确分离
  • 可配置日志记录:控制台日志记录,可选日志级别覆盖
  • 基于环境的配置:用途 .env 安装文件

快速开始

1.环境设置

复制 .env.example 向 .env 并更新您的配置:

cp .env.example .env

然后编辑 .env 与你的价值观:

# Required
MCP_SERVER_AUTH_KEY=your-mcp-auth-key

# Optional
LOG_LEVEL=INFO
ENVIRONMENT=development

# Docker/Deployment Specific (optional)
FILE_LOGGING=true

2.安装

首先,安装uv(参见 Astral安装指南):

# Create virtual environment and install dependencies
uv venv

# Activate the virtual environment
# On macOS/Linux:
source .venv/bin/activate
# On Windows:
# .venv\Scripts\activate

uv sync  # Install dependencies into virtual environment

3.运行服务器

uv run python mcp_server.py

部署

Docker部署

# Build and run
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

自定义部署选项

部署脚本现在为 BASE_NAME 和 REGION_CODE 从你的 .env 默认情况下,文件。对于一次性部署,您可以覆盖以下内容:

# Use custom names from environment variables
export BASE_NAME=myclient-mcp REGION_CODE=eastus
./deploy.sh

# Or override via command line (takes precedence over .env)
./deploy.sh --base-name myclient-mcp --environment prod --region-code eastus

# Update code only (no infrastructure changes)
./deploy.sh --update

已创建Azure资源

部署将创建以下Azure资源 标准命名约定:

资源 名称模式 示例
资源组 rg-{service}-{env}-{地区} rg-mcp-se-dev-weu
容器应用程序 ca-{service}-{env}-{地区} 加拿大中央银行
容器应用环境 cae-{service}-{env}-{地区} cae mcp sse-dev-weu
日志分析工作区 日志-{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 包含必需变量的文件

实时部署

该服务当前部署在:

  • 统一资源定位符: https://ca-mcp-sse-development-weu.mangosea-a4cea9ef.westeurope.azurecontainerapps.io
  • 环境:发展(西欧)
  • 资源组:rg mcp sse dev weu

Azure管理

门户访问:

  1. 引导到 Azure门户
  2. 搜索资源组: rg-mcp-sse-dev-weu
  3. 查找容器应用程序: ca-mcp-sse-development-weu

CLI命令:

# Get container app details
az containerapp show --name ca-mcp-sse-development-weu --resource-group rg-mcp-sse-dev-weu

# View logs
az containerapp logs show --name ca-mcp-sse-development-weu --resource-group rg-mcp-sse-dev-weu

# Restart app
az containerapp restart --name ca-mcp-sse-development-weu --resource-group rg-mcp-sse-dev-weu

故障排除

常见问题:

  1. 缺少环境变量:确保 .env 文件已存在,其中包含所有必需的变量
  2. Azure CLI问题:验证登录 az account show
  3. 容器故障:检查Docker守护进程是否正在运行
  4. 运行时问题:在Azure日志分析中查看容器日志

健康检查:

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

与ngrok的地方发展

对于使用web客户端的本地开发,您可以使用ngrok公开您的本地服务器:

  1. 安装ngrok:https://ngrok.com/download
  2. 启动本地MCP服务器:
    uv run python mcp_server.py
    
  3. 在另一个终端中,暴露服务器:
    ngrok http 8080
    
  4. 使用提供的HTTPS URL(例如。, https://abc123.ngrok.io)在您的web客户端中
  5. 记得包括你的 X-API-Key 发出请求时的标头

使用AI Buddy进行测试

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

  1. 启动本地服务器:

    uv run python mcp_server.py
    
  2. 创建ngrok隧道:

    ngrok http 8080
    
  3. 配置AI Buddy MCP连接器:

    • 打开AI Buddy并创建新的MCP连接器
    • 将服务器URL设置为您的ngrok公共URL /sse 端点
    • 例子: https://abc123.ngrok.io/sse
    • 设置 X-API-Key 标题值与您的 MCP_SERVER_AUTH_KEY 从 .env
  4. 测试连接:

    • AI Buddy现在应该能够连接到您的本地MCP服务器
    • 请专家透露它可以访问哪些工具调用
    • 您应该看到通过ngrok和应用服务器(在控制台中)输出的请求
    • 专家应在其工具清单中包括可用的行动

项目结构

mcp-sse-server/
├── mcp_server.py           # Main entry point and core application logic
├── src/                    # Main source code
│   ├── __init__.py         # Package marker
│   ├── config.py           # Configuration management
│   ├── mcp_tools.py        # MCP server and tools registration
│   ├── utils/              # Utility modules
│   │   ├── __init__.py     # Package marker
│   │   └── __init__.py     # Package marker
│   └── actions/            # MCP action implementations
│       ├── __init__.py     # Package marker
│       └── status.py       # Example action with no dependencies
├── tests/                  # Test files
│   ├── test_config.py      # Configuration tests
│   ├── test_email_utils.py # Email utility tests
│   ├── test_mcp_tools.py   # MCP tools tests
│   └── test_*.py           # Other test files
├── deployment/             # Deployment files
│   ├── Dockerfile          # Container configuration
│   └── bicep/              # Azure Bicep templates and scripts
│       ├── deploy.sh       # Automated deployment script
│       └── main.bicep      # Azure resource definitions
├── logs/                   # Runtime logs (created automatically)
├── pyproject.toml          # Dependencies and project config
└── README.md              # This file

配置

环境变量

必修的:

  • MCP_SERVER_AUTH_KEY:MCP请求的身份验证密钥

可选:

  • LOG_LEVEL:日志记录级别(默认值:INFO)
  • ENVIRONMENT:环境名称(默认:开发)
  • FILE_LOGGING:启用文件日志记录(在Docker容器中使用)

发展

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

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

目录结构

src/actions/
├── __init__.py          # Package marker
└── status.py            # Server status functionality (no dependencies)

运作原理

该系统使用 依赖关系注册表 方法:

  1. 中央登记处:所有服务器依赖关系都在中声明 src/mcp_tools.py:

    DEPENDENCIES: dict[str, object] = {
        "postmark_api_key": api_key,
        "sender_email": from_email,
        # Add new dependencies here ↓
        # "weather_api_key": os.getenv("WEATHER_API_KEY"),
    }
    
  2. 基于签名的注入:只有出现在函数签名中的依赖项才会被注入,没有隐藏的行为。

添加新操作(<60秒)

第一步:写动作

创建 src/actions/my_feature.py:

"""
My feature action implementation.
"""

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,    # Only injected if you need it
    sender_email: str,        # Only injected if you need it
) -> Any:
    """
    Description of what this action does.
    
    Args:
        user_param1: User-provided parameter
        user_param2: Another user-provided parameter
        postmark_api_key: Postmark API key (injected)
        sender_email: Sender email (injected)
        
    Returns:
        Result of the action
    """
    logger.info("My feature action called")
    
    # Your implementation here
    result = f"Processed {user_param1} with value {user_param2}"
    
    logger.info("My feature action completed")
    return result

步骤2:添加新的依赖项(如果需要)

如果您的操作需要额外的服务(如天气API密钥),请将它们添加到 DEPENDENCIES 注册表在 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"),  # ← Add this
}

步骤3:重新启动服务器

就是这样!该操作会自动注册为 my_feature_tool.

行动示例

简单操作(无依赖关系):

async def status_action() -> dict:
    """Get server status - needs no external dependencies."""
    return {"status": "ok", "version": "1.0.0"}

仅使用用户参数的操作:

async def greet_user_action(name: str, greeting: str = "Hello") -> str:
    """Greet a user - no server dependencies needed."""
    return f"{greeting}, {name}!"

使用服务器依赖关系的操作:

async def some_action(
    message: str,
    some_api_key: str,  # Injected because it's in DEPENDENCIES
) -> str:
    return f"Processed '{message}'"

使用自定义依赖项的操作:

async def fetch_weather_action(
    city: str,
    weather_api_key: str,   # Must be added to DEPENDENCIES first
) -> dict:
    """Fetch weather data using external API."""
    # Use weather_api_key to call external service
    return {"city": city, "temperature": "22°C"}

功能要求

命名约定:

  • 函数名必须以结尾 _action (例如。, status_action)
  • 注册的MCP工具将通过以下方式命名: _action 随着 _tool

参数:

  • 用户参数:必须记录暴露于MCP客户的信息
  • 依赖性参数:必须与中的名称匹配 DEPENDENCIES 注册表
  • 键入提示:所有参数都需要
  • 不 **kwargs:依赖关系作为显式命名参数传递

返回值:

  • 可以返回任何可序列化的类型(str、dict、list等)
  • 返回值将被发送回MCP客户端

异步函数:

  • 必须是 async 功能使用 async def
  • 可以使用 await 用于I/O操作

自动发现过程

服务器启动时:

  1. 这 register_tools() 函数填充 DEPENDENCIES 注册表(如果您的操作需要)
  2. 它扫描 src/actions/ Python模块包
  3. 它查找以结尾的异步函数 _action
  4. 对于每个操作,它都会检查函数签名
  5. 它创建了一个包装器,只注入操作请求的依赖关系
  6. 它将包装器注册为MCP工具

测试

# Run all tests
uv run python -m pytest tests/ -v

# Test action registration specifically
uv run python -m pytest tests/test_mcp_tools.py::TestRegisterTools -v

# Test individual actions

依赖项

  • httpx:HTTP客户端
  • mcp[cli]:模型上下文协议实现
  • 小星星:web服务器的ASGI框架
  • 优维康:ASGI服务器
  • python dotenv:环境变量加载
  • 媒染剂设置:配置管理

许可证

该项目根据 MIT许可证.