Skip to content

strands.multiagent

Multiagent capabilities for Strands Agents.

This module provides support for multiagent systems, including agent-to-agent (A2A) communication protocols and coordination mechanisms.

Submodules

a2a: Implementation of the Agent-to-Agent (A2A) protocol, which enables standardized communication between agents.

strands.multiagent.base

Multi-Agent Base Class.

Provides minimal foundation for multi-agent patterns (Swarm, Graph).

MultiAgentBase

Bases: ABC

Base class for multi-agent helpers.

This class integrates with existing Strands Agent instances and provides multi-agent orchestration capabilities.

Attributes:

Name Type Description
id str

Unique MultiAgent id for session management,etc.
Source code in strands/multiagent/base.py 
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
class MultiAgentBase(ABC):
    """Base class for multi-agent helpers.

    This class integrates with existing Strands Agent instances and provides
    multi-agent orchestration capabilities.

    Attributes:
        id: Unique MultiAgent id for session management,etc.
    """

    id: str

    @abstractmethod
    async def invoke_async(
        self, task: MultiAgentInput, invocation_state: dict[str, Any] | None = None, **kwargs: Any
    ) -> MultiAgentResult:
        """Invoke asynchronously.

        Args:
            task: The task to execute
            invocation_state: Additional state/context passed to underlying agents.
                Defaults to None to avoid mutable default argument issues.
            **kwargs: Additional keyword arguments passed to underlying agents.
        """
        raise NotImplementedError("invoke_async not implemented")

    async def stream_async(
        self, task: MultiAgentInput, invocation_state: dict[str, Any] | None = None, **kwargs: Any
    ) -> AsyncIterator[dict[str, Any]]:
        """Stream events during multi-agent execution.

        Default implementation executes invoke_async and yields the result as a single event.
        Subclasses can override this method to provide true streaming capabilities.

        Args:
            task: The task to execute
            invocation_state: Additional state/context passed to underlying agents.
                Defaults to None to avoid mutable default argument issues.
            **kwargs: Additional keyword arguments passed to underlying agents.

        Yields:
            Dictionary events containing multi-agent execution information including:
            - Multi-agent coordination events (node start/complete, handoffs)
            - Forwarded single-agent events with node context
            - Final result event
        """
        # Default implementation for backward compatibility
        # Execute invoke_async and yield the result as a single event
        result = await self.invoke_async(task, invocation_state, **kwargs)
        yield {"result": result}

    def __call__(
        self, task: MultiAgentInput, invocation_state: dict[str, Any] | None = None, **kwargs: Any
    ) -> MultiAgentResult:
        """Invoke synchronously.

        Args:
            task: The task to execute
            invocation_state: Additional state/context passed to underlying agents.
                Defaults to None to avoid mutable default argument issues.
            **kwargs: Additional keyword arguments passed to underlying agents.
        """
        if invocation_state is None:
            invocation_state = {}

        if kwargs:
            invocation_state.update(kwargs)
            warnings.warn("`**kwargs` parameter is deprecating, use `invocation_state` instead.", stacklevel=2)

        return run_async(lambda: self.invoke_async(task, invocation_state))

    def serialize_state(self) -> dict[str, Any]:
        """Return a JSON-serializable snapshot of the orchestrator state."""
        raise NotImplementedError

    def deserialize_state(self, payload: dict[str, Any]) -> None:
        """Restore orchestrator state from a session dict."""
        raise NotImplementedError

    def _parse_trace_attributes(
        self, attributes: Mapping[str, AttributeValue] | None = None
    ) -> dict[str, AttributeValue]:
        trace_attributes: dict[str, AttributeValue] = {}
        if attributes:
            for k, v in attributes.items():
                if isinstance(v, (str, int, float, bool)) or (
                    isinstance(v, list) and all(isinstance(x, (str, int, float, bool)) for x in v)
                ):
                    trace_attributes[k] = v
        return trace_attributes

__call__(task, invocation_state=None, **kwargs)

Invoke synchronously.

Parameters:

Name Type Description Default
task MultiAgentInput

The task to execute

 required
invocation_state dict[str, Any] | None

Additional state/context passed to underlying agents. Defaults to None to avoid mutable default argument issues.

 None
**kwargs Any

Additional keyword arguments passed to underlying agents.

 {}
Source code in strands/multiagent/base.py 
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
def __call__(
    self, task: MultiAgentInput, invocation_state: dict[str, Any] | None = None, **kwargs: Any
) -> MultiAgentResult:
    """Invoke synchronously.

    Args:
        task: The task to execute
        invocation_state: Additional state/context passed to underlying agents.
            Defaults to None to avoid mutable default argument issues.
        **kwargs: Additional keyword arguments passed to underlying agents.
    """
    if invocation_state is None:
        invocation_state = {}

    if kwargs:
        invocation_state.update(kwargs)
        warnings.warn("`**kwargs` parameter is deprecating, use `invocation_state` instead.", stacklevel=2)

    return run_async(lambda: self.invoke_async(task, invocation_state))

deserialize_state(payload)

Restore orchestrator state from a session dict.

Source code in strands/multiagent/base.py 
238
239
240
def deserialize_state(self, payload: dict[str, Any]) -> None:
    """Restore orchestrator state from a session dict."""
    raise NotImplementedError

invoke_async(task, invocation_state=None, **kwargs) abstractmethod async

Invoke asynchronously.

Parameters:

Name Type Description Default
task MultiAgentInput

The task to execute

 required
invocation_state dict[str, Any] | None

Additional state/context passed to underlying agents. Defaults to None to avoid mutable default argument issues.

 None
**kwargs Any

Additional keyword arguments passed to underlying agents.

 {}
Source code in strands/multiagent/base.py 
175
176
177
178
179
180
181
182
183
184
185
186
187
@abstractmethod
async def invoke_async(
    self, task: MultiAgentInput, invocation_state: dict[str, Any] | None = None, **kwargs: Any
) -> MultiAgentResult:
    """Invoke asynchronously.

    Args:
        task: The task to execute
        invocation_state: Additional state/context passed to underlying agents.
            Defaults to None to avoid mutable default argument issues.
        **kwargs: Additional keyword arguments passed to underlying agents.
    """
    raise NotImplementedError("invoke_async not implemented")

serialize_state()

Return a JSON-serializable snapshot of the orchestrator state.

Source code in strands/multiagent/base.py 
234
235
236
def serialize_state(self) -> dict[str, Any]:
    """Return a JSON-serializable snapshot of the orchestrator state."""
    raise NotImplementedError

stream_async(task, invocation_state=None, **kwargs) async

Stream events during multi-agent execution.

Default implementation executes invoke_async and yields the result as a single event. Subclasses can override this method to provide true streaming capabilities.

Parameters:

Name Type Description Default
task MultiAgentInput

The task to execute

 required
invocation_state dict[str, Any] | None

Additional state/context passed to underlying agents. Defaults to None to avoid mutable default argument issues.

 None
**kwargs Any

Additional keyword arguments passed to underlying agents.

 {}

Yields:

Type Description
AsyncIterator[dict[str, Any]]

Dictionary events containing multi-agent execution information including:
AsyncIterator[dict[str, Any]]
  • Multi-agent coordination events (node start/complete, handoffs)
AsyncIterator[dict[str, Any]]
  • Forwarded single-agent events with node context
AsyncIterator[dict[str, Any]]
  • Final result event
Source code in strands/multiagent/base.py 
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
async def stream_async(
    self, task: MultiAgentInput, invocation_state: dict[str, Any] | None = None, **kwargs: Any
) -> AsyncIterator[dict[str, Any]]:
    """Stream events during multi-agent execution.

    Default implementation executes invoke_async and yields the result as a single event.
    Subclasses can override this method to provide true streaming capabilities.

    Args:
        task: The task to execute
        invocation_state: Additional state/context passed to underlying agents.
            Defaults to None to avoid mutable default argument issues.
        **kwargs: Additional keyword arguments passed to underlying agents.

    Yields:
        Dictionary events containing multi-agent execution information including:
        - Multi-agent coordination events (node start/complete, handoffs)
        - Forwarded single-agent events with node context
        - Final result event
    """
    # Default implementation for backward compatibility
    # Execute invoke_async and yield the result as a single event
    result = await self.invoke_async(task, invocation_state, **kwargs)
    yield {"result": result}

MultiAgentResult dataclass

Result from multi-agent execution with accumulated metrics.

The status field represents the outcome of the MultiAgentBase execution: - COMPLETED: The execution was successfully accomplished - FAILED: The execution failed or produced an error

Source code in strands/multiagent/base.py 
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
@dataclass
class MultiAgentResult:
    """Result from multi-agent execution with accumulated metrics.

    The status field represents the outcome of the MultiAgentBase execution:
    - COMPLETED: The execution was successfully accomplished
    - FAILED: The execution failed or produced an error
    """

    status: Status = Status.PENDING
    results: dict[str, NodeResult] = field(default_factory=lambda: {})
    accumulated_usage: Usage = field(default_factory=lambda: Usage(inputTokens=0, outputTokens=0, totalTokens=0))
    accumulated_metrics: Metrics = field(default_factory=lambda: Metrics(latencyMs=0))
    execution_count: int = 0
    execution_time: int = 0

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> "MultiAgentResult":
        """Rehydrate a MultiAgentResult from persisted JSON."""
        if data.get("type") != "multiagent_result":
            raise TypeError(f"MultiAgentResult.from_dict: unexpected type {data.get('type')!r}")

        results = {k: NodeResult.from_dict(v) for k, v in data.get("results", {}).items()}
        usage = _parse_usage(data.get("accumulated_usage", {}))
        metrics = _parse_metrics(data.get("accumulated_metrics", {}))

        multiagent_result = cls(
            status=Status(data["status"]),
            results=results,
            accumulated_usage=usage,
            accumulated_metrics=metrics,
            execution_count=int(data.get("execution_count", 0)),
            execution_time=int(data.get("execution_time", 0)),
        )
        return multiagent_result

    def to_dict(self) -> dict[str, Any]:
        """Convert MultiAgentResult to JSON-serializable dict."""
        return {
            "type": "multiagent_result",
            "status": self.status.value,
            "results": {k: v.to_dict() for k, v in self.results.items()},
            "accumulated_usage": self.accumulated_usage,
            "accumulated_metrics": self.accumulated_metrics,
            "execution_count": self.execution_count,
            "execution_time": self.execution_time,
        }

from_dict(data) classmethod

Rehydrate a MultiAgentResult from persisted JSON.

Source code in strands/multiagent/base.py 
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
@classmethod
def from_dict(cls, data: dict[str, Any]) -> "MultiAgentResult":
    """Rehydrate a MultiAgentResult from persisted JSON."""
    if data.get("type") != "multiagent_result":
        raise TypeError(f"MultiAgentResult.from_dict: unexpected type {data.get('type')!r}")

    results = {k: NodeResult.from_dict(v) for k, v in data.get("results", {}).items()}
    usage = _parse_usage(data.get("accumulated_usage", {}))
    metrics = _parse_metrics(data.get("accumulated_metrics", {}))

    multiagent_result = cls(
        status=Status(data["status"]),
        results=results,
        accumulated_usage=usage,
        accumulated_metrics=metrics,
        execution_count=int(data.get("execution_count", 0)),
        execution_time=int(data.get("execution_time", 0)),
    )
    return multiagent_result

to_dict()

Convert MultiAgentResult to JSON-serializable dict.

Source code in strands/multiagent/base.py 
150
151
152
153
154
155
156
157
158
159
160
def to_dict(self) -> dict[str, Any]:
    """Convert MultiAgentResult to JSON-serializable dict."""
    return {
        "type": "multiagent_result",
        "status": self.status.value,
        "results": {k: v.to_dict() for k, v in self.results.items()},
        "accumulated_usage": self.accumulated_usage,
        "accumulated_metrics": self.accumulated_metrics,
        "execution_count": self.execution_count,
        "execution_time": self.execution_time,
    }

NodeResult dataclass

Unified result from node execution - handles both Agent and nested MultiAgentBase results.

The status field represents the semantic outcome of the node's work: - COMPLETED: The node's task was successfully accomplished - FAILED: The node's task failed or produced an error

Source code in strands/multiagent/base.py 
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
@dataclass
class NodeResult:
    """Unified result from node execution - handles both Agent and nested MultiAgentBase results.

    The status field represents the semantic outcome of the node's work:
    - COMPLETED: The node's task was successfully accomplished
    - FAILED: The node's task failed or produced an error
    """

    # Core result data - single AgentResult, nested MultiAgentResult, or Exception
    result: Union[AgentResult, "MultiAgentResult", Exception]

    # Execution metadata
    execution_time: int = 0
    status: Status = Status.PENDING

    # Accumulated metrics from this node and all children
    accumulated_usage: Usage = field(default_factory=lambda: Usage(inputTokens=0, outputTokens=0, totalTokens=0))
    accumulated_metrics: Metrics = field(default_factory=lambda: Metrics(latencyMs=0))
    execution_count: int = 0

    def get_agent_results(self) -> list[AgentResult]:
        """Get all AgentResult objects from this node, flattened if nested."""
        if isinstance(self.result, Exception):
            return []  # No agent results for exceptions
        elif isinstance(self.result, AgentResult):
            return [self.result]
        else:
            # Flatten nested results from MultiAgentResult
            flattened = []
            for nested_node_result in self.result.results.values():
                flattened.extend(nested_node_result.get_agent_results())
            return flattened

    def to_dict(self) -> dict[str, Any]:
        """Convert NodeResult to JSON-serializable dict, ignoring state field."""
        if isinstance(self.result, Exception):
            result_data: dict[str, Any] = {"type": "exception", "message": str(self.result)}
        elif isinstance(self.result, AgentResult):
            result_data = self.result.to_dict()
        else:
            # MultiAgentResult case
            result_data = self.result.to_dict()

        return {
            "result": result_data,
            "execution_time": self.execution_time,
            "status": self.status.value,
            "accumulated_usage": self.accumulated_usage,
            "accumulated_metrics": self.accumulated_metrics,
            "execution_count": self.execution_count,
        }

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> "NodeResult":
        """Rehydrate a NodeResult from persisted JSON."""
        if "result" not in data:
            raise TypeError("NodeResult.from_dict: missing 'result'")
        raw = data["result"]

        result: Union[AgentResult, "MultiAgentResult", Exception]
        if isinstance(raw, dict) and raw.get("type") == "agent_result":
            result = AgentResult.from_dict(raw)
        elif isinstance(raw, dict) and raw.get("type") == "exception":
            result = Exception(str(raw.get("message", "node failed")))
        elif isinstance(raw, dict) and raw.get("type") == "multiagent_result":
            result = MultiAgentResult.from_dict(raw)
        else:
            raise TypeError(f"NodeResult.from_dict: unsupported result payload: {raw!r}")

        usage = _parse_usage(data.get("accumulated_usage", {}))
        metrics = _parse_metrics(data.get("accumulated_metrics", {}))

        return cls(
            result=result,
            execution_time=int(data.get("execution_time", 0)),
            status=Status(data.get("status", "pending")),
            accumulated_usage=usage,
            accumulated_metrics=metrics,
            execution_count=int(data.get("execution_count", 0)),
        )

from_dict(data) classmethod

Rehydrate a NodeResult from persisted JSON.

Source code in strands/multiagent/base.py 
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
@classmethod
def from_dict(cls, data: dict[str, Any]) -> "NodeResult":
    """Rehydrate a NodeResult from persisted JSON."""
    if "result" not in data:
        raise TypeError("NodeResult.from_dict: missing 'result'")
    raw = data["result"]

    result: Union[AgentResult, "MultiAgentResult", Exception]
    if isinstance(raw, dict) and raw.get("type") == "agent_result":
        result = AgentResult.from_dict(raw)
    elif isinstance(raw, dict) and raw.get("type") == "exception":
        result = Exception(str(raw.get("message", "node failed")))
    elif isinstance(raw, dict) and raw.get("type") == "multiagent_result":
        result = MultiAgentResult.from_dict(raw)
    else:
        raise TypeError(f"NodeResult.from_dict: unsupported result payload: {raw!r}")

    usage = _parse_usage(data.get("accumulated_usage", {}))
    metrics = _parse_metrics(data.get("accumulated_metrics", {}))

    return cls(
        result=result,
        execution_time=int(data.get("execution_time", 0)),
        status=Status(data.get("status", "pending")),
        accumulated_usage=usage,
        accumulated_metrics=metrics,
        execution_count=int(data.get("execution_count", 0)),
    )

