Steps

Steps are the fundamental unit of durability in Polos. Each step’s execution is persisted, enabling workflows to resume from the last completed step after failures.

What is a step?

A step is a durable operation within a workflow. When a step completes, its output is saved to the database. If the workflow crashes, completed steps return their cached results on replay - no re-execution, no duplicate side effects.

from polos import workflow, WorkflowContext

@workflow
async def example_workflow(ctx: WorkflowContext, input: ExampleInput):
    # Step 1: Fetch data (cached on replay)
    data = await ctx.step.run("fetch", fetch_from_api, input.url)

    # Step 2: Process data (cached on replay)
    processed = await ctx.step.run("process", transform_data, data)

    # Step 3: Save result (executes only once)
    await ctx.step.run("save", save_to_db, processed)

    return {"status": "completed"}

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

const exampleWorkflow = defineWorkflow<ExampleInput, void, { status: string }>(
  { id: 'example-workflow' },
  async (ctx, input) => {
    // Step 1: Fetch data (cached on replay)
    const data = await ctx.step.run('fetch', () => fetchFromApi(input.url));

    // Step 2: Process data (cached on replay)
    const processed = await ctx.step.run('process', () => transformData(data));

    // Step 3: Save result (executes only once)
    await ctx.step.run('save', () => saveToDb(processed));

    return { status: 'completed' };
  }
);

On failure after step 2:

Replay:
  Step 1 ("fetch") → Returns cached result (no API call)
  Step 2 ("process") → Returns cached result (no re-processing)
  Step 3 ("save") → Executes for the first time

Step execution model

Unlike workflows, which are scheduled by the orchestrator and assigned to workers, steps execute directly on the same worker running the workflow. This means:

✅ Lower overhead - No scheduling delay
✅ Same execution context - Access to workflow variables
✅ Durability guaranteed - Output is persisted before continuing

Steps are not queued or orchestrated — they run inline with the workflow code.

Core step methods

`step.run()`

Run any Python function as a durable step:

@workflow
async def process_order(ctx: WorkflowContext, input: OrderInput):
    # Run async function
    order = await ctx.step.run(
        "validate_order",
        validate_order,
        input.order_id
    )

    # Run sync function (automatically wrapped)
    receipt = await ctx.step.run(
        "generate_receipt",
        generate_pdf,  # Sync function
        order
    )

    return receipt

const processOrder = defineWorkflow<OrderInput, void, Receipt>(
  { id: 'process-order' },
  async (ctx, input) => {
    // Run async function
    const order = await ctx.step.run(
      'validate_order',
      () => validateOrder(input.orderId)
    );

    // Run sync function
    const receipt = await ctx.step.run(
      'generate_receipt',
      () => generatePdf(order)
    );

    return receipt;
  }
);

With retry configuration:

result = await ctx.step.run(
    "api_call",
    call_external_api,
    url,
    max_retries=5,
    base_delay=2.0,
    max_delay=30.0
)

const result = await ctx.step.run(
  'api_call',
  () => callExternalApi(url),
  { maxRetries: 5, baseDelay: 2.0, maxDelay: 30.0 }
);

What should be a step:

✅ External API calls
✅ Database operations
✅ File I/O
✅ Non-deterministic operations (random(), datetime.now())
✅ Any operation that might fail

What should NOT be a step:

❌ Pure logic (if statements, loops)
❌ Variable assignments
❌ String manipulation

`step.invoke()`

Start a child workflow without waiting for it to complete:

@workflow
async def parent_workflow(ctx: WorkflowContext, input: ParentInput):
    # Start child workflow
    handle = await ctx.step.invoke(
        "start_child",
        child_workflow,
        {"data": input.data}
    )

    # Continue immediately (don't wait for child to complete)
    print(f"Started workflow: {handle.id}")
    return {"child_id": handle.id}

