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 — Платформа спортивной аналитики и оценки вероятности исходов

Tool Schema Changelog

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.

Maintenance

ActivityMaintained
ResponsivenessNo issues

Resources

Unclaimed servers have limited discoverability.

Looking for Admin?

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

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
    17
    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

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