nested_workflow_events.py
"""
Nested Workflow Example - Event Inspection
Runs a nested workflow and prints every workflow/step event with full details
so you can see workflow_id, workflow_name, nested_depth, and
how inner vs outer events differ.
"""
from agno.agent import Agent
from agno.db.postgres import PostgresDb
from agno.models.anthropic import Claude
from agno.run.workflow import (
BaseWorkflowRunOutputEvent,
StepCompletedEvent,
StepStartedEvent,
WorkflowCompletedEvent,
WorkflowStartedEvent,
)
from agno.workflow.step import Step
from agno.workflow.types import StepInput, StepOutput
from agno.workflow.workflow import Workflow
def create_summary(step_input: StepInput) -> StepOutput:
"""A simple function step that summarizes the previous step's output"""
previous_content = step_input.get_last_step_content()
summary = (
f"Summary of research:\n{previous_content[:500]}..."
if previous_content
else "No content to summarize"
)
return StepOutput(content=summary)
# --- Inner workflow ---
research_agent = Agent(
name="Research Agent",
model=Claude(id="claude-sonnet-4-20250514"),
instructions="You are a research assistant. Provide concise, factual information in 2-3 sentences.",
)
inner_workflow = Workflow(
name="Inner Workflow",
description="A simple workflow that researches a topic",
steps=[
Step(name="research", agent=research_agent),
Step(name="summary", executor=create_summary),
],
)
# --- Outer workflow ---
writer_agent = Agent(
name="Writer Agent",
model=Claude(id="claude-sonnet-4-20250514"),
instructions="You are a professional writer. Take the research provided and write a short polished paragraph.",
)
outer_workflow = Workflow(
name="Outer Workflow",
description="A workflow that researches a topic and then writes about it",
steps=[
Step(name="research_phase", workflow=inner_workflow),
Step(name="writing_phase", agent=writer_agent),
],
db=PostgresDb(db_url="postgresql+psycopg://ai:ai@localhost:5532/ai"),
)
def print_event_details(event: BaseWorkflowRunOutputEvent, label: str) -> None:
"""Print key fields of a workflow event."""
indent = " " * getattr(event, "nested_depth", 0)
print(f"\n{indent}{'=' * 60}")
print(f"{indent}[{label}]")
print(f"{indent} event type : {type(event).__name__}")
print(f"{indent} workflow_id : {getattr(event, 'workflow_id', None)}")
print(f"{indent} workflow_name : {getattr(event, 'workflow_name', None)}")
print(f"{indent} nested_depth : {getattr(event, 'nested_depth', None)}")
print(f"{indent} run_id : {getattr(event, 'run_id', None)}")
print(f"{indent} session_id : {getattr(event, 'session_id', None)}")
print(f"{indent} step_id : {getattr(event, 'step_id', None)}")
print(f"{indent} step_name : {getattr(event, 'step_name', None)}")
print(f"{indent} step_index : {getattr(event, 'step_index', None)}")
# Extra fields for specific event types
if isinstance(event, (StepCompletedEvent, WorkflowCompletedEvent)):
content = getattr(event, "content", None)
if content:
preview = str(content)[:120].replace("\n", " ")
print(f"{indent} content (preview) : {preview}...")
if isinstance(event, WorkflowCompletedEvent):
step_results = getattr(event, "step_results", None)
if step_results:
print(f"{indent} step_results count: {len(step_results)}")
for i, sr in enumerate(step_results):
sr_name = getattr(sr, "step_name", None) or f"step_{i}"
sr_type = getattr(sr, "executor_type", "?")
sr_metrics = getattr(sr, "metrics", None)
print(
f"{indent} [{i}] {sr_name} (type={sr_type}, metrics={sr_metrics is not None})"
)
print(f"{indent}{'=' * 60}")
if __name__ == "__main__":
print("Running nested workflow example with event inspection...")
print("=" * 60)
EVENT_TYPES = (
WorkflowStartedEvent,
WorkflowCompletedEvent,
StepStartedEvent,
StepCompletedEvent,
)
event_log = []
for event in outer_workflow.run(
input="Tell me about the history of artificial intelligence",
stream=True,
stream_events=True,
):
if isinstance(event, EVENT_TYPES):
label = type(event).__name__
# Tag inner vs outer
source = getattr(event, "workflow_name", None) or "?"
current = getattr(event, "workflow_name", None) or "?"
depth = getattr(event, "nested_depth", 0)
if depth > 0:
label = f"INNER (depth={depth}, source={source}) | {label}"
else:
label = f"OUTER (source={source}) | {label}"
print_event_details(event, label)
event_log.append(event)
# --- Summary ---
print("\n\n" + "=" * 60)
print("EVENT SUMMARY")
print("=" * 60)
print(f"Total workflow/step events captured: {len(event_log)}")
for i, ev in enumerate(event_log):
depth = getattr(ev, "nested_depth", 0)
source = getattr(ev, "workflow_name", None) or "?"
indent = " " * depth
print(
f" {i + 1}. {indent}{type(ev).__name__:<30s} depth={depth} source={source}"
)
Run the Example
1
Set up your virtual environment
uv venv --python 3.12
source .venv/bin/activate
uv venv --python 3.12
.venv\Scripts\activate
2
Install dependencies
uv pip install -U agno anthropic fastapi psycopg-binary sqlalchemy
3
Export your Anthropic API key
export ANTHROPIC_API_KEY="your_anthropic_api_key_here"
$Env:ANTHROPIC_API_KEY="your_anthropic_api_key_here"
4
Run PgVector
docker run -d \
-e POSTGRES_DB=ai \
-e POSTGRES_USER=ai \
-e POSTGRES_PASSWORD=ai \
-e PGDATA=/var/lib/postgresql/data/pgdata \
-v pgvolume:/var/lib/postgresql/data \
-p 5532:5432 \
--name pgvector \
agnohq/pgvector:18
5
Run the example
Save the code above as
nested_workflow_events.py, then run:python nested_workflow_events.py