Building Workflows

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

pipeline.py

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

Set up your environment

python3 -m venv .venv
source .venv/bin/activate

python -m venv .venv
.venv\Scripts\activate

uv venv --python 3.12
source .venv/bin/activate

Install Definable

pip install definable

uv pip install definable

poetry add definable

Export your API key

export OPENAI_API_KEY=sk-***

setx OPENAI_API_KEY sk-***

Run

python pipeline.py

Step Executors

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

# 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

str

required

Step name. Used to access results via result.get_step_output(name).

Agent

Agent to execute. Mutually exclusive with team and executor.

Team

Team to execute. Mutually exclusive with agent and executor.

Callable

Custom function (StepInput) -> str | RunOutput. Mutually exclusive with agent and team.

float

Maximum execution time in seconds.

int

default:"0"

Number of retry attempts on failure.

Callable[[StepInput], str]

Custom function to build the prompt from step context.

Composing Steps

All step types are composable. Nest them freely:

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
Sequential pipelines	Sequential pattern
Concurrent execution	Parallel pattern
If/else branching	Conditional pattern
Iterative refinement	Loop pattern
Dynamic routing	Router pattern

Get Started

Basics

Capabilities

Advanced

Run Your Workflow

Step Executors

Step Parameters

Composing Steps

Next Steps

​Run Your Workflow

​Step Executors

​Step Parameters

​Composing Steps

​Next Steps

Run Your Workflow

Step Executors

Step Parameters

Composing Steps

Next Steps