Skip to main content

Python SDK

The Python SDK mirrors the same authoring model as Go:

  • Service(name, *, display_name="", description="", lifecycle=Lifecycle.STABLE, settings=None)
  • svc.ingress(...)
  • svc.enable_dedup() / svc.enable_rate_limit()
  • svc.data_source(...)
  • svc.wiring()
  • @svc.action(...)
  • await svc.run()

Example​

from triggers import Lifecycle, NATS, Service

svc = Service("flow_lab", lifecycle=Lifecycle.BETA)
svc.ingress(NATS)

@svc.action("parse")
async def parse(ctx, payload: ParseInput) -> ParseResult:
    return ParseResult(message=payload.message, length=len(payload.message))

The facade still handles schema derivation, manifest building, runtime startup, retries, DLQ, and parity with Go's wire contract. Examples in this repo prefer Service("name", ...) directly. Reach for settings=TriggersSettings(...) only when a test or example needs to override process env locally. Use Lifecycle.STABLE, Lifecycle.BETA, or Lifecycle.DEPRECATED when you want catalog lifecycle metadata on the service or an action. For HTTP ingress, HTTP uses the canonical /v2/<service>/inbound path by default. Use HTTP.at("/custom/path") or HTTP.at("https://...") only when you need to override it. svc.wiring() gives app code the resolved runtime snapshot when it needs to inspect HTTP base URLs, ingress URLs, NATS stream names, or parsed Redis settings without re-reading env vars.