> ## Documentation Index
> Fetch the complete documentation index at: https://docs.timbal.ai/llms.txt
> Use this file to discover all available pages before exploring further.
# Overview
> Build multi-step AI pipelines with explicit control flow
Workflows orchestrate **steps** in a DAG: you define what runs, Timbal infers dependencies and runs independent steps in parallel. Use them when the path is predictable (ETL, gated deploys, fan-out/fan-in, LLM routing) rather than leaving every decision to the model.
## Workflow vs Agent
| | **Workflow** | **Agent** |
| ----------- | -------------------------------------------- | ---------------------------------- |
| Control | You define steps and order | The model decides what to do |
| Best for | Pipelines, batch jobs, deterministic routing | Open-ended reasoning, tool picking |
| Parallelism | Built-in across independent steps | Sequential tool loop |
| Composition | Nest workflows, mix agents + functions | Tools and sub-agents |
They share the same interface: call a Runnable to get an event stream, or chain `.collect()` for the final `OutputEvent`. A common pattern is a **workflow that routes to agents** (classify → specialist agent). See [Agents & LLMs](/workflows/agents-and-llms).
## Reading guide
1. **[Control flow](/workflows/control-flow)** — parallel execution, `step_span()`, `depends_on`, wiring data between steps
2. **[Branching](/workflows/branching)** — conditional steps with `when`
3. **[Errors & failure](/workflows/errors-and-failure)** — what happens when a step fails or is skipped
4. **[Agents & LLMs](/workflows/agents-and-llms)** — agents as steps, LLM routing patterns
5. **[Human in the loop](/workflows/human-in-the-loop)** — approval gates on workflow steps
6. **Examples** below — copy-paste pipelines for common shapes
## Quickstart
```python theme={"dark"}
import asyncio
from timbal import Workflow
from timbal.state import get_run_context
def celsius_to_fahrenheit(celsius: float) -> float:
return (celsius * 9 / 5) + 32
def format_result(temperature: float) -> str:
return f"Temperature: {temperature}°F"
workflow = (
Workflow(name="temperature_converter")
.step(celsius_to_fahrenheit, celsius=35)
.step(
format_result,
temperature=lambda: get_run_context().step_span("celsius_to_fahrenheit").output,
)
)
async def main():
result = await workflow().collect()
print(result.output)
asyncio.run(main())
```
Functions used as steps must accept and return Pydantic-serializable types (`str`, `int`, `float`, `bool`, `dict`, `list`, `BaseModel`). Custom classes that aren't Pydantic models cannot be passed between steps.
## Adding steps
Use `.step()` to add a Runnable (function, Tool, Agent, or nested Workflow):
```python theme={"dark"}
from timbal import Tool, Workflow
# Same handler twice? Give each Tool a unique name.
threshold_high = Tool(name="threshold_high", handler=check_threshold)
threshold_low = Tool(name="threshold_low", handler=check_threshold)
workflow = (
Workflow(name="monitoring")
.step(threshold_high, value=80, limit=100)
.step(threshold_low, value=80, limit=50)
)
```
Each step name must be unique within the workflow. Timbal detects cycles when you link steps and raises if the graph would loop.
## Workflow inputs
Keyword arguments passed to the workflow are forwarded to steps. Parameters without a default on a step become **workflow-level inputs** (merged into the workflow's params schema):
```python theme={"dark"}
workflow = (
Workflow(name="scraper")
.step(fetch) # fetch(url: str) → url is a workflow input
.step(process, raw=lambda: get_run_context().step_span("fetch").output)
)
result = await workflow(url="https://example.com").collect()
```
Fixed values passed in `.step(name, key=value)` are defaults for that step only and are not exposed as workflow inputs.
## Output
The workflow returns the **last executed step's output**. If you need multiple step results, add a final merge step:
```python theme={"dark"}
workflow = (
Workflow(name="pipeline")
.step(fetch_users)
.step(fetch_orders)
.step(
build_report,
users=lambda: get_run_context().step_span("fetch_users").output,
orders=lambda: get_run_context().step_span("fetch_orders").output,
)
)
# build_report's return value is the workflow output
```
## Running
```python theme={"dark"}
# Collect final OutputEvent
result = await workflow(url="https://example.com").collect()
print(result.output)
print(result.status.code) # "success" | "error" | "cancelled"
print(result.usage)
# Or stream events from every step
async for event in workflow(url="https://example.com"):
print(event)
```
## Composition
Nest workflows as steps. The inner workflow's final output becomes that step's output:
```python theme={"dark"}
data_pipeline = (
Workflow(name="data_pipeline")
.step(fetch_data, source="api")
.step(clean_data, raw=lambda: get_run_context().step_span("fetch_data").output)
)
report_pipeline = (
Workflow(name="report_pipeline")
.step(data_pipeline)
.step(generate_report, data=lambda: get_run_context().step_span("data_pipeline").output)
)
```
See [Workflow Composition](/examples/workflows/workflow-composition) for a full example.
## Examples
Chain steps with data passing between them
Fetch from multiple sources concurrently, then merge
Route to different handlers based on validation
Nest an inner workflow as a step