sipap-mcp
Click on "Install Server".
Wait a few minutes for the server to deploy. Once ready, it will show a "Started" state.
In the chat, type
@followed by the MCP server name and your instructions, e.g., "@sipap-mcplist all deployed MCP servers"
That's it! The server will respond to your query, and you can continue using it as needed.
Here is a step-by-step guide with screenshots.
sipap-mcp
Production-ready MCP server framework for AWS Lambda and ECS Fargate
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-mcpFor 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.handlerRuntime:
python3.12Timeout: 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,objectValidation:
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 onlyAPI 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 exitJSON-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 |
Simple calculator server | |
Lambda deployment with API key auth | |
HTTP server with Redis sessions | |
Advanced patterns & lifecycle hooks | |
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 RedisArchitecture
Design Patterns (from Sentinel)
This framework adapts proven patterns from the Sentinel architecture:
ExitStack + Generator Pattern: Resource management with context managers
Tool Auto-Discovery: Introspection-based tool registration
Structured Output Enforcement: JSON Schema validation on all inputs/outputs
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 validationDevelopment
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.htmlQuality 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.whlRequirements
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 requestlist_tools() -> list[dict]: Get registered toolsget_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:
passParameters:
description: Human-readable tool descriptioninput_schema: JSON Schema for input validation
SessionManager
class SessionManager(redis_client, ttl: int = 3600)Methods:
create_session(data, ttl=None) -> str: Create session, returns IDget_session(session_id) -> dict | None: Retrieve session dataupdate_session(session_id, data, ttl=None) -> bool: Update sessiondelete_session(session_id) -> bool: Delete sessionsession_exists(session_id) -> bool: Check if existsextend_ttl(session_id, ttl) -> bool: Extend expiration
Transport Functions
create_lambda_handler(server, auth=None) -> Callable
create_http_app(server, auth=None) -> FastAPITesting 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 responseIntegration 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"] == 200Production 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.handlerRuntime:
python3.12Memory: 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-alpineProduction:
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-mcpAuthentication Failing:
# Check API key header name (must be X-API-Key)
headers = {"X-API-Key": "your-key"} # Correct
headers = {"Api-Key": "your-key"} # WrongSession Not Found:
# Sessions expire after TTL
session_manager.session_exists(session_id) # Check first
session_manager.extend_ttl(session_id, 3600) # Extend if neededType Errors:
# Run mypy to catch type issues
mypy your_module.py --strictPerformance
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
Reduce cold starts: Use Lambda provisioned concurrency
Cache connections: Initialize in
_setup(), reuse across invocationsMinimize dependencies: Import only what you need
Use async: FastAPI transport supports async tools
Session TTL: Balance memory usage vs user experience
Contributing
Contributions are welcome! Please:
Follow the existing code style (ruff + mypy strict)
Add tests for new features (maintain 80%+ coverage)
Update documentation
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
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.
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