Workflow - Agno

The Workflow class executes steps, persists runs, and manages workflow sessions.

Parameters

Parameter	Type	Default	Description
`name`	`Optional[str]`	`None`	Workflow name
`id`	`Optional[str]`	`None`	Workflow ID (autogenerated if not set)
`description`	`Optional[str]`	`None`	Workflow description
`steps`	`Optional[WorkflowSteps]`	`None`	Callable, `Steps` object, or list of steps and nested workflows
`db`	`Optional[Union[BaseDb, AsyncBaseDb]]`	`None`	Database to use for this workflow
`agent`	`Optional[WorkflowAgent]`	`None`	Workflow agent that decides whether to execute the workflow or answer from session history
`session_id`	`Optional[str]`	`None`	Default session ID. Autogenerated if not set
`user_id`	`Optional[str]`	`None`	Default user ID
`session_state`	`Optional[Dict[str, Any]]`	`None`	Default session state persisted with the workflow session
`overwrite_db_session_state`	`bool`	`False`	Overwrite stored session state with the state provided to `run()` or `arun()`
`debug_mode`	`Optional[bool]`	`False`	If True, the workflow runs in debug mode
`debug_level`	`Literal[1, 2]`	`1`	Debug level: 1 = basic, 2 = detailed
`stream`	`Optional[bool]`	`None`	Stream the response from the Workflow
`stream_events`	`bool`	`False`	Stream intermediate workflow events
`stream_executor_events`	`bool`	`True`	Include agent and team executor events with workflow events
`store_events`	`bool`	`False`	Persist the events on the run response
`events_to_skip`	`Optional[List[Union[WorkflowRunEvent, RunEvent, TeamRunEvent]]]`	`None`	Events to skip when persisting the events on the run response
`store_executor_outputs`	`bool`	`True`	Control whether to store executor responses (agent/team responses) in flattened runs
`input_schema`	`Optional[Type[BaseModel]]`	`None`	Input schema to validate the input to the workflow
`metadata`	`Optional[Dict[str, Any]]`	`None`	Metadata stored with this workflow
`dependencies`	`Optional[Dict[str, Any]]`	`None`	Dependencies passed to downstream agents and teams
`add_dependencies_to_context`	`Optional[bool]`	`None`	If True, add the dependencies to the context of downstream agents and teams
`add_session_state_to_context`	`Optional[bool]`	`None`	If True, add the session_state to the context of downstream agents and teams
`add_workflow_history_to_steps`	`bool`	`False`	If True, add the workflow history to the steps
`num_history_runs`	`int`	`3`	Number of previous runs to include when `add_workflow_history_to_steps` is enabled
`cache_session`	`bool`	`False`	If True, cache the current workflow session in memory for faster access
`telemetry`	`bool`	`True`	Log minimal telemetry for analytics

Functions

`run`

Execute the workflow synchronously with optional streaming. Parameters:

input (Optional[Union[str, Dict[str, Any], List[Any], BaseModel]]): The input to send to the workflow
additional_data (Optional[Dict[str, Any]]): Additional data to include with the input
user_id (Optional[str]): User ID to use
run_id (Optional[str]): Run ID to use. A UUID is generated when omitted
session_id (Optional[str]): Session ID to use
session_state (Optional[Dict[str, Any]]): Session state to use
audio (Optional[List[Audio]]): Audio files to include
images (Optional[List[Image]]): Image files to include
videos (Optional[List[Video]]): Video files to include
files (Optional[List[File]]): Files to include
stream (Optional[bool]): Whether to stream the response. Defaults to the workflow setting
stream_events (Optional[bool]): Whether to stream intermediate steps
background (Optional[bool]): Reserved for API parity. Synchronous background execution is unsupported
background_tasks (Optional[Any]): Background task manager passed to step executors
dependencies (Optional[Dict[str, Any]]): Run-level dependencies merged with workflow dependencies
metadata (Optional[Dict[str, Any]]): Run-level metadata merged with workflow metadata
add_dependencies_to_context (Optional[bool]): Override whether downstream agents and teams receive dependencies in context
add_session_state_to_context (Optional[bool]): Override whether downstream agents and teams receive session state in context
**kwargs: Additional arguments passed to the workflow agent or step executors

Returns:

