适用于 AWS Lambda 和 ECS Fargate 的生产就绪型 MCP 服务器框架
sipap-mcp 提供了用于构建实现 JSON-RPC 2.0 的 Model Context Protocol（MCP）服务器的基础类和基础设施，可运行于：
AWS Lambda：用于轻量级、零散工作负载的无服务器函数
ECS Fargate：用于长时间运行、有状态工作负载的容器化服务
该框架驱动 Valo（体育智能平台）架构中的全部 5 个数据服务器，处理体育数据、赔率情报、新闻上下文、天气数据和历史统计数据。
Related MCP server: mcp-server-toolkit
✅ MCPServer 基类：提供工具注册和自动发现功能的抽象基类
✅ @mcp_tool 装饰器：将函数标记为 MCP 工具，并带有 JSON Schema 验证
✅ JSON-RPC 2.0 协议：完整实现，具备完善的错误处理
✅ 双传输方式：Lambda 处理器和 FastAPI HTTP 服务器
✅ 身份验证：可插拔策略（NoAuth、API 密钥、AWS SigV4）
✅ 会话管理：基于 Redis 的调用间状态保持
✅ 输入验证：对所有工具输入进行 JSON Schema 验证
✅ 类型安全：完全符合 mypy 严格模式（零错误）
✅ 测试覆盖率：96% 覆盖率，112 个测试全部通过
✅ 生产就绪：零 lint 错误，全面的错误处理
pip install sipap-mcp用于开发环境：
pip install sipap-mcp[dev]from sipap_mcp import MCPServer, mcp_tool
class WeatherMCP(MCPServer):
"""Weather data MCP server."""
def __init__(self):
super().__init__(name="weather-mcp", version="1.0.0")
@mcp_tool(
description="Get current weather for a location",
input_schema={
"type": "object",
"properties": {
"location": {"type": "string", "description": "City name"},
"units": {
"type": "string",
"enum": ["celsius", "fahrenheit"],
"default": "celsius"
}
},
"required": ["location"]
}
)
def get_weather(self, location: str, units: str = "celsius") -> dict:
"""Get current weather conditions."""
# Your implementation here
return {
"location": location,
"temperature": 22 if units == "celsius" else 72,
"units": units,
"condition": "partly cloudy"
}from sipap_mcp.transport import create_lambda_handler
from sipap_mcp.auth import APIKeyAuth
# Create server instance
server = WeatherMCP()
# Configure authentication
auth = APIKeyAuth(api_keys=["your-api-key"])
# Create Lambda handler (entry point for AWS)
handler = create_lambda_handler(server, auth=auth)使用 AWS CDK 或 Terraform 部署：
处理器：your_module.handler
运行时：python3.12
超时时间：30 秒
from sipap_mcp.transport import create_http_app
import uvicorn
# Create server instance
server = WeatherMCP()
# Create FastAPI app
app = create_http_app(server, auth=auth)
# Run with uvicorn
if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=8000)使用 Docker 部署：
FROM python:3.12-slim
WORKDIR /app
COPY . .
RUN pip install sipap-mcp
CMD ["uvicorn", "your_module:app", "--host", "0.0.0.0", "--port", "8000"]工具是使用 @mcp_tool 装饰的函数，可通过 MCP 协议进行调用：
@mcp_tool(
description="Description of what this tool does",
input_schema={
"type": "object",
"properties": {
"param": {"type": "string"}
},
"required": ["param"]
}
)
def my_tool(self, param: str) -> dict:
"""Docstring for the tool."""
return {"result": param}支持的 JSON Schema 类型：
string、number、integer、boolean、array、object
验证：minLength、maxLength、minimum、maximum、pattern、enum
选择适合您部署场景的身份验证策略：
from sipap_mcp.auth import NoAuth
auth = NoAuth() # No authentication - use for local dev onlyfrom sipap_mcp.auth import APIKeyAuth
auth = APIKeyAuth(api_keys=[
"client-a-key",
"client-b-key",
"client-c-key"
])客户端在 X-API-Key 请求头中发送 API 密钥。
from sipap_mcp.auth import SigV4Auth
auth = SigV4Auth(service="lambda", region="us-east-1")适用于启用了 IAM 身份验证的 Lambda Function URL。
使用 Redis 在多个请求之间保持状态：
import redis
from sipap_mcp.session import SessionManager
# Connect to Redis
redis_client = redis.Redis(host="localhost", port=6379)
# Create session manager
session_manager = SessionManager(
redis_client=redis_client,
ttl=3600 # 1 hour default
)
# Create session
session_id = session_manager.create_session(
data={"user_id": "123", "preferences": {...}},
ttl=1800 # 30 minutes custom TTL
)
# Retrieve session
session_data = session_manager.get_session(session_id)
# Update session
session_manager.update_session(session_id, updated_data)
# Extend TTL
session_manager.extend_ttl(session_id, ttl=7200)重写 _setup() 和 _cleanup() 以进行资源管理：
class MyServer(MCPServer):
def __init__(self):
super().__init__(name="my-server", version="1.0.0")
self.db_connection = None
def _setup(self) -> None:
"""Called when entering context manager."""
self.db_connection = connect_to_database()
def _cleanup(self) -> None:
"""Called when exiting context manager."""
if self.db_connection:
self.db_connection.close()与上下文管理器一起使用：
with server:
# Server is set up, resources initialized
response = server.handle_request(request)
# Cleanup happens automatically on exit{
"jsonrpc": "2.0",
"id": "req-1",
"method": "tools/list",
"params": {}
}响应：
{
"jsonrpc": "2.0",
"id": "req-1",
"result": {
"tools": [
{
"name": "get_weather",
"description": "Get current weather for a location",
"inputSchema": {
"type": "object",
"properties": {...},
"required": [...]
}
}
]
}
}{
"jsonrpc": "2.0",
"id": "req-2",
"method": "tools/call",
"params": {
"name": "get_weather",
"arguments": {
"location": "London",
"units": "celsius"
}
}
}响应：
{
"jsonrpc": "2.0",
"id": "req-2",
"result": {
"content": [{
"type": "text",
"text": "{\"location\": \"London\", \"temperature\": 15, ...}"
}]
}
}标准 JSON-RPC 2.0 错误码：
Code | 含义 | 触发时机 |
-32700 | 解析错误 | JSON 无效 |
-32600 | 无效请求 | 缺少必填字段 |
-32601 | 方法未找到 | 未知方法 |
-32602 | 参数无效 | 验证失败 |
-32603 | 内部错误 | 服务器错误 |
错误响应：
{
"jsonrpc": "2.0",
"id": "req-3",
"error": {
"code": -32602,
"message": "Invalid params: 'location' is required"
}
}查看 examples/ 目录获取完整示例：
Example | 描述 |
简单的计算器服务器 | |
带 API 密钥身份验证的 Lambda 部署 | |
带 Redis 会话的 HTTP 服务器 | |
高级模式与生命周期钩子 | |
所有身份验证策略 |
运行示例：
python examples/01_basic_server.py
python examples/02_lambda_with_auth.py
python examples/03_http_with_sessions.py # Requires Redis该框架借鉴了 Sentinel 架构中经过验证的模式：
ExitStack + 生成器模式：使用上下文管理器进行资源管理
工具自动发现：基于内省的工具注册
结构化输出强制：对所有输入/输出进行 JSON Schema 验证
基于 ContextVar 的日志记录：线程安全的上下文传播
sipap_mcp/
├── core/
│ ├── protocol.py # JSON-RPC 2.0 implementation
│ └── server.py # MCPServer base class
├── decorators/
│ └── tool.py # @mcp_tool decorator & registry
├── transport/
│ ├── lambda_handler.py # AWS Lambda adapter
│ └── http_handler.py # FastAPI adapter
├── auth/
│ └── middleware.py # Authentication strategies
├── session/
│ └── manager.py # Redis session management
└── validation/
└── schema.py # JSON Schema validation# Clone repository
git clone <repo-url>
cd sipap-mcp
# Create virtual environment
python3.12 -m venv .venv
source .venv/bin/activate
# Install in editable mode with dev dependencies
pip install -e ".[dev]"# Run all tests
pytest
# Run with coverage
pytest --cov=src/sipap_mcp --cov-report=html
# Open coverage report
open htmlcov/index.html提交前必须通过所有质量门禁：
# Type checking (strict mode)
mypy src/sipap_mcp --strict
# Linting
ruff check src/sipap_mcp tests/
# Auto-fix linting errors
ruff check --fix src/sipap_mcp tests/
# All gates at once
pytest && mypy src/sipap_mcp --strict && ruff check src/sipap_mcp tests/# Build wheel and source distribution
python -m build
# Install built package
pip install dist/sipap_mcp-0.1.0-py3-none-any.whlPython 3.12、3.13 或 3.14
pydantic >= 2.7.0
fastapi >= 0.111.0
uvicorn[standard] >= 0.30.0
jsonschema >= 4.22.0
sipap-common >= 0.1.0
typing-extensions >= 4.12.0
pytest >= 8.0.0
pytest-cov >= 5.0.0
mypy >= 1.10.0
ruff >= 0.4.0
class MCPServer(name: str, version: str)方法：
handle_request(request_data) -> dict：处理 JSON-RPC 请求
list_tools() -> list[dict]：获取已注册的工具
get_info() -> dict：获取服务器元数据
_setup() -> None：重写以进行初始化（可选）
_cleanup() -> None：重写以进行清理（可选）
@mcp_tool(description: str, input_schema: dict)
def tool_function(self, **kwargs) -> dict:
pass参数：
description：人类可读的工具描述
input_schema：用于输入验证的 JSON Schema
class SessionManager(redis_client, ttl: int = 3600)方法：
create_session(data, ttl=None) -> str：创建会话，返回 ID
get_session(session_id) -> dict | None：获取会话数据
update_session(session_id, data, ttl=None) -> bool：更新会话
delete_session(session_id) -> bool：删除会话
session_exists(session_id) -> bool：检查会话是否存在
extend_ttl(session_id, ttl) -> bool：延长过期时间
create_lambda_handler(server, auth=None) -> Callable
create_http_app(server, auth=None) -> FastAPIdef test_my_server():
server = MyServer()
# Test tool listing
tools = server.list_tools()
assert len(tools) > 0
# Test tool execution
request = {
"jsonrpc": "2.0",
"id": "1",
"method": "tools/call",
"params": {
"name": "my_tool",
"arguments": {"param": "value"}
}
}
with server:
response = server.handle_request(request)
assert "result" in responsedef test_lambda_handler():
from sipap_mcp.transport import create_lambda_handler
server = MyServer()
handler = create_lambda_handler(server)
event = {
"headers": {},
"body": json.dumps({
"jsonrpc": "2.0",
"id": "1",
"method": "tools/list",
"params": {}
})
}
response = handler(event, {})
assert response["statusCode"] == 200处理器设置：
# app.py
from sipap_mcp import MCPServer, mcp_tool
from sipap_mcp.transport import create_lambda_handler
from sipap_mcp.auth import APIKeyAuth
import os
class MyServer(MCPServer):
# ... server definition ...
server = MyServer()
auth = APIKeyAuth(api_keys=os.getenv("API_KEYS", "").split(","))
handler = create_lambda_handler(server, auth=auth)部署：
处理器：app.handler
运行时：python3.12
内存：512 MB（根据工作负载调整）
超时时间：30 秒（根据工具执行时间调整）
环境变量：API_KEYS=key1,key2,key3
Dockerfile：
FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8000"]app.py：
from sipap_mcp.transport import create_http_app
# ... server definition ...
app = create_http_app(server, auth=auth)任务定义：
容器端口：8000
健康检查：/health（如果已实现）
CPU：256（0.25 vCPU）
内存：512 MB
开发环境：
docker run -d -p 6379:6379 redis:7-alpine生产环境：
AWS ElastiCache for Redis
版本：Redis 7.x
节点类型：cache.t4g.micro（或更大）
加密：传输中加密和静态加密
多可用区：生产环境已启用
导入错误：
# Problem
from sipap_mcp import MCPServer # ImportError
# Solution
pip install sipap-mcp身份验证失败：
# Check API key header name (must be X-API-Key)
headers = {"X-API-Key": "your-key"} # Correct
headers = {"Api-Key": "your-key"} # Wrong找不到会话：
# Sessions expire after TTL
session_manager.session_exists(session_id) # Check first
session_manager.extend_ttl(session_id, 3600) # Extend if needed类型错误：
# Run mypy to catch type issues
mypy your_module.py --strict在 AWS Lambda（512 MB，Python 3.12）上测试：
操作 | 冷启动 | 热启动 |
tools/list | 850ms | 12ms |
tools/call（简单） | 900ms | 15ms |
tools/call（带数据库） | 1200ms | 45ms |
减少冷启动：使用 Lambda 预置并发
缓存连接：在 _setup() 中初始化，跨调用复用
减少依赖：只导入您需要的内容
使用异步：FastAPI 传输方式支持异步工具
会话 TTL：在内存使用与用户体验之间取得平衡
欢迎贡献！请遵循以下要求：
遵循现有代码风格（ruff + mypy 严格模式）
为新功能添加测试（保持 80% 以上覆盖率）
更新文档
提交前运行所有质量门禁
版权所有 © 2026 Valo Team
构建方式：
测试驱动开发（TDD）
类型安全（mypy 严格模式）
96% 测试覆盖率（112 个测试）
生产就绪的错误处理
全面的文档
Valo 平台的一部分 —— 体育智能与结果概率评估平台
This server cannot be deployed
MCP server for progressive tool usage at any scale (see https://klavis.ai)
MCP server for Superserve sandboxes: create, exec, and manage Firecracker microVMs
MCP server for interacting with the Supabase platform
An MCP server that let you interact with Cycloid.io Internal Development Portal and Platform
