feat(multiagent): add structured_output_model support to Graph

[guruhariharaun]

Add Structured Output Support to Graph Multi-Agent Pattern

Description

This PR adds support for structured_output_model parameter to the Graph multi-agent pattern, enabling structured output from all agent nodes in a graph. This addresses the feature request in #538.

The implementation allows users to:

• Pass a structured_output_model at the graph level when invoking the graph
• Set per-agent structured_output_model during agent initialization
• Access structured output from each node's result

Changes

Core Implementation

• Added structured_output_model parameter to:

  • Graph.__call__()
  • Graph.invoke_async()
  • Graph.stream_async()

• Execution chain updates:

  • _execute_graph() - accepts and passes structured_output_model
  • _execute_nodes_parallel() - propagates to parallel node execution
  • _stream_node_to_queue() - forwards to individual nodes
  • _execute_node() - uses agent's default or graph-level model

• Per-agent support:

  • Checks if agent has _default_structured_output_model set
  • Uses agent's default if available, otherwise falls back to graph-level
  • Works seamlessly with nested multi-agent systems (Graph/Swarm as nodes)

Code Changes

File: src/strands/multiagent/graph.py

• Added Type[BaseModel] import from pydantic
• Updated method signatures to accept structured_output_model parameter
• Modified execution chain to pass structured_output_model through all layers
• Added logic to prioritize agent's default model over graph-level model

File: tests/strands/multiagent/test_graph.py

• Added TestGraphStructuredOutput test class with 3 comprehensive tests:
  1. test_graph_passes_structured_output_model_to_agents - verifies parameter passing
  2. test_graph_structured_output_sync - tests synchronous invocation
  3. test_graph_structured_output_multiple_nodes - tests multiple nodes

Usage Examples

Example 1: Graph-Level Structured Output

from pydantic import BaseModel, Field
from strands import Agent
from strands.multiagent import GraphBuilder

class SummaryOutput(BaseModel):
    """Summary output structure"""
    summary: str = Field(description="A brief summary")
    key_points: list[str] = Field(description="List of key points")

# Create agents
researcher = Agent(name="researcher", system_prompt="You are a research specialist...")
analyst = Agent(name="analyst", system_prompt="You are a data analysis specialist...")

# Build graph
builder = GraphBuilder()
builder.add_node(researcher, "research")
builder.add_node(analyst, "analysis")
builder.add_edge("research", "analysis")
graph = builder.build()

# Execute with structured output
result = graph(
    "Research the impact of AI on healthcare",
    structured_output_model=SummaryOutput
)

# Access structured output from each node
for node_id, node_result in result.results.items():
    if node_result.result.structured_output:
        print(f"{node_id}: {node_result.result.structured_output}")

Example 2: Per-Agent Structured Output

from pydantic import BaseModel, Field
from strands import Agent
from strands.multiagent import GraphBuilder

class MathOutput(BaseModel):
    """Math-specific output"""
    topic: str = Field(description="The mathematical topic")
    explanation: str = Field(description="Clear explanation")
    formula: str = Field(description="Relevant formula")

class ComputerOutput(BaseModel):
    """Computer science-specific output"""
    technology: str = Field(description="The technology or concept")
    description: str = Field(description="Description")
    use_cases: list[str] = Field(description="Common use cases")

# Create agents with their own structured output models
math_agent = Agent(
    name="math_agent",
    system_prompt="You are a mathematics specialist...",
    structured_output_model=MathOutput  # Agent-specific model
)

computer_agent = Agent(
    name="computer_agent",
    system_prompt="You are a computer science specialist...",
    structured_output_model=ComputerOutput  # Agent-specific model
)

# Build graph
builder = GraphBuilder()
builder.add_node(math_agent, "math")
builder.add_node(computer_agent, "computer")
graph = builder.build()

# Execute - each agent uses its own structured output model
result = graph("Explain 2+2 and Python programming")

# Access structured output
math_output = result.results["math"].result.structured_output  # MathOutput instance
computer_output = result.results["computer"].result.structured_output  # ComputerOutput instance

Example 3: Mixed Approach (Agent Default + Graph Override)

# Agent with default structured output model
agent = Agent(
    name="researcher",
    structured_output_model=DefaultOutput  # Default model
)

builder = GraphBuilder()
builder.add_node(agent, "research")
graph = builder.build()

# Graph-level model overrides agent's default (if agent doesn't have one)
# But if agent has a default, it uses that instead
result = graph("Research topic", structured_output_model=OverrideOutput)

Testing

Test Coverage

Added comprehensive test suite in TestGraphStructuredOutput:

1. test_graph_passes_structured_output_model_to_agents

   • Verifies that structured_output_model is correctly passed from Graph to Agent
   • Confirms structured output is accessible in the result
   • Tests async invocation

2. test_graph_structured_output_sync

   • Tests synchronous __call__() method
   • Verifies structured output works with sync API

3. test_graph_structured_output_multiple_nodes

   • Tests multiple nodes executing in parallel
   • Verifies each node receives and uses structured_output_model
   • Confirms structured output is available for all nodes

Test Results

✅ All 3 tests pass
✅ Code formatted with `hatch fmt --formatter`
✅ Linting passes with `hatch fmt --linter`
✅ No breaking changes

Manual Testing

• Tested with real AWS Bedrock models using Claude Sonnet 4
• Verified structured output works with multiple agents in parallel
• Confirmed per-agent models work correctly
• Tested with nested graph structures

Implementation Details

Priority Order

When determining which structured_output_model to use:

1. Agent's default (agent._default_structured_output_model) - highest priority
2. Graph-level (graph(..., structured_output_model=Model)) - fallback
3. None - no structured output

This allows maximum flexibility:

• Set defaults per agent for consistent behavior
• Override at graph level when needed
• Mix both approaches in the same graph

Backward Compatibility

✅ Fully backward compatible

• All existing code continues to work without changes
• structured_output_model is an optional parameter
• Default behavior (no structured output) is preserved

Nested Multi-Agent Support

Works seamlessly with nested multi-agent patterns:

• Graph nodes containing other Graphs
• Graph nodes containing Swarms
• All receive and can use structured_output_model

Benefits

1. Type Safety: Get validated Pydantic models instead of raw text
2. Consistency: Same structured output API as Agent class
3. Flexibility: Per-agent or graph-level models
4. Composability: Works with all existing Graph features
5. Developer Experience: Clear, intuitive API

Related Issues

Fixes #538

Checklist

• Code follows style guidelines (formatted with hatch fmt --formatter)
• Linting passes (hatch fmt --linter)
• Tests added/updated (3 new comprehensive tests)
• All tests pass
• No breaking changes (fully backward compatible)
• Commit message follows Conventional Commits specification
• Documentation update Will be done once this PR gets merged

Additional Notes

• This implementation follows the same pattern as Agent's structured output support
• Aligns with project's development tenets: Simple, Extensible, Composability