Union[WorkflowRunOutput, Iterator[WorkflowRunOutputEvent]]: Either a WorkflowRunOutput or an iterator of WorkflowRunOutputEvents, depending on the stream parameter

`arun`

Execute the workflow asynchronously with optional streaming. Parameters:

input (Optional[Union[str, Dict[str, Any], List[Any], BaseModel, List[Message]]]): The input to send to the workflow
additional_data (Optional[Dict[str, Any]]): Additional data to include with the input
user_id (Optional[str]): User ID to use
run_id (Optional[str]): Run ID to use. A UUID is generated when omitted
session_id (Optional[str]): Session ID to use
session_state (Optional[Dict[str, Any]]): Session state to use
audio (Optional[List[Audio]]): Audio files to include
images (Optional[List[Image]]): Image files to include
videos (Optional[List[Video]]): Video files to include
files (Optional[List[File]]): Files to include
stream (Optional[bool]): Whether to stream the response. Defaults to the workflow setting
stream_events (Optional[bool]): Whether to stream intermediate steps
background (Optional[bool]): Whether to run in background
websocket (Optional[WebSocket]): WebSocket for real-time communication
enable_websocket (bool): Use WebSocket transport when background streaming. Requires websocket
background_tasks (Optional[Any]): Background task manager passed to step executors
dependencies (Optional[Dict[str, Any]]): Run-level dependencies merged with workflow dependencies
metadata (Optional[Dict[str, Any]]): Run-level metadata merged with workflow metadata
add_dependencies_to_context (Optional[bool]): Override whether downstream agents and teams receive dependencies in context
add_session_state_to_context (Optional[bool]): Override whether downstream agents and teams receive session state in context
**kwargs: Additional arguments passed to the workflow agent or step executors

Returns:

Union[WorkflowRunOutput, AsyncIterator[WorkflowRunOutputEvent]]: Either a WorkflowRunOutput or an iterator of WorkflowRunOutputEvents, depending on the stream parameter

`continue_run`

Resume a paused workflow run after HITL requirements have been resolved. Used for both step-level and executor-level pauses. See HITL overview. Parameters:

run_response (Optional[WorkflowRunOutput]): The paused WorkflowRunOutput to continue. If not provided, run_id and session_id must be supplied to load it from the database.
run_id (Optional[str]): The run_id of the paused run. Required if run_response is not provided.
session_id (Optional[str]): The session_id of the paused run. Required if run_response is not provided.
step_requirements (Optional[List[StepRequirement]]): Updated step requirements with confirmation status. Defaults to the requirements from run_response.
stream (Optional[bool]): Whether to stream the response. Defaults to the workflow’s stream setting.
stream_events (Optional[bool]): Whether to stream events. Defaults to the workflow’s stream_events setting.
**kwargs: Additional arguments passed to resumed step executors

Returns:

Union[WorkflowRunOutput, Iterator[WorkflowRunOutputEvent]]: A WorkflowRunOutput if stream=False, otherwise an iterator of events.

Raises:

ValueError: If neither run_response nor (run_id + session_id) are provided, the run is not paused, or step requirements have not been resolved.

`acontinue_run`

Asynchronous variant of continue_run. Required when the workflow’s database is async. It accepts the same parameters as continue_run, plus:

background (Optional[bool]): Whether to continue the run in the background
websocket (Optional[WebSocket]): WebSocket for real-time communication
enable_websocket (bool): Use WebSocket transport when background streaming. Requires websocket
**kwargs: Additional arguments passed to resumed step executors

Returns:

Union[WorkflowRunOutput, AsyncIterator[WorkflowRunOutputEvent]]: A WorkflowRunOutput if stream=False, otherwise an async iterator of events

`print_response`

Print workflow execution with rich formatting and optional streaming. Parameters:

input (Union[str, Dict[str, Any], List[Any], BaseModel, List[Message]]): The input to send to the workflow
additional_data (Optional[Dict[str, Any]]): Additional data to include with the input
user_id (Optional[str]): User ID to use
session_id (Optional[str]): Session ID to use
audio (Optional[List[Audio]]): Audio files to include
images (Optional[List[Image]]): Image files to include
videos (Optional[List[Video]]): Video files to include
files (Optional[List[File]]): Files to include
stream (Optional[bool]): Whether to stream the response content
markdown (bool): Whether to render content as markdown
show_time (bool): Whether to show execution time
show_step_details (bool): Whether to show individual step outputs
console (Optional[Any]): Rich console instance (optional)
**kwargs: Additional arguments passed to run()

