Workflows Development Guide

Agno Workflows provides a powerful, declarative way to orchestrate multi-step AI processes. Unlike traditional linear workflows, you can now create sophisticated branching logic, parallel execution, and dynamic routing based on content analysis. This guide covers all workflow patterns, from simple linear sequences to complex conditional logic with parallel execution.

Building Blocks

The core building blocks of Agno Workflows are:

Component	Purpose
Step	Basic execution unit
Agent	AI assistant with specific role
Team	Coordinated group of agents
Function	Custom Python logic
Parallel	Concurrent execution
Condition	Conditional execution
Loop	Iterative execution
Router	Dynamic routing

Workflow Patterns

1. Basic Sequential Workflows

When to use: Linear processes where each step depends on the previous one. Example Steps: Research → Process research data in a function before next step → Content Creation

from agno.workflow.v2 import Step, Workflow, StepOutput

def data_preprocessor(step_input):
    # Custom preprocessing logic

    # Or you can also run any agent/team over here itself
    # response = some_agent.run(...)
    return StepOutput(content=f"Processed: {step_input.message}") # <-- Now pass the agent/team response in content here

workflow = Workflow(
    name="Mixed Execution Pipeline",
    steps=[
        research_team,      # Team
        data_preprocessor,  # Function
        content_agent,      # Agent
    ]
)

workflow.print_response("Analyze the competitive landscape for fintech startups", markdown=True)

For more information on how to use custom functions, refer to the Advanced page.

See Example:

Sequence of Functions and Agents - Complete workflow with functions and agents

StepInput and StepOutput provides standardized interfaces for data flow between steps: So if you make a custom function as an executor for a step, make sure that the input and output types are compatible with the StepInput and StepOutput interfaces. This will ensure that your custom function can seamlessly integrate into the workflow system.Take a look at the schemas for StepInput and StepOutput.

2. Fully Python Workflow

Keep it Simple with Pure Python: If you prefer the Workflows 1.0 approach or need maximum flexibility, you can still use a single Python function to handle everything. This approach gives you complete control over the execution flow while still benefiting from workflow features like storage, streaming, and session management. Replace all the steps in the workflow with a single executable function where you can control everything.

def custom_workflow_function(workflow: Workflow, execution_input: WorkflowExecutionInput):
    # Custom orchestration logic
    research_result = research_team.run(execution_input.message)
    analysis_result = analysis_agent.run(research_result.content)
    return f"Final: {analysis_result.content}"

workflow = Workflow(
    name="Function-Based Workflow",
    steps=custom_workflow_function  # Single function replaces all steps
)

workflow.print_response("Evaluate the market potential for quantum computing applications", markdown=True)

See Example:

Function-Based Workflow - Complete function-based workflow

For migration from 1.0 style workflows, refer to the page for Migrating to Workflows 2.0

3. Step-Based Workflows

You can name your steps for better logging and future support on the Agno platform.
This also changes the name of a step when accessing that step’s output inside a StepInput object.

Parameters

Parameter	Type	Default	Description
`name`	`Optional[str]`	`None`	Name of the step for identification
`description`	`Optional[str]`	`None`	Description of the step's purpose
`executor`	`Optional[Callable]`	`None`	Custom function to execute for this step
`agent`	`Optional[Agent]`	`None`	Agent to execute for this step
`team`	`Optional[Team]`	`None`	Team to execute for this step

Example

from agno.workflow.v2 import Step, Workflow

# Named steps for better tracking
workflow = Workflow(
    name="Content Creation Pipeline",
    steps=[
        Step(name="Research Phase", team=researcher),
        Step(name="Analysis Phase", executor=custom_function), 
        Step(name="Writing Phase", agent=writer),
    ]
)

workflow.print_response(
    "AI trends in 2024",
    markdown=True,
)

See Examples:

4. Conditional Steps

When to use: Conditional step execution based on business logic. Example Use-Cases: Topic-specific research strategies, content type routing

Parameters

Parameter	Type	Default	Description
`evaluator`	`Union[Callable[[StepInput], bool], bool]`	Required	Function or boolean to evaluate the condition
`steps`	`WorkflowSteps`	Required	Steps to execute if the condition is met
`name`	`Optional[str]`	`None`	Name of the condition step
`description`	`Optional[str]`	`None`	Description of the condition step

Example

from agno.workflow.v2 import Condition, Step, Workflow

def is_tech_topic(step_input) -> bool:
    topic = step_input.message.lower()
    return any(keyword in topic for keyword in ["ai", "tech", "software"])

workflow = Workflow(
    name="Conditional Research",
    steps=[
        Condition(
            name="Tech Topic Check",
            evaluator=is_tech_topic,
            steps=[Step(name="Tech Research", agent=tech_researcher)]
        ),
        Step(name="General Analysis", agent=general_analyst),
    ]
)

workflow.print_response("Comprehensive analysis of AI and machine learning trends", markdown=True)

More Examples:

5. Parallel Execution

When to use: Independent tasks that can run simultaneously to save time. Example Use-Cases: Multiple research sources, parallel content creation

Parameters

