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

# Building Workflows

> Create workflows with composable step types.

To build a workflow, create agents and compose them into steps. Steps can be nested freely.

```python pipeline.py theme={null}
import asyncio
from definable.agent import Agent
from definable.agent.workflow import Workflow, Step, Parallel

researcher = Agent(model="gpt-4o", instructions="You are a research specialist.")
analyst = Agent(model="gpt-4o", instructions="You are a business analyst.")
writer = Agent(model="gpt-4o", instructions="You are a technical writer.")

workflow = Workflow(
    name="research-pipeline",
    steps=[
        Parallel(name="research", steps=[
            Step(name="web-research", agent=researcher),
            Step(name="analysis", agent=analyst),
        ]),
        Step(name="writer", agent=writer),
    ],
)

async def main():
    result = await workflow.arun("Write about the state of AI agents in 2026.")
    print(result.content)

asyncio.run(main())
```

## Run Your Workflow

<Steps>
  <Step title="Set up your environment">
    <Snippet file="setup-venv.mdx" />
  </Step>

  <Step title="Install Definable">
    <Snippet file="install-definable.mdx" />
  </Step>

  <Step title="Export your API key">
    <Snippet file="export-openai-key.mdx" />
  </Step>

  <Step title="Run">
    ```bash theme={null}
    python pipeline.py
    ```
  </Step>
</Steps>

## Step Executors

Each `Step` wraps exactly one executor. Set `agent=`, `team=`, or `executor=`.

```python theme={null}
# Agent step
Step(name="researcher", agent=my_agent)

# Team step
Step(name="analysis-team", team=my_team)

# Callable step (sync or async)
async def process(step_input):
    data = step_input.get_last_step_content()
    return f"Processed: {data}"

Step(name="processor", executor=process)
```

## Step Parameters

<ParamField path="name" type="str" required>
  Step name. Used to access results via `result.get_step_output(name)`.
</ParamField>

<ParamField path="agent" type="Agent">
  Agent to execute. Mutually exclusive with `team` and `executor`.
</ParamField>

<ParamField path="team" type="Team">
  Team to execute. Mutually exclusive with `agent` and `executor`.
</ParamField>

<ParamField path="executor" type="Callable">
  Custom function `(StepInput) -> str | RunOutput`. Mutually exclusive with `agent` and `team`.
</ParamField>

<ParamField path="timeout" type="float">
  Maximum execution time in seconds.
</ParamField>

<ParamField path="retries" type="int" default="0">
  Number of retry attempts on failure.
</ParamField>

<ParamField path="input_builder" type="Callable[[StepInput], str]">
  Custom function to build the prompt from step context.
</ParamField>

## Composing Steps

All step types are composable. Nest them freely:

```python theme={null}
from definable.agent.workflow import Workflow, Step, Parallel, Condition, Loop

workflow = Workflow(
    name="full-pipeline",
    steps=[
        Parallel(name="research", steps=[
            Step(name="web", agent=web_researcher),
            Step(name="papers", agent=paper_reader),
        ]),
        Step(name="writer", agent=writer),
        Condition(
            name="quality-check",
            condition=lambda ctx: "APPROVED" in (ctx.get_last_step_content() or ""),
            true_steps=Step(name="publish", agent=publisher),
            false_steps=Loop(
                name="revise",
                steps=[
                    Step(name="rewrite", agent=writer),
                    Step(name="review", agent=reviewer),
                ],
                end_condition=lambda outputs: any("APPROVED" in (o.content or "") for o in outputs),
                max_iterations=3,
            ),
        ),
    ],
)
```

## Next Steps

| Task                            | Guide                                                  |
| ------------------------------- | ------------------------------------------------------ |
| Execute and access step results | [Running workflows](/workflows/running-workflows)      |
| Sequential pipelines            | [Sequential pattern](/workflows/patterns/sequential)   |
| Concurrent execution            | [Parallel pattern](/workflows/patterns/parallel)       |
| If/else branching               | [Conditional pattern](/workflows/patterns/conditional) |
| Iterative refinement            | [Loop pattern](/workflows/patterns/loop)               |
| Dynamic routing                 | [Router pattern](/workflows/patterns/router)           |