stage-events-client

Python client for sending structured-mode CloudEvents 1.0.2 to the Stage Events API.

The package provides Pydantic models for the supported workflow events and synchronous and asynchronous clients built on HTTPX.

Requirements

• Python 3.10 or newer
• A Stage Events API endpoint
• A bearer token when authentication is enabled

Installation

python -m pip install stage-events-client

The command-line interface dependencies are optional. Enable the CLI explicitly when installing the package:

python -m pip install 'stage-events-client[cli]'

Command-line interface

The send-stage-event executable provides one command for every supported CloudEvent model:

send-stage-event --help
send-stage-event submitted --help

Pass the complete destination URL as the command argument. There is no separate base URL or endpoint-path setting:

send-stage-event submitted \
  https://events.example.com/hooks/cloud-events \
  --source workflows:example-process:submit \
  --subject workflows:2f660c57:example-workflow \
  --data '{"namespace":"workflows","time":"2026-07-18T12:00:00Z"}' \
  --x-kafka-topic workflows.2f660c57.submitted \
  --token your-bearer-token

The available commands are calendar, submitted, dismissed, prepared, completed, failed, piped, staged, and ordered. Each command validates --data against its corresponding Pydantic model before sending the request.

--data accepts inline JSON, a JSON file prefixed with @, or - to read from standard input:

send-stage-event prepared https://events.example.com/cloud-events \
  --source workflows:example-process:prepare \
  --subject workflows:2f660c57:example-workflow \
  --data @prepared-data.json

cat submitted-data.json | send-stage-event submitted \
  https://events.example.com/cloud-events \
  --source workflows:example-process:submit \
  --subject workflows:2f660c57:example-workflow \
  --data -

The partition key defaults to the subject. Override it with --partition-key when required. The --x-kafka-topic option is optional.

The bearer token can be supplied through --token or the STAGE_EVENTS_TOKEN environment variable. Run a command with --help for TLS, timeout, and all other options.

Quick start

Create a client, construct an event, and send it with the required Kafka topic:

from datetime import datetime, timezone
from uuid import uuid4

from stage_events_client import AuthenticatedClient
from stage_events_client.api.default import send_cloud_event
from stage_events_client.models import SubmittedCloudEvent, SubmittedData

client = AuthenticatedClient(
    base_url="https://events.example.com",
    token="your-bearer-token",
)

subject = "workflows:2f660c57:example-workflow"
event = SubmittedCloudEvent(
    source="workflows:example-process:submit",
    subject=subject,
    partitionkey=subject,
    specversion="1.0",
    id=str(uuid4()),
    data=SubmittedData(
        namespace="workflows",
        time=datetime.now(timezone.utc),
    ),
)

with client:
    result = send_cloud_event.sync(
        client=client,
        body=event,
        x_kafka_topic="workflows.2f660c57.submitted",
    )

print(result)

Use Client instead when the API does not require authentication:

from stage_events_client import Client

client = Client(base_url="https://events.example.com")

Detailed responses

sync returns the parsed response body. Use sync_detailed when the status, headers, or raw response body are also needed:

response = send_cloud_event.sync_detailed(
    client=client,
    body=event,
    x_kafka_topic="workflows.2f660c57.submitted",
)

print(response.status_code)
print(response.headers)
print(response.content)
print(response.parsed)

A successful 200 response is parsed as a string. Documented 400 responses are parsed into the corresponding problem-details model. An undocumented status returns None unless raise_on_unexpected_status=True is set on the client, in which case stage_events_client.errors.UnexpectedStatus is raised.

Async usage

Each endpoint has equivalent asyncio and asyncio_detailed functions:

import asyncio


async def main() -> None:
    async with client:
        result = await send_cloud_event.asyncio(
            client=client,
            body=event,
            x_kafka_topic="workflows.2f660c57.submitted",
        )
    print(result)


asyncio.run(main())

Do not use the same client instance in synchronous and asynchronous context managers at the same time.

Supported events

The request body can be any of the following models from stage_events_client.models:

The models validate event-specific payloads, timezone-aware timestamps, semantic versions, and GeoJSON structures where applicable. Additional standard CloudEvent attributes such as id and specversion are preserved by the models.

Event addressing

The API expects these values:

• source: three colon-separated components, normally {namespace}:{process-id}:{step-name}.
• subject: three colon-separated components, normally {namespace}:{workflow-uid}:{workflow-name}.
• partitionkey: usually the same value as subject.
• x_kafka_topic: a topic in the form {namespace}.{workflow-uid}.{event-suffix}, such as workflows.2f660c57.submitted.

Calendar topics use a duration followed by .calendar, for example workflows.2f660c57.10m.calendar.

Client configuration

Both Client and AuthenticatedClient accept shared headers, cookies, timeout, TLS verification, redirect handling, and additional HTTPX options:

import httpx

from stage_events_client import AuthenticatedClient

client = AuthenticatedClient(
    base_url="https://events.example.com",
    token="your-bearer-token",
    headers={"X-Correlation-ID": "request-id"},
    timeout=httpx.Timeout(30.0),
    verify_ssl="/path/to/ca-bundle.pem",
    follow_redirects=True,
    httpx_args={"proxy": "http://proxy.example.com:8080"},
    raise_on_unexpected_status=True,
)

TLS certificate verification is enabled by default. Setting verify_ssl=False disables server certificate validation and should only be used in controlled development environments.

Clients can be copied with updated settings:

client = client.with_headers({"X-Correlation-ID": "new-request-id"})
client = client.with_cookies({"session": "value"})
client = client.with_timeout(httpx.Timeout(10.0))

An existing httpx.Client or httpx.AsyncClient can also be supplied with set_httpx_client or set_async_httpx_client. Doing so overrides the generated client configuration, so the HTTPX instance must define its own base URL and other required settings.

Development

Install Hatch and run the unit tests:

hatch run test:test

Other useful checks are:

hatch run test:cov
hatch run types:check
hatch run dev:check
hatch run dev:lint

License