const parentWorkflow = defineWorkflow<ParentInput, void, { childId: string }>(
  { id: 'parent-workflow' },
  async (ctx, input) => {
    // Start child workflow
    const handle = await ctx.step.invoke(
      'start_child',
      childWorkflow,
      { data: input.data }
    );

    // Continue immediately (don't wait for child to complete)
    console.log(`Started workflow: ${handle.id}`);
    return { childId: handle.id };
  }
);

`step.invoke_and_wait()`

Start a child workflow and suspend until it completes:

@workflow
async def parent_workflow(ctx: WorkflowContext, input: ParentInput):
    # Call child and wait for result
    result = await ctx.step.invoke_and_wait(
        "call_child",
        child_workflow,
        {"data": input.data}
    )

    # Parent resumes here with child's result
    return {"child_result": result}

const parentWorkflow = defineWorkflow<ParentInput, void, { childResult: unknown }>(
  { id: 'parent-workflow' },
  async (ctx, input) => {
    // Call child and wait for result
    const result = await ctx.step.invokeAndWait(
      'call_child',
      childWorkflow,
      { data: input.data }
    );

    // Parent resumes here with child's result
    return { childResult: result };
  }
);

Worker behavior: The worker suspends the parent workflow (no compute consumed) until the child completes.

`step.batch_invoke()`

Start multiple workflows in parallel without waiting:

from polos import BatchWorkflowInput

@workflow
async def parallel_tasks(ctx: WorkflowContext, input: ParallelInput):
    # Start multiple workflows
    handles = await ctx.step.batch_invoke(
        "start_batch", [
            BatchWorkflowInput(id="task_workflow", payload={"task": "A"}),
            BatchWorkflowInput(id="task_workflow", payload={"task": "B"}),
            BatchWorkflowInput(id="log_workflow", payload={"input": input})
        ]
    )

    return {"started": len(handles)}

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

const parallelTasks = defineWorkflow<ParallelInput, void, { started: number }>(
  { id: 'parallel-tasks' },
  async (ctx, input) => {
    // Start multiple workflows
    const handles = await ctx.step.batchInvoke(
      'start_batch',
      [
        { workflow: 'task_workflow', payload: { task: 'A' } },
        { workflow: 'task_workflow', payload: { task: 'B' } },
        { workflow: 'log_workflow', payload: { input } },
      ]
    );

    return { started: handles.length };
  }
);

`step.batch_invoke_and_wait()`

Start multiple workflows in parallel and wait for all to complete:

@workflow
async def parallel_and_wait(ctx: WorkflowContext, input: ParallelInput):
    # Start batch and wait for all
    results = await ctx.step.batch_invoke_and_wait(
        "batch_process", [
            BatchWorkflowInput(id="processor", payload={"item": "A"}),
            BatchWorkflowInput(id="processor", payload={"item": "B"}),
            BatchWorkflowInput(id="log_workflow", payload={"input": input})
        ]
    )

    # Worker suspends until all complete
    for result in results:
        print(f"Result: {result.result}")

    return results

const parallelAndWait = defineWorkflow<ParallelInput, void, unknown[]>(
  { id: 'parallel-and-wait' },
  async (ctx, input) => {
    // Start batch and wait for all
    const results = await ctx.step.batchInvokeAndWait(
      'batch_process',
      [
        { workflow: 'processor', payload: { item: 'A' } },
        { workflow: 'processor', payload: { item: 'B' } },
        { workflow: 'log_workflow', payload: { input } },
      ]
    );

    // Worker suspends until all complete
    for (const result of results) {
      console.log(`Result: ${result.result}`);
    }

    return results;
  }
);

`step.agent_invoke()`

Start an agent execution without waiting for it to complete:

@workflow
async def parent_workflow(ctx: WorkflowContext, input: ParentInput):
    # Start agent execution
    handle = await ctx.step.agent_invoke(
        "research_agent",
        research_agent.with_input(input.query)
    )

    # Continue immediately (don't wait)
    print(f"Started agent execution: {handle.id}")
    return {"agent_execution_id": handle.id}

