feat(multiagent): Add stream_async

[mkmeral]

Description

This PR adds streaming support to the Swarm and Graph multi-agent systems, enabling real-time event emission during multi-agent execution. This brings multi-agent systems to feature parity with the single Agent class streaming capabilities.

Key Changes

New Event Types (src/strands/types/_events.py):

• MultiAgentNodeStartEvent: Emitted when a node begins execution
• MultiAgentNodeCompleteEvent: Emitted when a node completes execution
• MultiAgentNodeStreamEvent: Forwards agent events with node context
• MultiAgentHandoffEvent: Emitted during agent handoffs in Swarm (includes from_node, to_node, and message)

Swarm Streaming (src/strands/multiagent/swarm.py):

• Added stream_async() method that yields events during execution
• Refactored invoke_async() to use stream_async() internally (maintains backward compatibility)
• Events include node start/complete, forwarded agent events, handoff notifications, and final result
• Proper event emission even during failures

Graph Streaming (src/strands/multiagent/graph.py):

• Added stream_async() method for real-time event streaming
• Refactored invoke_async() to consume stream_async() events
• Supports streaming from parallel node execution
• Events maintain node context throughout execution

Testing:

• Comprehensive test coverage for streaming functionality in both Swarm and Graph
• Tests for parallel execution, handoffs, failures, and timeouts
• Backward compatibility tests to ensure existing code continues to work

Benefits

• Real-time visibility into multi-agent execution progress
• Consistent streaming API across single and multi-agent systems
• Better debugging and monitoring capabilities
• Foundation for UI progress indicators and live updates

Related Issues

• [FEATURE] Agent Graph to allow streamed response #608
• [FEATURE] Multi-Agent Streaming Responses Feature Request #912

Documentation PR

Type of Change

New feature

Testing

How have you tested the change?

• Added comprehensive unit tests for streaming in both Swarm and Graph (tests/strands/multiagent/test_swarm.py, tests/strands/multiagent/test_graph.py)
• Tests cover: basic streaming, parallel execution, handoffs, failures, timeouts, and backward compatibility
• All existing tests pass, confirming backward compatibility

Verify that the changes do not break functionality or introduce warnings in consuming repositories: agents-docs, agents-tools, agents-cli

• I ran hatch run prepare

Checklist

• I have read the CONTRIBUTING document
• I have added any necessary tests that prove my fix is effective or my feature works
• I have updated the documentation accordingly
• I have added an appropriate example to the documentation to outline the feature, or no new docs are needed
• My changes generate no new warnings
• Any dependent changes have been merged and published

---

By submitting this pull request, I confirm that you can use, modify, copy, and redistribute this contribution, under the terms of your choice.

[codecov]

Codecov Report

❌ Patch coverage is 89.05473% with 22 lines in your changes missing coverage. Please review.

Files with missing lines	Patch %	Lines
src/strands/multiagent/graph.py	91.37%	6 Missing and 4 partials ⚠️
src/strands/multiagent/swarm.py	84.12%	6 Missing and 4 partials ⚠️
src/strands/multiagent/base.py	50.00%	2 Missing ⚠️

📢 Thoughts on this report? Let us know!

[zastrowm]

The typed events shape & fields are a blocker for me

[mkmeral]

Multi-Agent System Events (Updated: Oct 17)

This document describes the events emitted by multi-agent systems (Graph and Swarm) during execution via their stream_async methods.

All examples are based on the actual event structure defined in the code.

Overview

Multi-agent systems emit events that provide visibility into:

• Node execution lifecycle (start, progress, stop)
• Agent coordination (handoffs between nodes)
• Nested agent events with context
• Final execution results

All events are emitted as dictionaries through the stream_async async generator.

Event Types

1. Node Start Event

Emitted when a node begins execution.

Structure:

{
    "type": "multiagent_node_start",
    "node_id": "agent1",
    "node_type": "agent"  # or "multiagent" for nested Graph/Swarm
}

