Event-triggered workflows

Event-triggered workflows execute automatically when events are published to specific topics. Perfect for building reactive systems, webhooks, and event-driven architectures.

Defining event-triggered workflows

Use trigger_on_event to specify which topic should trigger the workflow:

from polos import workflow, WorkflowContext, EventPayload

@workflow(
    id="user-signup-handler",
    trigger_on_event="user/signup"
)
async def user_signup_handler(ctx: WorkflowContext, payload: EventPayload):
    # Triggered when event published to "user.signup" topic
    user_id = payload.data["user_id"]
    email = payload.data["email"]

    # Send welcome email
    await ctx.step.run("send_welcome", send_welcome_email, user_id, email)

    # Create sample data
    await ctx.step.run("setup_account", create_sample_data, user_id)

    return {"status": "onboarded", "user_id": user_id}

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

const userSignupHandler = defineWorkflow<EventPayload, void, { status: string; userId: string }>(
  {
    id: 'user-signup-handler',
    triggerOnEvent: 'user/signup',
  },
  async (ctx, payload) => {
    // Triggered when event published to "user.signup" topic
    const userId = payload.data['user_id'];
    const email = payload.data['email'];

    // Send welcome email
    await ctx.step.run('send_welcome', () => sendWelcomeEmail(userId, email));

    // Create sample data
    await ctx.step.run('setup_account', () => createSampleData(userId));

    return { status: 'onboarded', userId };
  }
);

Event payload structure

Event-triggered workflows receive the event in their payload:

@workflow(
    id="notification-handler",
    trigger_on_event="notifications/new"
)
async def notification_handler(ctx: WorkflowContext, payload: EventPayload):
    # Single event
    print(f"Event ID: {payload.id}")
    print(f"Sequence ID: {payload.sequence_id}")
    print(f"Topic: {payload.topic}")
    print(f"Event Type: {payload.event_type}")
    print(f"Data: {payload.data}")  # This is a dict
    print(f"Created At: {payload.created_at}")

    # Process the event
    notification_data = payload.data
    await ctx.step.run("send", send_notification, notification_data)

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

const notificationHandler = defineWorkflow<EventPayload, void, void>(
  {
    id: 'notification-handler',
    triggerOnEvent: 'notifications/new',
  },
  async (ctx, payload) => {
    // Single event
    console.log(`Event ID: ${payload.id}`);
    console.log(`Sequence ID: ${payload.sequenceId}`);
    console.log(`Topic: ${payload.topic}`);
    console.log(`Event Type: ${payload.eventType}`);
    console.log(`Data: ${JSON.stringify(payload.data)}`);
    console.log(`Created At: ${payload.createdAt}`);

    // Process the event
    const notificationData = payload.data;
    await ctx.step.run('send', () => sendNotification(notificationData));
  }
);

Event structure:

{
  "id": "evt_123abc",
  "sequence_id": 456
  "topic": "notifications.new",
  "event_type": "notification_created",
  "data": {
    "user_id": "user_456",
    "message": "You have a new message"
  },
  "created_at": "2025-01-28T10:30:00Z"
}

Publishing events

Publish events from workflows or external systems:

From within a workflow

@workflow
async def create_user(ctx: WorkflowContext, input: CreateUserInput):
    # Create user
    user = await ctx.step.run("create", create_user_record, input)

    # Publish event (triggers event-triggered workflows)
    await ctx.step.publish_event(
        "publish_signup",
        topic="user/signup",
        data={
            "user_id": user.id,
            "email": user.email,
            "name": user.name
        },
        event_type="user_created"
    )

    return {"user_id": user.id}

import { defineWorkflow, WorkflowContext } from '@polos/sdk';

interface CreateUserInput {
  email: string;
  name: string;
}

const createUser = defineWorkflow<CreateUserInput, void, { userId: string }>(
  { id: 'create-user' },
  async (ctx, input) => {
    // Create user
    const user = await ctx.step.run('create', () => createUserRecord(input));

    // Publish event (triggers event-triggered workflows)
    await ctx.step.publishEvent('publish_signup', {
      topic: 'user/signup',
      data: {
        user_id: user.id,
        email: user.email,
        name: user.name,
      },
      type: 'user_created',
    });

    return { userId: user.id };
  }
);

From external systems (API)

import httpx