get_agent_results()

Get all AgentResult objects from this node, flattened if nested.

Source code in strands/multiagent/base.py 
52
53
54
55
56
57
58
59
60
61
62
63
def get_agent_results(self) -> list[AgentResult]:
    """Get all AgentResult objects from this node, flattened if nested."""
    if isinstance(self.result, Exception):
        return []  # No agent results for exceptions
    elif isinstance(self.result, AgentResult):
        return [self.result]
    else:
        # Flatten nested results from MultiAgentResult
        flattened = []
        for nested_node_result in self.result.results.values():
            flattened.extend(nested_node_result.get_agent_results())
        return flattened

to_dict()

Convert NodeResult to JSON-serializable dict, ignoring state field.

Source code in strands/multiagent/base.py 
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
def to_dict(self) -> dict[str, Any]:
    """Convert NodeResult to JSON-serializable dict, ignoring state field."""
    if isinstance(self.result, Exception):
        result_data: dict[str, Any] = {"type": "exception", "message": str(self.result)}
    elif isinstance(self.result, AgentResult):
        result_data = self.result.to_dict()
    else:
        # MultiAgentResult case
        result_data = self.result.to_dict()

    return {
        "result": result_data,
        "execution_time": self.execution_time,
        "status": self.status.value,
        "accumulated_usage": self.accumulated_usage,
        "accumulated_metrics": self.accumulated_metrics,
        "execution_count": self.execution_count,
    }

Status

Bases: Enum

Execution status for both graphs and nodes.

Source code in strands/multiagent/base.py 
22
23
24
25
26
27
28
class Status(Enum):
    """Execution status for both graphs and nodes."""

    PENDING = "pending"
    EXECUTING = "executing"
    COMPLETED = "completed"
    FAILED = "failed"

strands.multiagent.graph

Directed Graph Multi-Agent Pattern Implementation.

This module provides a deterministic graph-based agent orchestration system where agents or MultiAgentBase instances (like Swarm or Graph) are nodes in a graph, executed according to edge dependencies, with output from one node passed as input to connected nodes.

Key Features: - Agents and MultiAgentBase instances (Swarm, Graph, etc.) as graph nodes - Deterministic execution based on dependency resolution - Output propagation along edges - Support for cyclic graphs (feedback loops) - Clear dependency management - Supports nested graphs (Graph as a node in another Graph)

Graph

Bases: MultiAgentBase

Directed Graph multi-agent orchestration with configurable revisit behavior.

Source code in strands/multiagent/graph.py 
 403
 404
 405
 406
 407
 408
 409
 410
 411
 412
 413
 414
 415
 416
 417
 418
 419
 420
 421
 422
 423
 424
 425
 426
 427
 428
 429
 430
 431
 432
 433
 434
 435
 436
 437
 438
 439
 440
 441
 442
 443
 444
 445
 446
 447
 448
 449
 450
 451
 452
 453
 454
 455
 456
 457
 458
 459
 460
 461
 462
 463
 464
 465
 466
 467
 468
 469
 470
 471
 472
 473
 474
 475
 476
 477
 478
 479
 480
 481
 482
 483
 484
 485
 486
 487
 488
 489
 490
 491
 492
 493
 494
 495
 496
 497
 498
 499
 500
 501
 502
 503
 504
 505
 506
 507
 508
 509
 510
 511
 512
 513
 514
 515
 516
 517
 518
 519
 520
 521
 522
 523
 524
 525
 526
 527
 528
 529
 530
 531
 532
 533
 534
 535
 536
 537
 538
 539
 540
 541
 542
 543
 544
 545
 546
 547
 548
 549
 550
 551
 552
 553
 554
 555
 556
 557
 558
 559
 560
 561
 562
 563
 564
 565
 566
 567
 568
 569
 570
 571
 572
 573
 574
 575
 576
 577
 578
 579
 580
 581
 582
 583
 584
 585
 586
 587
 588
 589
 590
 591
 592
 593
 594
 595
 596
 597
 598
 599
 600
 601
 602
 603
 604
 605
 606
 607
 608
 609
 610
 611
 612
 613
 614
 615
 616
 617
 618
 619
 620
 621
 622
 623
 624
 625
 626
 627
 628
 629
 630
 631
 632
 633
 634
 635
 636
 637
 638
 639
 640
 641
 642
 643
 644
 645
 646
 647
 648
 649
 650
 651
 652
 653
 654
 655
 656
 657
 658
 659
 660
 661
 662
 663
 664
 665
 666
 667
 668
 669
 670
 671
 672
 673
 674
 675
 676
 677
 678
 679
 680
 681
 682
 683
 684
 685
 686
 687
 688
 689
 690
 691
 692
 693
 694
 695
 696
 697
 698
 699
 700
 701
 702
 703
 704
 705
 706
 707
 708
 709
 710
 711
 712
 713
 714
 715
 716
 717
 718
 719
 720
 721
 722
 723
 724
 725
 726
 727
 728
 729
 730
 731
 732
 733
 734
 735
 736
 737
 738
 739
 740
 741
 742
 743
 744
 745
 746
 747
 748
 749
 750
 751
 752
 753
 754
 755
 756
 757
 758
 759
 760
 761
 762
 763
 764
 765
 766
 767
 768
 769
 770
 771
 772
 773
 774
 775
 776
 777
 778
 779
 780
 781
 782
 783
 784
 785
 786
 787
 788
 789
 790
 791
 792
 793
 794
 795
 796
 797
 798
 799
 800
 801
 802
 803
 804
 805
 806
 807
 808
 809
 810
 811
 812
 813
 814
 815
 816
 817
 818
 819
 820
 821
 822
 823
 824
 825
 826
 827
 828
 829
 830
 831
 832
 833
 834
 835
 836
 837
 838
 839
 840
 841
 842
 843
 844
 845
 846
 847
 848
 849
 850
 851
 852
 853
 854
 855
 856
 857
 858
 859
 860
 861
 862
 863
 864
 865
 866
 867
 868
 869
 870
 871
 872
 873
 874
 875
 876
 877
 878
 879
 880
 881
 882
 883
 884
 885
 886
 887
 888
 889
 890
 891
 892
 893
 894
 895
 896
 897
 898
 899
 900
 901
 902
 903
 904
 905
 906
 907
 908
 909
 910
 911
 912
 913
 914
 915
 916
 917
 918
 919
 920
 921
 922
 923
 924
 925
 926
 927
 928
 929
 930
 931
 932
 933
 934
 935
 936
 937
 938
 939
 940
 941
 942
 943
 944
 945
 946
 947
 948
 949
 950
 951
 952
 953
 954
 955
 956
 957
 958
 959
 960
 961
 962
 963
 964
 965
 966
 967
 968
 969
 970
 971
 972
 973
 974
 975
 976
 977
 978
 979
 980
 981
 982
 983
 984
 985
 986
 987
 988
 989
 990
 991
 992
 993
 994
 995
 996
 997
 998
 999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