When emitted:

• Graph: Before each node (agent or nested multi-agent) starts executing
• Swarm: Before each agent in the swarm starts executing

Example:

{
    "type": "multiagent_node_start",
    "node_id": "researcher",
    "node_type": "agent"
}

---

2. Node Stream Event

Emitted during node execution to forward agent events with node context. This wraps ALL agent events.

Structure:

{
    "type": "multiagent_node_stream",
    "node_id": "agent1",
    "event": {
        # Original agent event (nested)
    }
}

When emitted:

• Graph: Forwards all events from executing agents/multi-agents
• Swarm: Forwards all events from the currently executing agent

Nested event types (in the "event" field):

• Lifecycle: {"init_event_loop": True}, {"start": True}, {"start_event_loop": True}
• Raw model events: {"event": {"messageStart": {...}}}, {"event": {"contentBlockDelta": {...}}}
• Text streaming: {"data": "text", "delta": {...}}
• Tool events: {"tool_result_event": {...}}
• Messages: {"message": {...}}
• Agent results: {"result": AgentResult(...)}

Examples:

Lifecycle event:

{
    "type": "multiagent_node_stream",
    "node_id": "researcher",
    "event": {
        "init_event_loop": True
    }
}

Text streaming:

{
    "type": "multiagent_node_stream",
    "node_id": "researcher",
    "event": {
        "data": "Hello! I'm researching...",
        "delta": {"text": "Hello! I'm researching..."},
        "agent": "<Agent object>",
        "event_loop_cycle_id": "uuid...",
        "request_state": {}
    }
}

Raw model event:

{
    "type": "multiagent_node_stream",
    "node_id": "researcher",
    "event": {
        "event": {
            "contentBlockDelta": {
                "delta": {"text": "Hello"},
                "contentBlockIndex": 0
            }
        }
    }
}

Agent result:

{
    "type": "multiagent_node_stream",
    "node_id": "researcher",
    "event": {
        "result": AgentResult(
            stop_reason="end_turn",
            message={"role": "assistant", "content": [{"text": "Research complete"}]},
            metrics=EventLoopMetrics(...),
            state={}
        )
    }
}

---

3. Node Stop Event

Emitted when a node stops execution (success or failure).

Similar to EventLoopStopEvent for single agents, this event provides the complete NodeResult which contains all execution details.

Structure:

{
    "type": "multiagent_node_stop",
    "node_id": "agent1",
    "node_result": NodeResult(...)
}

NodeResult Fields:

• result: AgentResult (for Agent nodes) or MultiAgentResult (for nested multi-agent nodes) or Exception (for failures)
• execution_time: Total execution time in milliseconds
• status: Status.COMPLETED or Status.FAILED
• accumulated_usage: Token usage dict with inputTokens, outputTokens, totalTokens
• accumulated_metrics: Performance metrics dict with latencyMs
• execution_count: Number of executions

For Agent nodes, AgentResult contains:

• stop_reason: Why the agent stopped (e.g., "end_turn", "max_tokens")
• message: Final message from the agent
• metrics: EventLoopMetrics with cycle_count, accumulated_usage, accumulated_metrics
• state: Agent state

When emitted:

• Graph: After each node finishes (even on timeout/failure)
• Swarm: After each agent finishes executing

Example (Success):

{
    "type": "multiagent_node_stop",
    "node_id": "researcher",
    "node_result": NodeResult(
        result=AgentResult(
            stop_reason="end_turn",
            message={
                "role": "assistant",
                "content": [{"text": "Research findings: ..."}]
            },
            metrics=EventLoopMetrics(
                cycle_count=1,
                accumulated_usage={"inputTokens": 18, "outputTokens": 23, "totalTokens": 41},
                accumulated_metrics={"latencyMs": 1975}
            ),
            state={}
        ),
        execution_time=2340,
        status=Status.COMPLETED,
        accumulated_usage={"inputTokens": 18, "outputTokens": 23, "totalTokens": 41},
        accumulated_metrics={"latencyMs": 1975},
        execution_count=1
    )
}

