Skip to main content
Glama

sipap-mcp

AWS Lambda および ECS Fargate 向けの本番対応 MCP サーバーフレームワーク

Python Version Type Checked Code Style Test Coverage Tests


概要

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 型:

  • stringnumberintegerbooleanarrayobject

  • 検証: minLengthmaxLengthminimummaximumpatternenum

認証

デプロイ環境に合った認証戦略を選択してください:

NoAuth (開発のみ)

from sipap_mcp.auth import NoAuth

auth = NoAuth()  # No authentication - use for local dev only

API キー認証

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 exit

JSON-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/ ディレクトリを参照してください:

説明

01_basic_server.py

シンプルな電卓サーバー

02_lambda_with_auth.py

API キー認証付き Lambda デプロイ

03_http_with_sessions.py

Redis セッション付き HTTP サーバー

04_advanced_server.py

高度なパターンとライフサイクルフック

05_authentication.py

すべての認証戦略

例の実行:

python examples/01_basic_server.py
python examples/02_lambda_with_auth.py
python examples/03_http_with_sessions.py  # Requires Redis

アーキテクチャ

設計パターン (Sentinel より)

このフレームワークは、Sentinel アーキテクチャの実証済みパターンを採用しています:

  1. ExitStack + ジェネレーターパターン: コンテキストマネージャーによるリソース管理

  2. ツールの自動検出: イントロスペクションに基づくツール登録

  3. 構造化出力の強制: すべての入力/出力に対する JSON Schema 検証

  4. 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

最適化のヒント

  1. コールドスタートを減らす: Lambda のプロビジョニング済み同時実行を使用

  2. 接続をキャッシュする: _setup() で初期化し、呼び出し間で再利用

  3. 依存関係を最小限に: 必要なものだけをインポート

  4. async を使用: FastAPI トランスポートは非同期ツールをサポート

  5. セッション TTL: メモリ使用量とユーザーエクスペリエンスのバランスを取る

コントリビューション

コントリビューションを歓迎します! 以下の点にご協力ください:

  1. 既存のコードスタイルに従う (ruff + mypy strict)

  2. 新機能のテストを追加する (カバレッジ 80% 以上を維持)

  3. ドキュメントを更新する

  4. 提出前にすべての品質ゲートを実行する

ライセンス

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.

Maintenance

ActivityMaintained
ResponsivenessNo issues

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

Related MCP Servers

  • A
    license
    A
    quality
    D
    maintenance
    A 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.
    1
    17
    MIT
  • A
    license
    Not graded
    quality
    D
    maintenance
    Production-ready MCP server starter with authentication, observability, and a plugin system for building and deploying MCP servers quickly.
    MIT

Latest Blog Posts

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