适用于 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 平台的一部分 —— 体育智能与结果概率评估平台
Recent tool additions, removals, and schema changes observed during successful MCP inspections. Dates show when Glama detected each change.
No tool schema history has been recorded yet.
This server cannot be installed
Unclaimed servers have limited discoverability.
If you are the server author, to access and configure the admin panel.
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
We provide all the information about MCP servers via our MCP API.
curl -X GET 'https://glama.ai/api/mcp/v1/servers/odirasamuel/sipap-serverlesshandler-mcp'
If you have feedback or need assistance with the MCP directory API, please join our Discord server
