event-bridge-client 1.4.0

Lightweight client for registering apps, batching lifecycle events, and handling HMAC-signed remote commands.

event-bridge-client (Python)

A lightweight client for connecting a Python application to a control backend: register the app, batch and push lifecycle events, handle HMAC-signed remote commands, and take payments. Wire-compatible with the Node event-bridge-client (same version, same signing scheme, same endpoints, same payloads).

Install

pip install event-bridge-client          # core (send events, verify commands)
pip install "event-bridge-client[fastapi]"  # + FastAPI inbound adapter

Requires Python 3.9+.

Quickstart (FastAPI)

from fastapi import FastAPI, Request
from event_bridge_client import create_client

client = create_client(
    base_url="https://bridge.example.com",
    api_key="...",
    callback_url="https://api.example.com/bridge/commands",
    callback_secret="...",
    capabilities=["user.ban", "user.unban"],
    env="PROD",
)

@client.on_command("user.ban")
async def _(data, ctx):
    await ban_account(data["externalUserId"], data["reason"])
    return {"ok": True}

app = FastAPI()

@app.post("/bridge/commands")
async def commands(request: Request):
    return await client.middleware.fastapi(request)

@app.on_event("startup")
async def _startup():
    await client.register()

@app.on_event("shutdown")
async def _shutdown():
    await client.aclose()

# Anywhere in your app:
client.events.emit("user.created", {"externalUserId": "usr_123", "email": "a@b.c"})

Any event type is accepted — emit a custom event the SDK doesn't model and it is batched and pushed like any other (see EVENT_TYPES for the built-ins):

client.events.emit("widget.exported", {"widgetId": "w_1", "rows": 4200})

Command handlers may be sync or async. Each receives (data, ctx) where data is the command payload (a dict) and ctx carries command_id, command_type, issued_at, and issued_by. Return {"ok": True, "result": ...} or {"ok": False, "error": "..."}.

Open dispatch + "*" catch-all

Any command type is accepted — the envelope is validated, but you can handle a capability the SDK doesn't model yet with zero SDK change. Register a handler under a specific type, or under "*" to catch every unregistered type:

@client.on_command("future.capability")          # specific type
async def _(data, ctx): ...

@client.on_command("*")                            # catch-all (anything else)
async def _(data, ctx):
    log.info("got %s", ctx.command_type)           # the actual command type
    return {"ok": True}

A specific handler always wins over "*". If neither a specific handler nor a "*" handler is registered, the callback returns 501 (NO_HANDLER).

Known-type validation

For the seven command types the SDK ships typed schemas for — user.ban, user.unban, user.extend_trial, subscription.change_plan, subscription.cancel, subscription.uncancel, subscription.extend (see KNOWN_COMMAND_TYPES / COMMAND_DATA_BY_TYPE) — the data payload is strictly validated (and defaults applied, e.g. atPeriodEnd → True) before your handler runs. Malformed data is rejected with 400 (invalid_command) and your handler is never called. Unknown types pass their data through untouched.

Payments

Create a payment (returns a hosted checkout URL) and receive the outcome as a signed webhook on the same callback endpoint as commands. Do not emit payment.* as events — they're rejected.

# Create → send the payer to hostedCheckoutUrl.
res = await client.payments.create(
    external_id="order_4821",        # your idempotency key
    currency="UZS",
    line_items=[{"name": "Pro plan — 1 month", "unit_amount": 120_000}],
    customer={"email": "[e-mail]", "external_id": "usr_123"},
)
checkout_url = res["hostedCheckoutUrl"]

# Receive the outcome. The same middleware verifies + dispatches here.
@client.on_payment
async def _(note, ctx):
    if note.type == "payment.settled":
        await fulfil(note.data["externalId"])
    elif note.type == "payment.failed":
        await mark_failed(note.data["externalId"])

# (Optional) poll instead of / alongside the webhook:
latest = await client.payments.get_by_external_id("order_4821")  # or .get(id); None if missing

Notes:

• external_id is your idempotency key — reusing it returns the existing payment.
• Line items take ad-hoc name + unit_amount, or a catalogue variant_id.
• A non-base currency requires rate_to_uzs.
• Make on_payment idempotent — delivery is at-least-once (a raised handler returns 500 so management retries). Dedupe on note.data["paymentId"] + note.type.
• The webhook defaults to your callback_url; override per payment with notify_url.

Managed resources

Declare an entity the backend can list / search / view / action — entirely from the descriptor, with no backend-side code change. Records are never shipped to the backend; it proxies list / get / action queries back over the same signed channel.

