sipap-mcp
sipap-mcp
Готовый к продакшену фреймворк MCP-серверов для AWS Lambda и ECS Fargate
Обзор
sipap-mcp предоставляет базовые классы и инфраструктуру для создания серверов Model Context Protocol (MCP), реализующих JSON-RPC 2.0 и способных работать на:
AWS Lambda: Бессерверные функции для лёгких, нерегулярных нагрузок
ECS Fargate: Контейнерные сервисы для длительных, сохраняющих состояние нагрузок
Этот фреймворк обеспечивает работу всех 5 серверов данных в архитектуре Valo (Sports Intelligence Platform), обрабатывая спортивные данные, аналитику коэффициентов, новостной контекст, данные о погоде и историческую статистику.
Related MCP server: mcp-server-toolkit
Возможности
Основная функциональность
✅ Базовый класс MCPServer: Абстрактный базовый класс с регистрацией инструментов и автоматическим обнаружением
✅ Декоратор @mcp_tool: Помечает функции как MCP-инструменты с валидацией JSON Schema
✅ Протокол JSON-RPC 2.0: Полная реализация с корректной обработкой ошибок
✅ Двойной транспорт: Обработчик Lambda и HTTP-сервер FastAPI
Безопасность и состояние
✅ Аутентификация: Подключаемые стратегии (NoAuth, API-ключ, AWS SigV4)
✅ Управление сессиями: Сохранение состояния между вызовами на базе Redis
✅ Валидация входных данных: Валидация JSON Schema для всех входных данных инструментов
Качество
✅ Типобезопасность: Полное соответствие строгому режиму mypy (ноль ошибок)
✅ Покрытие тестами: 96% покрытия, 112 проходящих тестов
✅ Готовность к продакшену: Ноль ошибок линтинга, всесторонняя обработка ошибок
Установка
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:
Handler:
your_module.handlerRuntime:
python3.12Timeout: 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 onlyАутентификация по API-ключу
from sipap_mcp.auth import APIKeyAuth
auth = APIKeyAuth(api_keys=[
"client-a-key",
"client-b-key",
"client-c-key"
])Клиенты отправляют API-ключ в заголовке X-API-Key.
Аутентификация AWS SigV4
from sipap_mcp.auth import SigV4Auth
auth = SigV4Auth(service="lambda", region="us-east-1")Для URL функций Lambda с аутентификацией IAM.
Управление сессиями
Поддерживайте состояние между несколькими запросами с помощью 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/ для полных примеров:
Пример | Описание |
Простой сервер-калькулятор | |
Развёртывание Lambda с аутентификацией по API-ключу | |
HTTP-сервер с сессиями Redis | |
Продвинутые паттерны и хуки жизненного цикла | |
Все стратегии аутентификации |
Запуск примеров:
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 + Generator: Управление ресурсами с помощью контекстных менеджеров
Автоматическое обнаружение инструментов: Регистрация инструментов на основе интроспекции
Принудительный структурированный вывод: Валидация 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: Создаёт сессию, возвращает IDget_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)Развёртывание:
Handler:
app.handlerRuntime:
python3.12Memory: 512 МБ (настройте в зависимости от нагрузки)
Timeout: 30 секунд (настройте в зависимости от времени выполнения инструментов)
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)Определение задачи:
Container port: 8000
Health check:
/health(если реализован)CPU: 256 (.25 vCPU)
Memory: 512 МБ
Redis для сессий
Разработка:
docker run -d -p 6379:6379 redis:7-alpineПродакшен:
AWS ElastiCache для 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 МБ, Python 3.12):
Операция | Холодный старт | Тёплый старт |
tools/list | 850мс | 12мс |
tools/call (простой) | 900мс | 15мс |
tools/call (с БД) | 1200мс | 45мс |
Советы по оптимизации
Уменьшите холодные старты: Используйте provisioned concurrency в Lambda
Кэшируйте соединения: Инициализируйте в
_setup(), переиспользуйте между вызовамиМинимизируйте зависимости: Импортируйте только то, что нужно
Используйте async: Транспорт FastAPI поддерживает асинхронные инструменты
TTL сессий: Балансируйте использование памяти и пользовательский опыт
Вклад в проект
Приветствуются любые вклады! Пожалуйста:
Следуйте существующему стилю кода (ruff + mypy strict)
Добавляйте тесты для новых функций (поддерживайте покрытие 80%+)
Обновляйте документацию
Запускайте все проверки качества перед отправкой
Лицензия
Copyright © 2026 Valo Team
Создано с помощью:
Разработка через тестирование (TDD)
Типобезопасность (строгий режим mypy)
96% покрытие тестами (112 тестов)
Готовая к продакшену обработка ошибок
Полная документация
Часть платформы Valo — Платформа спортивной аналитики и оценки вероятности исходов
This server cannot be deployed
Maintenance
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.112 npmMIT
- 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-