To build a workflow, create agents and compose them into steps. Steps can be nested freely.
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
Export your API key
export OPENAI_API_KEY=sk-***
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
Step name. Used to access results via result.get_step_output(name).
Agent to execute. Mutually exclusive with team and executor.
Team to execute. Mutually exclusive with agent and executor.
Custom function (StepInput) -> str | RunOutput. Mutually exclusive with agent and team.
Maximum execution time in seconds.
Number of retry attempts on failure.
input_builder
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 |