const parentWorkflow = defineWorkflow<ParentInput, void, { agentExecutionId: string }>(
  { id: 'parent-workflow' },
  async (ctx, input) => {
    // Start agent execution
    const handle = await ctx.step.agentInvoke(
      'research_agent',
      researchAgent.withInput(input.query)
    );

    // Continue immediately (don't wait)
    console.log(`Started agent execution: ${handle.id}`);
    return { agentExecutionId: handle.id };
  }
);

`step.agent_invoke_and_wait()`

Start an agent execution and suspend until it completes:

@workflow
async def review_workflow(ctx: WorkflowContext, input: ReviewInput):
    # Call agent and wait for result
    response = await ctx.step.agent_invoke_and_wait(
        "grammar_review_agent_invoke",
        grammar_review_agent.with_input(input.text)
    )

    # Parent resumes here with agent's response
    return {"review": response.result}

const reviewWorkflow = defineWorkflow<ReviewInput, void, { review: unknown }>(
  { id: 'review-workflow' },
  async (ctx, input) => {
    // Call agent and wait for result
    const response = await ctx.step.agentInvokeAndWait(
      'grammar_review_agent_invoke',
      grammarReviewAgent.withInput(input.text)
    );

    // Parent resumes here with agent's response
    return { review: response.result };
  }
);

Worker behavior: The worker suspends the parent workflow (no compute consumed) until the agent completes.

`step.batch_agent_invoke()`

Start multiple agent executions in parallel without waiting:

@workflow
async def parallel_agents(ctx: WorkflowContext, input: ParallelInput):
    # Start multiple agents
    handles = await ctx.step.batch_agent_invoke(
        "start_batch_agents",
        [
            grammar_review_agent.with_input(input.text),
            tone_review_agent.with_input(input.text),
            accuracy_review_agent.with_input(input.text),
        ]
    )

    return {"started": len(handles)}

const parallelAgents = defineWorkflow<ParallelInput, void, { started: number }>(
  { id: 'parallel-agents' },
  async (ctx, input) => {
    // Start multiple agents
    const handles = await ctx.step.batchAgentInvoke(
      'start_batch_agents',
      [
        grammarReviewAgent.withInput(input.text),
        toneReviewAgent.withInput(input.text),
        accuracyReviewAgent.withInput(input.text),
      ]
    );

    return { started: handles.length };
  }
);

`step.batch_agent_invoke_and_wait()`

Start multiple agent executions in parallel and wait for all to complete:

@workflow
async def parallel_reviews(ctx: WorkflowContext, input: ReviewInput):
    # Run three review agents in parallel
    results = await ctx.step.batch_agent_invoke_and_wait(
        "batch_invoke_grammar_tone_accuracy_reviews",
        [
            grammar_review_agent.with_input(input.text),
            tone_consistency_review_agent.with_input(input.text),
            accuracy_review_agent.with_input(input.text),
        ]
    )

    # Worker suspends until all agents complete
    for result in results:
        print(f"Review result: {result.result}")

    return results

const parallelReviews = defineWorkflow<ReviewInput, void, unknown[]>(
  { id: 'parallel-reviews' },
  async (ctx, input) => {
    // Run three review agents in parallel
    const results = await ctx.step.batchAgentInvokeAndWait(
      'batch_invoke_grammar_tone_accuracy_reviews',
      [
        grammarReviewAgent.withInput(input.text),
        toneConsistencyReviewAgent.withInput(input.text),
        accuracyReviewAgent.withInput(input.text),
      ]
    );

    // Worker suspends until all agents complete
    for (const result of results) {
      console.log(`Review result: ${result.result}`);
    }

    return results;
  }
);

`step.wait_for()`

Pause execution for a duration:

