Skip to main content
Glama

sipap-mcp

Production-ready MCP server framework for AWS Lambda and ECS Fargate

Python Version Type Checked Code Style Test Coverage Tests


Overview

sipap-mcp provides base classes and infrastructure for building Model Context Protocol (MCP) servers that implement JSON-RPC 2.0 and can run on:

  • AWS Lambda: Serverless functions for lightweight, sporadic workloads

  • ECS Fargate: Containerized services for long-running, stateful workloads

This framework powers all 5 data servers in the SIPAP (Sports Intelligence Platform) architecture, handling sports data, odds intelligence, news context, weather data, and historical statistics.

Related MCP server: Hello MCP Server

Features

Core Functionality

  • MCPServer Base Class: Abstract base with tool registration and auto-discovery

  • @mcp_tool Decorator: Mark functions as MCP tools with JSON Schema validation

  • JSON-RPC 2.0 Protocol: Complete implementation with proper error handling

  • Dual Transport: Lambda handler and FastAPI HTTP server

Security & State

  • Authentication: Pluggable strategies (NoAuth, API key, AWS SigV4)

  • Session Management: Redis-backed state preservation between calls

  • Input Validation: JSON Schema validation on all tool inputs

Quality

  • Type Safety: Full mypy strict mode compliance (zero errors)

  • Test Coverage: 96% coverage with 112 passing tests

  • Production Ready: Zero linting errors, comprehensive error handling

Installation

pip install sipap-mcp

For development:

pip install sipap-mcp[dev]

Quick Start

1. Define an MCP Server

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. Deploy to 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)

Deploy with AWS CDK or Terraform:

  • Handler: your_module.handler

  • Runtime: python3.12

  • Timeout: 30 seconds

3. Deploy to 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)

Deploy with 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"]

Core Concepts

Tools

Tools are functions decorated with @mcp_tool that become callable via the MCP protocol:

@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 types supported:

  • string, number, integer, boolean, array, object

  • Validation: minLength, maxLength, minimum, maximum, pattern, enum

Authentication

Choose the authentication strategy that fits your deployment:

NoAuth (Development Only)

from sipap_mcp.auth import NoAuth

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

API Key Authentication

from sipap_mcp.auth import APIKeyAuth

auth = APIKeyAuth(api_keys=[
    "client-a-key",
    "client-b-key",
    "client-c-key"
])

Clients send API key in X-API-Key header.

AWS SigV4 Authentication

from sipap_mcp.auth import SigV4Auth

auth = SigV4Auth(service="lambda", region="us-east-1")

For Lambda Function URLs with IAM authentication.

Session Management

Maintain state across multiple requests using 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)

Lifecycle Hooks

Override _setup() and _cleanup() for resource management:

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()

Use with context manager:

with server:
    # Server is set up, resources initialized
    response = server.handle_request(request)
    # Cleanup happens automatically on exit

JSON-RPC 2.0 Protocol

Request Format

List Available Tools

{
  "jsonrpc": "2.0",
  "id": "req-1",
  "method": "tools/list",
  "params": {}
}

Response:

{
  "jsonrpc": "2.0",
  "id": "req-1",
  "result": {
    "tools": [
      {
        "name": "get_weather",
        "description": "Get current weather for a location",
        "inputSchema": {
          "type": "object",
          "properties": {...},
          "required": [...]
        }
      }
    ]
  }
}

Call a Tool

{
  "jsonrpc": "2.0",
  "id": "req-2",
  "method": "tools/call",
  "params": {
    "name": "get_weather",
    "arguments": {
      "location": "London",
      "units": "celsius"
    }
  }
}

Response:

{
  "jsonrpc": "2.0",
  "id": "req-2",
  "result": {
    "content": [{
      "type": "text",
      "text": "{\"location\": \"London\", \"temperature\": 15, ...}"
    }]
  }
}

Error Handling

Standard JSON-RPC 2.0 error codes:

Code

Meaning

When

-32700

Parse error

Invalid JSON

-32600

Invalid Request

Missing required fields

-32601

Method not found

Unknown method

-32602

Invalid params

Validation failed

-32603

Internal error

Server error

Error Response:

{
  "jsonrpc": "2.0",
  "id": "req-3",
  "error": {
    "code": -32602,
    "message": "Invalid params: 'location' is required"
  }
}

Examples

See the examples/ directory for comprehensive examples:

Example

Description

01_basic_server.py

Simple calculator server

02_lambda_with_auth.py

Lambda deployment with API key auth

03_http_with_sessions.py

HTTP server with Redis sessions

04_advanced_server.py

Advanced patterns & lifecycle hooks

05_authentication.py

All authentication strategies

Run examples:

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

Architecture

Design Patterns (from Sentinel)