1113
1114
1115
class Graph(MultiAgentBase):
    """Directed Graph multi-agent orchestration with configurable revisit behavior."""

    def __init__(
        self,
        nodes: dict[str, GraphNode],
        edges: set[GraphEdge],
        entry_points: set[GraphNode],
        max_node_executions: Optional[int] = None,
        execution_timeout: Optional[float] = None,
        node_timeout: Optional[float] = None,
        reset_on_revisit: bool = False,
        session_manager: Optional[SessionManager] = None,
        hooks: Optional[list[HookProvider]] = None,
        id: str = _DEFAULT_GRAPH_ID,
        trace_attributes: Optional[Mapping[str, AttributeValue]] = None,
    ) -> None:
        """Initialize Graph with execution limits and reset behavior.

        Args:
            nodes: Dictionary of node_id to GraphNode
            edges: Set of GraphEdge objects
            entry_points: Set of GraphNode objects that are entry points
            max_node_executions: Maximum total node executions (default: None - no limit)
            execution_timeout: Total execution timeout in seconds (default: None - no limit)
            node_timeout: Individual node timeout in seconds (default: None - no limit)
            reset_on_revisit: Whether to reset node state when revisited (default: False)
            session_manager: Session manager for persisting graph state and execution history (default: None)
            hooks: List of hook providers for monitoring and extending graph execution behavior (default: None)
            id: Unique graph id (default: None)
            trace_attributes: Custom trace attributes to apply to the agent's trace span (default: None)
        """
        super().__init__()

        # Validate nodes for duplicate instances
        self._validate_graph(nodes)

        self.nodes = nodes
        self.edges = edges
        self.entry_points = entry_points
        self.max_node_executions = max_node_executions
        self.execution_timeout = execution_timeout
        self.node_timeout = node_timeout
        self.reset_on_revisit = reset_on_revisit
        self.state = GraphState()
        self.tracer = get_tracer()
        self.trace_attributes: dict[str, AttributeValue] = self._parse_trace_attributes(trace_attributes)
        self.session_manager = session_manager
        self.hooks = HookRegistry()
        if self.session_manager:
            self.hooks.add_hook(self.session_manager)
        if hooks:
            for hook in hooks:
                self.hooks.add_hook(hook)

        self._resume_next_nodes: list[GraphNode] = []
        self._resume_from_session = False
        self.id = id

        run_async(lambda: self.hooks.invoke_callbacks_async(MultiAgentInitializedEvent(self)))

    def __call__(
        self, task: MultiAgentInput, invocation_state: dict[str, Any] | None = None, **kwargs: Any
    ) -> GraphResult:
        """Invoke the graph synchronously.

        Args:
            task: The task to execute
            invocation_state: Additional state/context passed to underlying agents.
                Defaults to None to avoid mutable default argument issues.
            **kwargs: Keyword arguments allowing backward compatible future changes.
        """
        if invocation_state is None:
            invocation_state = {}

        return run_async(lambda: self.invoke_async(task, invocation_state))

    async def invoke_async(
        self, task: MultiAgentInput, invocation_state: dict[str, Any] | None = None, **kwargs: Any
    ) -> GraphResult:
        """Invoke the graph asynchronously.

        This method uses stream_async internally and consumes all events until completion,
        following the same pattern as the Agent class.

        Args:
            task: The task to execute
            invocation_state: Additional state/context passed to underlying agents.
                Defaults to None to avoid mutable default argument issues.
            **kwargs: Keyword arguments allowing backward compatible future changes.
        """
        events = self.stream_async(task, invocation_state, **kwargs)
        final_event = None
        async for event in events:
            final_event = event

        if final_event is None or "result" not in final_event:
            raise ValueError("Graph streaming completed without producing a result event")

        return cast(GraphResult, final_event["result"])

    async def stream_async(
        self, task: MultiAgentInput, invocation_state: dict[str, Any] | None = None, **kwargs: Any
    ) -> AsyncIterator[dict[str, Any]]:
        """Stream events during graph execution.

        Args:
            task: The task to execute
            invocation_state: Additional state/context passed to underlying agents.
                Defaults to None to avoid mutable default argument issues.
            **kwargs: Keyword arguments allowing backward compatible future changes.

        Yields:
            Dictionary events during graph execution, such as:
            - multi_agent_node_start: When a node begins execution
            - multi_agent_node_stream: Forwarded agent/multi-agent events with node context
            - multi_agent_node_stop: When a node stops execution
            - result: Final graph result
        """
        if invocation_state is None:
            invocation_state = {}

        await self.hooks.invoke_callbacks_async(BeforeMultiAgentInvocationEvent(self, invocation_state))

        logger.debug("task=<%s> | starting graph execution", task)

        # Initialize state
        start_time = time.time()
        if not self._resume_from_session:
            # Initialize state
            self.state = GraphState(
                status=Status.EXECUTING,
                task=task,
                total_nodes=len(self.nodes),
                edges=[(edge.from_node, edge.to_node) for edge in self.edges],
                entry_points=list(self.entry_points),
                start_time=start_time,
            )
        else:
            self.state.status = Status.EXECUTING
            self.state.start_time = start_time

        span = self.tracer.start_multiagent_span(task, "graph", custom_trace_attributes=self.trace_attributes)
        with trace_api.use_span(span, end_on_exit=True):
            try:
                logger.debug(
                    "max_node_executions=<%s>, execution_timeout=<%s>s, node_timeout=<%s>s | graph execution config",
                    self.max_node_executions or "None",
                    self.execution_timeout or "None",
                    self.node_timeout or "None",
                )

                async for event in self._execute_graph(invocation_state):
                    yield event.as_dict()

                # Set final status based on execution results
                if self.state.failed_nodes:
                    self.state.status = Status.FAILED
                elif self.state.status == Status.EXECUTING:
                    self.state.status = Status.COMPLETED

                logger.debug("status=<%s> | graph execution completed", self.state.status)

                # Yield final result (consistent with Agent's AgentResultEvent format)
                result = self._build_result()

                # Use the same event format as Agent for consistency
                yield MultiAgentResultEvent(result=result).as_dict()

            except Exception:
                logger.exception("graph execution failed")
                self.state.status = Status.FAILED
                raise
            finally:
                self.state.execution_time = round((time.time() - start_time) * 1000)
                await self.hooks.invoke_callbacks_async(AfterMultiAgentInvocationEvent(self))
                self._resume_from_session = False
                self._resume_next_nodes.clear()

    def _validate_graph(self, nodes: dict[str, GraphNode]) -> None:
        """Validate graph nodes for duplicate instances."""
        # Check for duplicate node instances
        seen_instances = set()
        for node in nodes.values():
            if id(node.executor) in seen_instances:
                raise ValueError("Duplicate node instance detected. Each node must have a unique object instance.")
            seen_instances.add(id(node.executor))

            # Validate Agent-specific constraints for each node
            _validate_node_executor(node.executor)

    async def _execute_graph(self, invocation_state: dict[str, Any]) -> AsyncIterator[Any]:
        """Execute graph and yield TypedEvent objects."""
        ready_nodes = self._resume_next_nodes if self._resume_from_session else list(self.entry_points)

        while ready_nodes:
            # Check execution limits before continuing
            should_continue, reason = self.state.should_continue(
                max_node_executions=self.max_node_executions,
                execution_timeout=self.execution_timeout,
            )
            if not should_continue:
                self.state.status = Status.FAILED
                logger.debug("reason=<%s> | stopping execution", reason)
                return  # Let the top-level exception handler deal with it

            current_batch = ready_nodes.copy()
            ready_nodes.clear()

            # Execute current batch
            async for event in self._execute_nodes_parallel(current_batch, invocation_state):
                yield event

            # Find newly ready nodes after batch execution
            # We add all nodes in current batch as completed batch,
            # because a failure would throw exception and code would not make it here
            newly_ready = self._find_newly_ready_nodes(current_batch)

            # Emit handoff event for batch transition if there are nodes to transition to
            if newly_ready:
                handoff_event = MultiAgentHandoffEvent(
                    from_node_ids=[node.node_id for node in current_batch],
                    to_node_ids=[node.node_id for node in newly_ready],
                )
                yield handoff_event
                logger.debug(
                    "from_node_ids=<%s>, to_node_ids=<%s> | batch transition",
                    [node.node_id for node in current_batch],
                    [node.node_id for node in newly_ready],
                )

            ready_nodes.extend(newly_ready)

    async def _execute_nodes_parallel(
        self, nodes: list["GraphNode"], invocation_state: dict[str, Any]
    ) -> AsyncIterator[Any]:
        """Execute multiple nodes in parallel and merge their event streams in real-time.

        Uses a shared queue where each node's stream runs independently and pushes events
        as they occur, enabling true real-time event propagation without round-robin delays.
        """
        event_queue: asyncio.Queue[Any | None | Exception] = asyncio.Queue()

        # Start all node streams as independent tasks
        tasks = [asyncio.create_task(self._stream_node_to_queue(node, event_queue, invocation_state)) for node in nodes]

        try:
            # Consume events from the queue as they arrive
            # Continue until all tasks are done
            while any(not task.done() for task in tasks):
                try:
                    # Use timeout to avoid race condition: if all tasks complete between
                    # checking task.done() and calling queue.get(), we'd hang forever.
                    # The 0.1s timeout allows us to periodically re-check task completion
                    # while still being responsive to incoming events.
                    event = await asyncio.wait_for(event_queue.get(), timeout=0.1)
                except asyncio.TimeoutError:
                    # No event available, continue checking tasks
                    continue

                # Check if it's an exception - fail fast
                if isinstance(event, Exception):
                    # Cancel all other tasks immediately
                    for task in tasks:
                        if not task.done():
                            task.cancel()
                    raise event

                if event is not None:
                    yield event

            # Process any remaining events in the queue after all tasks complete
            while not event_queue.empty():
                event = await event_queue.get()
                if isinstance(event, Exception):
                    raise event
                if event is not None:
                    yield event
        finally:
            # Cancel any remaining tasks
            remaining_tasks = [task for task in tasks if not task.done()]
            if remaining_tasks:
                logger.warning(
                    "remaining_task_count=<%d> | cancelling remaining tasks in finally block",
                    len(remaining_tasks),
                )
                for task in remaining_tasks:
                    task.cancel()
            await asyncio.gather(*tasks, return_exceptions=True)

    async def _stream_node_to_queue(
        self,
        node: GraphNode,
        event_queue: asyncio.Queue[Any | None | Exception],
        invocation_state: dict[str, Any],
    ) -> None:
        """Stream events from a node to the shared queue with optional timeout."""
        try:
            # Apply timeout to the entire streaming process if configured
            if self.node_timeout is not None:

                async def stream_node() -> None:
                    async for event in self._execute_node(node, invocation_state):
                        await event_queue.put(event)

                try:
                    await asyncio.wait_for(stream_node(), timeout=self.node_timeout)
                except asyncio.TimeoutError:
                    # Handle timeout and send exception through queue
                    timeout_exc = await self._handle_node_timeout(node, event_queue)
                    await event_queue.put(timeout_exc)
            else:
                # No timeout - stream normally
                async for event in self._execute_node(node, invocation_state):
                    await event_queue.put(event)
        except Exception as e:
            # Send exception through queue for fail-fast behavior
            await event_queue.put(e)
        finally:
            await event_queue.put(None)

    async def _handle_node_timeout(self, node: GraphNode, event_queue: asyncio.Queue[Any | None]) -> Exception:
        """Handle a node timeout by creating a failed result and emitting events.

        Returns:
            The timeout exception to be re-raised for fail-fast behavior
        """
        assert self.node_timeout is not None
        timeout_exception = Exception(f"Node '{node.node_id}' execution timed out after {self.node_timeout}s")

        node_result = NodeResult(
            result=timeout_exception,
            execution_time=round(self.node_timeout * 1000),
            status=Status.FAILED,
            accumulated_usage=Usage(inputTokens=0, outputTokens=0, totalTokens=0),
            accumulated_metrics=Metrics(latencyMs=round(self.node_timeout * 1000)),
            execution_count=1,
        )

        node.execution_status = Status.FAILED
        node.result = node_result
        node.execution_time = node_result.execution_time
        self.state.failed_nodes.add(node)
        self.state.results[node.node_id] = node_result

        complete_event = MultiAgentNodeStopEvent(
            node_id=node.node_id,
            node_result=node_result,
        )
        await event_queue.put(complete_event)

        return timeout_exception

    def _find_newly_ready_nodes(self, completed_batch: list["GraphNode"]) -> list["GraphNode"]:
        """Find nodes that became ready after the last execution."""
        newly_ready = []
        for _node_id, node in self.nodes.items():
            if self._is_node_ready_with_conditions(node, completed_batch):
                newly_ready.append(node)
        return newly_ready

    def _is_node_ready_with_conditions(self, node: GraphNode, completed_batch: list["GraphNode"]) -> bool:
        """Check if a node is ready considering conditional edges."""
        # Get incoming edges to this node
        incoming_edges = [edge for edge in self.edges if edge.to_node == node]

        # Check if at least one incoming edge condition is satisfied
        for edge in incoming_edges:
            if edge.from_node in completed_batch:
                if edge.should_traverse(self.state):
                    logger.debug(
                        "from=<%s>, to=<%s> | edge ready via satisfied condition", edge.from_node.node_id, node.node_id
                    )
                    return True
                else:
                    logger.debug(
                        "from=<%s>, to=<%s> | edge condition not satisfied", edge.from_node.node_id, node.node_id
                    )
        return False

    async def _execute_node(self, node: GraphNode, invocation_state: dict[str, Any]) -> AsyncIterator[Any]:
        """Execute a single node and yield TypedEvent objects."""
        # Reset the node's state if reset_on_revisit is enabled, and it's being revisited
        if self.reset_on_revisit and node in self.state.completed_nodes:
            logger.debug("node_id=<%s> | resetting node state for revisit", node.node_id)
            node.reset_executor_state()
            self.state.completed_nodes.remove(node)

        node.execution_status = Status.EXECUTING
        logger.debug("node_id=<%s> | executing node", node.node_id)

        # Emit node start event
        start_event = MultiAgentNodeStartEvent(
            node_id=node.node_id, node_type="agent" if isinstance(node.executor, Agent) else "multiagent"
        )
        yield start_event

        before_event, _ = await self.hooks.invoke_callbacks_async(
            BeforeNodeCallEvent(self, node.node_id, invocation_state)
        )

        start_time = time.time()
        try:
            if before_event.cancel_node:
                cancel_message = (
                    before_event.cancel_node if isinstance(before_event.cancel_node, str) else "node cancelled by user"
                )
                logger.debug("reason=<%s> | cancelling execution", cancel_message)
                yield MultiAgentNodeCancelEvent(node.node_id, cancel_message)
                raise RuntimeError(cancel_message)

            # Build node input from satisfied dependencies
            node_input = self._build_node_input(node)

            # Execute and stream events (timeout handled at task level)
            if isinstance(node.executor, MultiAgentBase):
                # For nested multi-agent systems, stream their events and collect result
                multi_agent_result = None
                async for event in node.executor.stream_async(node_input, invocation_state):
                    # Forward nested multi-agent events with node context
                    wrapped_event = MultiAgentNodeStreamEvent(node.node_id, event)
                    yield wrapped_event
                    # Capture the final result event
                    if "result" in event:
                        multi_agent_result = event["result"]

                # Use the captured result from streaming (no double execution)
                if multi_agent_result is None:
                    raise ValueError(f"Node '{node.node_id}' did not produce a result event")

                node_result = NodeResult(
                    result=multi_agent_result,
                    execution_time=multi_agent_result.execution_time,
                    status=Status.COMPLETED,
                    accumulated_usage=multi_agent_result.accumulated_usage,
                    accumulated_metrics=multi_agent_result.accumulated_metrics,
                    execution_count=multi_agent_result.execution_count,
                )

            elif isinstance(node.executor, Agent):
                # For agents, stream their events and collect result
                agent_response = None
                async for event in node.executor.stream_async(node_input, invocation_state=invocation_state):
                    # Forward agent events with node context
                    wrapped_event = MultiAgentNodeStreamEvent(node.node_id, event)
                    yield wrapped_event
                    # Capture the final result event
                    if "result" in event:
                        agent_response = event["result"]

                # Use the captured result from streaming (no double execution)
                if agent_response is None:
                    raise ValueError(f"Node '{node.node_id}' did not produce a result event")

                # Check for interrupt (from main branch)
                if agent_response.stop_reason == "interrupt":
                    node.executor.messages.pop()  # remove interrupted tool use message
                    node.executor._interrupt_state.deactivate()

                    raise RuntimeError("user raised interrupt from agent | interrupts are not yet supported in graphs")

                # Extract metrics with defaults
                response_metrics = getattr(agent_response, "metrics", None)
                usage = getattr(
                    response_metrics, "accumulated_usage", Usage(inputTokens=0, outputTokens=0, totalTokens=0)
                )
                metrics = getattr(response_metrics, "accumulated_metrics", Metrics(latencyMs=0))

                node_result = NodeResult(
                    result=agent_response,
                    execution_time=round((time.time() - start_time) * 1000),
                    status=Status.COMPLETED,
                    accumulated_usage=usage,
                    accumulated_metrics=metrics,
                    execution_count=1,
                )
            else:
                raise ValueError(f"Node '{node.node_id}' of type '{type(node.executor)}' is not supported")

            # Mark as completed
            node.execution_status = Status.COMPLETED
            node.result = node_result
            node.execution_time = node_result.execution_time
            self.state.completed_nodes.add(node)
            self.state.results[node.node_id] = node_result
            self.state.execution_order.append(node)

            # Accumulate metrics
            self._accumulate_metrics(node_result)

            # Emit node stop event with full NodeResult
            complete_event = MultiAgentNodeStopEvent(
                node_id=node.node_id,
                node_result=node_result,
            )
            yield complete_event

            logger.debug(
                "node_id=<%s>, execution_time=<%dms> | node completed successfully",
                node.node_id,
                node.execution_time,
            )

        except Exception as e:
            # All failures (programming errors and execution failures) stop graph execution
            # This matches the old fail-fast behavior
            logger.error("node_id=<%s>, error=<%s> | node failed", node.node_id, e)
            execution_time = round((time.time() - start_time) * 1000)

            # Create a NodeResult for the failed node
            node_result = NodeResult(
                result=e,
                execution_time=execution_time,
                status=Status.FAILED,
                accumulated_usage=Usage(inputTokens=0, outputTokens=0, totalTokens=0),
                accumulated_metrics=Metrics(latencyMs=execution_time),
                execution_count=1,
            )

            node.execution_status = Status.FAILED
            node.result = node_result
            node.execution_time = execution_time
            self.state.failed_nodes.add(node)
            self.state.results[node.node_id] = node_result

            # Emit stop event even for failures
            complete_event = MultiAgentNodeStopEvent(
                node_id=node.node_id,
                node_result=node_result,
            )
            yield complete_event

            # Re-raise to stop graph execution (fail-fast behavior)
            raise

        finally:
            await self.hooks.invoke_callbacks_async(AfterNodeCallEvent(self, node.node_id, invocation_state))

    def _accumulate_metrics(self, node_result: NodeResult) -> None:
        """Accumulate metrics from a node result."""
        self.state.accumulated_usage["inputTokens"] += node_result.accumulated_usage.get("inputTokens", 0)
        self.state.accumulated_usage["outputTokens"] += node_result.accumulated_usage.get("outputTokens", 0)
        self.state.accumulated_usage["totalTokens"] += node_result.accumulated_usage.get("totalTokens", 0)
        self.state.accumulated_metrics["latencyMs"] += node_result.accumulated_metrics.get("latencyMs", 0)
        self.state.execution_count += node_result.execution_count

    def _build_node_input(self, node: GraphNode) -> list[ContentBlock]:
        """Build input text for a node based on dependency outputs.

        Example formatted output:
        ```
        Original Task: Analyze the quarterly sales data and create a summary report

        Inputs from previous nodes:

        From data_processor:
          - Agent: Sales data processed successfully. Found 1,247 transactions totaling $89,432.
          - Agent: Key trends: 15% increase in Q3, top product category is Electronics.

        From validator:
          - Agent: Data validation complete. All records verified, no anomalies detected.
        ```
        """
        # Get satisfied dependencies
        dependency_results = {}
        for edge in self.edges:
            if (
                edge.to_node == node
                and edge.from_node in self.state.completed_nodes
                and edge.from_node.node_id in self.state.results
            ):
                if edge.should_traverse(self.state):
                    dependency_results[edge.from_node.node_id] = self.state.results[edge.from_node.node_id]

        if not dependency_results:
            # No dependencies - return task as ContentBlocks
            if isinstance(self.state.task, str):
                return [ContentBlock(text=self.state.task)]
            else:
                return self.state.task

        # Combine task with dependency outputs
        node_input = []

        # Add original task
        if isinstance(self.state.task, str):
            node_input.append(ContentBlock(text=f"Original Task: {self.state.task}"))
        else:
            # Add task content blocks with a prefix
            node_input.append(ContentBlock(text="Original Task:"))
            node_input.extend(self.state.task)

        # Add dependency outputs
        node_input.append(ContentBlock(text="\nInputs from previous nodes:"))

        for dep_id, node_result in dependency_results.items():
            node_input.append(ContentBlock(text=f"\nFrom {dep_id}:"))
            # Get all agent results from this node (flattened if nested)
            agent_results = node_result.get_agent_results()
            for result in agent_results:
                agent_name = getattr(result, "agent_name", "Agent")
                result_text = str(result)
                node_input.append(ContentBlock(text=f"  - {agent_name}: {result_text}"))

        return node_input

    def _build_result(self) -> GraphResult:
        """Build graph result from current state."""
        return GraphResult(
            status=self.state.status,
            results=self.state.results,
            accumulated_usage=self.state.accumulated_usage,
            accumulated_metrics=self.state.accumulated_metrics,
            execution_count=self.state.execution_count,
            execution_time=self.state.execution_time,
            total_nodes=self.state.total_nodes,
            completed_nodes=len(self.state.completed_nodes),
            failed_nodes=len(self.state.failed_nodes),
            execution_order=self.state.execution_order,
            edges=self.state.edges,
            entry_points=self.state.entry_points,
        )

    def serialize_state(self) -> dict[str, Any]:
        """Serialize the current graph state to a dictionary."""
        compute_nodes = self._compute_ready_nodes_for_resume()
        next_nodes = [n.node_id for n in compute_nodes] if compute_nodes else []
        return {
            "type": "graph",
            "id": self.id,
            "status": self.state.status.value,
            "completed_nodes": [n.node_id for n in self.state.completed_nodes],
            "failed_nodes": [n.node_id for n in self.state.failed_nodes],
            "node_results": {k: v.to_dict() for k, v in (self.state.results or {}).items()},
            "next_nodes_to_execute": next_nodes,
            "current_task": self.state.task,
            "execution_order": [n.node_id for n in self.state.execution_order],
        }

    def deserialize_state(self, payload: dict[str, Any]) -> None:
        """Restore graph state from a session dict and prepare for execution.

        This method handles two scenarios:
        1. If the graph execution ended (no next_nodes_to_execute, eg: Completed, or Failed with dead end nodes),
        resets all nodes and graph state to allow re-execution from the beginning.
        2. If the graph execution was interrupted mid-execution (has next_nodes_to_execute),
           restores the persisted state and prepares to resume execution from the next ready nodes.

        Args:
            payload: Dictionary containing persisted state data including status,
                    completed nodes, results, and next nodes to execute.
        """
        if not payload.get("next_nodes_to_execute"):
            # Reset all nodes
            for node in self.nodes.values():
                node.reset_executor_state()
            # Reset graph state
            self.state = GraphState()
            self._resume_from_session = False
            return
        else:
            self._from_dict(payload)
            self._resume_from_session = True

    def _compute_ready_nodes_for_resume(self) -> list[GraphNode]:
        if self.state.status == Status.PENDING:
            return []
        ready_nodes: list[GraphNode] = []
        completed_nodes = set(self.state.completed_nodes)
        for node in self.nodes.values():
            if node in completed_nodes:
                continue
            incoming = [e for e in self.edges if e.to_node is node]
            if not incoming:
                ready_nodes.append(node)
            elif all(e.from_node in completed_nodes and e.should_traverse(self.state) for e in incoming):
                ready_nodes.append(node)

        return ready_nodes

    def _from_dict(self, payload: dict[str, Any]) -> None:
        self.state.status = Status(payload["status"])
        # Hydrate completed nodes & results
        raw_results = payload.get("node_results") or {}
        results: dict[str, NodeResult] = {}
        for node_id, entry in raw_results.items():
            if node_id not in self.nodes:
                continue
            try:
                results[node_id] = NodeResult.from_dict(entry)
            except Exception:
                logger.exception("Failed to hydrate NodeResult for node_id=%s; skipping.", node_id)
                raise
        self.state.results = results

        self.state.failed_nodes = set(
            self.nodes[node_id] for node_id in (payload.get("failed_nodes") or []) if node_id in self.nodes
        )

        # Restore completed nodes from persisted data
        completed_node_ids = payload.get("completed_nodes") or []
        self.state.completed_nodes = {self.nodes[node_id] for node_id in completed_node_ids if node_id in self.nodes}

        # Execution order (only nodes that still exist)
        order_node_ids = payload.get("execution_order") or []
        self.state.execution_order = [self.nodes[node_id] for node_id in order_node_ids if node_id in self.nodes]

        # Task
        self.state.task = payload.get("current_task", self.state.task)

        # next nodes to execute
        next_nodes = [self.nodes[nid] for nid in (payload.get("next_nodes_to_execute") or []) if nid in self.nodes]
        self._resume_next_nodes = next_nodes

__call__(task, invocation_state=None, **kwargs)

Invoke the graph synchronously.

Parameters:

Name Type Description Default
task MultiAgentInput

The task to execute

 required
invocation_state dict[str, Any] | None

Additional state/context passed to underlying agents. Defaults to None to avoid mutable default argument issues.

 None