`aprint_response`

Print workflow execution with rich formatting and optional streaming asynchronously. Parameters:

input (Union[str, Dict[str, Any], List[Any], BaseModel, List[Message]]): The input to send to the workflow
additional_data (Optional[Dict[str, Any]]): Additional data to include with the input
user_id (Optional[str]): User ID to use
session_id (Optional[str]): Session ID to use
audio (Optional[List[Audio]]): Audio files to include
images (Optional[List[Image]]): Image files to include
videos (Optional[List[Video]]): Video files to include
files (Optional[List[File]]): Files to include
stream (Optional[bool]): Whether to stream the response content
markdown (bool): Whether to render content as markdown
show_time (bool): Whether to show execution time
show_step_details (bool): Whether to show individual step outputs
console (Optional[Any]): Rich console instance (optional)
**kwargs: Additional arguments passed to arun()

`cli_app`

Run an interactive command-line interface to interact with the workflow. Parameters:

input (Optional[str]): The input to send to the workflow
session_id (Optional[str]): Session ID to use
user_id (Optional[str]): User ID to use
user (str): Name for the user (default: “User”)
emoji (str): Emoji for the user (default: “:technologist:”)
stream (Optional[bool]): Whether to stream the response content
markdown (bool): Whether to render content as markdown (default: True)
show_time (bool): Whether to show execution time (default: True)
show_step_details (bool): Whether to show individual step outputs (default: True)
exit_on (Optional[List[str]]): List of commands to exit the CLI
**kwargs: Additional keyword arguments

`acli_app`

Run an interactive command-line interface to interact with the workflow asynchronously. Parameters:

input (Optional[str]): The input to send to the workflow
session_id (Optional[str]): Session ID to use
user_id (Optional[str]): User ID to use
user (str): Name for the user (default: “User”)
emoji (str): Emoji for the user (default: “:technologist:”)
stream (Optional[bool]): Whether to stream the response content
markdown (bool): Whether to render content as markdown (default: True)
show_time (bool): Whether to show execution time (default: True)
show_step_details (bool): Whether to show individual step outputs (default: True)
exit_on (Optional[List[str]]): List of commands to exit the CLI
**kwargs: Additional keyword arguments

`cancel_run`

Cancel a running workflow execution. Parameters:

run_id (str): The run_id to cancel

Returns:

bool: True if the run was found and marked for cancellation, False otherwise

`acancel_run`

Cancel a running workflow asynchronously. Parameters:

run_id (str): The run ID to cancel

Returns:

bool: True if the run was found and marked for cancellation, False otherwise

`get_run`

Get the status and details of a background workflow run. Parameters:

run_id (str): The run ID to get
session_id (Optional[str]): Session ID used for database lookup

Returns:

Optional[WorkflowRunOutput]: The workflow run output if found

`get_run_output`

Get a WorkflowRunOutput from the database. Parameters:

run_id (str): The run ID
session_id (Optional[str]): Session ID to use
user_id (Optional[str]): User ID used to scope the session lookup

Returns:

Optional[WorkflowRunOutput]: The run output

`get_last_run_output`

Get the last run response from the database for the given session ID. Parameters:

session_id (Optional[str]): Session ID to use

Returns:

Optional[WorkflowRunOutput]: The last run output

`get_chat_history`

Return the input and output interactions for each run in the session. Parameters:

session_id (Optional[str]): The session ID to get the chat history for. If not provided, the current cached session ID is used
last_n_runs (Optional[int]): Number of recent runs to include. If None, all runs will be considered

Returns:

List[WorkflowChatInteraction]: The input and output interactions for each run

`get_session`

Get the session for the given session ID. Parameters:

session_id (Optional[str]): Session ID to use
user_id (Optional[str]): User ID used to scope the session lookup

Returns:

Optional[WorkflowSession]: The workflow session

`get_session_state`

Get the session state for the given session ID. Parameters:

session_id (Optional[str]): Session ID to use

Returns:

