Skip to main content
The Workflow class enables you to build structured, multi-step AI workflows with agents, teams, or custom functions. Workflows provide explicit control over execution flow with support for loops, conditionals, parallel execution, and human-in-the-loop interactions.

​
Constructor

from agno import Workflow

workflow = Workflow(
    name="data_pipeline",
    steps=[
        extract_step,
        transform_step,
        load_step
    ]
)

​
Core Parameters

​
name
str
default:"None"
A descriptive name for the workflow.
​
id
str
default:"None"
Unique identifier for the workflow. Auto-generated if not provided.
​
description
str
default:"None"
A description of what the workflow does.
​
steps
WorkflowSteps
default:"None"
The steps to execute in the workflow. Can be a callable, Steps object, or list of Step objects.

​
Database

​
db
BaseDb | AsyncBaseDb
default:"None"
Database for storing workflow sessions and run history.

​
Session & State

​
session_id
str
default:"None"
Default session ID for the workflow.
​
user_id
str
default:"None"
Default user ID for the workflow.
​
session_state
Dict[str, Any]
default:"None"
Session state stored in database to persist across runs.
​
overwrite_db_session_state
bool
default:"False"
If True, overwrites stored session_state with provided session_state.

​
History

​
add_workflow_history_to_steps
bool
default:"False"
If True, adds workflow history to step executors.
​
num_history_runs
int
default:"3"
Number of historical runs to include in messages.

​
Streaming

​
stream
bool
default:"None"
Stream the response from the workflow.
​
stream_events
bool
default:"False"
Stream intermediate step events.
​
stream_executor_events
bool
default:"True"
Stream events from executors (agents/teams) within steps.

​
Storage

​
store_events
bool
default:"False"
Persist events on the run response.
​
store_executor_outputs
bool
default:"True"
Store executor responses (agent/team responses) in flattened runs.
​
events_to_skip
List[WorkflowRunEvent | RunEvent | TeamRunEvent]
default:"None"
Events to skip when persisting.

​
Validation

​
input_schema
Type[BaseModel]
default:"None"
Pydantic model to validate workflow input.
​
metadata
Dict[str, Any]
default:"None"
Metadata stored with the workflow.

​
Methods

​
run()

Execute the workflow. 
result = workflow.run(message="Process this data")
print(result.content)
Parameters:
  • message (str): The input message (if workflow uses default parameters)
  • Or pass custom parameters matching your workflow definition
  • stream (bool): Whether to stream the response
  • session_id (str): Optional session ID
  • user_id (str): Optional user ID
Returns: WorkflowRunOutput or Iterator of events if streaming

​
arun()

Async version of run(). 
result = await workflow.arun(message="Process this data")

​
continue_run()

Continue a paused workflow run (for HITL scenarios). 
# Initial run that pauses for approval
result = workflow.run(message="Execute trade")

if result.requirements:
    # ... handle requirements ...
    # Continue execution
    final_result = workflow.continue_run(result)

​
save()

Save the workflow configuration. 
workflow.save(db=db, stage="published")

​
load()

Load a workflow from the database. 
workflow = Workflow.load(id="my_workflow", db=db)

​
get_session()

Retrieve a workflow session. 
session = workflow.get_session(session_id="abc123")

​
Example Usage

from agno import Agent, Workflow
from agno.workflow import Step

researcher = Agent(name="Researcher")
writer = Agent(name="Writer")

workflow = Workflow(
    name="content_creation",
    steps=[
        Step(agent=researcher, name="research"),
        Step(agent=writer, name="write")
    ]
)

result = workflow.run(message="Write about AI")

Build docs developers (and LLMs) love

Get started for freeTalk to us