**kwargs Any

Keyword arguments allowing backward compatible future changes.

 {}
Source code in strands/multiagent/graph.py 
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
def __call__(
    self, task: MultiAgentInput, invocation_state: dict[str, Any] | None = None, **kwargs: Any
) -> GraphResult:
    """Invoke the graph synchronously.

    Args:
        task: The task to execute
        invocation_state: Additional state/context passed to underlying agents.
            Defaults to None to avoid mutable default argument issues.
        **kwargs: Keyword arguments allowing backward compatible future changes.
    """
    if invocation_state is None:
        invocation_state = {}

    return run_async(lambda: self.invoke_async(task, invocation_state))

__init__(nodes, edges, entry_points, max_node_executions=None, execution_timeout=None, node_timeout=None, reset_on_revisit=False, session_manager=None, hooks=None, id=_DEFAULT_GRAPH_ID, trace_attributes=None)

Initialize Graph with execution limits and reset behavior.

Parameters:

Name Type Description Default
nodes dict[str, GraphNode]

Dictionary of node_id to GraphNode

 required
edges set[GraphEdge]

Set of GraphEdge objects

 required
entry_points set[GraphNode]

Set of GraphNode objects that are entry points

 required
max_node_executions Optional[int]

Maximum total node executions (default: None - no limit)

 None
execution_timeout Optional[float]

Total execution timeout in seconds (default: None - no limit)

 None
node_timeout Optional[float]

Individual node timeout in seconds (default: None - no limit)

 None
reset_on_revisit bool

Whether to reset node state when revisited (default: False)

 False
session_manager Optional[SessionManager]

Session manager for persisting graph state and execution history (default: None)

 None
hooks Optional[list[HookProvider]]

List of hook providers for monitoring and extending graph execution behavior (default: None)

 None
id str

Unique graph id (default: None)

 _DEFAULT_GRAPH_ID
trace_attributes Optional[Mapping[str, AttributeValue]]

Custom trace attributes to apply to the agent's trace span (default: None)

 None
Source code in strands/multiagent/graph.py 
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
def __init__(
    self,
    nodes: dict[str, GraphNode],
    edges: set[GraphEdge],
    entry_points: set[GraphNode],
    max_node_executions: Optional[int] = None,
    execution_timeout: Optional[float] = None,
    node_timeout: Optional[float] = None,
    reset_on_revisit: bool = False,
    session_manager: Optional[SessionManager] = None,
    hooks: Optional[list[HookProvider]] = None,
    id: str = _DEFAULT_GRAPH_ID,
    trace_attributes: Optional[Mapping[str, AttributeValue]] = None,
) -> None:
    """Initialize Graph with execution limits and reset behavior.

    Args:
        nodes: Dictionary of node_id to GraphNode
        edges: Set of GraphEdge objects
        entry_points: Set of GraphNode objects that are entry points
        max_node_executions: Maximum total node executions (default: None - no limit)
        execution_timeout: Total execution timeout in seconds (default: None - no limit)
        node_timeout: Individual node timeout in seconds (default: None - no limit)
        reset_on_revisit: Whether to reset node state when revisited (default: False)
        session_manager: Session manager for persisting graph state and execution history (default: None)
        hooks: List of hook providers for monitoring and extending graph execution behavior (default: None)
        id: Unique graph id (default: None)
        trace_attributes: Custom trace attributes to apply to the agent's trace span (default: None)
    """
    super().__init__()

    # Validate nodes for duplicate instances
    self._validate_graph(nodes)

    self.nodes = nodes
    self.edges = edges
    self.entry_points = entry_points
    self.max_node_executions = max_node_executions
    self.execution_timeout = execution_timeout
    self.node_timeout = node_timeout
    self.reset_on_revisit = reset_on_revisit
    self.state = GraphState()
    self.tracer = get_tracer()
    self.trace_attributes: dict[str, AttributeValue] = self._parse_trace_attributes(trace_attributes)
    self.session_manager = session_manager
    self.hooks = HookRegistry()
    if self.session_manager:
        self.hooks.add_hook(self.session_manager)
    if hooks:
        for hook in hooks:
            self.hooks.add_hook(hook)

    self._resume_next_nodes: list[GraphNode] = []
    self._resume_from_session = False
    self.id = id

    run_async(lambda: self.hooks.invoke_callbacks_async(MultiAgentInitializedEvent(self)))

deserialize_state(payload)

Restore graph state from a session dict and prepare for execution.

This method handles two scenarios: 1. If the graph execution ended (no next_nodes_to_execute, eg: Completed, or Failed with dead end nodes), resets all nodes and graph state to allow re-execution from the beginning. 2. If the graph execution was interrupted mid-execution (has next_nodes_to_execute), restores the persisted state and prepares to resume execution from the next ready nodes.

Parameters:

Name Type Description Default
payload dict[str, Any]

Dictionary containing persisted state data including status, completed nodes, results, and next nodes to execute.

 required
Source code in strands/multiagent/graph.py 
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
def deserialize_state(self, payload: dict[str, Any]) -> None:
    """Restore graph state from a session dict and prepare for execution.

    This method handles two scenarios:
    1. If the graph execution ended (no next_nodes_to_execute, eg: Completed, or Failed with dead end nodes),
    resets all nodes and graph state to allow re-execution from the beginning.
    2. If the graph execution was interrupted mid-execution (has next_nodes_to_execute),
       restores the persisted state and prepares to resume execution from the next ready nodes.

    Args:
        payload: Dictionary containing persisted state data including status,
                completed nodes, results, and next nodes to execute.
    """
    if not payload.get("next_nodes_to_execute"):
        # Reset all nodes
        for node in self.nodes.values():
            node.reset_executor_state()
        # Reset graph state
        self.state = GraphState()
        self._resume_from_session = False
        return
    else:
        self._from_dict(payload)
        self._resume_from_session = True

invoke_async(task, invocation_state=None, **kwargs) async

Invoke the graph asynchronously.

This method uses stream_async internally and consumes all events until completion, following the same pattern as the Agent class.

Parameters:

Name Type Description Default
task MultiAgentInput

The task to execute

 required
invocation_state dict[str, Any] | None

Additional state/context passed to underlying agents. Defaults to None to avoid mutable default argument issues.

 None
**kwargs Any

Keyword arguments allowing backward compatible future changes.

 {}
Source code in strands/multiagent/graph.py 
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
async def invoke_async(
    self, task: MultiAgentInput, invocation_state: dict[str, Any] | None = None, **kwargs: Any
) -> GraphResult:
    """Invoke the graph asynchronously.

    This method uses stream_async internally and consumes all events until completion,
    following the same pattern as the Agent class.

    Args:
        task: The task to execute
        invocation_state: Additional state/context passed to underlying agents.
            Defaults to None to avoid mutable default argument issues.
        **kwargs: Keyword arguments allowing backward compatible future changes.
    """
    events = self.stream_async(task, invocation_state, **kwargs)
    final_event = None
    async for event in events:
        final_event = event

    if final_event is None or "result" not in final_event:
        raise ValueError("Graph streaming completed without producing a result event")

    return cast(GraphResult, final_event["result"])

serialize_state()

Serialize the current graph state to a dictionary.

Source code in strands/multiagent/graph.py 
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
def serialize_state(self) -> dict[str, Any]:
    """Serialize the current graph state to a dictionary."""
    compute_nodes = self._compute_ready_nodes_for_resume()
    next_nodes = [n.node_id for n in compute_nodes] if compute_nodes else []
    return {
        "type": "graph",
        "id": self.id,
        "status": self.state.status.value,
        "completed_nodes": [n.node_id for n in self.state.completed_nodes],
        "failed_nodes": [n.node_id for n in self.state.failed_nodes],
        "node_results": {k: v.to_dict() for k, v in (self.state.results or {}).items()},
        "next_nodes_to_execute": next_nodes,
        "current_task": self.state.task,
        "execution_order": [n.node_id for n in self.state.execution_order],
    }

stream_async(task, invocation_state=None, **kwargs) async

Stream events during graph execution.

Parameters:

Name Type Description Default
task MultiAgentInput

The task to execute

 required
invocation_state dict[str, Any] | None

Additional state/context passed to underlying agents. Defaults to None to avoid mutable default argument issues.

 None
**kwargs Any

Keyword arguments allowing backward compatible future changes.

 {}

Yields:

Type Description
AsyncIterator[dict[str, Any]]

Dictionary events during graph execution, such as:
AsyncIterator[dict[str, Any]]
  • multi_agent_node_start: When a node begins execution
AsyncIterator[dict[str, Any]]
  • multi_agent_node_stream: Forwarded agent/multi-agent events with node context
AsyncIterator[dict[str, Any]]
  • multi_agent_node_stop: When a node stops execution
AsyncIterator[dict[str, Any]]
  • result: Final graph result
Source code in strands/multiagent/graph.py 
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
async def stream_async(
    self, task: MultiAgentInput, invocation_state: dict[str, Any] | None = None, **kwargs: Any
) -> AsyncIterator[dict[str, Any]]:
    """Stream events during graph execution.

    Args:
        task: The task to execute
        invocation_state: Additional state/context passed to underlying agents.
            Defaults to None to avoid mutable default argument issues.
        **kwargs: Keyword arguments allowing backward compatible future changes.

    Yields:
        Dictionary events during graph execution, such as:
        - multi_agent_node_start: When a node begins execution
        - multi_agent_node_stream: Forwarded agent/multi-agent events with node context
        - multi_agent_node_stop: When a node stops execution
        - result: Final graph result
    """
    if invocation_state is None:
        invocation_state = {}

    await self.hooks.invoke_callbacks_async(BeforeMultiAgentInvocationEvent(self, invocation_state))

    logger.debug("task=<%s> | starting graph execution", task)

    # Initialize state
    start_time = time.time()
    if not self._resume_from_session:
        # Initialize state
        self.state = GraphState(
            status=Status.EXECUTING,
            task=task,
            total_nodes=len(self.nodes),
            edges=[(edge.from_node, edge.to_node) for edge in self.edges],
            entry_points=list(self.entry_points),
            start_time=start_time,
        )
    else:
        self.state.status = Status.EXECUTING
        self.state.start_time = start_time

    span = self.tracer.start_multiagent_span(task, "graph", custom_trace_attributes=self.trace_attributes)
    with trace_api.use_span(span, end_on_exit=True):
        try:
            logger.debug(
                "max_node_executions=<%s>, execution_timeout=<%s>s, node_timeout=<%s>s | graph execution config",
                self.max_node_executions or "None",
                self.execution_timeout or "None",
                self.node_timeout or "None",
            )

            async for event in self._execute_graph(invocation_state):
                yield event.as_dict()

            # Set final status based on execution results
            if self.state.failed_nodes:
                self.state.status = Status.FAILED
            elif self.state.status == Status.EXECUTING:
                self.state.status = Status.COMPLETED

            logger.debug("status=<%s> | graph execution completed", self.state.status)

            # Yield final result (consistent with Agent's AgentResultEvent format)
            result = self._build_result()

            # Use the same event format as Agent for consistency
            yield MultiAgentResultEvent(result=result).as_dict()

        except Exception:
            logger.exception("graph execution failed")
            self.state.status = Status.FAILED
            raise
        finally:
            self.state.execution_time = round((time.time() - start_time) * 1000)
            await self.hooks.invoke_callbacks_async(AfterMultiAgentInvocationEvent(self))
            self._resume_from_session = False
            self._resume_next_nodes.clear()

GraphBuilder

Builder pattern for constructing graphs.

Source code in strands/multiagent/graph.py 
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
class GraphBuilder:
    """Builder pattern for constructing graphs."""

    def __init__(self) -> None:
        """Initialize GraphBuilder with empty collections."""
        self.nodes: dict[str, GraphNode] = {}
        self.edges: set[GraphEdge] = set()
        self.entry_points: set[GraphNode] = set()

        # Configuration options
        self._max_node_executions: Optional[int] = None
        self._execution_timeout: Optional[float] = None
        self._node_timeout: Optional[float] = None
        self._reset_on_revisit: bool = False
        self._id: str = _DEFAULT_GRAPH_ID
        self._session_manager: Optional[SessionManager] = None
        self._hooks: Optional[list[HookProvider]] = None

    def add_node(self, executor: Agent | MultiAgentBase, node_id: str | None = None) -> GraphNode:
        """Add an Agent or MultiAgentBase instance as a node to the graph."""
        _validate_node_executor(executor, self.nodes)

        # Auto-generate node_id if not provided
        if node_id is None:
            node_id = getattr(executor, "id", None) or getattr(executor, "name", None) or f"node_{len(self.nodes)}"

        if node_id in self.nodes:
            raise ValueError(f"Node '{node_id}' already exists")

        node = GraphNode(node_id=node_id, executor=executor)
        self.nodes[node_id] = node
        return node

    def add_edge(
        self,
        from_node: str | GraphNode,
        to_node: str | GraphNode,
        condition: Callable[[GraphState], bool] | None = None,
    ) -> GraphEdge:
        """Add an edge between two nodes with optional condition function that receives full GraphState."""

        def resolve_node(node: str | GraphNode, node_type: str) -> GraphNode:
            if isinstance(node, str):
                if node not in self.nodes:
                    raise ValueError(f"{node_type} node '{node}' not found")
                return self.nodes[node]
            else:
                if node not in self.nodes.values():
                    raise ValueError(f"{node_type} node object has not been added to the graph, use graph.add_node")
                return node

        from_node_obj = resolve_node(from_node, "Source")
        to_node_obj = resolve_node(to_node, "Target")

        # Add edge and update dependencies
        edge = GraphEdge(from_node=from_node_obj, to_node=to_node_obj, condition=condition)
        self.edges.add(edge)
        to_node_obj.dependencies.add(from_node_obj)
        return edge

    def set_entry_point(self, node_id: str) -> "GraphBuilder":
        """Set a node as an entry point for graph execution."""
        if node_id not in self.nodes:
            raise ValueError(f"Node '{node_id}' not found")
        self.entry_points.add(self.nodes[node_id])
        return self

    def reset_on_revisit(self, enabled: bool = True) -> "GraphBuilder":
        """Control whether nodes reset their state when revisited.

        When enabled, nodes will reset their messages and state to initial values
        each time they are revisited (re-executed). This is useful for stateless
        behavior where nodes should start fresh on each revisit.

        Args:
            enabled: Whether to reset node state when revisited (default: True)
        """
        self._reset_on_revisit = enabled
        return self

    def set_max_node_executions(self, max_executions: int) -> "GraphBuilder":
        """Set maximum number of node executions allowed.

        Args:
            max_executions: Maximum total node executions (None for no limit)
        """
        self._max_node_executions = max_executions
        return self

    def set_execution_timeout(self, timeout: float) -> "GraphBuilder":
        """Set total execution timeout.

        Args:
            timeout: Total execution timeout in seconds (None for no limit)
        """
        self._execution_timeout = timeout
        return self

    def set_node_timeout(self, timeout: float) -> "GraphBuilder":
        """Set individual node execution timeout.

        Args:
            timeout: Individual node timeout in seconds (None for no limit)
        """
        self._node_timeout = timeout
        return self

    def set_graph_id(self, graph_id: str) -> "GraphBuilder":
        """Set graph id.

        Args:
            graph_id: Unique graph id
        """
        self._id = graph_id
        return self

    def set_session_manager(self, session_manager: SessionManager) -> "GraphBuilder":
        """Set session manager for the graph.

        Args:
            session_manager: SessionManager instance
        """
        self._session_manager = session_manager
        return self

    def set_hook_providers(self, hooks: list[HookProvider]) -> "GraphBuilder":
        """Set hook providers for the graph.

        Args:
            hooks: Customer hooks user passes in
        """
        self._hooks = hooks
        return self

    def build(self) -> "Graph":
        """Build and validate the graph with configured settings."""
        if not self.nodes:
            raise ValueError("Graph must contain at least one node")

        # Auto-detect entry points if none specified
        if not self.entry_points:
            self.entry_points = {node for node_id, node in self.nodes.items() if not node.dependencies}
            logger.debug(
                "entry_points=<%s> | auto-detected entrypoints", ", ".join(node.node_id for node in self.entry_points)
            )
            if not self.entry_points:
                raise ValueError("No entry points found - all nodes have dependencies")

        # Validate entry points and check for cycles
        self._validate_graph()

        return Graph(
            nodes=self.nodes.copy(),
            edges=self.edges.copy(),
            entry_points=self.entry_points.copy(),
            max_node_executions=self._max_node_executions,
            execution_timeout=self._execution_timeout,
            node_timeout=self._node_timeout,
            reset_on_revisit=self._reset_on_revisit,
            session_manager=self._session_manager,
            hooks=self._hooks,
            id=self._id,
        )

    def _validate_graph(self) -> None:
        """Validate graph structure."""
        # Validate entry points exist
        entry_point_ids = {node.node_id for node in self.entry_points}
        invalid_entries = entry_point_ids - set(self.nodes.keys())
        if invalid_entries:
            raise ValueError(f"Entry points not found in nodes: {invalid_entries}")

        # Warn about potential infinite loops if no execution limits are set
        if self._max_node_executions is None and self._execution_timeout is None:
            logger.warning("Graph without execution limits may run indefinitely if cycles exist")