Example (Failure):

{
    "type": "multiagent_node_stop",
    "node_id": "researcher",
    "node_result": NodeResult(
        result=Exception("Node execution failed"),
        execution_time=500,
        status=Status.FAILED,
        accumulated_usage={"inputTokens": 0, "outputTokens": 0, "totalTokens": 0},
        accumulated_metrics={"latencyMs": 500},
        execution_count=1
    )
}

---

4. Handoff Event

Emitted during node transitions in multi-agent systems.

Structure:

{
    "type": "multiagent_handoff",
    "from_nodes": ["node1"],  # List of completing nodes
    "to_nodes": ["node2"],    # List of starting nodes
    "message": "Optional handoff message"  # Only present if provided
}

When emitted:

• Swarm: When an agent uses the handoff_to_agent tool to transfer control
  • from_nodes: Single-element list with current agent
  • to_nodes: Single-element list with target agent
  • message: Handoff reason/message
• Graph: During batch transitions between graph layers
  • from_nodes: List of nodes that just completed
  • to_nodes: List of nodes starting next
  • message: Not typically used

Example (Swarm):

{
    "type": "multiagent_handoff",
    "from_nodes": ["researcher"],
    "to_nodes": ["analyst"],
    "message": "Need calculations for the data"
}

Example (Graph batch transition):

{
    "type": "multiagent_handoff",
    "from_nodes": ["node1", "node2"],
    "to_nodes": ["node3", "node4"]
}

---

5. Final Result Event

Emitted when multi-agent execution completes with final result.

Structure:

{
    "type": "multiagent_result",
    "result": GraphResult(...) or SwarmResult(...)
}

When emitted:

• Graph: After all nodes complete or execution stops
• Swarm: After swarm completes or reaches termination condition

Key Result Fields:

• status: Status.COMPLETED or Status.FAILED
• results: Dict mapping node_id to NodeResult
• accumulated_usage: Total tokens across all nodes (inputTokens, outputTokens, totalTokens)
• accumulated_metrics: Total latency across all nodes (latencyMs)
• execution_count: Total number of node executions
• execution_time: Total wall-clock time in milliseconds

GraphResult additional fields:

• total_nodes: Total number of nodes in graph
• completed_nodes: Number of successfully completed nodes
• failed_nodes: Number of failed nodes
• execution_order: List of node IDs in execution order
• edges: Graph edges
• entry_points: Graph entry point node IDs

SwarmResult additional fields:

• node_history: List of SwarmNode objects showing execution path

Example (Graph):

{
    "type": "multiagent_result",
    "result": GraphResult(
        status=Status.COMPLETED,
        results={
            "researcher": NodeResult(...),
            "analyst": NodeResult(...)
        },
        accumulated_usage={"inputTokens": 75, "outputTokens": 49, "totalTokens": 124},
        accumulated_metrics={"latencyMs": 3897},
        execution_count=2,
        execution_time=4586,
        total_nodes=2,
        completed_nodes=2,
        failed_nodes=0,
        execution_order=["researcher", "analyst"],
        edges=[("researcher", "analyst")],
        entry_points=["researcher"]
    )
}

Example (Swarm):

{
    "type": "multiagent_result",
    "result": SwarmResult(
        status=Status.COMPLETED,
        results={
            "researcher": NodeResult(...)
        },
        accumulated_usage={"inputTokens": 602, "outputTokens": 33, "totalTokens": 635},
        accumulated_metrics={"latencyMs": 2641},
        execution_count=1,
        execution_time=2958,
        node_history=[SwarmNode(node_id="researcher")]
    )
}

---

Complete Event Flow Examples

Graph Execution

