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

# Workflows and Steps

> Make a subagent follow ordered steps and collect each answer.

A workflow is the ordered list of steps on your agent. A step is one thing a agent must finish, such as getting the caller's name. A step only ends when your tool records its answer, so the model can't skip ahead.

![A workflow left to right: the call starts in the intake subagent, which runs its patient and visit steps in order. on\_exit moves the call to the scheduling subagent, which runs its slot and confirm steps, and then the call ends.](https://mintcdn.com/zeroruntime/pM4kwzbs2SE47G3b/images/workflow-diagram.svg?fit=max\&auto=format\&n=pM4kwzbs2SE47G3b\&q=85\&s=6becd1dc82751b91a2267b688af366f1)

Each arrow between steps is your tool calling `self.complete(...)`. After a subagent's last step, its `on_exit` moves the call on.

## Add Steps

List a subagent's steps in `steps`. Each answer is a dataclass, and a tool ends the step with `self.complete(...)`.

```python theme={null}
from dataclasses import dataclass

from zeroruntime import Step, SubAgent, function_tool


@dataclass
class Patient:
    full_name: str


@dataclass
class Visit:
    kind: str
    details: str


class Intake(SubAgent):
    name = "intake"
    instructions = "You're the clinic's front desk. Ask one thing at a time."
    steps = [
        Step(
            id="patient",
            description="who is calling",
            instructions="Get the patient's full name, then call record_patient.",
            opening="Ask for the patient's full name.",
            result=Patient,
        ),
        Step(
            id="visit",
            description="why they're coming in",
            instructions="Find out why they're coming in, then call record_visit.",
            opening="Ask what they'd like to come in for.",
            result=Visit,
        ),
    ]

    @function_tool
    async def record_patient(self, full_name: str) -> dict:
        """Record the patient's name.

        Args:
            full_name: The patient's full name.
        """
        return await self.complete(Patient(full_name))

    @function_tool
    async def record_visit(self, kind: str, details: str) -> dict:
        """Record why they're coming in.

        Args:
            kind: checkup, illness or specialist.
            details: A few words about it.
        """
        return await self.complete(Visit(kind, details))

    async def on_exit(self):
        return await self.transfer_to("scheduling", reason="intake done")
```

| Option | What it does |
| :- | :- |
| `id` | The step's name. Must be a valid Python name. |
| `description` | A few words on what the step is for. |
| `instructions` | What the model does during the step. |
| `opening` | What the model should say first. It's a guide, not a script. |
| `result` | The dataclass of the answer. |
| `optional` | Lets the caller decline the step. |
| `condition` | Runs the step only when this is true. |
| `max_attempts` | Errors allowed before `on_step_failed`. |

Callers often answer ahead. If your tool completes a later step early, the answer is kept and that step isn't asked.

## Steps Without Subagents

If one agent does the whole call, give the agent its steps with `workflow=`. Its own tools finish them with `self.complete(...)`.

```python theme={null}
class Feedback(Agent):
    def __init__(self) -> None:
        super().__init__(
            instructions="You call customers for feedback on a recent delivery.",
            pipeline=Pipeline(...),
            workflow=[
                Step(id="rating", description="their rating",
                     instructions="Ask for a rating from 1 to 5, then call record_rating.",
                     opening="Ask how they'd rate their delivery, from 1 to 5.", result=Rating),
            ],
        )

    @function_tool
    async def record_rating(self, score: int) -> dict:
        """Record their rating.

        Args:
            score: 1 to 5.
        """
        return await self.complete(Rating(score))
```

An agent takes `workflow=` or `subagents=`, not both. Subagents with no steps don't need a workflow at all: each `transfer_to` is a plain agent handoff, and any runtime runs it.

## Read Answers

Use the step's `id` on `self.workflow`. For another subagent's step, add its name. A step with no answer reads as `None`.

```python theme={null}
class Scheduling(SubAgent):
    name = "scheduling"
    instructions = "You book the appointment."

    async def on_enter(self, reason: str):
        patient = self.workflow.intake.patient
        return f"Thank {patient.full_name}, and say you'll find them a time."
```

## Optional and Conditional Steps

Add `optional=True` to let the caller decline a step. The model skips it for you.

Add a `condition` to run a step only for some callers:

```python theme={null}
Step(
    id="referral",
    description="referral details",
    instructions="Get the referring doctor's name, then call record_referral.",
    opening="Ask who referred them.",
    condition={"field": "visit.kind", "eq": "specialist"},
)
```

A field is `<step>.<key>` from the same subagent, or `userdata.<key>` from the call's metadata. You can use `eq`, `ne`, `in`, `not_in`, or `exists`, and combine them with `all`, `any`, and `not`.

## Retry a Step

Set `max_attempts` to limit retries. Each `{"error": ...}` from a subagent's tool counts as one attempt. At the limit, `on_step_failed` runs:

```python theme={null}
async def on_step_failed(self, step_id: str, reason: str):
    return await self.transfer_to("front_desk", reason=f"{step_id} failed")
```

## Change an Answer

Use `go_to_step` when the caller changes their mind. Use `"<subagent>.<step>"` to go back to another subagent's step.

```python theme={null}
@function_tool
async def change_answer(self, step_id: str) -> dict:
    """Go back to change an earlier answer.

    Args:
        step_id: patient or visit.
    """
    return await self.go_to_step(step_id, reason="caller correction")
```

Only that step is asked again. If the answer changes, the steps in later subagents are done again.

If the caller gives the new answer straight away, skip `go_to_step`: call `self.complete(...)` with it. The finished step is updated and the call carries on from where it was.

## Track Progress

Events tell your code what just happened on the call, so you can log progress or save answers to your own systems. They stay in your agent: the model doesn't see them, and they aren't sent to the room.

| Event | Fires when | Data |
| :- | :- | :- |
| `agent_handoff` | A subagent takes over, including the first one when the call starts | `to_agent`, `from_agent`, `reason` |
| `step_start` | A step begins | `agent`, `step` |
| `step_end` | A step's answer is recorded | `agent`, `step`, `result`, `progress` |
| `step_back` | A step is reopened to be done again | `agent`, `step`, `reason` |
| `step_failed` | A step runs out of attempts | `agent`, `step`, `attempts`, `reason` |

`result` is the answer as a dict. `progress` lists that subagent's steps as `done`, `skipped`, and `to_do`. A skipped step has no event of its own, so look for it in `progress`.

Listen on your agent:

```python theme={null}
agent.on("step_end", lambda data: print(data["step"], data["result"]))
```

Or use `@on_workflow` on a subagent. It hears every subagent's events, so check `data["agent"]` if you only want its own.

```python theme={null}
from zeroruntime.workflow import on_workflow


class Scheduling(SubAgent):
    name = "scheduling"
    instructions = "You book the appointment."

    @on_workflow("step_end")
    def track(self, data: dict):
        if data["agent"] == self.name:
            self.log.info("%s done, to do: %s", data["step"], data["progress"]["to_do"])
```

## What's Next

> [card] [Subagents](https://docs.zeroruntime.ai/runtime-sdk/build/subagents.md)
>
> Roles, hooks, and moving the call.

> [card] [Clinic Example](https://github.com/ZeroRuntimeAI/zeroruntime-python-examples/blob/main/workflow/clinic_showcase.py)
>
> A full call with retries, corrections, and events.


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.