eventpilot 0.0.1

EventPilot Python SDK

EventPilot Python SDK

Durable, class-based workflow handlers for Python. Production-ready primitives for retries, rate limiting, cancellation, signals, queries, versioning, heartbeats, scheduling, and deterministic replay.

Install:

pip install eventpilot              # core
pip install 'eventpilot[typed]'     # + Pydantic v2 payload validation
pip install 'eventpilot[otel]'      # + OpenTelemetry span emission

Core concepts

• Handler — a class decorated with @handler(name) and an async def run(self, state) entrypoint. One class = one workflow type.
• Step — an async def method decorated with @step(name). Steps are the durable unit: each one checkpoints its result before run() continues.
• State — the WorkflowState passed to run() and each step. Carries the trigger payload, accumulated context, and durability helpers.
• Closures are never required. All data flows through self and state.

Minimal example

import asyncio
from eventpilot import Worker, handler, step, StepResult, WorkflowState

@handler("order.processing", version="1.0.0", max_retries=5)
class OrderProcessing:
    async def run(self, state: WorkflowState):
        await self.next_step(self.validate)
        await self.next_step(self.charge)

    @step("validate")
    async def validate(self, state: WorkflowState) -> StepResult:
        payload = state.get_payload_map()
        return StepResult(data={"order_id": payload["order_id"]})

    @step("charge", rate_limit_scope="gateway", rate_limit_rate=100)
    async def charge(self, state: WorkflowState) -> StepResult:
        charge_id = await self.side_effect(
            "gateway-charge", lambda: gateway.charge(state.get_string("order_id"))
        )
        return StepResult(data={"charge_id": charge_id})

async def main():
    w = Worker("localhost:50052", http_addr="http://localhost:8080")
    w.register(OrderProcessing)
    await w.start()

asyncio.run(main())

Feature matrix

Feature	API	Notes
Retries	@handler(max_retries=...) / @step(retry_policy=...)	exp/linear/fibonacci/fixed backoff, jitter, non-retryable codes
Rate limiting	@step(rate_limit_scope=, rate_limit_rate=)	worker-side semaphore
Cancellation	client.cancel_workflow(wf_id, reason=)	cascade to children via cancel_policy="CASCADE"
Durable sleep	await self.sleep(name, duration_ms)	survives restart
Sleep until	await self.sleep_until(name, deadline)	absolute UTC
Wait for event	await self.wait_for_event(name, event=, timeout_ms=, correlation_key=)	with timeout
Wait for signal	await self.wait_for_signal(name, signal=, timeout_ms=)	sugar over wait_for_event
Send signal	pub.send_signal(wf_id, name, payload)	wakes wait_for_signal / @signal
Child workflow (sync)	await self.child_workflow(name, event_name=, payload=, timeout_ms=)	returns ChildWorkflowResult
Child workflow (async)	await self.child_workflow_async(name, event_name=, payload=)	fire-and-forget
Idempotency	StepResult(side_effect_key=...)	dedup external side effects across retries
Timeouts	StepTimeouts(schedule_to_start_ms=, start_to_close_ms=, schedule_to_close_ms=, heartbeat_timeout_ms=)	+ on_timeout= callback
Heartbeat	await self.heartbeat(progress=...)	for long-running steps; surfaces as state.last_progress on retry
Determinism	await self.now() / self.random() / self.side_effect(key, fn)	replay-safe time / random / IO
Versioning	await self.get_version("change-id", default=2)	Temporal-style patched branching
Queries	@query(name) method	read-only introspection; dispatched when backend support lands
Typed signals	@signal(name, model=Pydantic) method	declarative handler + payload validation
Scheduling	@scheduled(cron=..., interval_sec=..., ...) class decorator	auto-registers with orchestrator on worker start

Handler authoring: class + methods, no closures

from pydantic import BaseModel
from eventpilot import handler, step, query, signal, scheduled, StepResult, WorkflowState

class OrderPayload(BaseModel):
    order_id: str
    amount: float
    is_vip: bool = False

@scheduled(cron="0 3 * * *", name="nightly-audit")
@handler("order.processing")
class OrderProcessing:

    async def run(self, state: WorkflowState):
        payload = state.parse_payload(OrderPayload)

        await self.next_step(self.validate)

        if await self.get_version("fraud-check", default=1) >= 2:
            await self.next_step(self.fraud_check)

        approved = await self.wait_for_signal("manager-approval", signal="approval", timeout_ms=3_600_000)

        await self.next_step(self.charge)

    @step("validate")
    async def validate(self, state) -> StepResult: ...

    @step("fraud-check")
    async def fraud_check(self, state) -> StepResult: ...

    @step("charge", rate_limit_scope="gateway", rate_limit_rate=100)
    async def charge(self, state) -> StepResult:
        charge_id = await self.side_effect("charge", lambda: gateway.charge(...))
        await self.heartbeat(progress={"stage": "committed"})
        return StepResult(data={"charge_id": charge_id}, side_effect_key="charge")

    @query("get_stage")
    async def get_stage(self, state) -> dict:
        return {"stage": state.context.get("stage", "init")}

    @signal("approval", model=dict)
    async def on_approval(self, state, payload): ...

Observability

from eventpilot import Worker, MetricsRecorder

class PromMetrics:
    def record_step_started(self, handler, step, workflow_id): ...
    def record_step_completed(self, handler, step, workflow_id, duration_ms): ...
    def record_step_failed(self, handler, step, workflow_id, error_code, retryable): ...
    def record_step_deferred(self, handler, step, workflow_id, delay_ms): ...
    def record_workflow_cancelled(self, handler, workflow_id, reason): ...

w = Worker("localhost:50052", metrics_recorder=PromMetrics())

If opentelemetry-api is installed and a tracer provider is configured, every step execution is automatically wrapped in an eventpilot.step.<name> span with eventpilot.{handler,step,workflow_id} attributes.

Structured logs use extra={workflow_id, ep_handler, ep_attempt, ep_step} — compatible with any JSON log aggregator.

Clients

# Sync — safe from scripts / non-async code.
from eventpilot import EventPilotClient
client = EventPilotClient(base_url="http://localhost:8080")
client.publish_event("order.created", {"order_id": "ord_123"})

# Async — non-blocking; use from async workers.
from eventpilot import AsyncEventPilotClient
async with AsyncEventPilotClient(base_url="http://localhost:8080") as c:
    await c.publish_event(...)

Testing

from eventpilot import TestEnv, TestStatus, new_state

env = TestEnv()
env.register(OrderProcessing)

# Whole-workflow trigger
result = await env.trigger("order.processing", {"order_id": "ord_1", "amount": 10, "is_vip": False})
assert result.status == TestStatus.COMPLETED

# Query invocation
state = new_state("order.processing").with_context(stage="charging").build()
response = await env.query("order.processing", "get_stage", state)

# Deliver signals / events
env.signal(state.workflow_id, "approval", {"ok": True})
env.deliver_event(state.workflow_id, "payment.succeeded", {"id": "p_1"})

# Inspect heartbeats
assert env.heartbeats

What changed from v0.1

• @step.compensate was removed until backend dispatch lands — declare rollback logic explicitly or emit a compensation event.
• EventPilotClient is now async-aware. Sync calls stay sync; inside an async loop use AsyncEventPilotClient to avoid blocking.
• Step result context values are strictly JSON-validated — non-JSON types now raise TypeError at checkpoint time instead of silently corrupting.
• Checkpoints now retry with exponential backoff and raise CheckpointWriteError on persistent failure, guaranteeing at-least-once step delivery.

License

Apache 2.0 — part of EventPilot.