shop-mcp

Локальный MCP-сервер (Model Context Protocol) только для чтения, который позволяет ИИ-агенту анализировать SQLite-базу данных электронной коммерции (shop.db) — клиентов, товары, заказы и позиции заказов — через транспорт stdio. Никакого HTTP-сервера, никакого отдельного процесса базы данных: сервер открывает shop.db напрямую и предоставляет два небольших инструмента общего назначения, которые агент может использовать для изучения схемы и выполнения собственных аналитических SQL-запросов.

Создан с использованием официального Python MCP SDK (mcp на PyPI).

Структура проекта

mcp-sql/
├── server.py                    # the MCP server (stdio transport)
├── shop.db                      # SQLite database (not modified by this project)
├── requirements.txt
├── .env.example
├── mcp-config.example.json
├── tests/
│   ├── conftest.py
│   ├── test_server.py           # unit tests (call tool functions directly)
│   └── test_stdio_integration.py# protocol-level test (spawns server.py over stdio)
└── README.md

Схема базы данных (как она есть в shop.db)

customers(id PK, first_name, last_name, email UNIQUE, phone, created_at)
products(id PK, name, category, price, stock_quantity, created_at)
orders(id PK, customer_id -> customers.id, order_date, status, total_amount)
order_items(id PK, order_id -> orders.id, product_id -> products.id, quantity, unit_price)

orders.status ограничен значениями: new, processing, shipped, completed, cancelled. products.category в настоящее время имеет 5 различных значений. Внешние ключи: orders.customer_id → customers.id, order_items.order_id → orders.id, order_items.product_id → products.id. Сервер получает всё это из живой базы данных во время запроса (через sqlite_master / PRAGMA table_info / PRAGMA foreign_key_list) — здесь ничего не захардкожено, поэтому если shop.db заменить на другой файл с другой схемой, get_database_schema автоматически это отразит.

Известные особенности данных в предоставленном shop.db: в customers нет столбца country, поэтому вопросы типа «клиенты из Германии» не могут быть обработаны — инструмент схемы позволяет это обнаружить, а query_database возвращает понятную ошибку no such column: country вместо предположений. Все 750 заказов в базе данных датированы 2026 годом (ни одного в 2025), поэтому запрос «выручка за 2025» корректно возвращает 0/null, а не ошибку.

Установка

cd mcp-sql
python3 -m venv .venv
source .venv/bin/activate        # on Windows: .venv\Scripts\activate
pip install -r requirements.txt

Конфигурация

Путь к базе данных никогда не зашит в исходном коде. Он определяется следующим образом:

1. переменная окружения SHOP_DB_PATH, если она задана;

2. в противном случае shop.db рядом с server.py.

Скопируйте .env.example в .env и отредактируйте его, если хотите указать серверу на другой файл базы данных (вам нужно будет загрузить его в свою оболочку/лаунчер агента самостоятельно, например export $(cat .env | xargs), или просто установите SHOP_DB_PATH напрямую):

cp .env.example .env
# edit .env, or simply:
export SHOP_DB_PATH=/absolute/path/to/shop.db

Запуск

source .venv/bin/activate
python server.py

Процесс общается по MCP через stdio и ожидает клиента — он будет выглядеть «зависшим» без вывода, что ожидаемо: подключите MCP-клиента (ИИ-агента или mcp-inspector, см. ниже), а не запускайте его отдельно в терминале.

Быстрая ручная проверка с помощью официального MCP Inspector (установка не требуется):

npx @modelcontextprotocol/inspector --cli .venv/bin/python server.py --method tools/list

Подключение к ИИ-агенту

Большинство MCP-совместимых клиентов (Claude Desktop, Claude Code и т. д.) читают JSON-блок конфигурации, например mcp-config.example.json:

{
  "mcpServers": {
    "shop-mcp": {
      "command": "/absolute/path/to/mcp-sql/.venv/bin/python",
      "args": ["/absolute/path/to/mcp-sql/server.py"],
      "env": {
        "SHOP_DB_PATH": "/absolute/path/to/mcp-sql/shop.db"
      }
    }
  }
}

Примечания:

• Используйте абсолютный путь к интерпретатору Python в venv (как указано выше), чтобы пакет mcp был найден без ручной активации venv; использование простого python3 также работает, если mcp установлен в том окружении, на которое он указывает.

• SHOP_DB_PATH необязателен — опустите его, чтобы использовать встроенный shop.db.

• Абсолютные пути должны находиться в этом конфигурационном файле, который предоставляет тот, кто подключает сервер, — никогда внутри самого server.py.

• Размещение этого блока зависит от конкретного клиента (например, Claude Desktop использует claude_desktop_config.json с той же структурой mcpServers; другим клиентам может понадобиться только внутренний объект {"command": ..., "args": ..., "env": ...}). Проверьте документацию вашего клиента, чтобы узнать, где находится файл.

Тестирование

source .venv/bin/activate
python -m pytest tests/ -v

Это запускает 48 тестов, включая:

• обнаружение схемы (таблицы, столбцы, PK/FK, связи, количество строк);

