> ## 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.

# Workflows

Workflow is the core building block in Polos. A workflow is durable code - it survives failures and resumes exactly where it stopped.

# What is a workflow?

A workflow is a Python function decorated with `@workflow`. It receives a `WorkflowContext` (ctx) and your input data.

<CodeGroup>
  ```python Python theme={null}
  from polos import workflow, WorkflowContext

  @workflow
  async def research(ctx: WorkflowContext, input: ResearchInput):
      # Step 1: Search for information
      results = await ctx.step.run("search_web", search_web, input.topic)

      # Step 2: Analyze results (dynamic logic)
      if len(results) > 10:
          summary = await ctx.step.run("summarize", summarize_results, results)
      else:
          summary = await ctx.step.run("detailed_analysis", detailed_analysis, results)

      # Step 3: Generate report
      report = await ctx.step.run("generate_report", generate_report, summary)

      return report

  async def search_web(query: str):
      response = await http_client.get(f"https://api.search.com?q={query}")
      return response.json()
  ```

  ```typescript TypeScript theme={null}
  import { defineWorkflow } from '@polos/sdk';

  const research = defineWorkflow<ResearchInput, void, Report>(
    { id: 'research' },
    async (ctx, input) => {
      // Step 1: Search for information
      const results = await ctx.step.run('search_web', () => searchWeb(input.topic));

      // Step 2: Analyze results (dynamic logic)
      let summary;
      if (results.length > 10) {
        summary = await ctx.step.run('summarize', () => summarizeResults(results));
      } else {
        summary = await ctx.step.run('detailed_analysis', () => detailedAnalysis(results));
      }

      // Step 3: Generate report
      const report = await ctx.step.run('generate_report', () => generateReport(summary));

      return report;
    }
  );

  async function searchWeb(query: string) {
    const response = await httpClient.get(`https://api.search.com?q=${query}`);
    return response.json();
  }
  ```
</CodeGroup>

Let's unpack what's happening here:

`ctx.step.run("search_web", search_web, input.topic)` tells Polos to execute `search_web` (a regular Python function) as a **durable step**. If the workflow crashes and replays, completed steps are skipped.

# Step: The unit of durability

Steps are how Polos achieves durability. Each step has a unique **step key** (like `"search_web"` or `"generate_report"`). When a workflow replays after a failure, Polos checks which steps already completed and skips them.

**What should be a step?**

* ✅ External API calls (OpenAI, Stripe, databases)
* ✅ Non-deterministic operations (LLM calls, `time.time()`, `random()`)
* ✅ Side effects (sending emails, charging cards, writing to DB)

**Critical rule:** Each step in a workflow must have a unique step key. In loops, use variables:

<CodeGroup>
  ```python Python theme={null}
  @workflow
  async def process_items(ctx: WorkflowContext, input: Input):
      # ✅ GOOD: Unique step key per iteration
      for i, item in enumerate(input.items):
          await ctx.step.run(f"process_{i}", process_item, item)
  ```

  ```typescript TypeScript theme={null}
  const processItems = defineWorkflow<Input, void, void>(
    { id: 'process-items' },
    async (ctx, input) => {
      // ✅ GOOD: Unique step key per iteration
      for (let i = 0; i < input.items.length; i++) {
        await ctx.step.run(`process_${i}`, () => processItem(input.items[i]));
      }
    }
  );
  ```
</CodeGroup>

# Workflow composition

Workflows can invoke other workflows. The parent suspends while children execute - no compute is consumed during waits.

<CodeGroup>
  ```python Python theme={null}
  @workflow
  async def parent_workflow(ctx: WorkflowContext, input: ParentInput):
      # Invoke child and wait for result
      result = await ctx.step.invoke_and_wait(
          "call_child",
          child_workflow,
          ChildInput(message="Hello!")
      )

      # Parent resumes here with result
      return result
  ```

  ```typescript TypeScript theme={null}
  const parentWorkflow = defineWorkflow<ParentInput, void, Result>(
    { id: 'parent-workflow' },
    async (ctx, input) => {
      // Invoke child and wait for result
      const result = await ctx.step.invokeAndWait(
        'call_child',
        childWorkflow,
        { message: 'Hello!' }
      );

      // Parent resumes here with result
      return result;
    }
  );
  ```
</CodeGroup>

# Waiting

Workflows can pause for time or events. Workers suspend during waits consuming no compute.

<CodeGroup>
  ```python Python theme={null}
  @workflow
  async def approval_workflow(ctx: WorkflowContext, input: Input):
      # Request approval
      await ctx.step.run("request", request_approval, input.data)

      # Wait for event (hours or days) - no compute consumed
      decision = await ctx.step.wait_for_event("wait_approval", topic=f"approval/{input.id}")

      # Resumes here when event arrives
      if decision.data["approved"]:
          await ctx.step.run("execute", execute_action, input.data)
      else:
          await ctx.step.run("handle_rejection", handle_rejection, decision.data)
  ```

  ```typescript TypeScript theme={null}
  const approvalWorkflow = defineWorkflow<Input, void, void>(
    { id: 'approval-workflow' },
    async (ctx, input) => {
      // Request approval
      await ctx.step.run('request', () => requestApproval(input.data));

      // Wait for event (hours or days) - no compute consumed
      const decision = await ctx.step.waitForEvent('wait_approval', {
        topic: `approval/${input.id}`,
      });

      // Resumes here when event arrives
      if (decision.data.approved) {
        await ctx.step.run('execute', () => executeAction(input.data));
      } else {
        await ctx.step.run('handle_rejection', () => handleRejection(decision.data));
      }
    }
  );
  ```
</CodeGroup>

# Starting workflows

Workflows can be triggered in three ways:

**1. Direct invocation**

<CodeGroup>
  ```python Python theme={null}
  result = await research.invoke(ResearchInput(topic="AI agents"))
  ```

  ```typescript TypeScript theme={null}
  const result = await research.invoke({ topic: 'AI agents' });
  ```
</CodeGroup>

**2. Event-triggered**

<CodeGroup>
  ```python Python theme={null}
  @workflow(trigger_on_event="user/signup")
  async def onboard_user(ctx: WorkflowContext, event: UserSignUpEvent):
      # runs when event occurs
  ```

  ```typescript TypeScript theme={null}
  const onboardUser = defineWorkflow<UserSignUpEvent, void, void>(
    { id: 'onboard-user', triggerOnEvent: 'user/signup' },
    async (ctx, event) => {
      // runs when event occurs
    }
  );
  ```
</CodeGroup>

**3. Scheduled**

<CodeGroup>
  ```python Python theme={null}
  @workflow(schedule="0 9 * * *")  # Daily at 9am
  async def daily_report(ctx: WorkflowContext, input: SchedulePayload):
      # runs at the defined schedule
  ```

  ```typescript TypeScript theme={null}
  const dailyReport = defineWorkflow<SchedulePayload, void, void>(
    { id: 'daily-report', schedule: '0 9 * * *' }, // Daily at 9am
    async (ctx, input) => {
      // runs at the defined schedule
    }
  );
  ```
</CodeGroup>

# Key takeaways

* Workflows are durable - they survive failures and resume from the last completed step
* Steps are the unit of durability - use them for API calls, side effects, and non-deterministic operations
* Step keys must be unique per execution
* Workers suspend during waits (child workflows, events, timeouts) - no compute consumed

# Learn more

For detailed guides on workflows and steps, see:

* **[Workflows](/docs/workflows/overview)** – Complete workflow reference
* **[Steps](/docs/workflows/steps)** – Built-in step functions