output_schema on each agent to validate that step’s output as a Pydantic model. The workflow passes each result to the next step. This example moves from ResearchFindings to ContentStrategy to FinalContentPlan.
Install dependencies:
uv pip install agno openai yfinance
structured_io_at_each_step_level_agent.py
from typing import List
from agno.agent import Agent
from agno.models.openai import OpenAIResponses
from agno.tools.hackernews import HackerNewsTools
from agno.tools.yfinance import YFinanceTools
from agno.workflow.step import Step
from agno.workflow.workflow import Workflow
from pydantic import BaseModel, Field
# Define structured models for each step
class ResearchFindings(BaseModel):
"""Structured research findings with key insights"""
topic: str = Field(description="The research topic")
key_insights: List[str] = Field(description="Main insights discovered", min_items=3)
trending_technologies: List[str] = Field(
description="Technologies that are trending", min_items=2
)
market_impact: str = Field(description="Potential market impact analysis")
sources_count: int = Field(description="Number of sources researched")
confidence_score: float = Field(
description="Confidence in findings (0.0-1.0)", ge=0.0, le=1.0
)
class ContentStrategy(BaseModel):
"""Structured content strategy based on research"""
target_audience: str = Field(description="Primary target audience")
content_pillars: List[str] = Field(description="Main content themes", min_items=3)
posting_schedule: List[str] = Field(description="Recommended posting schedule")
key_messages: List[str] = Field(
description="Core messages to communicate", min_items=3
)
hashtags: List[str] = Field(description="Recommended hashtags", min_items=5)
engagement_tactics: List[str] = Field(
description="Ways to increase engagement", min_items=2
)
class FinalContentPlan(BaseModel):
"""Final content plan with specific deliverables"""
campaign_name: str = Field(description="Name for the content campaign")
content_calendar: List[str] = Field(
description="Specific content pieces planned", min_items=6
)
success_metrics: List[str] = Field(
description="How to measure success", min_items=3
)
budget_estimate: str = Field(description="Estimated budget range")
timeline: str = Field(description="Implementation timeline")
risk_factors: List[str] = Field(
description="Potential risks and mitigation", min_items=2
)
# Define agents with response models
research_agent = Agent(
name="AI Research Specialist",
model=OpenAIResponses(id="gpt-5.2"),
tools=[HackerNewsTools(), YFinanceTools()],
role="Research AI trends and extract structured insights",
output_schema=ResearchFindings,
instructions=[
"Research the given topic thoroughly using available tools",
"Provide structured findings with confidence scores",
"Focus on recent developments and market trends",
"Make sure to structure your response according to the ResearchFindings model",
],
)
strategy_agent = Agent(
name="Content Strategy Expert",
model=OpenAIResponses(id="gpt-5.2"),
role="Create content strategies based on research findings",
output_schema=ContentStrategy,
instructions=[
"Analyze the research findings provided from the previous step",
"Create a comprehensive content strategy based on the structured research data",
"Focus on audience engagement and brand building",
"Structure your response according to the ContentStrategy model",
],
)
planning_agent = Agent(
name="Content Planning Specialist",
model=OpenAIResponses(id="gpt-5.2"),
role="Create detailed content plans and calendars",
output_schema=FinalContentPlan,
instructions=[
"Use the content strategy from the previous step to create a detailed implementation plan",
"Include specific timelines and success metrics",
"Consider budget and resource constraints",
"Structure your response according to the FinalContentPlan model",
],
)
# Define steps
research_step = Step(
name="research_insights",
agent=research_agent,
)
strategy_step = Step(
name="content_strategy",
agent=strategy_agent,
)
planning_step = Step(
name="final_planning",
agent=planning_agent,
)
# Create workflow
structured_workflow = Workflow(
name="Structured Content Creation Pipeline",
description="AI-powered content creation with structured data flow",
steps=[research_step, strategy_step, planning_step],
)
if __name__ == "__main__":
print("=== Testing Structured Output Flow Between Steps ===")
# Test with simple string input
structured_workflow.print_response(
input="Latest developments in artificial intelligence and machine learning"
)