Skip to main content
Glama

sipap-mcp

Готовый к продакшену фреймворк MCP-серверов для AWS Lambda и ECS Fargate

Python Version Type Checked Code Style Test Coverage Tests


Обзор

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

  • Runtime: python3.12

  • Timeout: 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/ для полных примеров:

Пример

Описание

01_basic_server.py

Простой сервер-калькулятор

02_lambda_with_auth.py

Развёртывание Lambda с аутентификацией по API-ключу

03_http_with_sessions.py

HTTP-сервер с сессиями Redis

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 + Generator: Управление ресурсами с помощью контекстных менеджеров

  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)

Развёртывание:

  • Handler: app.handler

  • Runtime: python3.12

  • Memory: 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мс

Советы по оптимизации

  1. Уменьшите холодные старты: Используйте provisioned concurrency в Lambda

  2. Кэшируйте соединения: Инициализируйте в _setup(), переиспользуйте между вызовами

  3. Минимизируйте зависимости: Импортируйте только то, что нужно

  4. Используйте async: Транспорт FastAPI поддерживает асинхронные инструменты

  5. TTL сессий: Балансируйте использование памяти и пользовательский опыт

Вклад в проект

Приветствуются любые вклады! Пожалуйста:

  1. Следуйте существующему стилю кода (ruff + mypy strict)

  2. Добавляйте тесты для новых функций (поддерживайте покрытие 80%+)

  3. Обновляйте документацию

  4. Запускайте все проверки качества перед отправкой

Лицензия

Copyright © 2026 Valo Team


Создано с помощью:

  • Разработка через тестирование (TDD)

  • Типобезопасность (строгий режим mypy)

  • 96% покрытие тестами (112 тестов)

  • Готовая к продакшену обработка ошибок

  • Полная документация

Часть платформы Valo — Платформа спортивной аналитики и оценки вероятности исходов

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
    12 npm
    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