sipap-mcp
sipap-mcp
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]クイックスタート
1. MCP サーバーを定義する
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"
}2. AWS Lambda にデプロイする
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 秒
3. ECS Fargate にデプロイする (HTTP)
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
認証
デプロイ環境に合った認証戦略を選択してください:
NoAuth (開発のみ)
from sipap_mcp.auth import NoAuth
auth = NoAuth() # No authentication - use for local dev onlyAPI キー認証
from sipap_mcp.auth import APIKeyAuth
auth = APIKeyAuth(api_keys=[
"client-a-key",
"client-b-key",
"client-c-key"
])クライアントは X-API-Key ヘッダーで API キーを送信します。
AWS SigV4 認証
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 exitJSON-RPC 2.0 プロトコル
リクエスト形式
利用可能なツールの一覧表示
{
"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 より)
このフレームワークは、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.whl要件
ランタイム
Python 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
API リファレンス
MCPServer
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
@mcp_tool(description: str, input_schema: dict)
def tool_function(self, **kwargs) -> dict:
passパラメータ:
description: 人間が読めるツールの説明input_schema: 入力検証用の JSON Schema
SessionManager
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) -> FastAPIサーバーのテスト
ユニットテスト
def 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 response統合テスト
def 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本番デプロイ
AWS Lambda
ハンドラーのセットアップ:
# 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
ECS Fargate
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
セッション用 Redis
開発:
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 --strictパフォーマンス
ベンチマーク
AWS 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 プラットフォームの一部 - スポーツインテリジェンスおよび結果確率評価プラットフォーム
Tool Schema Changelog
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
Maintenance
Resources
Unclaimed servers have limited discoverability.
Looking for Admin?
If you are the server author, to access and configure the admin panel.
Related MCP Connectors
MCP server for progressive tool usage at any scale (see https://klavis.ai)
MCP server for Superserve sandboxes: create, exec, and manage Firecracker microVMs
- SupabaseOAuthcom.supabase
MCP server for interacting with the Supabase platform
An MCP server that let you interact with Cycloid.io Internal Development Portal and Platform
Related MCP Servers
- AlicenseAqualityDmaintenanceA simple MCP server that provides a basic greeting tool and serves as a starter template for AWS Lambda deployment. Demonstrates how to build and deploy MCP servers with both local development and cloud deployment capabilities.117MIT
- AlicenseNot gradedqualityDmaintenanceProduction-ready MCP server starter with authentication, observability, and a plugin system for building and deploying MCP servers quickly.MIT
- AlicenseNot gradedqualityDmaintenanceA minimal, production-ready MCP server running on AWS Lambda with Streamable HTTP transport, enabling deployment of custom tools behind API Gateway.1MIT
- FlicenseNot gradedqualityDmaintenanceA minimal MCP server deployed on AWS Lambda and API Gateway using AWS CDK, enabling tool execution via JSON-RPC (e.g., an add tool).3-
Latest Blog Posts
- Who's Calling? MCP Hosts Are an Identity Blind Spot (And the Spec Knows It)By Om-Shree-0709 on .mcpAgent IdentityOAuth 2.1
- Your AI Chatbot Just Exposed Your CEO's Salary to an InternBy Om-Shree-0709 on .Agent IdentityMCP SecurityOAuth Delegation
- Why MCP Servers Need Execution Sandboxing (And Why Your Current Stack Isn't Enough)By Om-Shree-0709 on .Agentic AiPrompt InjectionWebAssembly
MCP directory API
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