Framework de servidores MCP listo para producción en AWS Lambda y ECS Fargate
sipap-mcp proporciona clases base e infraestructura para construir servidores Model Context Protocol (MCP) que implementan JSON-RPC 2.0 y pueden ejecutarse en:
AWS Lambda: Funciones serverless para cargas de trabajo ligeras y esporádicas
ECS Fargate: Servicios contenedorizados para cargas de trabajo de larga duración y con estado
Este framework impulsa los 5 servidores de datos de la arquitectura de Valo (Plataforma de Inteligencia Deportiva), que gestionan datos deportivos, inteligencia de cuotas, contexto de noticias, datos meteorológicos y estadísticas históricas.
Related MCP server: mcp-server-toolkit
✅ Clase base MCPServer: Base abstracta con registro de herramientas y autodetección
✅ Decorador @mcp_tool: Marca funciones como herramientas MCP con validación de JSON Schema
✅ Protocolo JSON-RPC 2.0: Implementación completa con manejo de errores adecuado
✅ Transporte dual: Handler de Lambda y servidor HTTP FastAPI
✅ Autenticación: Estrategias conectables (NoAuth, API key, AWS SigV4)
✅ Gestión de sesiones: Preservación de estado entre llamadas respaldada por Redis
✅ Validación de entradas: Validación JSON Schema en todas las entradas de las herramientas
✅ Seguridad de tipos: Cumplimiento total del modo estricto de mypy (cero errores)
✅ Cobertura de pruebas: 96 % de cobertura con 112 pruebas superadas
✅ Listo para producción: Cero errores de linting, manejo de errores exhaustivo
pip install sipap-mcpPara desarrollo:
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)Despliega con AWS CDK o Terraform:
Handler: your_module.handler
Runtime: python3.12
Timeout: 30 segundos
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)Despliega con 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"]Las herramientas son funciones decoradas con @mcp_tool que se pueden invocar a través del protocolo 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}Tipos de JSON Schema compatibles:
string, number, integer, boolean, array, object
Validación: minLength, maxLength, minimum, maximum, pattern, enum
Elige la estrategia de autenticación que se adapte a tu despliegue:
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"
])Los clientes envían la API key en la cabecera X-API-Key.
from sipap_mcp.auth import SigV4Auth
auth = SigV4Auth(service="lambda", region="us-east-1")Para URLs de Lambda Function con autenticación IAM.
Mantén el estado entre múltiples solicitudes usando 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)Sobrescribe _setup() y _cleanup() para la gestión de recursos:
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()Úsalo con el gestor de contexto:
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": {}
}Respuesta:
{
"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"
}
}
}Respuesta:
{
"jsonrpc": "2.0",
"id": "req-2",
"result": {
"content": [{
"type": "text",
"text": "{\"location\": \"London\", \"temperature\": 15, ...}"
}]
}
}Códigos de error estándar de JSON-RPC 2.0:
Código | Significado | Cuándo |
-32700 | Error de análisis | JSON no válido |
-32600 | Solicitud no válida | Faltan campos obligatorios |
-32601 | Método no encontrado | Método desconocido |
-32602 | Parámetros no válidos | Falló la validación |
-32603 | Error interno | Error del servidor |
Respuesta de error:
{
"jsonrpc": "2.0",
"id": "req-3",
"error": {
"code": -32602,
"message": "Invalid params: 'location' is required"
}
}Consulta el directorio examples/ para ver ejemplos completos:
Ejemplo | Descripción |
Servidor simple de calculadora | |
Despliegue en Lambda con autenticación por API key | |
Servidor HTTP con sesiones Redis | |
Patrones avanzados y hooks del ciclo de vida | |
Todas las estrategias de autenticación |
Ejecuta los ejemplos:
python examples/01_basic_server.py
python examples/02_lambda_with_auth.py
python examples/03_http_with_sessions.py # Requires RedisEste framework adapta patrones probados de la arquitectura Sentinel:
Patrón ExitStack + Generator: Gestión de recursos con gestores de contexto
Autodetección de herramientas: Registro de herramientas basado en introspección
Aplicación de salida estructurada: Validación JSON Schema en todas las entradas/salidas
Registro basado en ContextVar: Propagación de contexto segura para hilos
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.htmlTodos los controles de calidad deben superarse antes de hacer commit:
# 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 o 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)Métodos:
handle_request(request_data) -> dict: Procesa una solicitud JSON-RPC
list_tools() -> list[dict]: Obtiene las herramientas registradas
get_info() -> dict: Obtiene los metadatos del servidor
_setup() -> None: Sobrescribir para la inicialización (opcional)
_cleanup() -> None: Sobrescribir para la limpieza (opcional)
@mcp_tool(description: str, input_schema: dict)
def tool_function(self, **kwargs) -> dict:
passParámetros:
description: Descripción de la herramienta legible para humanos
input_schema: JSON Schema para la validación de entradas
class SessionManager(redis_client, ttl: int = 3600)Métodos:
create_session(data, ttl=None) -> str: Crea una sesión y devuelve el ID
get_session(session_id) -> dict | None: Recupera los datos de la sesión
update_session(session_id, data, ttl=None) -> bool: Actualiza la sesión
delete_session(session_id) -> bool: Elimina la sesión
session_exists(session_id) -> bool: Comprueba si existe
extend_ttl(session_id, ttl) -> bool: Extiende la expiración
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"] == 200Configuración del handler:
# 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)Despliegue:
Handler: app.handler
Runtime: python3.12
Memoria: 512 MB (ajústala según la carga de trabajo)
Timeout: 30 segundos (ajústalo según el tiempo de ejecución de las herramientas)
Variables de entorno: 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)Definición de tarea:
Puerto del contenedor: 8000
Comprobación de salud: /health (si está implementada)
CPU: 256 (.25 vCPU)
Memoria: 512 MB
Desarrollo:
docker run -d -p 6379:6379 redis:7-alpineProducción:
AWS ElastiCache para Redis
Versión: Redis 7.x
Tipo de nodo: cache.t4g.micro (o superior)
Cifrado: en tránsito y en reposo
Multi-AZ: habilitado para producción
Error de importación:
# Problem
from sipap_mcp import MCPServer # ImportError
# Solution
pip install sipap-mcpFallo de autenticación:
# Check API key header name (must be X-API-Key)
headers = {"X-API-Key": "your-key"} # Correct
headers = {"Api-Key": "your-key"} # WrongSesión no encontrada:
# Sessions expire after TTL
session_manager.session_exists(session_id) # Check first
session_manager.extend_ttl(session_id, 3600) # Extend if neededErrores de tipo:
# Run mypy to catch type issues
mypy your_module.py --strictProbado en AWS Lambda (512 MB, Python 3.12):
Operación | Arranque en frío | Arranque en caliente |
tools/list | 850ms | 12ms |
tools/call (simple) | 900ms | 15ms |
tools/call (con BD) |
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
Unclaimed servers have limited discoverability.
If you are the server author, to access and configure the admin panel.
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
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
