Reference
Sync and async Python client
Use the useeventstack package to emit events, manage workflows, replay, and query projections from Python 3.10+.
Installation
pip install useeventstackVersion 0.3.0. Requires Python 3.10+ and httpx.
To pin it in a project, add useeventstack>=0.3,<0.4 to your requirements. The SDK is MIT licensed; the platform itself is not open source.
Synchronous client
import os
from useeventstack import UseEventStackClient
with UseEventStackClient(
api_key=os.environ["USEEVENTSTACK_API_KEY"],
project_id=os.environ["USEEVENTSTACK_PROJECT_ID"],
base_url="http://localhost:3001",
organization_id=os.environ.get("USEEVENTSTACK_ORGANIZATION_ID"),
environment="production",
) as client:
result = client.events.emit(
"deployment.completed",
payload={
"service": "api",
"version": "2.1.0",
"environment": "production",
"status": "healthy",
},
idempotency_key="deploy-api-2.1.0",
)
print(f"Event: {result['event']['id']}")
# Query events
events = client.events.query(event_type="deployment.completed")
# Replay
client.replay.single("event-uuid", skip_side_effects=True)Asynchronous client
from useeventstack import AsyncUseEventStackClient
async with AsyncUseEventStackClient(
api_key="useeventstack_key",
project_id="project-uuid",
) as client:
result = await client.events.emit(
"deployment.completed",
payload={"service": "worker", "version": "1.5.0",
"environment": "sandbox", "status": "healthy"},
environment="sandbox",
)
events = await client.events.query(event_type="deployment.completed")
workflows = await client.workflows.list()Available methods
Both sync and async clients expose identical namespaces:
| Namespace | Methods |
|---|---|
events | emit, query, get, trace |
workflows | query, list, get, create, update, delete, enable, disable |
custom_actions | list, get, delete, simulate |
replay | single, bulk |
dlq | list, reprocess |
api_keys | list, create, delete |
projections | query, service_status |
Responses are Python dictionaries. Non-2xx responses raise UseEventStackApiError.
Error handling
from useeventstack.errors import UseEventStackApiError
try:
client.events.emit("bad.event", payload={})
except UseEventStackApiError as e:
print(f"[{e.status}] {e.code}: {e.message}")
print(e.details) # Optional dict