client.define_resource(
    {
        "key": "widgetUser",
        "label": "Widget User",
        "labelPlural": "Widget Users",
        "titleField": "email",
        "fields": [
            {"key": "id", "label": "ID", "type": "string", "listVisible": False},
            {"key": "email", "label": "Email", "type": "email", "filterable": True},
            {"key": "plan", "label": "Plan", "type": "enum", "enumValues": ["free", "pro"]},
        ],
        "actions": [
            {
                "capability": "widgetUser.ban",
                "label": "Ban",
                "confirm": True,
                "destructive": True,
                "fields": [{"name": "reason", "label": "Reason", "kind": "textarea", "required": True}],
            }
        ],
    },
    # `query` is a ResourceListQuery: query.page, query.page_size, query.q, ...
    list=lambda query: {"records": db.search(query.q, query.page, query.page_size), "total": db.count()},
    get=lambda record_id: db.find(record_id),  # optional
    action=lambda inp: ban(inp["recordId"], inp["params"]["reason"]),  # optional
)

Descriptor keys accept either snake_case or camelCase; they're sent to the backend as camelCase. Field type is one of string, number, boolean, date, datetime, enum, currency, badge, email, url, json.

Options

create_client(...) keyword arguments:

Option	Default	Notes
base_url	required	Control backend base URL
api_key	required	API key minted by the backend admin
callback_url	required	HTTPS URL the backend POSTs commands to
callback_secret	required	HMAC shared secret minted alongside the API key
capabilities	()	Strings matching command types, e.g. user.ban
env	"PROD"	PROD / STAGING / DEV
enabled	True	If False, all methods are no-ops (staged rollout)
batch_interval_ms	1500	Event batcher flush interval
batch_max_size	100	Force-flush when this many events are queued
max_buffer_size	10000	Hard cap on buffered events; oldest dropped past it
max_retries	6	Retries for one-shot calls + event batch re-queue cap
request_timeout	15.0	Per-request timeout (seconds) for every HTTP call
heartbeat_interval_ms	60000	Heartbeat cadence
shutdown_timeout	5.0	Max seconds to drain buffered events on aclose()
nonce_store	in-memory	Replay store; supply a shared one for multi-instance

Reliability: retries, timeouts, shutdown

Every one-shot call (register, heartbeat, command ack, and payments.*) goes through a shared HTTP core that:

• applies the request_timeout per request;
• retries on transport/timeout errors and 408 / 429 / 5xx with exponential backoff + full jitter (other 4xx are never retried), honoring a Retry-After header when present;
• stamps an x-request-id correlation header on every request.

On exhaustion it raises RequestTimeoutError (a timeout; code="TIMEOUT", status=408) or ManagementError (network error / HttpError with code="http_<status>"). The event batcher keeps its own re-queue/backoff and so issues its POST with retries disabled (it still gets the timeout + correlation id).

await client.aclose() cancels the heartbeat and drains buffered events under shutdown_timeout; if the backend is unreachable it logs a warning and returns rather than hanging.

Flask / WSGI

The async fastapi() adapter is preferred, but a sync bridge is provided for WSGI frameworks. Flask is imported lazily, so it stays an optional dependency:

from flask import Flask, request

app = Flask(__name__)

@app.post("/bridge/commands")
def commands():
    return client.middleware.flask(request)   # returns a Flask Response

There is also client.middleware.handle_sync(raw_body, headers, method, path) returning (status, body) if you build the response yourself.

Constraints (best-effort by design): handle_sync/flask spin up a short-lived event loop via asyncio.run, so they must not be called from inside a running event loop — use fastapi() / handle in async apps. Resource queries and payment webhooks are handled inline and fully supported; command callbacks are dispatched and acked on that short-lived loop (best-effort).

Replay protection across instances

The default replay cache is in-process — it only protects a single instance. If you run more than one instance behind a load balancer, supply a shared nonce_store (e.g. Redis) so a captured command can't be replayed against another instance inside the 300-second signature window:

class RedisNonceStore:
    def __init__(self, redis): self.r = redis
    async def has(self, nonce: str) -> bool:
        return await self.r.exists(f"bridge:nonce:{nonce}") > 0
    async def add(self, nonce: str, ttl_ms: int) -> None:
        await self.r.set(f"bridge:nonce:{nonce}", "1", px=ttl_ms, nx=True)

client = create_client(..., nonce_store=RedisNonceStore(redis))

Send-only usage (no FastAPI)

If the app only emits events and never receives commands, you don't need FastAPI — pip install event-bridge-client and use register() / events.emit() / aclose(). The middleware is only needed to receive commands.