Workflow Sessions

Track multi-step workflow executions with session history.

Before running the examples, create and activate a virtual environment:

uv pip install agno openai sqlalchemy
export OPENAI_API_KEY="your-api-key"

Workflow sessions store run records with inputs, outputs, step results, state, and status. A run may be partial if it pauses, fails, or is cancelled. Injected workflow history selects prior completed runs and uses their input/final-output pairs.

Session TypeStores
Agent/Team sessionsConversation history (messages)
Workflow sessionsExecution records, including partial runs

When to Use Workflow Sessions

Use workflow sessions when you need to:

  • Track workflow execution history across multiple runs
  • Share state between steps in a workflow (like passing data between pipeline stages)
  • Enable workflows to learn from previous runs by accessing past inputs and outputs
  • Persist workflow results for analysis and debugging
  • Maintain context across multiple workflow executions

In most cases it is recommended to add session persistence to your workflow.

The fragments below reuse this setup:

from agno.agent import Agent
from agno.models.openai import OpenAIResponses
from agno.workflow import Step

summary_agent = Agent(
    name="Summarizer",
    model=OpenAIResponses(id="gpt-5.2"),
    instructions="Explain the supplied topic using the available context.",
)

How Workflow Sessions Work

When you create a workflow with a database, Agno automatically manages sessions for you:

from agno.workflow import Workflow
from agno.db.sqlite import SqliteDb

workflow = Workflow(
    name="Research Pipeline",
    db=SqliteDb(db_file="workflows.db"),
    steps=[Step("Summary", agent=summary_agent, num_history_runs=5)],
)

# Each run creates or updates the workflow session
result = workflow.run(input="AI trends", session_id="session_123")

Each time you run the workflow, Agno:

  1. Creates a unique run_id for this execution
  2. Stores the input, available output and step results, and status
  3. Updates the session with the new run
  4. Makes the history available for future runs

Workflow Session Structure

This abbreviated schema illustrates the stored fields; import WorkflowSession from agno.session in application code:

@dataclass
class WorkflowSession:
    session_id: str           # Unique session identifier
    user_id: str | None       # User who owns this session
    workflow_id: str | None   # Which workflow this belongs to
    workflow_name: str | None # Name of the workflow
    
    # List of all workflow runs (executions)
    runs: List[WorkflowRunOutput] | None
    
    # Session-specific data
    session_data: Dict | None    # Includes session_name, session_state
    workflow_data: Dict | None   # Workflow configuration
    metadata: Dict | None        # Custom metadata
    
    created_at: int | None    # Unix timestamp
    updated_at: int | None    # Unix timestamp

Unlike agent sessions, workflow sessions don't have a summary field. Workflows store run records instead of creating summaries.

What Gets Stored

Each workflow run can store:

  • Input: The data passed to workflow.run()
  • Output: The final result from the workflow
  • Step results: Output from each step in the pipeline
  • Run data: Execution time, status, metrics
  • Session state: Shared data between steps (if used)

Key Differences from Agent/Team Sessions

If you're familiar with agent or team sessions, here are the main differences:

FeatureAgent/Team SessionsWorkflow Sessions
What's storedMessages and conversation turnsWorkflow run records with available step results
History typeMessage-based (chat history)Run-based (execution history)
SummariesSupported with enable_session_summariesNot supported (stores run records)
History formatMessages in LLM contextPrevious run results prepended to step inputs

Database Options

Workflow sessions require a database to persist execution history. Agno supports multiple database types:

from agno.db.sqlite import SqliteDb

# Quick start - SQLite
workflow = Workflow(
    name="Research Pipeline",
    db=SqliteDb(db_file="workflows.db"),
    steps=[Step("Summary", agent=summary_agent, num_history_runs=5)],
)

Database Configuration Guide

See all supported databases, connection options, and production recommendations

Workflow History

Workflow history supplies previous completed runs' input/final-output pairs to agent/team step inputs. It excludes paused, error, and cancelled runs and does not include every intermediate message.

Enable history for your steps

from agno.workflow import Workflow
from agno.db.sqlite import SqliteDb

workflow = Workflow(
    name="Content Pipeline",
    db=SqliteDb(db_file="workflows.db"),
    steps=[Step("Summary", agent=summary_agent, num_history_runs=5)],
    add_workflow_history_to_steps=True,  # Include previous runs
    num_history_runs=5,                  # Also set explicitly on Step above
)

Why it helps:

  • Access previous runs instead of repeating work
  • Reference past decisions to keep outputs consistent
  • Maintain context across multi-run workflows
  • Build on prior results for richer analysis

Set the window explicitly on each Step: its default of three currently overrides a larger Workflow-level count. Custom functions choose their own num_runs through StepInput history helpers.

History format

Agno wraps past runs in a structured XML block before inserting it into each step input:

<workflow_history_context>
[Workflow Run-1]
User input: Create a blog post about AI
Workflow output: [Full output from run]

[Workflow Run-2]
User input: Write about machine learning
Workflow output: [Full output from run]
</workflow_history_context>

See the workflow history implementation guide for advanced controls, per-step overrides, and programmatic access patterns.

Session Naming

Give your workflow sessions meaningful names for easier identification:

Manual Naming

from agno.workflow import Workflow
from agno.db.sqlite import SqliteDb

workflow = Workflow(
    name="Research Pipeline",
    db=SqliteDb(db_file="workflows.db"),
    steps=[Step("Summary", agent=summary_agent, num_history_runs=5)],
)

workflow.run(input="Analyze AI trends", session_id="session_123")
workflow.set_session_name(session_id="session_123", session_name="AI Trends Analysis Q4 2024")

# Retrieve the name
name = workflow.get_session_name(session_id="session_123")
print(name)  # "AI Trends Analysis Q4 2024"

Auto-Generation

Workflow sessions can auto-generate timestamp-based names:

workflow = Workflow(
    name="Research Pipeline",
    description="Automated research and analysis pipeline",
    db=SqliteDb(db_file="workflows.db"),
    steps=[Step("Summary", agent=summary_agent, num_history_runs=5)],
)

workflow.run(input="Research topic", session_id="session_123")
workflow.set_session_name(session_id="session_123", autogenerate=True)

name = workflow.get_session_name(session_id="session_123")
print(name)  # "Automated research and analysis pipeline - 2024-11-19 14:30"

Next Steps

Now that you understand workflow sessions, explore these features:

Workflow History

Enable workflows to learn from previous runs

Session State

Share data between workflow steps across runs

Session Management

Manage session names, IDs, and organization

Conversational Workflows

Make workflows interactive with WorkflowAgent

Metrics

Track workflow performance and usage

Database Setup

Configure databases for session storage

Developer Resources