Parameter	Type	Default	Description
`steps`	`WorkflowSteps`	Required	List of steps to execute in parallel
`name`	`Optional[str]`	`None`	Name of the parallel execution block
`description`	`Optional[str]`	`None`	Description of the parallel execution

Example

from agno.workflow.v2 import Parallel, Step, Workflow

workflow = Workflow(
    name="Parallel Research Pipeline",
    steps=[
        Parallel(
            Step(name="HackerNews Research", agent=hn_researcher),
            Step(name="Web Research", agent=web_researcher),
            Step(name="Academic Research", agent=academic_researcher),
            name="Research Step"
        ),
        Step(name="Synthesis", agent=synthesizer),  # Combines the results and produces a report
    ]
)

workflow.print_response("Write about the latest AI developments", markdown=True)

More Examples:

Parallel Steps Workflow

6. Loop/Iteration Workflows

When to use: Quality-driven processes, iterative refinement, or retry logic. Example Use-Cases: Research until sufficient quality, iterative improvement

Parameters

Parameter	Type	Default	Description
`steps`	`WorkflowSteps`	Required	Steps to execute in each loop iteration
`name`	`Optional[str]`	`None`	Name of the loop step
`description`	`Optional[str]`	`None`	Description of the loop step
`max_iterations`	`int`	`3`	Maximum number of iterations for the loop
`end_condition`	`Optional[Callable[[List[StepOutput]], bool]]`	`None`	Function to evaluate if the loop should end

Example

from agno.workflow.v2 import Loop, Step, Workflow

def quality_check(outputs) -> bool:
    # Return True to break loop, False to continue
    return any(len(output.content) > 500 for output in outputs)

workflow = Workflow(
    name="Quality-Driven Research",
    steps=[
        Loop(
            name="Research Loop",
            steps=[Step(name="Deep Research", agent=researcher)],
            end_condition=quality_check,
            max_iterations=3
        ),
        Step(name="Final Analysis", agent=analyst),
    ]
)

workflow.print_response("Research the impact of renewable energy on global markets", markdown=True)

More Examples:

Loop Steps Workflow

7. Branching Workflows

When to use: Complex decision trees, topic-specific workflows, dynamic routing. Example Use-Cases: Content type detection, expertise routing

Parameters

Parameter	Type	Default	Description
`selector`	`Callable[[StepInput], Union[WorkflowSteps, List[WorkflowSteps]]]`	Required	Function to select steps dynamically
`choices`	`WorkflowSteps`	Required	Available steps for selection
`name`	`Optional[str]`	`None`	Name of the router step
`description`	`Optional[str]`	`None`	Description of the router step

Example

from agno.workflow.v2 import Router, Step, Workflow

def route_by_topic(step_input) -> List[Step]:
    topic = step_input.message.lower()
    
    if "tech" in topic:
        return [Step(name="Tech Research", agent=tech_expert)]
    elif "business" in topic:
        return [Step(name="Business Research", agent=biz_expert)]
    else:
        return [Step(name="General Research", agent=generalist)]

workflow = Workflow(
    name="Expert Routing",
    steps=[
        Router(
            name="Topic Router",
            selector=route_by_topic,
            choices=[tech_step, business_step, general_step]
        ),
        Step(name="Synthesis", agent=synthesizer),
    ]
)

workflow.print_response("Latest developments in artificial intelligence and machine learning", markdown=True)

More Examples:

Router Steps Workflow

8. Steps: Grouping a list of steps

When to use: When you need to group multiple steps into logical sequences, create reusable workflows, or organize complex workflows with multiple branching paths. Better Routing: Use with Router for clean branching logic

Parameters

Parameter	Type	Default	Description
`steps`	`WorkflowSteps`	Required	List of steps to execute sequentially
`name`	`Optional[str]`	`None`	Name of the steps step
`description`	`Optional[str]`	`None`	Description of the steps step

Basic Example

from agno.workflow.v2 import Steps, Step, Workflow

# Create a reusable content creation sequence
article_creation_sequence = Steps(
    name="ArticleCreation",
    description="Complete article creation workflow from research to final edit",
    steps=[
        Step(name="research", agent=researcher),
        Step(name="writing", agent=writer), 
        Step(name="editing", agent=editor),
    ],
)

# Use the sequence in a workflow
workflow = Workflow(
    name="Article Creation Workflow",
    steps=[article_creation_sequence]  # Single sequence
)

workflow.print_response("Write an article about renewable energy", markdown=True)

Steps with Router

This is where Steps really shines - creating distinct sequences for different content types or workflows:

from agno.workflow.v2 import Steps, Router, Step, Workflow

# Define two completely different workflows as Steps
image_sequence = Steps(
    name="image_generation",
    description="Complete image generation and analysis workflow",
    steps=[
        Step(name="generate_image", agent=image_generator),
        Step(name="describe_image", agent=image_describer),
    ],
)

video_sequence = Steps(
    name="video_generation", 
    description="Complete video production and analysis workflow",
    steps=[
        Step(name="generate_video", agent=video_generator),
        Step(name="describe_video", agent=video_describer),
    ],
)

