Accessing Multiple Previous Steps

Access any previous step's output by name with StepInput methods.

StepInput can retrieve any previous step output by name, including outputs nested in workflow primitives.

from agno.workflow import Step, Workflow
from agno.workflow.types import StepInput, StepOutput


def research_hackernews(step_input: StepInput) -> StepOutput:
    return StepOutput(content=f"HackerNews findings for {step_input.input}")


def research_web(step_input: StepInput) -> StepOutput:
    return StepOutput(content=f"Web findings for {step_input.input}")


def create_comprehensive_report(step_input: StepInput) -> StepOutput:
    original_topic = step_input.input or ""
    hackernews_data = step_input.get_step_content("research_hackernews") or ""
    web_data = step_input.get_step_content("research_web") or ""
    all_research = step_input.get_all_previous_content()

    return StepOutput(
        content=(
            f"Report: {original_topic}\n"
            f"HackerNews: {hackernews_data}\n"
            f"Web: {web_data}\n\n"
            f"All previous content:\n{all_research}"
        )
    )


workflow = Workflow(
    name="research_workflow",
    steps=[
        Step(name="research_hackernews", executor=research_hackernews),
        Step(name="research_web", executor=research_web),
        Step(name="comprehensive_report", executor=create_comprehensive_report),
    ],
)

workflow.print_response("agent infrastructure")

Available Methods

Method or FieldReturns
get_step_content("step_name")Content from a named previous step
get_step_output("step_name")Full StepOutput from a named previous step
get_all_previous_content()Top-level previous outputs joined into one string
inputOriginal workflow input
previous_step_contentContent from the immediate previous step

Accessing Nested Steps

You can directly access steps nested inside Parallel, Condition, Router, Loop, and Steps groups using their step name. The lookup performs a recursive search to find nested steps at any depth.

Direct Access to Steps in Parallel Groups

from agno.workflow.parallel import Parallel


def run_step_a(step_input: StepInput) -> StepOutput:
    return StepOutput(content="output from step_a")


def run_step_b(step_input: StepInput) -> StepOutput:
    return StepOutput(content="output from step_b")


def run_step_c(step_input: StepInput) -> StepOutput:
    return StepOutput(content="output from step_c")


def aggregator(step_input: StepInput) -> StepOutput:
    parallel_data = step_input.get_step_content("parallel_processing")
    step_a_output = step_input.get_step_output("step_a")
    step_a_content = step_input.get_step_content("step_a")

    return StepOutput(
        content=f"Parallel: {parallel_data}\nStep A: {step_a_content}\nFound: {step_a_output is not None}"
    )

workflow = Workflow(
    name="parallel_access",
    steps=[
        Parallel(
            Step(name="step_a", executor=run_step_a),
            Step(name="step_b", executor=run_step_b),
            Step(name="step_c", executor=run_step_c),
            name="parallel_processing",
        ),
        Step(name="aggregator", executor=aggregator),
    ],
)

Deeply Nested Steps

The recursive search works at any depth level:

from agno.workflow.condition import Condition
from agno.workflow.parallel import Parallel
from agno.workflow.steps import Steps


def deep_nested_step(step_input: StepInput) -> StepOutput:
    return StepOutput(content="Deep nested content")


def verify_nested_access(step_input: StepInput) -> StepOutput:
    nested_output = step_input.get_step_output("deep_nested_step")
    nested_content = step_input.get_step_content("deep_nested_step")

    return StepOutput(
        content=f"Nested output found: {nested_output is not None}. Content: {nested_content}"
    )


def condition_evaluator(step_input: StepInput) -> bool:
    return True

workflow = Workflow(
    name="multiple_depth_access",
    steps=[
        Parallel(
            Condition(
                name="condition_layer",
                evaluator=condition_evaluator,
                steps=[
                    Steps(
                        name="steps_layer",
                        steps=[deep_nested_step],
                    )
                ],
            ),
            name="parallel_processing",
        ),
        Step(name="verifier", executor=verify_nested_access),
    ],
)

A top-level output takes priority when the same step name also exists in a nested group.

Parallel Group Output Format

When accessing a Parallel group by its name, the output is a dictionary with each nested step's name as the key:

parallel_output = step_input.get_step_content("parallel_processing")
# Returns:
# {
#     "step_a": "output from step_a",
#     "step_b": "output from step_b",
#     "step_c": "output from step_c",
# }

Developer Resources