@workflow
async def delayed_workflow(ctx: WorkflowContext, input: dict):
    # Wait 1 hour (worker suspends, no compute cost)
    await ctx.step.wait_for("wait_1_hour", hours=1)

    # Resume after 1 hour
    result = await ctx.step.run("process", process_data, input)
    return result

const delayedWorkflow = defineWorkflow<Record<string, unknown>, void, unknown>(
  { id: 'delayed-workflow' },
  async (ctx, input) => {
    // Wait 1 hour (worker suspends, no compute cost)
    await ctx.step.waitFor('wait_1_hour', { hours: 1 });

    // Resume after 1 hour
    const result = await ctx.step.run('process', () => processData(input));
    return result;
  }
);

Available units: seconds, minutes, hours, days, weeks

`step.wait_until()`

Wait until a specific datetime:

from datetime import datetime, timezone

@workflow
async def scheduled_action(ctx: WorkflowContext, input: dict):
    # Schedule for specific time
    target_time = datetime(2025, 12, 31, 23, 59, 0, tzinfo=timezone.utc)

    await ctx.step.wait_until("wait_until_new_year", target_time)

    # Executes at 23:59 on Dec 31, 2025
    await ctx.step.run("celebrate", send_celebration)

const scheduledAction = defineWorkflow<Record<string, unknown>, void, void>(
  { id: 'scheduled-action' },
  async (ctx, input) => {
    // Schedule for specific time
    const targetTime = new Date('2025-12-31T23:59:00Z');

    await ctx.step.waitUntil('wait_until_new_year', targetTime);

    // Executes at 23:59 on Dec 31, 2025
    await ctx.step.run('celebrate', () => sendCelebration());
  }
);

`step.wait_for_event()`

Wait for an external event:

@workflow
async def approval_workflow(ctx: WorkflowContext, input: dict):
    # Submit for user confirmation
    await ctx.step.run("submit", submit_for_user_confirmation, input)

    # Wait for user.confirmation event (hours or days)
    approval = await ctx.step.wait_for_event(
        "user_confirmation",
        topic="user.confirmation",
        timeout=86400  # 24 hour timeout
    )

    # Resume when event arrives
    if approval.data["approved"]:
        await ctx.step.run("execute", execute_action, input)

const approvalWorkflow = defineWorkflow<Record<string, unknown>, void, void>(
  { id: 'approval-workflow' },
  async (ctx, input) => {
    // Submit for user confirmation
    await ctx.step.run('submit', () => submitForUserConfirmation(input));

    // Wait for user.confirmation event (hours or days)
    const approval = await ctx.step.waitForEvent(
      'user_confirmation',
      { topic: 'user.confirmation', timeout: 86400 }  // 24 hour timeout
    );

    // Resume when event arrives
    if (approval.data.approved) {
      await ctx.step.run('execute', () => executeAction(input));
    }
  }
);

`step.suspend()`

Suspend execution and wait for manual resume. This is syntactic sugar around wait_for_event that’s tailored for agent human-in-the-loop flows.

@workflow
async def manual_approval(ctx: WorkflowContext, input: dict):
    # Prepare action
    action = await ctx.step.run("prepare", prepare_action, input)

    # Suspend with context data
    resume_data = await ctx.step.suspend(
        "approval",
        data={
            "action": action.description,
            "cost": action.estimated_cost
        },
        timeout=3600  # 1 hour timeout
    )

    # Resume with approval decision
    decision = resume_data.get("data", {})
    if decision.get("approved"):
        await ctx.step.run("execute", execute_action, action)

const manualApproval = defineWorkflow<Record<string, unknown>, void, void>(
  { id: 'manual-approval' },
  async (ctx, input) => {
    // Prepare action
    const action = await ctx.step.run('prepare', () => prepareAction(input));

    // Suspend with context data
    const resumeData = await ctx.step.suspend(
      'approval',
      {
        data: {
          action: action.description,
          cost: action.estimatedCost,
        },
        timeout: 3600,  // 1 hour timeout
      }
    );

    // Resume with approval decision
    const decision = resumeData?.data ?? {};
    if (decision.approved) {
      await ctx.step.run('execute', () => executeAction(action));
    }
  }
);