async def publish_event(topic: str, data: dict):
    """Publish an event via API."""
    async with httpx.AsyncClient() as client:
        response = await client.post(
            "https://api.polos.ai/api/v1/events/publish",
            headers={
                "Authorization": "Bearer YOUR_API_KEY",
                "Content-Type": "application/json"
            },
            json={
                "topic": topic,
                "events": [{
                    "data": data,
                    "event_type": "custom_event"
                }]
            }
        )
        response.raise_for_status()

# Trigger workflows listening to "order.completed"
await publish_event(
    "order.completed",
    {"order_id": "ord_123", "total": 99.99}
)

async function publishEvent(topic: string, data: Record<string, unknown>) {
  /** Publish an event via API. */
  const response = await fetch('https://api.polos.ai/api/v1/events/publish', {
    method: 'POST',
    headers: {
      Authorization: 'Bearer YOUR_API_KEY',
      'Content-Type': 'application/json',
    },
    body: JSON.stringify({
      topic,
      events: [
        {
          data,
          event_type: 'custom_event',
        },
      ],
    }),
  });

  if (!response.ok) {
    throw new Error(`Request failed: ${response.status}`);
  }
}

// Trigger workflows listening to "order.completed"
await publishEvent('order.completed', { order_id: 'ord_123', total: 99.99 });

Batch events for processing

Process multiple events together for efficiency:

@workflow(
    id="batch-processor",
    trigger_on_event="analytics/events",
    batch_size=10,              # Process up to 10 events
    batch_timeout_seconds=30    # Or wait max 30 seconds
)
async def batch_processor(ctx: WorkflowContext, payload: BatchEventPayload):
    # Multiple events in the payload
    events = payload.events

    print(f"Processing batch of {len(events)} events")

    # Extract all event data
    analytics_data = [event.data for event in events]

    # Process batch
    await ctx.step.run("batch_insert", insert_analytics, analytics_data)

    return {"processed": len(events)}

import { defineWorkflow, WorkflowContext, BatchEventPayload } from '@polos/sdk';

const batchProcessor = defineWorkflow<BatchEventPayload, void, { processed: number }>(
  {
    id: 'batch-processor',
    triggerOnEvent: 'analytics/events',
    batchSize: 10,              // Process up to 10 events
    batchTimeoutSeconds: 30,    // Or wait max 30 seconds
  },
  async (ctx, payload) => {
    // Multiple events in the payload
    const events = payload.events;

    console.log(`Processing batch of ${events.length} events`);

    // Extract all event data
    const analyticsData = events.map((event) => event.data);

    // Process batch
    await ctx.step.run('batch_insert', () => insertAnalytics(analyticsData));

    return { processed: events.length };
  }
);

Batching behavior:

Workflow triggers when either batch_size is reached or batch_timeout_seconds elapses
If 10 events arrive in 5 seconds → triggers immediately with 10 events
If only 3 events arrive in 30 seconds → triggers with 3 events after timeout

Batch payload structure:

{
  "events": [
    {
      "id": "evt_1",
      "sequence_id": 1001,
      "topic": "analytics.events",
      "data": {"action": "click"},
      "created_at": "2025-01-28T10:30:00Z"
    },
    {
      "id": "evt_2",
      "sequence_id": 1002,
      "topic": "analytics.events",
      "data": {"action": "view"},
      "created_at": "2025-01-28T10:30:05Z"
    }
  ]
}

Multiple handlers for one topic

Multiple workflows can listen to the same topic:

@workflow(
    id="immediate-handler",
    trigger_on_event="order/created"
)
async def immediate_handler(ctx: WorkflowContext, payload: EventPayload):
    """Process each order immediately."""
    order_id = payload.data["order_id"]

    await ctx.step.run("send_confirmation", send_order_confirmation, order_id)
    return {"handler": "immediate"}

@workflow(
    id="batched-handler",
    trigger_on_event="order/created",
    batch_size=5,
    batch_timeout_seconds=60
)
async def batched_handler(ctx: WorkflowContext, payload: BatchEventPayload):
    """Process orders in batches for analytics."""
    events = payload.events

    order_ids = [e.data["order_id"] for e in events]
    await ctx.step.run("batch_analytics", update_analytics, order_ids)

    return {"handler": "batched", "count": len(events)}

