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

Метод	Что делает	Ошибки
`send(source, subject, action, *, priority, payload, dedup_key, ttl_seconds, scheduled_at)`	POST /api/v1/events → `Event(id, status, deduplicated)`	типизированные, см. ниже
`emit(...)`	тот же, ловит SynapseError → warning в логгер, возвращает `Event \	None`	не бросает
`send_batch(events)`	массив конвертов; локально битые не посылаются; rejected — не исключение (частичный успех норма); транспортный сбой/5xx — исключение	`SynapseError`
`status(event_id)`	GET /api/v1/events/{id} → `EventStatus` (в т.ч. `deliveries`)	`SynapseNotFoundError`
`health()` / `ready()`	диагностика без ключа	—
`close()`	закрыть свой httpx.Client (инжектный не трогает)	—

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

Класс	Значение
`SynapseConfigError`	нет url/api_key/source в аргументах и env
`SynapseConnectionError`	Synapse недоступен (DNS/timeout/refused), `__cause__` — оригинал
`SynapseAuthError`	401 — ключ отсутствует/неизвестен/отозван (или источник в архиве)
`SynapseForbiddenError`	403 — `source` конверта ≠ источник ключа; задайте `default_source`
`SynapseValidationError`	422 сервера или локальная ошибка конверта (`status_code is None`)
`SynapseNotFoundError`	404 — событие не у источника ключа
`SynapseServerError`	5xx — событие не принято, повторите при желании
`SynapseWebhookError`	приём s2s: подпись/freshness/не-JSON — не прошло проверку `verify_webhook`

Локальная валидация зеркалит серверную (^[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