async for event in graph.stream_async("Analyze this data"):
    # Event 1: First node starts
    # {
    #     "type": "multiagent_node_start",
    #     "node_id": "researcher",
    #     "node_type": "agent"
    # }
    
    # Events 2-15: Researcher agent lifecycle and streaming events (wrapped)
    # {
    #     "type": "multiagent_node_stream",
    #     "node_id": "researcher",
    #     "event": {"init_event_loop": True}
    # }
    # {
    #     "type": "multiagent_node_stream",
    #     "node_id": "researcher",
    #     "event": {"start": True}
    # }
    # {
    #     "type": "multiagent_node_stream",
    #     "node_id": "researcher",
    #     "event": {"data": "Researching...", "delta": {...}}
    # }
    # ... (more streaming events)
    # {
    #     "type": "multiagent_node_stream",
    #     "node_id": "researcher",
    #     "event": {"result": AgentResult(...)}
    # }
    
    # Event 16: First node stops
    # {
    #     "type": "multiagent_node_stop",
    #     "node_id": "researcher",
    #     "node_result": NodeResult(...)
    # }
    
    # Event 17: Handoff to next batch
    # {
    #     "type": "multiagent_handoff",
    #     "from_nodes": ["researcher"],
    #     "to_nodes": ["analyst"]
    # }
    
    # Event 18: Second node starts
    # {
    #     "type": "multiagent_node_start",
    #     "node_id": "analyst",
    #     "node_type": "agent"
    # }
    
    # Events 19-32: Analyst agent events (similar to researcher)
    # ... (lifecycle, streaming, result)
    
    # Event 33: Second node stops
    # {
    #     "type": "multiagent_node_stop",
    #     "node_id": "analyst",
    #     "node_result": NodeResult(...)
    # }
    
    # Event 34: Final result
    # {
    #     "type": "multiagent_result",
    #     "result": GraphResult(...)
    # }

Swarm Execution

async for event in swarm.stream_async("Calculate 10 + 5"):
    # Event 1: First agent starts
    # {
    #     "type": "multiagent_node_start",
    #     "node_id": "researcher",
    #     "node_type": "agent"
    # }
    
    # Events 2-20: Researcher agent events (wrapped)
    # {
    #     "type": "multiagent_node_stream",
    #     "node_id": "researcher",
    #     "event": {"data": "I need calculations...", "delta": {...}}
    # }
    # ... (more streaming events including handoff tool use)
    
    # Event 21: Researcher stops
    # {
    #     "type": "multiagent_node_stop",
    #     "node_id": "researcher",
    #     "node_result": NodeResult(...)
    # }
    
    # Event 22: Handoff to analyst
    # {
    #     "type": "multiagent_handoff",
    #     "from_nodes": ["researcher"],
    #     "to_nodes": ["analyst"],
    #     "message": "Need calculations"
    # }
    
    # Event 23: Analyst starts
    # {
    #     "type": "multiagent_node_start",
    #     "node_id": "analyst",
    #     "node_type": "agent"
    # }
    
    # Events 24-40: Analyst agent events
    # ... (lifecycle, streaming, result)
    
    # Event 41: Analyst stops
    # {
    #     "type": "multiagent_node_stop",
    #     "node_id": "analyst",
    #     "node_result": NodeResult(...)
    # }
    
    # Event 42: Final result
    # {
    #     "type": "multiagent_result",
    #     "result": SwarmResult(...)
    # }

---

Usage Patterns

Monitoring Progress

async for event in graph.stream_async(task):
    if event.get("type") == "multiagent_node_start":
        print(f"Starting: {event['node_id']}")
    
    elif event.get("type") == "multiagent_node_stop":
        node_result = event["node_result"]
        print(f"Stopped: {event['node_id']} in {node_result.execution_time}ms")

Collecting Text Output