import {
  defineWorkflow,
  WorkflowContext,
  EventPayload,
  BatchEventPayload,
} from '@polos/sdk';

const immediateHandler = defineWorkflow<EventPayload, void, { handler: string }>(
  {
    id: 'immediate-handler',
    triggerOnEvent: 'order/created',
  },
  async (ctx, payload) => {
    /** Process each order immediately. */
    const orderId = payload.data['order_id'];

    await ctx.step.run('send_confirmation', () => sendOrderConfirmation(orderId));
    return { handler: 'immediate' };
  }
);

const batchedHandler = defineWorkflow<BatchEventPayload, void, { handler: string; count: number }>(
  {
    id: 'batched-handler',
    triggerOnEvent: 'order/created',
    batchSize: 5,
    batchTimeoutSeconds: 60,
  },
  async (ctx, payload) => {
    /** Process orders in batches for analytics. */
    const events = payload.events;

    const orderIds = events.map((e) => e.data['order_id']);
    await ctx.step.run('batch_analytics', () => updateAnalytics(orderIds));

    return { handler: 'batched', count: events.length };
  }
);

When an event is published:

immediate-handler triggers once per event
batched-handler triggers once per batch (up to 5 events or 60 seconds)

Event filtering

Filter events by event_type:

@workflow(
    id="high-priority-handler",
    trigger_on_event="notifications/system"
)
async def high_priority_handler(ctx: WorkflowContext, payload: PayloadEvent):
    # Only process high priority notifications
    if payload.event_type == "high_priority":
        await ctx.step.run("alert", send_alert, payload.data)
    else:
        print(f"Skipping event type: {payload.event_type}")

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

const highPriorityHandler = defineWorkflow<EventPayload, void, void>(
  {
    id: 'high-priority-handler',
    triggerOnEvent: 'notifications/system',
  },
  async (ctx, payload) => {
    // Only process high priority notifications
    if (payload.eventType === 'high_priority') {
      await ctx.step.run('alert', () => sendAlert(payload.data));
    } else {
      console.log(`Skipping event type: ${payload.eventType}`);
    }
  }
);

Alternative approach: Use multiple topics

# Publish to specific topics
await ctx.step.publish_event(
    "publish",
    topic="notifications/high_priority",  # Specific topic
    data=notification_data
)

@workflow(
    id="high-priority-handler",
    trigger_on_event="notifications/high_priority"  # Only high priority
)
async def high_priority_handler(ctx: WorkflowContext, payload: EventPayload):
    # All events on this topic are high priority
    await ctx.step.run("alert", send_alert, payload.data)

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

// Publish to specific topics
await ctx.step.publishEvent('publish', {
  topic: 'notifications/high_priority', // Specific topic
  data: notificationData,
});

const highPriorityHandler = defineWorkflow<EventPayload, void, void>(
  {
    id: 'high-priority-handler',
    triggerOnEvent: 'notifications/high_priority', // Only high priority
  },
  async (ctx, payload) => {
    // All events on this topic are high priority
    await ctx.step.run('alert', () => sendAlert(payload.data));
  }
);

Event topic patterns

Use topic hierarchies for organization:

# User events
"user/signup"
"user/login"
"user/deleted"

# Order events
"order/created"
"order/completed"
"order/cancelled"

# System events
"system/error"
"system/warning"
"system/maintenance"

Key takeaways

Event-triggered workflows execute automatically when events are published
Use trigger_on_event to specify the topic
Publish events from workflows or external systems via API
Batch events for efficiency with batch_size and batch_timeout_seconds
Multiple handlers can listen to the same topic

Introduction

Getting Started

Fundamentals

Agents

Workflows

Observability

Guides and Examples

Community

Defining event-triggered workflows

Event payload structure

Publishing events

From within a workflow

From external systems (API)

Batch events for processing

Multiple handlers for one topic

Event filtering

Event topic patterns

Key takeaways

​Defining event-triggered workflows

​Event payload structure

​Publishing events

​From within a workflow

​From external systems (API)

​Batch events for processing

​Multiple handlers for one topic

​Event filtering

​Event topic patterns

​Key takeaways

Defining event-triggered workflows

Event payload structure

Publishing events

From within a workflow

From external systems (API)

Batch events for processing

Multiple handlers for one topic

Event filtering

Event topic patterns

Key takeaways