Workflows and Steps

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.

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(...).

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(...).

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.

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:

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:

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.

@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:

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.

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"])

Get started

Quickstarts

Build

Monitor

Deploy

Plugins

Add Steps

Steps Without Subagents

Read Answers

Optional and Conditional Steps

Retry a Step

Change an Answer

Track Progress

What’s Next

Subagents

Clinic Example

​Add Steps

​Steps Without Subagents

​Read Answers

​Optional and Conditional Steps

​Retry a Step

​Change an Answer

​Track Progress

​What’s Next

Subagents

Clinic Example

Add Steps

Steps Without Subagents

Read Answers

Optional and Conditional Steps

Retry a Step

Change an Answer

Track Progress

What’s Next