__init__()

Initialize GraphBuilder with empty collections.

Source code in strands/multiagent/graph.py 
229
230
231
232
233
234
235
236
237
238
239
240
241
242
def __init__(self) -> None:
    """Initialize GraphBuilder with empty collections."""
    self.nodes: dict[str, GraphNode] = {}
    self.edges: set[GraphEdge] = set()
    self.entry_points: set[GraphNode] = set()

    # Configuration options
    self._max_node_executions: Optional[int] = None
    self._execution_timeout: Optional[float] = None
    self._node_timeout: Optional[float] = None
    self._reset_on_revisit: bool = False
    self._id: str = _DEFAULT_GRAPH_ID
    self._session_manager: Optional[SessionManager] = None
    self._hooks: Optional[list[HookProvider]] = None

add_edge(from_node, to_node, condition=None)

Add an edge between two nodes with optional condition function that receives full GraphState.

Source code in strands/multiagent/graph.py 
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
def add_edge(
    self,
    from_node: str | GraphNode,
    to_node: str | GraphNode,
    condition: Callable[[GraphState], bool] | None = None,
) -> GraphEdge:
    """Add an edge between two nodes with optional condition function that receives full GraphState."""

    def resolve_node(node: str | GraphNode, node_type: str) -> GraphNode:
        if isinstance(node, str):
            if node not in self.nodes:
                raise ValueError(f"{node_type} node '{node}' not found")
            return self.nodes[node]
        else:
            if node not in self.nodes.values():
                raise ValueError(f"{node_type} node object has not been added to the graph, use graph.add_node")
            return node

    from_node_obj = resolve_node(from_node, "Source")
    to_node_obj = resolve_node(to_node, "Target")

    # Add edge and update dependencies
    edge = GraphEdge(from_node=from_node_obj, to_node=to_node_obj, condition=condition)
    self.edges.add(edge)
    to_node_obj.dependencies.add(from_node_obj)
    return edge

add_node(executor, node_id=None)

Add an Agent or MultiAgentBase instance as a node to the graph.

Source code in strands/multiagent/graph.py 
244
245
246
247
248
249
250
251
252
253
254
255
256
257
def add_node(self, executor: Agent | MultiAgentBase, node_id: str | None = None) -> GraphNode:
    """Add an Agent or MultiAgentBase instance as a node to the graph."""
    _validate_node_executor(executor, self.nodes)

    # Auto-generate node_id if not provided
    if node_id is None:
        node_id = getattr(executor, "id", None) or getattr(executor, "name", None) or f"node_{len(self.nodes)}"

    if node_id in self.nodes:
        raise ValueError(f"Node '{node_id}' already exists")

    node = GraphNode(node_id=node_id, executor=executor)
    self.nodes[node_id] = node
    return node

build()

Build and validate the graph with configured settings.

Source code in strands/multiagent/graph.py 
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
def build(self) -> "Graph":
    """Build and validate the graph with configured settings."""
    if not self.nodes:
        raise ValueError("Graph must contain at least one node")

    # Auto-detect entry points if none specified
    if not self.entry_points:
        self.entry_points = {node for node_id, node in self.nodes.items() if not node.dependencies}
        logger.debug(
            "entry_points=<%s> | auto-detected entrypoints", ", ".join(node.node_id for node in self.entry_points)
        )
        if not self.entry_points:
            raise ValueError("No entry points found - all nodes have dependencies")

    # Validate entry points and check for cycles
    self._validate_graph()

    return Graph(
        nodes=self.nodes.copy(),
        edges=self.edges.copy(),
        entry_points=self.entry_points.copy(),
        max_node_executions=self._max_node_executions,
        execution_timeout=self._execution_timeout,
        node_timeout=self._node_timeout,
        reset_on_revisit=self._reset_on_revisit,
        session_manager=self._session_manager,
        hooks=self._hooks,
        id=self._id,
    )

reset_on_revisit(enabled=True)

Control whether nodes reset their state when revisited.

When enabled, nodes will reset their messages and state to initial values each time they are revisited (re-executed). This is useful for stateless behavior where nodes should start fresh on each revisit.

Parameters:

Name Type Description Default
enabled bool

Whether to reset node state when revisited (default: True)

 True
Source code in strands/multiagent/graph.py 
293
294
295
296
297
298
299
300
301
302
303
304
def reset_on_revisit(self, enabled: bool = True) -> "GraphBuilder":
    """Control whether nodes reset their state when revisited.

    When enabled, nodes will reset their messages and state to initial values
    each time they are revisited (re-executed). This is useful for stateless
    behavior where nodes should start fresh on each revisit.

    Args:
        enabled: Whether to reset node state when revisited (default: True)
    """
    self._reset_on_revisit = enabled
    return self

set_entry_point(node_id)

Set a node as an entry point for graph execution.

Source code in strands/multiagent/graph.py 
286
287
288
289
290
291
def set_entry_point(self, node_id: str) -> "GraphBuilder":
    """Set a node as an entry point for graph execution."""
    if node_id not in self.nodes:
        raise ValueError(f"Node '{node_id}' not found")
    self.entry_points.add(self.nodes[node_id])
    return self

set_execution_timeout(timeout)

Set total execution timeout.

Parameters:

Name Type Description Default
timeout float

Total execution timeout in seconds (None for no limit)

 required
Source code in strands/multiagent/graph.py 
315
316
317
318
319
320
321
322
def set_execution_timeout(self, timeout: float) -> "GraphBuilder":
    """Set total execution timeout.

    Args:
        timeout: Total execution timeout in seconds (None for no limit)
    """
    self._execution_timeout = timeout
    return self

set_graph_id(graph_id)

Set graph id.

Parameters:

Name Type Description Default
graph_id str

Unique graph id

 required
Source code in strands/multiagent/graph.py 
333
334
335
336
337
338
339
340
def set_graph_id(self, graph_id: str) -> "GraphBuilder":
    """Set graph id.

    Args:
        graph_id: Unique graph id
    """
    self._id = graph_id
    return self

set_hook_providers(hooks)

Set hook providers for the graph.

Parameters:

Name Type Description Default
hooks list[HookProvider]

Customer hooks user passes in

 required
Source code in strands/multiagent/graph.py 
351
352
353
354
355
356
357
358
def set_hook_providers(self, hooks: list[HookProvider]) -> "GraphBuilder":
    """Set hook providers for the graph.

    Args:
        hooks: Customer hooks user passes in
    """
    self._hooks = hooks
    return self

set_max_node_executions(max_executions)

Set maximum number of node executions allowed.

Parameters:

Name Type Description Default
max_executions int

Maximum total node executions (None for no limit)

 required
Source code in strands/multiagent/graph.py 
306
307
308
309
310
311
312
313
def set_max_node_executions(self, max_executions: int) -> "GraphBuilder":
    """Set maximum number of node executions allowed.

    Args:
        max_executions: Maximum total node executions (None for no limit)
    """
    self._max_node_executions = max_executions
    return self

set_node_timeout(timeout)

Set individual node execution timeout.

Parameters:

Name Type Description Default
timeout float

Individual node timeout in seconds (None for no limit)

 required
Source code in strands/multiagent/graph.py 
324
325
326
327
328
329
330
331
def set_node_timeout(self, timeout: float) -> "GraphBuilder":
    """Set individual node execution timeout.

    Args:
        timeout: Individual node timeout in seconds (None for no limit)
    """
    self._node_timeout = timeout
    return self

set_session_manager(session_manager)

Set session manager for the graph.

Parameters:

Name Type Description Default
session_manager SessionManager

SessionManager instance

 required
Source code in strands/multiagent/graph.py 
342
343
344
345
346
347
348
349
def set_session_manager(self, session_manager: SessionManager) -> "GraphBuilder":
    """Set session manager for the graph.

    Args:
        session_manager: SessionManager instance
    """
    self._session_manager = session_manager
    return self

GraphEdge dataclass

Represents an edge in the graph with an optional condition.

Source code in strands/multiagent/graph.py 
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
@dataclass
class GraphEdge:
    """Represents an edge in the graph with an optional condition."""

    from_node: "GraphNode"
    to_node: "GraphNode"
    condition: Callable[[GraphState], bool] | None = None

    def __hash__(self) -> int:
        """Return hash for GraphEdge based on from_node and to_node."""
        return hash((self.from_node.node_id, self.to_node.node_id))

    def should_traverse(self, state: GraphState) -> bool:
        """Check if this edge should be traversed based on condition."""
        if self.condition is None:
            return True
        return self.condition(state)

__hash__()

Return hash for GraphEdge based on from_node and to_node.

Source code in strands/multiagent/graph.py 
138
139
140
def __hash__(self) -> int:
    """Return hash for GraphEdge based on from_node and to_node."""
    return hash((self.from_node.node_id, self.to_node.node_id))

should_traverse(state)

Check if this edge should be traversed based on condition.

Source code in strands/multiagent/graph.py 
142
143
144
145
146
def should_traverse(self, state: GraphState) -> bool:
    """Check if this edge should be traversed based on condition."""
    if self.condition is None:
        return True
    return self.condition(state)

GraphNode dataclass

Represents a node in the graph.

The execution_status tracks the node's lifecycle within graph orchestration: - PENDING: Node hasn't started executing yet - EXECUTING: Node is currently running - COMPLETED/FAILED: Node finished executing (regardless of result quality)

Source code in strands/multiagent/graph.py 
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
@dataclass
class GraphNode:
    """Represents a node in the graph.

    The execution_status tracks the node's lifecycle within graph orchestration:
    - PENDING: Node hasn't started executing yet
    - EXECUTING: Node is currently running
    - COMPLETED/FAILED: Node finished executing (regardless of result quality)
    """

    node_id: str
    executor: Agent | MultiAgentBase
    dependencies: set["GraphNode"] = field(default_factory=set)
    execution_status: Status = Status.PENDING
    result: NodeResult | None = None
    execution_time: int = 0
    _initial_messages: Messages = field(default_factory=list, init=False)
    _initial_state: AgentState = field(default_factory=AgentState, init=False)

    def __post_init__(self) -> None:
        """Capture initial executor state after initialization."""
        # Deep copy the initial messages and state to preserve them
        if hasattr(self.executor, "messages"):
            self._initial_messages = copy.deepcopy(self.executor.messages)

        if hasattr(self.executor, "state") and hasattr(self.executor.state, "get"):
            self._initial_state = AgentState(self.executor.state.get())

    def reset_executor_state(self) -> None:
        """Reset GraphNode executor state to initial state when graph was created.

        This is useful when nodes are executed multiple times and need to start
        fresh on each execution, providing stateless behavior.
        """
        if hasattr(self.executor, "messages"):
            self.executor.messages = copy.deepcopy(self._initial_messages)

        if hasattr(self.executor, "state"):
            self.executor.state = AgentState(self._initial_state.get())

        # Reset execution status
        self.execution_status = Status.PENDING
        self.result = None

    def __hash__(self) -> int:
        """Return hash for GraphNode based on node_id."""
        return hash(self.node_id)

    def __eq__(self, other: Any) -> bool:
        """Return equality for GraphNode based on node_id."""
        if not isinstance(other, GraphNode):
            return False
        return self.node_id == other.node_id

__eq__(other)

Return equality for GraphNode based on node_id.

Source code in strands/multiagent/graph.py 
197
198
199
200
201
def __eq__(self, other: Any) -> bool:
    """Return equality for GraphNode based on node_id."""
    if not isinstance(other, GraphNode):
        return False
    return self.node_id == other.node_id

__hash__()

Return hash for GraphNode based on node_id.

Source code in strands/multiagent/graph.py 
193
194
195
def __hash__(self) -> int:
    """Return hash for GraphNode based on node_id."""
    return hash(self.node_id)

__post_init__()

Capture initial executor state after initialization.

Source code in strands/multiagent/graph.py 
168
169
170
171
172
173
174
175
def __post_init__(self) -> None:
    """Capture initial executor state after initialization."""
    # Deep copy the initial messages and state to preserve them
    if hasattr(self.executor, "messages"):
        self._initial_messages = copy.deepcopy(self.executor.messages)

    if hasattr(self.executor, "state") and hasattr(self.executor.state, "get"):
        self._initial_state = AgentState(self.executor.state.get())

reset_executor_state()

Reset GraphNode executor state to initial state when graph was created.

This is useful when nodes are executed multiple times and need to start fresh on each execution, providing stateless behavior.

Source code in strands/multiagent/graph.py 
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
def reset_executor_state(self) -> None:
    """Reset GraphNode executor state to initial state when graph was created.

    This is useful when nodes are executed multiple times and need to start
    fresh on each execution, providing stateless behavior.
    """
    if hasattr(self.executor, "messages"):
        self.executor.messages = copy.deepcopy(self._initial_messages)

    if hasattr(self.executor, "state"):
        self.executor.state = AgentState(self._initial_state.get())

    # Reset execution status
    self.execution_status = Status.PENDING
    self.result = None

GraphResult dataclass

Bases: MultiAgentResult

Result from graph execution - extends MultiAgentResult with graph-specific details.

Source code in strands/multiagent/graph.py 
118
119
120
121
122
123
124
125
126
127
@dataclass
class GraphResult(MultiAgentResult):
    """Result from graph execution - extends MultiAgentResult with graph-specific details."""

    total_nodes: int = 0
    completed_nodes: int = 0
    failed_nodes: int = 0
    execution_order: list["GraphNode"] = field(default_factory=list)
    edges: list[Tuple["GraphNode", "GraphNode"]] = field(default_factory=list)
    entry_points: list["GraphNode"] = field(default_factory=list)

GraphState dataclass

Graph execution state.

Attributes:

Name Type Description
status Status

Current execution status of the graph.
completed_nodes set[GraphNode]

Set of nodes that have completed execution.
failed_nodes set[GraphNode]

Set of nodes that failed during execution.
execution_order list[GraphNode]

List of nodes in the order they were executed.
task MultiAgentInput

The original input prompt/query provided to the graph execution. This represents the actual work to be performed by the graph as a whole. Entry point nodes receive this task as their input if they have no dependencies.
Source code in strands/multiagent/graph.py 
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
@dataclass
class GraphState:
    """Graph execution state.

    Attributes:
        status: Current execution status of the graph.
        completed_nodes: Set of nodes that have completed execution.
        failed_nodes: Set of nodes that failed during execution.
        execution_order: List of nodes in the order they were executed.
        task: The original input prompt/query provided to the graph execution.
              This represents the actual work to be performed by the graph as a whole.
              Entry point nodes receive this task as their input if they have no dependencies.
    """

    # Task (with default empty string)
    task: MultiAgentInput = ""

    # Execution state
    status: Status = Status.PENDING
    completed_nodes: set["GraphNode"] = field(default_factory=set)
    failed_nodes: set["GraphNode"] = field(default_factory=set)
    execution_order: list["GraphNode"] = field(default_factory=list)
    start_time: float = field(default_factory=time.time)

    # Results
    results: dict[str, NodeResult] = field(default_factory=dict)

    # Accumulated metrics
    accumulated_usage: Usage = field(default_factory=lambda: Usage(inputTokens=0, outputTokens=0, totalTokens=0))
    accumulated_metrics: Metrics = field(default_factory=lambda: Metrics(latencyMs=0))
    execution_count: int = 0
    execution_time: int = 0

    # Graph structure info
    total_nodes: int = 0
    edges: list[Tuple["GraphNode", "GraphNode"]] = field(default_factory=list)
    entry_points: list["GraphNode"] = field(default_factory=list)

    def should_continue(
        self,
        max_node_executions: Optional[int],
        execution_timeout: Optional[float],
    ) -> Tuple[bool, str]:
        """Check if the graph should continue execution.

        Returns: (should_continue, reason)
        """
        # Check node execution limit (only if set)
        if max_node_executions is not None and len(self.execution_order) >= max_node_executions:
            return False, f"Max node executions reached: {max_node_executions}"

        # Check timeout (only if set)
        if execution_timeout is not None:
            elapsed = time.time() - self.start_time
            if elapsed > execution_timeout:
                return False, f"Execution timed out: {execution_timeout}s"

        return True, "Continuing"