outputs = []
async for event in graph.stream_async(task):
    if event.get("type") == "multiagent_node_stream":
        nested = event["event"]
        if "data" in nested:
            outputs.append({
                "node": event["node_id"],
                "text": nested["data"]
            })

Getting Node Results

results = {}
async for event in graph.stream_async(task):
    if event.get("type") == "multiagent_node_stop":
        node_result = event["node_result"]
        results[event["node_id"]] = node_result

Tracking Handoffs

handoffs = []
async for event in swarm.stream_async(task):
    if event.get("type") == "multiagent_handoff":
        handoffs.append({
            "from": event["from_nodes"],
            "to": event["to_nodes"],
            "message": event.get("message")
        })

Real-time Streaming Display

async for event in graph.stream_async(task):
    if event.get("type") == "multiagent_node_stream":
        nested = event["event"]
        
        # Display text as it streams
        if "data" in nested:
            print(nested["data"], end="", flush=True)
        
        # Show when agent completes
        if "result" in nested:
            print(f"\n[{event['node_id']} completed]")

Access Node Results Immediately

async for event in graph.stream_async(task):
    if event.get("type") == "multiagent_node_stop":
        node_id = event["node_id"]
        node_result = event["node_result"]
        
        # Check status
        if node_result.status == Status.FAILED:
            print(f"❌ {node_id} failed: {node_result.result}")
            continue
        
        # For successful Agent nodes, access AgentResult
        agent_result = node_result.result
        if hasattr(agent_result, "stop_reason"):
            print(f"{node_id} stopped: {agent_result.stop_reason}")
        
        if hasattr(agent_result, "message"):
            content = agent_result.message.get("content", [])
            for item in content:
                if "text" in item:
                    print(f"{node_id} output: {item['text']}")
        
        # Access accumulated metrics
        print(f"{node_id} tokens: {node_result.accumulated_usage}")

---

Event Identification Patterns

To identify event types, check the "type" field:

async for event in graph.stream_async(task):
    event_type = event.get("type")
    
    if event_type == "multiagent_node_start":
        # Node starting
        node_id = event["node_id"]
        node_type = event["node_type"]
    
    elif event_type == "multiagent_node_stream":
        # Forwarded agent event
        node_id = event["node_id"]
        agent_event = event["event"]
        
        # Check nested event type
        if "init_event_loop" in agent_event:
            # Agent lifecycle start
            pass
        elif "data" in agent_event:
            # Text streaming
            text = agent_event["data"]
        elif "result" in agent_event:
            # Agent result
            result = agent_event["result"]
        elif "message" in agent_event:
            # Agent message
            message = agent_event["message"]
    
    elif event_type == "multiagent_node_stop":
        # Node stopped
        node_id = event["node_id"]
        node_result = event["node_result"]
    
    elif event_type == "multiagent_handoff":
        # Node transition
        from_nodes = event["from_nodes"]
        to_nodes = event["to_nodes"]
        message = event.get("message")  # Optional
    
    elif event_type == "multiagent_result":
        # Final result
        final_result = event["result"]

---

Nested Multi-Agent Systems

When a Graph or Swarm contains another Graph or Swarm as a node, events are nested:

# Outer graph with nested graph as a node
async for event in outer_graph.stream_async(task):
    if event.get("type") == "multiagent_node_stream":
        nested = event["event"]
        
        # The nested event might itself be a multi-agent event
        if nested.get("type") == "multiagent_node_start":
            # Nested graph's node is starting
            print(f"Nested node starting: {nested['node_id']}")
        
        elif nested.get("type") == "multiagent_node_stream":
            # Doubly nested event from agent in nested graph
            doubly_nested = nested["event"]
            if "data" in doubly_nested:
                print(f"Text from nested agent: {doubly_nested['data']}")

This allows you to track execution at any depth of nesting while maintaining context about which node at each level is generating the events.

[Unshure]

Reviewed the logic and the integ tests. Will review the unit tests after the comments are addressed.