Dict[str, Any]: The session state

`update_session_state`

Merge values into the session state and persist the session. Parameters:

session_state_updates (Dict[str, Any]): Values to merge into session state
session_id (Optional[str]): Session ID to use

Returns:

Dict[str, Any]: The updated session state

`get_session_name`

Get the session name for the given session ID. Parameters:

session_id (Optional[str]): Session ID to use

Returns:

str: The session name

`set_session_name`

Set the session name and save to storage. Parameters:

session_id (Optional[str]): Session ID to use
autogenerate (bool): Whether to autogenerate the name
session_name (Optional[str]): The name to set

Returns:

WorkflowSession: The updated session

`get_session_metrics`

Get the session metrics for the given session ID. Parameters:

session_id (Optional[str]): Session ID to use

Returns:

Optional[SessionMetrics]: The session metrics

`delete_session`

Delete a session. Parameters:

session_id (str): Session ID to delete
user_id (Optional[str]): User ID used to scope the session deletion

`save_session`

Save the WorkflowSession to storage. Parameters:

session (WorkflowSession): The session to save

`to_dict`

Convert workflow to dictionary representation. Returns:

Dict[str, Any]: Dictionary representation of the workflow

`run_parameters`

Return the input parameters accepted by the workflow. Callable workflows expose their function signature. Step-based workflows expose a required string message parameter. Returns:

Dict[str, Any]: Parameter names mapped to their defaults, annotations, and required status

`from_dict`

Create a workflow from a dictionary. Parameters:

data (Dict[str, Any]): Serialized workflow configuration
db (Optional[BaseDb]): Database used to reconstruct step executors
links (Optional[List[Dict[str, Any]]]): Links for the workflow version
registry (Optional[Registry]): Registry used to resolve schemas and executors

Returns:

Workflow: Reconstructed workflow

`save`

Save the workflow component and configuration. Parameters:

db (Optional[BaseDb]): Database to save to. Defaults to the workflow database
stage (str): Component stage. Defaults to "published"
label (Optional[str]): Configuration label
notes (Optional[str]): Configuration notes

Returns:

Optional[int]: Saved configuration version, or None when saving fails

`load`

Load a workflow configuration from a database. Parameters:

id (str): Workflow ID
db (BaseDb): Database to load from
registry (Optional[Registry]): Registry used to reconstruct schemas and executors
label (Optional[str]): Configuration label
version (Optional[int]): Configuration version

Returns:

Optional[Workflow]: Loaded workflow, or None when no matching configuration exists

`delete`

Delete the workflow component. Parameters:

db (Optional[BaseDb]): Database to delete from. Defaults to the workflow database
hard_delete (bool): Permanently delete the component when True

Returns:

bool: Whether the workflow was deleted

`deep_copy`

Create a copy with isolated mutable state while sharing heavy resources such as database connections. Parameters:

update (Optional[Dict[str, Any]]): Fields to override on the copied workflow

Returns:

Workflow: Copied workflow

Async database variants are available for session and run methods, including aget_run(), aget_run_output(), aget_last_run_output(), aget_chat_history(), aget_session(), aget_session_state(), aupdate_session_state(), aget_session_name(), aset_session_name(), aget_session_metrics(), adelete_session(), and asave_session().

​Parameters

​Functions

​run

​arun

​continue_run

​acontinue_run

​print_response

​aprint_response

​cli_app

​acli_app

​cancel_run

​acancel_run

​get_run

​get_run_output

​get_last_run_output

​get_chat_history

​get_session

​get_session_state

​update_session_state

​get_session_name

​set_session_name

​get_session_metrics

​delete_session

​save_session

​to_dict

​run_parameters

​from_dict

​save

​load

​delete

​deep_copy

Parameters

Functions

`run`

`arun`

`continue_run`

`acontinue_run`

`print_response`

`aprint_response`

`cli_app`

`acli_app`

`cancel_run`

`acancel_run`

`get_run`

`get_run_output`

`get_last_run_output`

`get_chat_history`

`get_session`

`get_session_state`

`update_session_state`

`get_session_name`

`set_session_name`

`get_session_metrics`

`delete_session`

`save_session`

`to_dict`

`run_parameters`

`from_dict`

`save`

`load`

`delete`

`deep_copy`