gnexus-synapse — Python-клиент Synapse

Тонкий клиент Ingestion API v1 хаба уведомлений Gnexus Synapse (git+https://git.gnexus.space/git/root/gn-synapse.git): в обе стороны — сборка и отправка событий (Ingestion) и приём s2s-доставок (проверка подписи webhook'а) — без ретраев и локальных очередей (очередь у Synapse своя, приём отвечает мгновенно). Контракт — docs/05-ingestion-api.md в репозитории gn-synapse, порядок интеграции — docs/09-integration-guide.md.

Установка

pip install "git+https://git.gnexus.space/git/root/gn-synapse-client-py.git@v0.1.1"

Серверный репозиторий — https://git.gnexus.space/git/root/gn-synapse.git. Зависимость одна: httpx.

Quickstart

from gnexus_synapse import SynapseClient

# аргументы сильнее env; фолбэк — SYNAPSE_URL / SYNAPSE_API_KEY
syn = SynapseClient("http://synapse.gnexus.space:8013", "syn_...")

event = syn.send("bugtrail", "test", "failed",
                 payload={"user_id": "<sub gnexus-auth>", "error": "..."},
                 priority="high", dedup_key="run-2026-10-03")
print(event.id, event.status)  # queued

# пожарная отправка: ошибки Synapse не ломают сервис, пишутся в warning
syn.emit("gntodo", "task", "created", payload={"user_id": uid})

status = syn.status(event.id)          # статусы доставок по каналам
result = syn.send_batch([...])         # batch: result.accepted / result.rejected

env-фолбэк (для сервисов без своей конфигурации):

SYNAPSE_URL=http://synapse:8013
SYNAPSE_API_KEY=syn_...          # выдаёт админ (create-key / MCP key_issue), печатается один раз
SYNAPSE_DEFAULT_SOURCE=bugtrail  # необязательно: не указывать source в каждом вызове
SYNAPSE_TIMEOUT=10               # секунды

default_source — лучшая защита от частой ошибки 403: ключ принадлежит одному источнику, и source в конверте обязан совпадать с ним.

API

Исключения (все — наследники SynapseError с полями detail, status_code):

Локальная валидация зеркалит серверную (^[a-z0-9]([a-z0-9._-]*[a-z0-9])?$, 1..64; priority low/normal/high/critical; ttl 1..604800; dedup_key ≤ 255; payload — JSON-объект), поэтому SynapseValidationError от клиента — почти всегда ошибка кода, ловить её нужно в тестах, а не в рантайме. Сервер игнорирует незарегистрированные пары (subject, action) — их регистрирует админ (add-type / MCP type_register).

Асинхронный близнец для FastAPI-сервисов:

from gnexus_synapse import AsyncSynapseClient
syn = AsyncSynapseClient()          # httpx.AsyncClient, await syn.aclose()
event = await syn.send("gntodo", "task", "created", payload={"user_id": uid})

Приём s2s-доставки (webhook-приёмник)

Сервисы, которые и шлют события, и принимают доставки Synapse (Navi, …), используют тот же SDK для второй стороны:

from gnexus_synapse import verify_webhook, SIGNATURE_HEADER

@router.post("/webhooks/synapse")
async def synapse_webhook(request: Request):
    envelope = verify_webhook(await request.body(), request.headers, S2S_SECRET)
    ...  # envelope["event_id"], envelope["subject"], envelope["payload"]

Один вход: не-подпись/чужой секрет/replay (тело старше max_skew, дефолт 300 с)/не-JSON — всё ловится одним SynapseWebhookError (детали в detail). Схема подписи — одна на всю экосистему (совместима с gnexus-auth и с app/signature.py Synapse); проверка константным сравнением. make_signature() — для тестов приёмника. Секрет per-target — S2S_SECRET_<token_ref> (docs/05).

Конвенции и оговорки

• Дедуп best-effort (окно 24 ч, сервер не даёт гарантии уникальности): не стройте бизнес-логику на отсутствии дублей.
• Дедуп вернул 202 с deduplicated=true — это id первого события, не ошибки.
• payload.user_id — конвенция «о ком событие» (нестроковое значение → адресные доставки skipped): клиент пишет warning, но не блокирует.
• scheduled_at — резерв контракта v1: передаётся, сервер пока не обрабатывает.
• Свой httpx.Client создаётся с явным timeout (у httpx дефолт — «бесконечно»). Инжектный клиент (http_client=...) не трогается — таймаут настраивайте сами.
• user_agent gnexus-synapse-py/<версия> — по нему Synapse видит, кто источался.

Разработка

uv venv -p 3.12 && uv pip install -e . -e '.[dev]'
pytest            # юнит-тесты (MockTransport)
ruff check src tests && mypy src
# интеграционный смок против живого стека:
SYNAPSE_URL=http://localhost:8013 SYNAPSE_API_KEY=syn_... python examples/plain/smoke.py