event-bridge-client 1.0.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, and handle HMAC-signed remote commands. Wire-compatible with the Node event-bridge-client — 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"})

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

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	Exponential-backoff retries for event batch POSTs
nonce_store	in-memory	Replay store; supply a shared one for multi-instance

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.