should_continue(max_node_executions, execution_timeout)

Check if the graph should continue execution.

Returns: (should_continue, reason)

Source code in strands/multiagent/graph.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
def should_continue(
    self,
    max_node_executions: Optional[int],
    execution_timeout: Optional[float],
) -> Tuple[bool, str]:
    """Check if the graph should continue execution.

    Returns: (should_continue, reason)
    """
    # Check node execution limit (only if set)
    if max_node_executions is not None and len(self.execution_order) >= max_node_executions:
        return False, f"Max node executions reached: {max_node_executions}"

    # Check timeout (only if set)
    if execution_timeout is not None:
        elapsed = time.time() - self.start_time
        if elapsed > execution_timeout:
            return False, f"Execution timed out: {execution_timeout}s"

    return True, "Continuing"

strands.multiagent.swarm

Swarm Multi-Agent Pattern Implementation.

This module provides a collaborative agent orchestration system where agents work together as a team to solve complex tasks, with shared context and autonomous coordination.

Key Features: - Self-organizing agent teams with shared working memory - Tool-based coordination - Autonomous agent collaboration without central control - Dynamic task distribution based on agent capabilities - Collective intelligence through shared context

SharedContext dataclass

Shared context between swarm nodes.

Source code in strands/multiagent/swarm.py 
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
@dataclass
class SharedContext:
    """Shared context between swarm nodes."""

    context: dict[str, dict[str, Any]] = field(default_factory=dict)

    def add_context(self, node: SwarmNode, key: str, value: Any) -> None:
        """Add context."""
        self._validate_key(key)
        self._validate_json_serializable(value)

        if node.node_id not in self.context:
            self.context[node.node_id] = {}
        self.context[node.node_id][key] = value

    def _validate_key(self, key: str) -> None:
        """Validate that a key is valid.

        Args:
            key: The key to validate

        Raises:
            ValueError: If key is invalid
        """
        if key is None:
            raise ValueError("Key cannot be None")
        if not isinstance(key, str):
            raise ValueError("Key must be a string")
        if not key.strip():
            raise ValueError("Key cannot be empty")

    def _validate_json_serializable(self, value: Any) -> None:
        """Validate that a value is JSON serializable.

        Args:
            value: The value to validate

        Raises:
            ValueError: If value is not JSON serializable
        """
        try:
            json.dumps(value)
        except (TypeError, ValueError) as e:
            raise ValueError(
                f"Value is not JSON serializable: {type(value).__name__}. "
                f"Only JSON-compatible types (str, int, float, bool, list, dict, None) are allowed."
            ) from e

add_context(node, key, value)

Add context.

Source code in strands/multiagent/swarm.py 
103
104
105
106
107
108
109
110
def add_context(self, node: SwarmNode, key: str, value: Any) -> None:
    """Add context."""
    self._validate_key(key)
    self._validate_json_serializable(value)

    if node.node_id not in self.context:
        self.context[node.node_id] = {}
    self.context[node.node_id][key] = value

Swarm

Bases: MultiAgentBase

Self-organizing collaborative agent teams with shared working memory.

Source code in strands/multiagent/swarm.py 
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
class Swarm(MultiAgentBase):
    """Self-organizing collaborative agent teams with shared working memory."""

    def __init__(
        self,
        nodes: list[Agent],
        *,
        entry_point: Agent | None = None,
        max_handoffs: int = 20,
        max_iterations: int = 20,
        execution_timeout: float = 900.0,
        node_timeout: float = 300.0,
        repetitive_handoff_detection_window: int = 0,
        repetitive_handoff_min_unique_agents: int = 0,
        session_manager: Optional[SessionManager] = None,
        hooks: Optional[list[HookProvider]] = None,
        id: str = _DEFAULT_SWARM_ID,
        trace_attributes: Optional[Mapping[str, AttributeValue]] = None,
    ) -> None:
        """Initialize Swarm with agents and configuration.

        Args:
            id : Unique swarm id (default: None)
            nodes: List of nodes (e.g. Agent) to include in the swarm
            entry_point: Agent to start with. If None, uses the first agent (default: None)
            max_handoffs: Maximum handoffs to agents and users (default: 20)
            max_iterations: Maximum node executions within the swarm (default: 20)
            execution_timeout: Total execution timeout in seconds (default: 900.0)
            node_timeout: Individual node timeout in seconds (default: 300.0)
            repetitive_handoff_detection_window: Number of recent nodes to check for repetitive handoffs
                Disabled by default (default: 0)
            repetitive_handoff_min_unique_agents: Minimum unique agents required in recent sequence
                Disabled by default (default: 0)
            session_manager: Session manager for persisting graph state and execution history (default: None)
            hooks: List of hook providers for monitoring and extending graph execution behavior (default: None)
            trace_attributes: Custom trace attributes to apply to the agent's trace span (default: None)
        """
        super().__init__()
        self.id = id
        self.entry_point = entry_point
        self.max_handoffs = max_handoffs
        self.max_iterations = max_iterations
        self.execution_timeout = execution_timeout
        self.node_timeout = node_timeout
        self.repetitive_handoff_detection_window = repetitive_handoff_detection_window
        self.repetitive_handoff_min_unique_agents = repetitive_handoff_min_unique_agents

        self.shared_context = SharedContext()
        self.nodes: dict[str, SwarmNode] = {}
        self.state = SwarmState(
            current_node=None,  # Placeholder, will be set properly
            task="",
            completion_status=Status.PENDING,
        )
        self.tracer = get_tracer()
        self.trace_attributes: dict[str, AttributeValue] = self._parse_trace_attributes(trace_attributes)

        self.session_manager = session_manager
        self.hooks = HookRegistry()
        if hooks:
            for hook in hooks:
                self.hooks.add_hook(hook)
        if self.session_manager:
            self.hooks.add_hook(self.session_manager)

        self._resume_from_session = False

        self._setup_swarm(nodes)
        self._inject_swarm_tools()
        run_async(lambda: self.hooks.invoke_callbacks_async(MultiAgentInitializedEvent(self)))

    def __call__(
        self, task: MultiAgentInput, invocation_state: dict[str, Any] | None = None, **kwargs: Any
    ) -> SwarmResult:
        """Invoke the swarm synchronously.

        Args:
            task: The task to execute
            invocation_state: Additional state/context passed to underlying agents.
                Defaults to None to avoid mutable default argument issues.
            **kwargs: Keyword arguments allowing backward compatible future changes.
        """
        if invocation_state is None:
            invocation_state = {}
        return run_async(lambda: self.invoke_async(task, invocation_state))

    async def invoke_async(
        self, task: MultiAgentInput, invocation_state: dict[str, Any] | None = None, **kwargs: Any
    ) -> SwarmResult:
        """Invoke the swarm asynchronously.

        This method uses stream_async internally and consumes all events until completion,
        following the same pattern as the Agent class.

        Args:
            task: The task to execute
            invocation_state: Additional state/context passed to underlying agents.
                Defaults to None to avoid mutable default argument issues.
            **kwargs: Keyword arguments allowing backward compatible future changes.
        """
        events = self.stream_async(task, invocation_state, **kwargs)
        final_event = None
        async for event in events:
            final_event = event

        if final_event is None or "result" not in final_event:
            raise ValueError("Swarm streaming completed without producing a result event")

        return cast(SwarmResult, final_event["result"])

    async def stream_async(
        self, task: MultiAgentInput, invocation_state: dict[str, Any] | None = None, **kwargs: Any
    ) -> AsyncIterator[dict[str, Any]]:
        """Stream events during swarm execution.

        Args:
            task: The task to execute
            invocation_state: Additional state/context passed to underlying agents.
                Defaults to None to avoid mutable default argument issues.
            **kwargs: Keyword arguments allowing backward compatible future changes.

        Yields:
            Dictionary events during swarm execution, such as:
            - multi_agent_node_start: When a node begins execution
            - multi_agent_node_stream: Forwarded agent events with node context
            - multi_agent_handoff: When control is handed off between agents
            - multi_agent_node_stop: When a node stops execution
            - result: Final swarm result
        """
        if invocation_state is None:
            invocation_state = {}

        await self.hooks.invoke_callbacks_async(BeforeMultiAgentInvocationEvent(self, invocation_state))

        logger.debug("starting swarm execution")

        if not self._resume_from_session:
            # Initialize swarm state with configuration
            initial_node = self._initial_node()

            self.state = SwarmState(
                current_node=initial_node,
                task=task,
                completion_status=Status.EXECUTING,
                shared_context=self.shared_context,
            )
        else:
            self.state.completion_status = Status.EXECUTING
            self.state.start_time = time.time()

        span = self.tracer.start_multiagent_span(task, "swarm", custom_trace_attributes=self.trace_attributes)
        with trace_api.use_span(span, end_on_exit=True):
            try:
                current_node = cast(SwarmNode, self.state.current_node)
                logger.debug("current_node=<%s> | starting swarm execution with node", current_node.node_id)
                logger.debug(
                    "max_handoffs=<%d>, max_iterations=<%d>, timeout=<%s>s | swarm execution config",
                    self.max_handoffs,
                    self.max_iterations,
                    self.execution_timeout,
                )

                async for event in self._execute_swarm(invocation_state):
                    yield event.as_dict()

            except Exception:
                logger.exception("swarm execution failed")
                self.state.completion_status = Status.FAILED
                raise
            finally:
                self.state.execution_time = round((time.time() - self.state.start_time) * 1000)
                await self.hooks.invoke_callbacks_async(AfterMultiAgentInvocationEvent(self, invocation_state))
                self._resume_from_session = False

            # Yield final result after execution_time is set
            result = self._build_result()
            yield MultiAgentResultEvent(result=result).as_dict()

    async def _stream_with_timeout(
        self, async_generator: AsyncIterator[Any], timeout: float | None, timeout_message: str
    ) -> AsyncIterator[Any]:
        """Wrap an async generator with timeout for total execution time.

        Tracks elapsed time from start and enforces timeout across all events.
        Each event wait uses remaining time from the total timeout budget.

        Args:
            async_generator: The generator to wrap
            timeout: Total timeout in seconds for entire stream, or None for no timeout
            timeout_message: Message to include in timeout exception

        Yields:
            Events from the wrapped generator as they arrive

        Raises:
            Exception: If total execution time exceeds timeout
        """
        if timeout is None:
            # No timeout - just pass through
            async for event in async_generator:
                yield event
        else:
            # Track start time for total timeout
            start_time = asyncio.get_event_loop().time()

            while True:
                # Calculate remaining time from total timeout budget
                elapsed = asyncio.get_event_loop().time() - start_time
                remaining = timeout - elapsed

                if remaining <= 0:
                    raise Exception(timeout_message)

                try:
                    event = await asyncio.wait_for(async_generator.__anext__(), timeout=remaining)
                    yield event
                except StopAsyncIteration:
                    break
                except asyncio.TimeoutError as err:
                    raise Exception(timeout_message) from err

    def _setup_swarm(self, nodes: list[Agent]) -> None:
        """Initialize swarm configuration."""
        # Validate nodes before setup
        self._validate_swarm(nodes)

        # Validate agents have names and create SwarmNode objects
        for i, node in enumerate(nodes):
            if not node.name:
                node_id = f"node_{i}"
                node.name = node_id
                logger.debug("node_id=<%s> | agent has no name, dynamically generating one", node_id)

            node_id = str(node.name)

            # Ensure node IDs are unique
            if node_id in self.nodes:
                raise ValueError(f"Node ID '{node_id}' is not unique. Each agent must have a unique name.")

            self.nodes[node_id] = SwarmNode(node_id=node_id, executor=node)

        # Validate entry point if specified
        if self.entry_point is not None:
            entry_point_node_id = str(self.entry_point.name)
            if (
                entry_point_node_id not in self.nodes
                or self.nodes[entry_point_node_id].executor is not self.entry_point
            ):
                available_agents = [
                    f"{node_id} ({type(node.executor).__name__})" for node_id, node in self.nodes.items()
                ]
                raise ValueError(f"Entry point agent not found in swarm nodes. Available agents: {available_agents}")

        swarm_nodes = list(self.nodes.values())
        logger.debug("nodes=<%s> | initialized swarm with nodes", [node.node_id for node in swarm_nodes])

        if self.entry_point:
            entry_point_name = getattr(self.entry_point, "name", "unnamed_agent")
            logger.debug("entry_point=<%s> | configured entry point", entry_point_name)
        else:
            first_node = next(iter(self.nodes.keys()))
            logger.debug("entry_point=<%s> | using first node as entry point", first_node)

    def _validate_swarm(self, nodes: list[Agent]) -> None:
        """Validate swarm structure and nodes."""
        # Check for duplicate object instances
        seen_instances = set()
        for node in nodes:
            if id(node) in seen_instances:
                raise ValueError("Duplicate node instance detected. Each node must have a unique object instance.")
            seen_instances.add(id(node))

            # Check for session persistence
            if node._session_manager is not None:
                raise ValueError("Session persistence is not supported for Swarm agents yet.")

    def _inject_swarm_tools(self) -> None:
        """Add swarm coordination tools to each agent."""
        # Create tool functions with proper closures
        swarm_tools = [
            self._create_handoff_tool(),
        ]

        for node in self.nodes.values():
            # Check for existing tools with conflicting names
            existing_tools = node.executor.tool_registry.registry
            conflicting_tools = []

            if "handoff_to_agent" in existing_tools:
                conflicting_tools.append("handoff_to_agent")

            if conflicting_tools:
                raise ValueError(
                    f"Agent '{node.node_id}' already has tools with names that conflict with swarm coordination tools: "
                    f"{', '.join(conflicting_tools)}. Please rename these tools to avoid conflicts."
                )

            # Use the agent's tool registry to process and register the tools
            node.executor.tool_registry.process_tools(swarm_tools)

        logger.debug(
            "tool_count=<%d>, node_count=<%d> | injected coordination tools into agents",
            len(swarm_tools),
            len(self.nodes),
        )

    def _create_handoff_tool(self) -> Callable[..., Any]:
        """Create handoff tool for agent coordination."""
        swarm_ref = self  # Capture swarm reference

        @tool
        def handoff_to_agent(agent_name: str, message: str, context: dict[str, Any] | None = None) -> dict[str, Any]:
            """Transfer control to another agent in the swarm for specialized help.

            Args:
                agent_name: Name of the agent to hand off to
                message: Message explaining what needs to be done and why you're handing off
                context: Additional context to share with the next agent

            Returns:
                Confirmation of handoff initiation
            """
            try:
                context = context or {}

                # Validate target agent exists
                target_node = swarm_ref.nodes.get(agent_name)
                if not target_node:
                    return {"status": "error", "content": [{"text": f"Error: Agent '{agent_name}' not found in swarm"}]}

                # Execute handoff
                swarm_ref._handle_handoff(target_node, message, context)

                return {"status": "success", "content": [{"text": f"Handing off to {agent_name}: {message}"}]}
            except Exception as e:
                return {"status": "error", "content": [{"text": f"Error in handoff: {str(e)}"}]}

        return handoff_to_agent

    def _handle_handoff(self, target_node: SwarmNode, message: str, context: dict[str, Any]) -> None:
        """Handle handoff to another agent."""
        # If task is already completed, don't allow further handoffs
        if self.state.completion_status != Status.EXECUTING:
            logger.debug(
                "task_status=<%s> | ignoring handoff request - task already completed",
                self.state.completion_status,
            )
            return

        current_node = cast(SwarmNode, self.state.current_node)

        self.state.handoff_node = target_node
        self.state.handoff_message = message

        # Store handoff context as shared context
        if context:
            for key, value in context.items():
                self.shared_context.add_context(current_node, key, value)

        logger.debug(
            "from_node=<%s>, to_node=<%s> | handing off from agent to agent",
            current_node.node_id,
            target_node.node_id,
        )

    def _build_node_input(self, target_node: SwarmNode) -> str:
        """Build input text for a node based on shared context and handoffs.

        Example formatted output:
        ```
        Handoff Message: The user needs help with Python debugging - I've identified the issue but need someone with more expertise to fix it.

        User Request: My Python script is throwing a KeyError when processing JSON data from an API

        Previous agents who worked on this: data_analyst → code_reviewer

        Shared knowledge from previous agents:
        • data_analyst: {"issue_location": "line 42", "error_type": "missing key validation", "suggested_fix": "add key existence check"}
        • code_reviewer: {"code_quality": "good overall structure", "security_notes": "API key should be in environment variable"}

        Other agents available for collaboration:
        Agent name: data_analyst. Agent description: Analyzes data and provides deeper insights
        Agent name: code_reviewer.
        Agent name: security_specialist. Agent description: Focuses on secure coding practices and vulnerability assessment

        You have access to swarm coordination tools if you need help from other agents. If you don't hand off to another agent, the swarm will consider the task complete.
        ```
        """  # noqa: E501
        context_info: dict[str, Any] = {
            "task": self.state.task,
            "node_history": [node.node_id for node in self.state.node_history],
            "shared_context": {k: v for k, v in self.shared_context.context.items()},
        }
        context_text = ""

        # Include handoff message prominently at the top if present
        if self.state.handoff_message:
            context_text += f"Handoff Message: {self.state.handoff_message}\n\n"

        # Include task information if available
        if "task" in context_info:
            task = context_info.get("task")
            if isinstance(task, str):
                context_text += f"User Request: {task}\n\n"
            elif isinstance(task, list):
                context_text += "User Request: Multi-modal task\n\n"

        # Include detailed node history
        if context_info.get("node_history"):
            context_text += f"Previous agents who worked on this: {' → '.join(context_info['node_history'])}\n\n"

        # Include actual shared context, not just a mention
        shared_context = context_info.get("shared_context", {})
        if shared_context:
            context_text += "Shared knowledge from previous agents:\n"
            for node_name, context in shared_context.items():
                if context:  # Only include if node has contributed context
                    context_text += f"• {node_name}: {context}\n"
            context_text += "\n"

        # Include available nodes with descriptions if available
        other_nodes = [node_id for node_id in self.nodes.keys() if node_id != target_node.node_id]
        if other_nodes:
            context_text += "Other agents available for collaboration:\n"
            for node_id in other_nodes:
                node = self.nodes.get(node_id)
                context_text += f"Agent name: {node_id}."
                if node and hasattr(node.executor, "description") and node.executor.description:
                    context_text += f" Agent description: {node.executor.description}"
                context_text += "\n"
            context_text += "\n"

        context_text += (
            "You have access to swarm coordination tools if you need help from other agents. "
            "If you don't hand off to another agent, the swarm will consider the task complete."
        )

        return context_text

    async def _execute_swarm(self, invocation_state: dict[str, Any]) -> AsyncIterator[Any]:
        """Execute swarm and yield TypedEvent objects."""
        try:
            # Main execution loop
            while True:
                if self.state.completion_status != Status.EXECUTING:
                    reason = f"Completion status is: {self.state.completion_status}"
                    logger.debug("reason=<%s> | stopping streaming execution", reason)
                    break

                should_continue, reason = self.state.should_continue(
                    max_handoffs=self.max_handoffs,
                    max_iterations=self.max_iterations,
                    execution_timeout=self.execution_timeout,
                    repetitive_handoff_detection_window=self.repetitive_handoff_detection_window,
                    repetitive_handoff_min_unique_agents=self.repetitive_handoff_min_unique_agents,
                )
                if not should_continue:
                    self.state.completion_status = Status.FAILED
                    logger.debug("reason=<%s> | stopping execution", reason)
                    break

                current_node = self.state.current_node
                if not current_node or current_node.node_id not in self.nodes:
                    logger.error("node=<%s> | node not found", current_node.node_id if current_node else "None")
                    self.state.completion_status = Status.FAILED
                    break

                logger.debug(
                    "current_node=<%s>, iteration=<%d> | executing node",
                    current_node.node_id,
                    len(self.state.node_history) + 1,
                )

                before_event, _ = await self.hooks.invoke_callbacks_async(
                    BeforeNodeCallEvent(self, current_node.node_id, invocation_state)
                )

                # TODO: Implement cancellation token to stop _execute_node from continuing
                try:
                    if before_event.cancel_node:
                        cancel_message = (
                            before_event.cancel_node
                            if isinstance(before_event.cancel_node, str)
                            else "node cancelled by user"
                        )
                        logger.debug("reason=<%s> | cancelling execution", cancel_message)
                        yield MultiAgentNodeCancelEvent(current_node.node_id, cancel_message)
                        self.state.completion_status = Status.FAILED
                        break

                    node_stream = self._stream_with_timeout(
                        self._execute_node(current_node, self.state.task, invocation_state),
                        self.node_timeout,
                        f"Node '{current_node.node_id}' execution timed out after {self.node_timeout}s",
                    )
                    async for event in node_stream:
                        yield event

                    self.state.node_history.append(current_node)

                except Exception:
                    logger.exception("node=<%s> | node execution failed", current_node.node_id)
                    self.state.completion_status = Status.FAILED
                    break

                finally:
                    await self.hooks.invoke_callbacks_async(
                        AfterNodeCallEvent(self, current_node.node_id, invocation_state)
                    )

                logger.debug("node=<%s> | node execution completed", current_node.node_id)

                # Check if handoff requested during execution
                if self.state.handoff_node:
                    previous_node = current_node
                    current_node = self.state.handoff_node

                    self.state.handoff_node = None
                    self.state.current_node = current_node

                    handoff_event = MultiAgentHandoffEvent(
                        from_node_ids=[previous_node.node_id],
                        to_node_ids=[current_node.node_id],
                        message=self.state.handoff_message or "Agent handoff occurred",
                    )
                    yield handoff_event
                    logger.debug(
                        "from_node=<%s>, to_node=<%s> | handoff detected",
                        previous_node.node_id,
                        current_node.node_id,
                    )

                else:
                    logger.debug("node=<%s> | no handoff occurred, marking swarm as complete", current_node.node_id)
                    self.state.completion_status = Status.COMPLETED
                    break

        except Exception:
            logger.exception("swarm execution failed")
            self.state.completion_status = Status.FAILED
        finally:
            elapsed_time = time.time() - self.state.start_time
            logger.debug("status=<%s> | swarm execution completed", self.state.completion_status)
            logger.debug(
                "node_history_length=<%d>, time=<%s>s | metrics",
                len(self.state.node_history),
                f"{elapsed_time:.2f}",
            )

    async def _execute_node(
        self, node: SwarmNode, task: MultiAgentInput, invocation_state: dict[str, Any]
    ) -> AsyncIterator[Any]:
        """Execute swarm node and yield TypedEvent objects."""
        start_time = time.time()
        node_name = node.node_id

        # Emit node start event
        start_event = MultiAgentNodeStartEvent(node_id=node_name, node_type="agent")
        yield start_event

        try:
            # Prepare context for node
            context_text = self._build_node_input(node)
            node_input = [ContentBlock(text=f"Context:\n{context_text}\n\n")]

            # Clear handoff message after it's been included in context
            self.state.handoff_message = None

            if not isinstance(task, str):
                # Include additional ContentBlocks in node input
                node_input = node_input + task

            # Execute node with streaming
            node.reset_executor_state()

            # Stream agent events with node context and capture final result
            result = None
            async for event in node.executor.stream_async(node_input, invocation_state=invocation_state):
                # Forward agent events with node context
                wrapped_event = MultiAgentNodeStreamEvent(node_name, event)
                yield wrapped_event
                # Capture the final result event
                if "result" in event:
                    result = event["result"]

            if result is None:
                raise ValueError(f"Node '{node_name}' did not produce a result event")

            if result.stop_reason == "interrupt":
                node.executor.messages.pop()  # remove interrupted tool use message
                node.executor._interrupt_state.deactivate()

                raise RuntimeError("user raised interrupt from agent | interrupts are not yet supported in swarms")

            execution_time = round((time.time() - start_time) * 1000)

            # Create NodeResult with extracted metrics
            result_metrics = getattr(result, "metrics", None)
            usage = getattr(result_metrics, "accumulated_usage", Usage(inputTokens=0, outputTokens=0, totalTokens=0))
            metrics = getattr(result_metrics, "accumulated_metrics", Metrics(latencyMs=execution_time))

            node_result = NodeResult(
                result=result,
                execution_time=execution_time,
                status=Status.COMPLETED,
                accumulated_usage=usage,
                accumulated_metrics=metrics,
                execution_count=1,
            )

            # Store result in state
            self.state.results[node_name] = node_result

            # Accumulate metrics
            self._accumulate_metrics(node_result)

            # Emit node stop event with full NodeResult
            complete_event = MultiAgentNodeStopEvent(
                node_id=node_name,
                node_result=node_result,
            )
            yield complete_event

        except Exception as e:
            execution_time = round((time.time() - start_time) * 1000)
            logger.exception("node=<%s> | node execution failed", node_name)

            # Create a NodeResult for the failed node
            node_result = NodeResult(
                result=e,
                execution_time=execution_time,
                status=Status.FAILED,
                accumulated_usage=Usage(inputTokens=0, outputTokens=0, totalTokens=0),
                accumulated_metrics=Metrics(latencyMs=execution_time),
                execution_count=1,
            )

            # Store result in state
            self.state.results[node_name] = node_result

            # Emit node stop event even for failures
            complete_event = MultiAgentNodeStopEvent(
                node_id=node_name,
                node_result=node_result,
            )
            yield complete_event

            raise

    def _accumulate_metrics(self, node_result: NodeResult) -> None:
        """Accumulate metrics from a node result."""
        self.state.accumulated_usage["inputTokens"] += node_result.accumulated_usage.get("inputTokens", 0)
        self.state.accumulated_usage["outputTokens"] += node_result.accumulated_usage.get("outputTokens", 0)
        self.state.accumulated_usage["totalTokens"] += node_result.accumulated_usage.get("totalTokens", 0)
        self.state.accumulated_metrics["latencyMs"] += node_result.accumulated_metrics.get("latencyMs", 0)

    def _build_result(self) -> SwarmResult:
        """Build swarm result from current state."""
        return SwarmResult(
            status=self.state.completion_status,
            results=self.state.results,
            accumulated_usage=self.state.accumulated_usage,
            accumulated_metrics=self.state.accumulated_metrics,
            execution_count=len(self.state.node_history),
            execution_time=self.state.execution_time,
            node_history=self.state.node_history,
        )

    def serialize_state(self) -> dict[str, Any]:
        """Serialize the current swarm state to a dictionary."""
        status_str = self.state.completion_status.value
        if self.state.handoff_node:
            next_nodes = [self.state.handoff_node.node_id]
        elif self.state.completion_status == Status.EXECUTING and self.state.current_node:
            next_nodes = [self.state.current_node.node_id]
        else:
            next_nodes = []

        return {
            "type": "swarm",
            "id": self.id,
            "status": status_str,
            "node_history": [n.node_id for n in self.state.node_history],
            "node_results": {k: v.to_dict() for k, v in self.state.results.items()},
            "next_nodes_to_execute": next_nodes,
            "current_task": self.state.task,
            "context": {
                "shared_context": getattr(self.state.shared_context, "context", {}) or {},
                "handoff_message": self.state.handoff_message,
            },
        }

    def deserialize_state(self, payload: dict[str, Any]) -> None:
        """Restore swarm state from a session dict and prepare for execution.

        This method handles two scenarios:
        1. If the persisted status is COMPLETED, FAILED resets all nodes and graph state
           to allow re-execution from the beginning.
        2. Otherwise, restores the persisted state and prepares to resume execution
           from the next ready nodes.

        Args:
            payload: Dictionary containing persisted state data including status,
                    completed nodes, results, and next nodes to execute.
        """
        if not payload.get("next_nodes_to_execute"):
            for node in self.nodes.values():
                node.reset_executor_state()
            self.state = SwarmState(
                current_node=SwarmNode("", Agent()),
                task="",
                completion_status=Status.PENDING,
            )
            self._resume_from_session = False
            return
        else:
            self._from_dict(payload)
            self._resume_from_session = True

    def _from_dict(self, payload: dict[str, Any]) -> None:
        self.state.completion_status = Status(payload["status"])
        # Hydrate completed nodes & results
        context = payload["context"] or {}
        self.shared_context.context = context.get("shared_context") or {}
        self.state.handoff_message = context.get("handoff_message")

        self.state.node_history = [self.nodes[nid] for nid in (payload.get("node_history") or []) if nid in self.nodes]

        raw_results = payload.get("node_results") or {}
        results: dict[str, NodeResult] = {}
        for node_id, entry in raw_results.items():
            if node_id not in self.nodes:
                continue
            try:
                results[node_id] = NodeResult.from_dict(entry)
            except Exception:
                logger.exception("Failed to hydrate NodeResult for node_id=%s; skipping.", node_id)
                raise
        self.state.results = results
        self.state.task = payload.get("current_task", self.state.task)

        next_node_ids = payload.get("next_nodes_to_execute") or []
        if next_node_ids:
            self.state.current_node = self.nodes[next_node_ids[0]] if next_node_ids[0] else self._initial_node()

    def _initial_node(self) -> SwarmNode:
        if self.entry_point:
            return self.nodes[str(self.entry_point.name)]
        return next(iter(self.nodes.values()))  # First SwarmNode