def media_sequence_selector(step_input) -> List[Step]:
    """Route to appropriate media generation pipeline"""
    if not step_input.message:
        return [image_sequence]
        
    message_lower = step_input.message.lower()
    
    if "video" in message_lower:
        return [video_sequence]
    elif "image" in message_lower:
        return [image_sequence]
    else:
        return [image_sequence]  # Default

# Clean workflow with clear branching
media_workflow = Workflow(
    name="AI Media Generation Workflow",
    description="Generate and analyze images or videos using AI agents",
    steps=[
        Router(
            name="Media Type Router",
            description="Routes to appropriate media generation pipeline",
            selector=media_sequence_selector,
            choices=[image_sequence, video_sequence],  # Clear choices
        )
    ],
)

# Usage examples
media_workflow.print_response("Create an image of a magical forest", markdown=True)
media_workflow.print_response("Create a cinematic video of city timelapse", markdown=True)

More Examples:

9. Advanced Workflow Patterns

You can use the patterns above to construct sophisticated workflows. Example Usage: Conditions + Parallel + Loops + Custom Post-Processing Function + Routing

from agno.workflow.v2 import Condition, Loop, Parallel, Router, Step, Workflow

def research_post_processor(step_input) -> StepOutput:
    """Post-process and consolidate research data from parallel conditions"""
    research_data = step_input.previous_step_content or ""
    
    try:
        # Analyze research quality and completeness
        word_count = len(research_data.split())
        has_tech_content = any(keyword in research_data.lower() 
                              for keyword in ["technology", "ai", "software", "tech"])
        has_business_content = any(keyword in research_data.lower() 
                                  for keyword in ["market", "business", "revenue", "strategy"])
        
        # Create enhanced research summary
        enhanced_summary = f"""
            ## Research Analysis Report
            
            **Data Quality:** {"✓ High-quality" if word_count > 200 else "⚠ Limited data"}
            
            **Content Coverage:**
            - Technical Analysis: {"✓ Completed" if has_tech_content else "✗ Not available"}
            - Business Analysis: {"✓ Completed" if has_business_content else "✗ Not available"}
            
            **Research Findings:**
            {research_data}
        """.strip()
        
        return StepOutput(
            content=enhanced_summary,
            success=True,
        )
        
    except Exception as e:
        return StepOutput(
            content=f"Research post-processing failed: {str(e)}",
            success=False,
            error=str(e)
        )

# Complex workflow combining multiple patterns
workflow = Workflow(
    name="Advanced Multi-Pattern Workflow",
    steps=[
        Parallel(
            Condition(
                name="Tech Check",
                evaluator=is_tech_topic,
                steps=[Step(name="Tech Research", agent=tech_researcher)]
            ),
            Condition(
                name="Business Check", 
                evaluator=is_business_topic,
                steps=[
                    Loop(
                        name="Deep Business Research",
                        steps=[Step(name="Market Research", agent=market_researcher)],
                        end_condition=research_quality_check,
                        max_iterations=3
                    )
                ]
            ),
            name="Conditional Research Phase"
        ),
        Step(
            name="Research Post-Processing",
            executor=research_post_processor,
            description="Consolidate and analyze research findings with quality metrics"
        ),
        Router(
            name="Content Type Router",
            selector=content_type_selector,
            choices=[blog_post_step, social_media_step, report_step]
        ),
        Step(name="Final Review", agent=reviewer),
    ]
)

workflow.print_response("Create a comprehensive analysis of sustainable technology trends and their business impact for 2024", markdown=True)

More Examples:

Introduction

Concepts

Other

How to

Workflows Development Guide

Building Blocks

Workflow Patterns

1. Basic Sequential Workflows

2. Fully Python Workflow

3. Step-Based Workflows

Parameters

Example

4. Conditional Steps

Parameters

Example

5. Parallel Execution

Parameters

Example

6. Loop/Iteration Workflows

Parameters

Example

7. Branching Workflows

Parameters

Example

8. Steps: Grouping a list of steps

Parameters

Basic Example

Steps with Router

9. Advanced Workflow Patterns

​Building Blocks

​Workflow Patterns

​1. Basic Sequential Workflows

​2. Fully Python Workflow

​3. Step-Based Workflows

​Parameters

​Example

​4. Conditional Steps

​Parameters

​Example

​5. Parallel Execution

​Parameters

​Example

​6. Loop/Iteration Workflows

​Parameters

​Example

​7. Branching Workflows

​Parameters

​Example

​8. Steps: Grouping a list of steps

​Parameters

​Basic Example

​Steps with Router

​9. Advanced Workflow Patterns

Building Blocks

Workflow Patterns

1. Basic Sequential Workflows

2. Fully Python Workflow

3. Step-Based Workflows

Parameters

Example

4. Conditional Steps

Parameters

Example

5. Parallel Execution

Parameters

Example

6. Loop/Iteration Workflows

Parameters

Example

7. Branching Workflows

Parameters

Example

8. Steps: Grouping a list of steps

Parameters

Basic Example

Steps with Router

9. Advanced Workflow Patterns