This framework adapts proven patterns from the Sentinel architecture:

  1. ExitStack + Generator Pattern: Resource management with context managers

  2. Tool Auto-Discovery: Introspection-based tool registration

  3. Structured Output Enforcement: JSON Schema validation on all inputs/outputs

  4. ContextVar-Based Logging: Thread-safe context propagation

Module Structure

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

Development

Setup

# 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]"

Running Tests

# Run all tests
pytest

# Run with coverage
pytest --cov=src/sipap_mcp --cov-report=html

# Open coverage report
open htmlcov/index.html

Quality Gates

All quality gates must pass before committing:

# 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/

Building

# Build wheel and source distribution
python -m build

# Install built package
pip install dist/sipap_mcp-0.1.0-py3-none-any.whl

Requirements

Runtime

  • Python 3.12, 3.13, or 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

Development

  • pytest >= 8.0.0

  • pytest-cov >= 5.0.0

  • mypy >= 1.10.0

  • ruff >= 0.4.0

API Reference

MCPServer

class MCPServer(name: str, version: str)

Methods:

  • handle_request(request_data) -> dict: Process JSON-RPC request

  • list_tools() -> list[dict]: Get registered tools

  • get_info() -> dict: Get server metadata

  • _setup() -> None: Override for initialization (optional)

  • _cleanup() -> None: Override for cleanup (optional)

@mcp_tool

@mcp_tool(description: str, input_schema: dict)
def tool_function(self, **kwargs) -> dict:
    pass

Parameters:

  • description: Human-readable tool description

  • input_schema: JSON Schema for input validation

SessionManager

class SessionManager(redis_client, ttl: int = 3600)

Methods:

  • create_session(data, ttl=None) -> str: Create session, returns ID

  • get_session(session_id) -> dict | None: Retrieve session data

  • update_session(session_id, data, ttl=None) -> bool: Update session

  • delete_session(session_id) -> bool: Delete session

  • session_exists(session_id) -> bool: Check if exists

  • extend_ttl(session_id, ttl) -> bool: Extend expiration

Transport Functions

create_lambda_handler(server, auth=None) -> Callable
create_http_app(server, auth=None) -> FastAPI

Testing Your Server

Unit Tests

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

Integration Tests

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

Production Deployment

AWS Lambda

Handler setup:

# 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)

Deploy:

  • Handler: app.handler

  • Runtime: python3.12

  • Memory: 512 MB (adjust based on workload)

  • Timeout: 30 seconds (adjust based on tool execution time)

  • Environment: 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)

Task definition:

  • Container port: 8000

  • Health check: /health (if implemented)

  • CPU: 256 (.25 vCPU)

  • Memory: 512 MB

Redis for Sessions

Development:

docker run -d -p 6379:6379 redis:7-alpine

Production:

  • AWS ElastiCache for Redis

  • Version: Redis 7.x

  • Node type: cache.t4g.micro (or larger)

  • Encryption: In-transit and at-rest

  • Multi-AZ: Enabled for production

Troubleshooting

Common Issues

Import Error:

# Problem
from sipap_mcp import MCPServer  # ImportError

# Solution
pip install sipap-mcp

Authentication Failing:

# Check API key header name (must be X-API-Key)
headers = {"X-API-Key": "your-key"}  # Correct
headers = {"Api-Key": "your-key"}    # Wrong

Session Not Found:

# Sessions expire after TTL
session_manager.session_exists(session_id)  # Check first
session_manager.extend_ttl(session_id, 3600)  # Extend if needed

Type Errors:

# Run mypy to catch type issues
mypy your_module.py --strict

Performance

Benchmarks

Tested on AWS Lambda (512 MB, Python 3.12):

Operation

Cold Start

Warm Start

tools/list

850ms

12ms

tools/call (simple)

900ms

15ms

tools/call (with DB)

1200ms

45ms

Optimization Tips

  1. Reduce cold starts: Use Lambda provisioned concurrency

  2. Cache connections: Initialize in _setup(), reuse across invocations

  3. Minimize dependencies: Import only what you need

  4. Use async: FastAPI transport supports async tools

  5. Session TTL: Balance memory usage vs user experience

Contributing

Contributions are welcome! Please:

  1. Follow the existing code style (ruff + mypy strict)

  2. Add tests for new features (maintain 80%+ coverage)

  3. Update documentation

  4. Run all quality gates before submitting

License

Copyright © 2026 SIPAP Team


Built with:

  • Test-Driven Development (TDD)

  • Type safety (mypy strict mode)

  • 96% test coverage (112 tests)

  • Production-ready error handling

  • Comprehensive documentation

Part of the SIPAP platform - Sports Intelligence and Outcome Probability Assessment Platform

F
license - not found
-
quality - not tested
C
maintenance

Maintenance

Maintainers
Response time
Release cycle
Releases (12mo)
Commit activity

Resources

Unclaimed servers have limited discoverability.

Looking for Admin?

If you are the server author, to access and configure the admin panel.

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