__call__(task, invocation_state=None, **kwargs)

Invoke the swarm synchronously.

Parameters:

Name Type Description Default
task MultiAgentInput

The task to execute

 required
invocation_state dict[str, Any] | None

Additional state/context passed to underlying agents. Defaults to None to avoid mutable default argument issues.

 None
**kwargs Any

Keyword arguments allowing backward compatible future changes.

 {}
Source code in strands/multiagent/swarm.py 
285
286
287
288
289
290
291
292
293
294
295
296
297
298
def __call__(
    self, task: MultiAgentInput, invocation_state: dict[str, Any] | None = None, **kwargs: Any
) -> SwarmResult:
    """Invoke the swarm synchronously.

    Args:
        task: The task to execute
        invocation_state: Additional state/context passed to underlying agents.
            Defaults to None to avoid mutable default argument issues.
        **kwargs: Keyword arguments allowing backward compatible future changes.
    """
    if invocation_state is None:
        invocation_state = {}
    return run_async(lambda: self.invoke_async(task, invocation_state))

__init__(nodes, *, entry_point=None, max_handoffs=20, max_iterations=20, execution_timeout=900.0, node_timeout=300.0, repetitive_handoff_detection_window=0, repetitive_handoff_min_unique_agents=0, session_manager=None, hooks=None, id=_DEFAULT_SWARM_ID, trace_attributes=None)

Initialize Swarm with agents and configuration.

Parameters:

Name Type Description Default
id

Unique swarm id (default: None)

 required
nodes list[Agent]

List of nodes (e.g. Agent) to include in the swarm

 required
entry_point Agent | None

Agent to start with. If None, uses the first agent (default: None)

 None
max_handoffs int

Maximum handoffs to agents and users (default: 20)

 20
max_iterations int

Maximum node executions within the swarm (default: 20)

 20
execution_timeout float

Total execution timeout in seconds (default: 900.0)

 900.0
node_timeout float

Individual node timeout in seconds (default: 300.0)

 300.0
repetitive_handoff_detection_window int

Number of recent nodes to check for repetitive handoffs Disabled by default (default: 0)

 0
repetitive_handoff_min_unique_agents int

Minimum unique agents required in recent sequence Disabled by default (default: 0)

 0
session_manager Optional[SessionManager]

Session manager for persisting graph state and execution history (default: None)

 None
hooks Optional[list[HookProvider]]

List of hook providers for monitoring and extending graph execution behavior (default: None)

 None
trace_attributes Optional[Mapping[str, AttributeValue]]

Custom trace attributes to apply to the agent's trace span (default: None)

 None
