Event-Triggered Workflows

from polos import workflow, WorkflowContext, EventPayload

@workflow(id="on_order_created", trigger_on_event="orders/created")
async def on_order_created(ctx: WorkflowContext, payload: EventPayload):
    order = payload.data
    await ctx.step.run("validate", validate_order, order)
    await ctx.step.run("reserve", reserve_inventory, order)
    await ctx.step.run("notify", send_confirmation, order)
    return {"order_id": order["order_id"], "status": "processed"}

import { defineWorkflow } from '@polos/sdk';
import type { EventPayload } from '@polos/sdk';

const onOrderCreated = defineWorkflow<EventPayload, unknown, OrderProcessedResult>(
  {
    id: 'on_order_created',
    triggerOnEvent: 'orders/created',
  },
  async (ctx, payload) => {
    const orderData = payload.data;
    const orderId = (orderData['order_id'] as string) ?? 'unknown';

    await ctx.step.run('validate_order', () => ({ valid: true, orderId }));
    await ctx.step.run('reserve_inventory', () => ({ reserved: true }));
    await ctx.step.run('send_confirmation', () => ({ sent: true }));

    return { orderId, status: 'processed' };
  },
);

from polos import BatchEventPayload

@workflow(
    id="batch_processor",
    trigger_on_event="data/updates",
    batch_size=10,
    batch_timeout_seconds=30,
)
async def batch_processor(ctx: WorkflowContext, payload: BatchEventPayload):
    # Receives up to 10 events at once, or triggers after 30s
    for event in payload.events:
        await ctx.step.run(f"process_{event.sequence_id}", process, event.data)
    return {"processed": len(payload.events)}

import type { BatchEventPayload } from '@polos/sdk';

const batchProcessor = defineWorkflow<BatchEventPayload, unknown, BatchProcessResult>(
  {
    id: 'batch_processor',
    triggerOnEvent: 'data/updates',
    batchSize: 10,
    batchTimeoutSeconds: 30,
  },
  async (ctx, payload) => {
    // Receives up to 10 events at once, or triggers after 30s
    const processed: Record<string, unknown>[] = [];
    for (const event of payload.events) {
      const result = await ctx.step.run(
        `process_event_${String(event.sequenceId)}`,
        () => processEventData(event.data),
      );
      processed.push(result);
    }
    return { batchSize: payload.events.length, processed };
  },
);

@workflow(id="order_pipeline")
async def order_pipeline(ctx: WorkflowContext, payload):
    # Publish event to trigger other workflows
    await ctx.step.publish_event(
        "publish_order",
        topic="orders/created",
        data={"order_id": "123", "items": [...]},
    )
    return {"published": True}

const eventPublisher = defineWorkflow<PublishEventPayload, unknown, EventPublishedResult>(
  { id: 'event_publisher' },
  async (ctx, payload) => {
    // Publish event to trigger other workflows
    await ctx.step.publishEvent(
      'publish_event',
      {
        topic: payload.topic,
        data: payload.eventData,
        type: payload.eventType,
      },
    );
    return { published: true, topic: payload.topic, eventType: payload.eventType };
  },
);

@workflow(id="request_response")
async def request_response(ctx: WorkflowContext, payload):
    request_id = await ctx.step.uuid("request_id")

    await ctx.step.publish_event("request", topic=f"requests/{request_id}", data={...})

    # Wait for response event
    response = await ctx.step.wait_for_event(
        "wait_response",
        topic=f"responses/{request_id}",
        timeout=300,
    )
    return {"response": response.data}

const chainWithEvents = defineWorkflow<ChainPayload, unknown, ChainResult>(
  { id: 'chain_with_events' },
  async (ctx, payload) => {
    const requestId = await ctx.step.uuid('request_id');

    await ctx.step.publishEvent('publish_request', {
      topic: `requests/${requestId}`,
      data: { requestId, action: payload.action },
      type: 'request',
    });

    // Wait for response event
    const response = await ctx.step.waitForEvent<EventPayload>(
      'wait_for_response',
      {
        topic: `responses/${requestId}`,
        timeout: 300,
      },
    );
    return { requestId, response: response.data };
  },
);

git clone https://github.com/polos-dev/polos.git
cd polos/python-examples/15-event-triggered
cp .env.example .env  # Add your POLOS_PROJECT_ID and API key
uv sync
python main.py

git clone https://github.com/polos-dev/polos.git
cd polos/typescript-examples/15-event-triggered
cp .env.example .env  # Add your POLOS_PROJECT_ID and API key
npm install
npx tsx main.ts