Workflow Metrics

Workflow, step, and session metrics for token usage and performance.

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

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

Token usage, timing, and step breakdowns from workflow runs.

from agno.agent import Agent
from agno.db.sqlite import SqliteDb
from agno.models.openai import OpenAIResponses
from agno.team import Team
from agno.tools.hackernews import HackerNewsTools
from agno.tools.yfinance import YFinanceTools
from agno.workflow import Step, Workflow
from rich.pretty import pprint

hackernews_agent = Agent(
    name="HackerNews Agent",
    model=OpenAIResponses(id="gpt-5.2"),
    tools=[HackerNewsTools()],
    role="Extract key insights from HackerNews posts",
)

finance_agent = Agent(
    name="Finance Agent",
    model=OpenAIResponses(id="gpt-5.2"),
    tools=[YFinanceTools()],
    role="Get stock prices and financial data",
)

research_team = Team(
    name="Research Team",
    members=[hackernews_agent, finance_agent],
    instructions="Research tech topics from HackerNews and financial data",
)

content_planner = Agent(
    name="Content Planner",
    model=OpenAIResponses(id="gpt-5.2"),
    instructions="Plan a content schedule based on research",
)

workflow = Workflow(
    name="Content Creation Workflow",
    db=SqliteDb(db_file="tmp/workflow.db"),
    steps=[
        Step(name="Research Step", team=research_team),
        Step(name="Content Planning Step", agent=content_planner),
    ],
)

response = workflow.run(input="AI trends in 2024")

# Workflow metrics
if response.metrics:
    pprint(response.metrics.to_dict())

# Step metrics
if response.metrics:
    for step_name, step_metrics in response.metrics.steps.items():
        print(f"{step_name}: {step_metrics.executor_name} ({step_metrics.executor_type})")
        if step_metrics.metrics:
            print(f"  Duration: {step_metrics.metrics.duration:.2f}s, Tokens: {step_metrics.metrics.total_tokens}")

# Session metrics
pprint(workflow.get_session_metrics().to_dict())

Metrics are tracked at three levels:

LevelTypeAccess
WorkflowWorkflowMetricsresponse.metrics
Per stepRunMetrics (via StepMetrics)response.metrics.steps["step_name"].metrics
Per sessionSessionMetricsworkflow.get_session_metrics()

Workflow fields (WorkflowMetrics)

FieldDescription
durationTotal workflow execution time in seconds (includes orchestration overhead).
stepsDictionary mapping step names to StepMetrics.

Step fields (StepMetrics)

FieldDescription
step_nameName of the step.
executor_type"agent" or "team". Other step types (functions, parallel blocks, nested workflows) are excluded from aggregated workflow metrics.
executor_nameName of the executor.
metricsRunMetrics with tokens, duration, and model information. See Metrics reference.

Session metrics aggregate token usage and cost from step metrics across all runs in the session. They do not include timing.

Developer Resources