Source code in strands/multiagent/swarm.py 
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
def __init__(
    self,
    nodes: list[Agent],
    *,
    entry_point: Agent | None = None,
    max_handoffs: int = 20,
    max_iterations: int = 20,
    execution_timeout: float = 900.0,
    node_timeout: float = 300.0,
    repetitive_handoff_detection_window: int = 0,
    repetitive_handoff_min_unique_agents: int = 0,
    session_manager: Optional[SessionManager] = None,
    hooks: Optional[list[HookProvider]] = None,
    id: str = _DEFAULT_SWARM_ID,
    trace_attributes: Optional[Mapping[str, AttributeValue]] = None,
) -> None:
    """Initialize Swarm with agents and configuration.

    Args:
        id : Unique swarm id (default: None)
        nodes: List of nodes (e.g. Agent) to include in the swarm
        entry_point: Agent to start with. If None, uses the first agent (default: None)
        max_handoffs: Maximum handoffs to agents and users (default: 20)
        max_iterations: Maximum node executions within the swarm (default: 20)
        execution_timeout: Total execution timeout in seconds (default: 900.0)
        node_timeout: Individual node timeout in seconds (default: 300.0)
        repetitive_handoff_detection_window: Number of recent nodes to check for repetitive handoffs
            Disabled by default (default: 0)
        repetitive_handoff_min_unique_agents: Minimum unique agents required in recent sequence
            Disabled by default (default: 0)
        session_manager: Session manager for persisting graph state and execution history (default: None)
        hooks: List of hook providers for monitoring and extending graph execution behavior (default: None)
        trace_attributes: Custom trace attributes to apply to the agent's trace span (default: None)
    """
    super().__init__()
    self.id = id
    self.entry_point = entry_point
    self.max_handoffs = max_handoffs
    self.max_iterations = max_iterations
    self.execution_timeout = execution_timeout
    self.node_timeout = node_timeout
    self.repetitive_handoff_detection_window = repetitive_handoff_detection_window
    self.repetitive_handoff_min_unique_agents = repetitive_handoff_min_unique_agents

    self.shared_context = SharedContext()
    self.nodes: dict[str, SwarmNode] = {}
    self.state = SwarmState(
        current_node=None,  # Placeholder, will be set properly
        task="",
        completion_status=Status.PENDING,
    )
    self.tracer = get_tracer()
    self.trace_attributes: dict[str, AttributeValue] = self._parse_trace_attributes(trace_attributes)

    self.session_manager = session_manager
    self.hooks = HookRegistry()
    if hooks:
        for hook in hooks:
            self.hooks.add_hook(hook)
    if self.session_manager:
        self.hooks.add_hook(self.session_manager)

    self._resume_from_session = False

    self._setup_swarm(nodes)
    self._inject_swarm_tools()
    run_async(lambda: self.hooks.invoke_callbacks_async(MultiAgentInitializedEvent(self)))

deserialize_state(payload)

Restore swarm state from a session dict and prepare for execution.

This method handles two scenarios: 1. If the persisted status is COMPLETED, FAILED resets all nodes and graph state to allow re-execution from the beginning. 2. Otherwise, restores the persisted state and prepares to resume execution from the next ready nodes.

Parameters:

Name Type Description Default
payload dict[str, Any]

Dictionary containing persisted state data including status, completed nodes, results, and next nodes to execute.

 required
Source code in strands/multiagent/swarm.py 
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
def deserialize_state(self, payload: dict[str, Any]) -> None:
    """Restore swarm state from a session dict and prepare for execution.

    This method handles two scenarios:
    1. If the persisted status is COMPLETED, FAILED resets all nodes and graph state
       to allow re-execution from the beginning.
    2. Otherwise, restores the persisted state and prepares to resume execution
       from the next ready nodes.

    Args:
        payload: Dictionary containing persisted state data including status,
                completed nodes, results, and next nodes to execute.
    """
    if not payload.get("next_nodes_to_execute"):
        for node in self.nodes.values():
            node.reset_executor_state()
        self.state = SwarmState(
            current_node=SwarmNode("", Agent()),
            task="",
            completion_status=Status.PENDING,
        )
        self._resume_from_session = False
        return
    else:
        self._from_dict(payload)
        self._resume_from_session = True

invoke_async(task, invocation_state=None, **kwargs) async

Invoke the swarm asynchronously.

This method uses stream_async internally and consumes all events until completion, following the same pattern as the Agent class.

Parameters:

Name Type Description Default
task MultiAgentInput

The task to execute

 required
invocation_state dict[str, Any] | None

Additional state/context passed to underlying agents. Defaults to None to avoid mutable default argument issues.

 None
**kwargs Any

Keyword arguments allowing backward compatible future changes.

 {}
Source code in strands/multiagent/swarm.py 
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
async def invoke_async(
    self, task: MultiAgentInput, invocation_state: dict[str, Any] | None = None, **kwargs: Any
) -> SwarmResult:
    """Invoke the swarm asynchronously.

    This method uses stream_async internally and consumes all events until completion,
    following the same pattern as the Agent class.

    Args:
        task: The task to execute
        invocation_state: Additional state/context passed to underlying agents.
            Defaults to None to avoid mutable default argument issues.
        **kwargs: Keyword arguments allowing backward compatible future changes.
    """
    events = self.stream_async(task, invocation_state, **kwargs)
    final_event = None
    async for event in events:
        final_event = event

    if final_event is None or "result" not in final_event:
        raise ValueError("Swarm streaming completed without producing a result event")

    return cast(SwarmResult, final_event["result"])

serialize_state()

Serialize the current swarm state to a dictionary.

Source code in strands/multiagent/swarm.py 
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
def serialize_state(self) -> dict[str, Any]:
    """Serialize the current swarm state to a dictionary."""
    status_str = self.state.completion_status.value
    if self.state.handoff_node:
        next_nodes = [self.state.handoff_node.node_id]
    elif self.state.completion_status == Status.EXECUTING and self.state.current_node:
        next_nodes = [self.state.current_node.node_id]
    else:
        next_nodes = []

    return {
        "type": "swarm",
        "id": self.id,
        "status": status_str,
        "node_history": [n.node_id for n in self.state.node_history],
        "node_results": {k: v.to_dict() for k, v in self.state.results.items()},
        "next_nodes_to_execute": next_nodes,
        "current_task": self.state.task,
        "context": {
            "shared_context": getattr(self.state.shared_context, "context", {}) or {},
            "handoff_message": self.state.handoff_message,
        },
    }

stream_async(task, invocation_state=None, **kwargs) async

Stream events during swarm execution.

Parameters:

Name Type Description Default
task MultiAgentInput

The task to execute

 required
invocation_state dict[str, Any] | None

Additional state/context passed to underlying agents. Defaults to None to avoid mutable default argument issues.

 None
**kwargs Any

Keyword arguments allowing backward compatible future changes.

 {}

Yields:

Type Description
AsyncIterator[dict[str, Any]]

Dictionary events during swarm execution, such as:
AsyncIterator[dict[str, Any]]
  • multi_agent_node_start: When a node begins execution
AsyncIterator[dict[str, Any]]
  • multi_agent_node_stream: Forwarded agent events with node context
AsyncIterator[dict[str, Any]]
  • multi_agent_handoff: When control is handed off between agents
AsyncIterator[dict[str, Any]]
  • multi_agent_node_stop: When a node stops execution
AsyncIterator[dict[str, Any]]
  • result: Final swarm result
Source code in strands/multiagent/swarm.py 
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
async def stream_async(
    self, task: MultiAgentInput, invocation_state: dict[str, Any] | None = None, **kwargs: Any
) -> AsyncIterator[dict[str, Any]]:
    """Stream events during swarm execution.

    Args:
        task: The task to execute
        invocation_state: Additional state/context passed to underlying agents.
            Defaults to None to avoid mutable default argument issues.
        **kwargs: Keyword arguments allowing backward compatible future changes.

    Yields:
        Dictionary events during swarm execution, such as:
        - multi_agent_node_start: When a node begins execution
        - multi_agent_node_stream: Forwarded agent events with node context
        - multi_agent_handoff: When control is handed off between agents
        - multi_agent_node_stop: When a node stops execution
        - result: Final swarm result
    """
    if invocation_state is None:
        invocation_state = {}

    await self.hooks.invoke_callbacks_async(BeforeMultiAgentInvocationEvent(self, invocation_state))

    logger.debug("starting swarm execution")

    if not self._resume_from_session:
        # Initialize swarm state with configuration
        initial_node = self._initial_node()

        self.state = SwarmState(
            current_node=initial_node,
            task=task,
            completion_status=Status.EXECUTING,
            shared_context=self.shared_context,
        )
    else:
        self.state.completion_status = Status.EXECUTING
        self.state.start_time = time.time()

    span = self.tracer.start_multiagent_span(task, "swarm", custom_trace_attributes=self.trace_attributes)
    with trace_api.use_span(span, end_on_exit=True):
        try:
            current_node = cast(SwarmNode, self.state.current_node)
            logger.debug("current_node=<%s> | starting swarm execution with node", current_node.node_id)
            logger.debug(
                "max_handoffs=<%d>, max_iterations=<%d>, timeout=<%s>s | swarm execution config",
                self.max_handoffs,
                self.max_iterations,
                self.execution_timeout,
            )

            async for event in self._execute_swarm(invocation_state):
                yield event.as_dict()

        except Exception:
            logger.exception("swarm execution failed")
            self.state.completion_status = Status.FAILED
            raise
        finally:
            self.state.execution_time = round((time.time() - self.state.start_time) * 1000)
            await self.hooks.invoke_callbacks_async(AfterMultiAgentInvocationEvent(self, invocation_state))
            self._resume_from_session = False

        # Yield final result after execution_time is set
        result = self._build_result()
        yield MultiAgentResultEvent(result=result).as_dict()

SwarmNode dataclass

Represents a node (e.g. Agent) in the swarm.

Source code in strands/multiagent/swarm.py 
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
@dataclass
class SwarmNode:
    """Represents a node (e.g. Agent) in the swarm."""

    node_id: str
    executor: Agent
    _initial_messages: Messages = field(default_factory=list, init=False)
    _initial_state: AgentState = field(default_factory=AgentState, init=False)

    def __post_init__(self) -> None:
        """Capture initial executor state after initialization."""
        # Deep copy the initial messages and state to preserve them
        self._initial_messages = copy.deepcopy(self.executor.messages)
        self._initial_state = AgentState(self.executor.state.get())

    def __hash__(self) -> int:
        """Return hash for SwarmNode based on node_id."""
        return hash(self.node_id)

    def __eq__(self, other: Any) -> bool:
        """Return equality for SwarmNode based on node_id."""
        if not isinstance(other, SwarmNode):
            return False
        return self.node_id == other.node_id

    def __str__(self) -> str:
        """Return string representation of SwarmNode."""
        return self.node_id

    def __repr__(self) -> str:
        """Return detailed representation of SwarmNode."""
        return f"SwarmNode(node_id='{self.node_id}')"

    def reset_executor_state(self) -> None:
        """Reset SwarmNode executor state to initial state when swarm was created."""
        self.executor.messages = copy.deepcopy(self._initial_messages)
        self.executor.state = AgentState(self._initial_state.get())

__eq__(other)

Return equality for SwarmNode based on node_id.

Source code in strands/multiagent/swarm.py 
77
78
79
80
81
def __eq__(self, other: Any) -> bool:
    """Return equality for SwarmNode based on node_id."""
    if not isinstance(other, SwarmNode):
        return False
    return self.node_id == other.node_id

__hash__()

Return hash for SwarmNode based on node_id.

Source code in strands/multiagent/swarm.py 
73
74
75
def __hash__(self) -> int:
    """Return hash for SwarmNode based on node_id."""
    return hash(self.node_id)

__post_init__()

Capture initial executor state after initialization.

Source code in strands/multiagent/swarm.py 
67
68
69
70
71
def __post_init__(self) -> None:
    """Capture initial executor state after initialization."""
    # Deep copy the initial messages and state to preserve them
    self._initial_messages = copy.deepcopy(self.executor.messages)
    self._initial_state = AgentState(self.executor.state.get())

__repr__()

Return detailed representation of SwarmNode.

Source code in strands/multiagent/swarm.py 
87
88
89
def __repr__(self) -> str:
    """Return detailed representation of SwarmNode."""
    return f"SwarmNode(node_id='{self.node_id}')"

__str__()

Return string representation of SwarmNode.

Source code in strands/multiagent/swarm.py 
83
84
85
def __str__(self) -> str:
    """Return string representation of SwarmNode."""
    return self.node_id

reset_executor_state()

Reset SwarmNode executor state to initial state when swarm was created.

Source code in strands/multiagent/swarm.py 
91
92
93
94
def reset_executor_state(self) -> None:
    """Reset SwarmNode executor state to initial state when swarm was created."""
    self.executor.messages = copy.deepcopy(self._initial_messages)
    self.executor.state = AgentState(self._initial_state.get())

SwarmResult dataclass

Bases: MultiAgentResult

Result from swarm execution - extends MultiAgentResult with swarm-specific details.

Source code in strands/multiagent/swarm.py 
207
208
209
210
211
@dataclass
class SwarmResult(MultiAgentResult):
    """Result from swarm execution - extends MultiAgentResult with swarm-specific details."""

    node_history: list[SwarmNode] = field(default_factory=list)

SwarmState dataclass

Current state of swarm execution.

Source code in strands/multiagent/swarm.py 
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
@dataclass
class SwarmState:
    """Current state of swarm execution."""

    current_node: SwarmNode | None  # The agent currently executing
    task: MultiAgentInput  # The original task from the user that is being executed
    completion_status: Status = Status.PENDING  # Current swarm execution status
    shared_context: SharedContext = field(default_factory=SharedContext)  # Context shared between agents
    node_history: list[SwarmNode] = field(default_factory=list)  # Complete history of agents that have executed
    start_time: float = field(default_factory=time.time)  # When swarm execution began
    results: dict[str, NodeResult] = field(default_factory=dict)  # Results from each agent execution
    # Total token usage across all agents
    accumulated_usage: Usage = field(default_factory=lambda: Usage(inputTokens=0, outputTokens=0, totalTokens=0))
    # Total metrics across all agents
    accumulated_metrics: Metrics = field(default_factory=lambda: Metrics(latencyMs=0))
    execution_time: int = 0  # Total execution time in milliseconds
    handoff_node: SwarmNode | None = None  # The agent to execute next
    handoff_message: str | None = None  # Message passed during agent handoff

    def should_continue(
        self,
        *,
        max_handoffs: int,
        max_iterations: int,
        execution_timeout: float,
        repetitive_handoff_detection_window: int,
        repetitive_handoff_min_unique_agents: int,
    ) -> Tuple[bool, str]:
        """Check if the swarm should continue.

        Returns: (should_continue, reason)
        """
        # Check handoff limit
        if len(self.node_history) >= max_handoffs:
            return False, f"Max handoffs reached: {max_handoffs}"

        # Check iteration limit
        if len(self.node_history) >= max_iterations:
            return False, f"Max iterations reached: {max_iterations}"

        # Check timeout
        elapsed = time.time() - self.start_time
        if elapsed > execution_timeout:
            return False, f"Execution timed out: {execution_timeout}s"

        # Check for repetitive handoffs (agents passing back and forth)
        if repetitive_handoff_detection_window > 0 and len(self.node_history) >= repetitive_handoff_detection_window:
            recent = self.node_history[-repetitive_handoff_detection_window:]
            unique_nodes = len(set(recent))
            if unique_nodes < repetitive_handoff_min_unique_agents:
                return (
                    False,
                    (
                        f"Repetitive handoff: {unique_nodes} unique nodes "
                        f"out of {repetitive_handoff_detection_window} recent iterations"
                    ),
                )

        return True, "Continuing"

should_continue(*, max_handoffs, max_iterations, execution_timeout, repetitive_handoff_detection_window, repetitive_handoff_min_unique_agents)

Check if the swarm should continue.

Returns: (should_continue, reason)

Source code in strands/multiagent/swarm.py 
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
def should_continue(
    self,
    *,
    max_handoffs: int,
    max_iterations: int,
    execution_timeout: float,
    repetitive_handoff_detection_window: int,
    repetitive_handoff_min_unique_agents: int,
) -> Tuple[bool, str]:
    """Check if the swarm should continue.

    Returns: (should_continue, reason)
    """
    # Check handoff limit
    if len(self.node_history) >= max_handoffs:
        return False, f"Max handoffs reached: {max_handoffs}"

    # Check iteration limit
    if len(self.node_history) >= max_iterations:
        return False, f"Max iterations reached: {max_iterations}"

    # Check timeout
    elapsed = time.time() - self.start_time
    if elapsed > execution_timeout:
        return False, f"Execution timed out: {execution_timeout}s"

    # Check for repetitive handoffs (agents passing back and forth)
    if repetitive_handoff_detection_window > 0 and len(self.node_history) >= repetitive_handoff_detection_window:
        recent = self.node_history[-repetitive_handoff_detection_window:]
        unique_nodes = len(set(recent))
        if unique_nodes < repetitive_handoff_min_unique_agents:
            return (
                False,
                (
                    f"Repetitive handoff: {unique_nodes} unique nodes "
                    f"out of {repetitive_handoff_detection_window} recent iterations"
                ),
            )

    return True, "Continuing"