Router Steps

Parameters

Parameter	Type	Default	Description
`selector`	`Callable[[StepInput], ...] \| Callable[[StepInput, list], ...] \| str`	`None`	Function or CEL expression string to select steps dynamically. Functions can optionally accept `step_choices` as second parameter
`choices`	`WorkflowSteps`	Required	Available steps for selection. Supports nested lists (becomes Steps container)
`name`	`Optional[str]`	`None`	Name of the router step
`description`	`Optional[str]`	`None`	Description of the router step
`human_review`	`HumanReview`	`HumanReview()`	Human review configuration. See the fields below.

Selector Return Types

The selector function can return any of the following:

Return Type	Description
`str`	Step name as string - Router resolves it from choices
`Step`	Step object directly
`List[Step]`	List of steps for chaining execution

selector also accepts a CEL expression string in place of a function. The expression must evaluate to a step name from choices.

Selector Function Signatures

Basic Signature

def selector(step_input: StepInput) -> Union[str, Step, List[Step]]:
    ...

Extended Signature (with step_choices)

def selector(step_input: StepInput, step_choices: list) -> Union[str, Step, List[Step]]:
    ...

The step_choices parameter provides access to the prepared Step objects from Router.choices, enabling dynamic selection based on available options.

Extended Signature (with run_context)

The workflow injects RunContext into a parameter named run_context. Read session state through that object; a callable parameter named session_state is not injected.

from agno.run import RunContext
from agno.workflow.types import StepInput

def selector(step_input: StepInput, run_context: RunContext) -> str:
    state = run_context.session_state or {}
    return "priority" if state.get("priority") else "standard"

This selector requires choices named priority and standard. CEL expressions separately expose a session_state variable; that does not change callable injection.

Async Support

All selector signatures support async functions:

async def selector(step_input: StepInput, step_choices: list) -> Union[str, Step, List[Step]]:
    ...

HumanReview configuration

Pass these settings inside human_review=HumanReview(...). They are not direct constructor arguments. See HumanReview for the complete configuration and supported combinations.

Parameter	Type	Default	Description
`requires_confirmation`	`bool`	`False`	Pause for user confirmation before executing selected route
`confirmation_message`	`Optional[str]`	`None`	Message shown to user when requesting confirmation
`requires_user_input`	`bool`	`False`	Pause for user to select route(s) instead of using selector. See Router HITL.
`user_input_message`	`Optional[str]`	`None`	Message shown to user when requesting route selection
`allow_multiple_selections`	`bool`	`False`	Allow user to select multiple routes
`user_input_schema`	`Optional[List[Dict[str, Any]]]`	`None`	Custom input schema for the user-input pause. If unset, the pause presents the available `choices` for route selection
`requires_output_review`	`bool`	`False`	Pause after the selected route completes to review its output. See Output Review.
`output_review_message`	`Optional[str]`	`None`	Message shown to user during output review
`max_retries`	`int`	`3`	Max re-executions when an output review is rejected with `OnReject.retry`
`on_reject`	`Union[OnReject, str]`	`OnReject.skip`	Action when rejected: `skip`, `cancel`, `retry`

Constructor example (install agno first):

from agno.workflow.router import Router
from agno.workflow.step import Step
from agno.workflow.types import HumanReview, StepInput, StepOutput

def prepare(step_input: StepInput) -> StepOutput:
    return StepOutput(content=step_input.input)

prepare_step = Step(name="prepare", executor=prepare)
step = Router(
    choices=[prepare_step], selector=lambda step_input: "prepare",
    human_review=HumanReview(requires_confirmation=True),
)

On this page