See Human-in-the-Loop for details.

`step.publish_event()`

Publish events durably:

@workflow
async def event_publisher(ctx: WorkflowContext, input: dict):
    # Process data
    result = await ctx.step.run("process", process_data, input)

    # Publish event (guaranteed once)
    await ctx.step.publish_event(
        "notify_complete",
        topic="processing.completed",
        data={"result": result},
        event_type="completion"
    )

const eventPublisher = defineWorkflow<Record<string, unknown>, void, void>(
  { id: 'event-publisher' },
  async (ctx, input) => {
    // Process data
    const result = await ctx.step.run('process', () => processData(input));

    // Publish event (guaranteed once)
    await ctx.step.publishEvent(
      'notify_complete',
      {
        topic: 'processing.completed',
        data: { result },
        type: 'completion',
      }
    );
  }
);

`step.uuid()`

Generate a UUID that persists across replays:

@workflow
async def create_entity(ctx: WorkflowContext, input: dict):
    # Generate stable UUID (same on replay)
    entity_id = await ctx.step.uuid("entity_id")

    # Use the generated UUID in API calls
    await ctx.step.run("create", create_in_api, entity_id, input)

    return {"id": entity_id}

const createEntity = defineWorkflow<Record<string, unknown>, void, { id: string }>(
  { id: 'create-entity' },
  async (ctx, input) => {
    // Generate stable UUID (same on replay)
    const entityId = await ctx.step.uuid('entity_id');

    // Use the generated UUID in API calls
    await ctx.step.run('create', () => createInApi(entityId, input));

    return { id: entityId };
  }
);

Why this matters:

# ❌ BAD: New UUID on each replay
entity_id = str(uuid.uuid4())  # Different every time
await ctx.step.run("create", create_entity, entity_id)

# ✅ GOOD: Same UUID on replay
entity_id = await ctx.step.uuid("entity_id")  # Cached
await ctx.step.run("create", create_entity, entity_id)

// ❌ BAD: New UUID on each replay
const entityId = crypto.randomUUID();  // Different every time
await ctx.step.run('create', () => createEntity(entityId));

// ✅ GOOD: Same UUID on replay
const entityId = await ctx.step.uuid('entity_id');  // Cached
await ctx.step.run('create', () => createEntity(entityId));

`step.now()`

Get current timestamp that persists across replays:

@workflow
async def timestamped_workflow(ctx: WorkflowContext, input: dict):
    # Get stable timestamp (same on replay)
    created_at = await ctx.step.now("created_at")

    # Use in conditional logic
    if created_at % 2 == 0:
        await ctx.step.run("even_path", process_even)
    else:
        await ctx.step.run("odd_path", process_odd)

const timestampedWorkflow = defineWorkflow<Record<string, unknown>, void, void>(
  { id: 'timestamped-workflow' },
  async (ctx, input) => {
    // Get stable timestamp (same on replay)
    const createdAt = await ctx.step.now('created_at');

    // Use in conditional logic
    if (createdAt % 2 === 0) {
      await ctx.step.run('even_path', () => processEven());
    } else {
      await ctx.step.run('odd_path', () => processOdd());
    }
  }
);

`step.trace()`

Add custom spans for observability:

@workflow
async def monitored_workflow(ctx: WorkflowContext, input: dict):
    # Custom span for database operations
    with ctx.step.trace("database_query", {"table": "users", "limit": 100}):
        users = await db.query("SELECT * FROM users LIMIT 100")

    # Custom span for external API
    with ctx.step.trace("external_api", {"endpoint": "/data"}):
        data = await api.fetch("/data")

    return {"users": len(users), "data": data}

