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 デコレータ: JSON Schema 検証付きで関数を MCP ツールとしてマーク
✅ JSON-RPC 2.0 プロトコル: 適切なエラー処理を備えた完全な実装
✅ デュアルトランスポート: Lambda ハンドラーと FastAPI HTTP サーバー
✅ 認証: プラグ可能な戦略 (NoAuth、API キー、AWS SigV4)
✅ セッション管理: Redis による呼び出し間の状態保持
✅ 入力検証: すべてのツール入力に対する JSON Schema 検証
✅ 型安全性: mypy strict モード完全準拠 (エラーゼロ)
✅ テストカバレッジ: 112 件のテストが成功し、96% のカバレッジ
✅ 本番対応: リンティングエラーゼロ、包括的なエラー処理
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 関数 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 エラーコード:
コード | 意味 | 発生条件 |
-32700 | 解析エラー | 無効な JSON |
-32600 | 無効なリクエスト | 必須フィールドの欠落 |
-32601 | メソッドが見つからない | 不明なメソッド |
-32602 | 無効なパラメータ | 検証の失敗 |
-32603 | 内部エラー | サーバーエラー |
エラーレスポンス:
{
"jsonrpc": "2.0",
"id": "req-3",
"error": {
"code": -32602,
"message": "Invalid params: 'location' is required"
}
}包括的な例については examples/ ディレクトリを参照してください:
例 | 説明 |
シンプルな電卓サーバー | |
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 (.25 vCPU)
メモリ: 512 MB
開発:
docker run -d -p 6379:6379 redis:7-alpine本番:
AWS ElastiCache for Redis
バージョン: Redis 7.x
ノードタイプ: cache.t4g.micro (またはそれ以上)
暗号化: 転送中および保存時
Multi-AZ: 本番環境では有効化
インポートエラー:
# 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 --strictAWS Lambda (512 MB、Python 3.12) でテスト済み:
操作 | コールドスタート | ウォームスタート |
tools/list | 850ms | 12ms |
tools/call (simple) | 900ms | 15ms |
tools/call (with DB) | 1200ms | 45ms |
コールドスタートを減らす: Lambda のプロビジョニング済み同時実行を使用
接続をキャッシュする: _setup() で初期化し、呼び出し間で再利用
依存関係を最小限に: 必要なものだけをインポート
async を使用: FastAPI トランスポートは非同期ツールをサポート
セッション TTL: メモリ使用量とユーザーエクスペリエンスのバランスを取る
コントリビューションを歓迎します! 以下の点にご協力ください:
既存のコードスタイルに従う (ruff + mypy strict)
新機能のテストを追加する (カバレッジ 80% 以上を維持)
ドキュメントを更新する
提出前にすべての品質ゲートを実行する
Copyright © 2026 Valo Team
以下を使用して構築:
テスト駆動開発 (TDD)
型安全性 (mypy strict モード)
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