• SELECT, JOIN, WHERE, GROUP BY, ORDER BY, агрегатные функции (COUNT/SUM/AVG/MIN/MAX), подзапросы, безопасный CTE WITH ... SELECT и фильтрацию по дате (strftime);

• ограничение количества строк и постраничную навигацию на основе offset;

• понятную обработку ошибок для неверного SQL, неизвестных таблиц/столбцов, пустого запроса и отсутствующего файла базы данных;

• безопасность только для чтения: каждый тип оператора, перечисленный в задании (DELETE, UPDATE, DROP, CREATE, INSERT, а также ALTER, REPLACE, TRUNCATE, ATTACH, DETACH, VACUUM, REINDEX, разрушительный PRAGMA, составной SELECT 1; DROP TABLE ... и удаление, замаскированное под CTE WITH x AS (...) DELETE ...), отклоняется, и после этого проверяется, что количество строк и SHA-256 хэш файла базы данных не изменились;

• tests/test_stdio_integration.py запускает server.py как реальный подпроцесс и управляет им через настоящий MCP client SDK по stdio (initialize → list_tools → call_tool), а не вызывает функции Python напрямую — это тот же путь, который использует реальный агент.

MCP-инструменты

get_database_schema()

Без параметров. Вызывайте его первым, если вы ещё не знаете точных имён таблиц/столбцов — не угадывайте их. Возвращает для каждой таблицы: row_count, columns (имя, тип SQLite, not_null, default_value, is_primary_key), primary_key, foreign_keys (столбец, ссылающаяся таблица/столбец, ON DELETE/ON UPDATE) и несколько sample_rows, чтобы агент мог увидеть реальные форматы дат, значения статусов, величины цен и т. д. Список relationships верхнего уровня даёт строки table.column -> other_table.column, полученные из живых внешних ключей.

query_database(sql, limit=100, offset=0)

Выполняет один оператор SQL только для чтения (SELECT или WITH ... SELECT) и возвращает {columns, rows, row_count, limit, offset, truncated, total_matching_rows}. Поддерживает JOIN, WHERE, GROUP BY, ORDER BY, агрегатные функции, подзапросы и CTE. limit ограничивается диапазоном 1..500 (по умолчанию 100); используйте offset для постраничного просмотра больших результатов. total_matching_rows и truncated сообщают вызывающей стороне, является ли текущая страница полным результатом или есть ещё данные для получения. Ошибки (неверный синтаксис, неизвестная таблица/столбец или отклонённая попытка записи) выдаются в виде короткого конкретного сообщения — никогда в виде сырого Python traceback.

Безопасность: как обеспечивается режим только для чтения

В задании явно указано не полагаться на одну проверку regex/ключевых слов, поэтому этот сервер использует четыре независимых уровня защиты — проверено в tests/test_server.py:

1. Файловый дескриптор только для чтения на уровне ОС. Файл SQLite открывается с URI file:<path>?mode=ro. Затем сам SQLite отказывает в любой записи (OperationalError: attempt to write a readonly database) независимо от выполняемого SQL — это работает, даже если все проверки ниже содержат ошибку.

2. PRAGMA query_only = ON устанавливается на каждом соединении как вторая независимая защита на уровне SQLite от записи.

3. Обратный вызов авторизатора sqlite3 (Connection.set_authorizer) разрешает только действия SELECT / READ / FUNCTION / RECURSIVE на уровне движка SQLite и запрещает всё остальное — INSERT, UPDATE, DELETE, DROP, ALTER, CREATE, REPLACE, TRUNCATE, ATTACH, DETACH, VACUUM, REINDEX, PRAGMA, транзакции и т. д. Это выполняется на разобранном операторе, поэтому также перехватывает классический обход через CTE WITH x AS (SELECT 1) DELETE FROM ..., который наивная текстовая проверка «должен начинаться с SELECT» пропустила бы.

4. Проверки формы оператора в server.py: отправленный текст должен начинаться с SELECT/WITH (быстрое и понятное отклонение до обращения к SQLite), и каждый запрос выполняется в обёртке SELECT * FROM (<query>) LIMIT :limit OFFSET :offset — для разбора требуется один оператор, поэтому составной SELECT 1; DROP TABLE customers становится обычной синтаксической ошибкой SQL, а не двумя выполненными операторами.

Поскольку уровень 1 (mode=ro) обеспечивается SQLite/ОС независимо от собственной логики сервера, shop.db не может быть изменён через этот сервер, даже если в уровнях 2-4 есть ошибка.

Известные ограничения

• В customers нет столбца country/местоположения в предоставленном shop.db, поэтому вопросы типа «клиенты из Германии» не могут быть обработаны на основе этих данных — инструмент схемы показывает это, а не сервер придумывает столбец.

• Все заказы в предоставленных данных датированы 2026 годом; запрос выручки за 2025 год корректно возвращает 0, а не ошибку.

• total_matching_rows в query_database вычисляется с помощью второго COUNT(*), оборачивающего тот же запрос; для очень дорогих запросов это примерно удваивает работу. Учитывая размер этой базы данных (от сотен до нескольких тысяч строк на таблицу), это не является практической проблемой.