const monitoredWorkflow = defineWorkflow<Record<string, unknown>, void, { users: number; data: unknown }>(
  { id: 'monitored-workflow' },
  async (ctx, input) => {
    // Custom span for database operations
    const dbSpan = ctx.step.trace('database_query', { table: 'users', limit: 100 });
    const users = await db.query('SELECT * FROM users LIMIT 100');
    dbSpan.end();

    // Custom span for external API
    const apiSpan = ctx.step.trace('external_api', { endpoint: '/data' });
    const data = await api.fetch('/data');
    apiSpan.end();

    return { users: users.length, data };
  }
);

See Tracing for details.

Step keys must be unique

Each step needs a unique step_key per execution:

@workflow
async def process_items(ctx: WorkflowContext, input: dict):
    # ❌ BAD: Same key in loop
    for item in input.items:
        await ctx.step.run("process", process_item, item)  # Collision!

    # ✅ GOOD: Unique key per iteration
    for i, item in enumerate(input.items):
        await ctx.step.run(f"process_{i}", process_item, item)

const processItems = defineWorkflow<{ items: unknown[] }, void, void>(
  { id: 'process-items' },
  async (ctx, input) => {
    // ❌ BAD: Same key in loop
    for (const item of input.items) {
      await ctx.step.run('process', () => processItem(item));  // Collision!
    }

    // ✅ GOOD: Unique key per iteration
    for (let i = 0; i < input.items.length; i++) {
      await ctx.step.run(`process_${i}`, () => processItem(input.items[i]));
    }
  }
);

Agents and steps

In Polos, agents are special workflows. Everything an agent does is broken into steps: Agent actions as steps:

✅ LLM calls - Each call is a step
✅ Guardrail evaluation - Each guardrail is a step
✅ Stop condition checks - Each check is a step
✅ Tool calls - Tools are subworkflows (not steps)

This means:

Agent failures resume from the last completed step (not the beginning)
Tool calls are durable subworkflows (not lost on failure)
Every agent action is observable via step events

Introduction

Getting Started

Fundamentals

Agents

Workflows

Observability

Guides and Examples

Community

What is a step?

Step execution model

Core step methods

`step.run()`

`step.invoke()`

`step.invoke_and_wait()`

`step.batch_invoke()`

`step.batch_invoke_and_wait()`

`step.agent_invoke()`

`step.agent_invoke_and_wait()`

`step.batch_agent_invoke()`

`step.batch_agent_invoke_and_wait()`

`step.wait_for()`

`step.wait_until()`

`step.wait_for_event()`

`step.suspend()`

`step.publish_event()`

`step.uuid()`

`step.now()`

`step.trace()`

Step keys must be unique

Agents and steps

​What is a step?

​Step execution model

​Core step methods

​step.run()

​step.invoke()

​step.invoke_and_wait()

​step.batch_invoke()

​step.batch_invoke_and_wait()

​step.agent_invoke()

​step.agent_invoke_and_wait()

​step.batch_agent_invoke()

​step.batch_agent_invoke_and_wait()

​step.wait_for()

​step.wait_until()

​step.wait_for_event()

​step.suspend()

​step.publish_event()

​step.uuid()

​step.now()

​step.trace()

​Step keys must be unique

​Agents and steps

What is a step?

Step execution model

Core step methods

`step.run()`

`step.invoke()`

`step.invoke_and_wait()`

`step.batch_invoke()`

`step.batch_invoke_and_wait()`

`step.agent_invoke()`

`step.agent_invoke_and_wait()`

`step.batch_agent_invoke()`

`step.batch_agent_invoke_and_wait()`

`step.wait_for()`

`step.wait_until()`

`step.wait_for_event()`

`step.suspend()`

`step.publish_event()`

`step.uuid()`

`step.now()`

`step.trace()`

Step keys must be unique

Agents and steps