> ## Documentation Index
> Fetch the complete documentation index at: https://polos.dev/docs/llms.txt
> Use this file to discover all available pages before exploring further.

# Event-Triggered Workflows

> Trigger workflows automatically from events

Workflows that run automatically when events are published.

## Trigger on event

<CodeGroup>
  ```python Python theme={null}
  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"}
  ```

  ```typescript TypeScript theme={null}
  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' };
    },
  );
  ```
</CodeGroup>

## Batch events

<CodeGroup>
  ```python Python theme={null}
  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)}
  ```

  ```typescript TypeScript theme={null}
  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 };
    },
  );
  ```
</CodeGroup>

## Publish events

<CodeGroup>
  ```python Python theme={null}
  @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}
  ```

  ```typescript TypeScript theme={null}
  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 };
    },
  );
  ```
</CodeGroup>

## Wait for events

<CodeGroup>
  ```python Python theme={null}
  @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}
  ```

  ```typescript TypeScript theme={null}
  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 };
    },
  );
  ```
</CodeGroup>

## Run it

<CodeGroup>
  ```bash Python theme={null}
  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
  ```

  ```bash TypeScript theme={null}
  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
  ```
</CodeGroup>

Open [http://localhost:5173](http://localhost:5173) to view your agents and workflows, run them from the UI, and see execution traces.

[Python example on GitHub](https://github.com/polos-dev/polos/tree/main/python-examples/15-event-triggered) | [TypeScript example on GitHub](https://github.com/polos-dev/polos/tree/main/typescript-examples/15-event-triggered)