feat(multi-agent): introduce Graph orchestrator

Author: awsarron

.gitignore

@@ -8,4 +8,5 @@ __pycache__*
 .ruff_cache
 *.bak
 .vscode
-dist
+dist
+repl_state

pyproject.toml

@@ -89,13 +89,57 @@ a2a = [
     "fastapi>=0.115.12",
     "starlette>=0.46.2",
 ]
+all = [
+    # anthropic
+    "anthropic>=0.21.0,<1.0.0",
+
+    # dev
+    "commitizen>=4.4.0,<5.0.0",
+    "hatch>=1.0.0,<2.0.0",
+    "moto>=5.1.0,<6.0.0",
+    "mypy>=1.15.0,<2.0.0",
+    "pre-commit>=3.2.0,<4.2.0",
+    "pytest>=8.0.0,<9.0.0",
+    "pytest-asyncio>=0.26.0,<0.27.0",
+    "ruff>=0.4.4,<0.5.0",
+
+    # docs
+    "sphinx>=5.0.0,<6.0.0",
+    "sphinx-rtd-theme>=1.0.0,<2.0.0",
+    "sphinx-autodoc-typehints>=1.12.0,<2.0.0",
+
+    # litellm
+    "litellm>=1.72.6,<1.73.0",
+
+    # llama
+    "llama-api-client>=0.1.0,<1.0.0",
+
+    # mistral
+    "mistralai>=1.8.2",
+
+    # ollama
+    "ollama>=0.4.8,<1.0.0",
+
+    # openai
+    "openai>=1.68.0,<2.0.0",
+
+    # otel
+    "opentelemetry-exporter-otlp-proto-http>=1.30.0,<2.0.0",
+
+    # a2a
+    "a2a-sdk>=0.2.6",
+    "uvicorn>=0.34.2",
+    "httpx>=0.28.1",
+    "fastapi>=0.115.12",
+    "starlette>=0.46.2",
+]
 
 [tool.hatch.version]
 # Tells Hatch to use your version control system (git) to determine the version.
 source = "vcs"
 
 [tool.hatch.envs.hatch-static-analysis]
-features = ["anthropic", "litellm", "llamaapi", "ollama", "openai", "otel","mistral"]
+features = ["anthropic", "litellm", "llamaapi", "ollama", "openai", "otel", "mistral", "a2a"]
 dependencies = [
   "mypy>=1.15.0,<2.0.0",
   "ruff>=0.11.6,<0.12.0",
@@ -111,15 +155,14 @@ format-fix = [
 ]
 lint-check = [
     "ruff check",
-    # excluding due to A2A and OTEL http exporter dependency conflict
-    "mypy -p src --exclude src/strands/multiagent"
+    "mypy -p src"
 ]
 lint-fix = [
     "ruff check --fix"
 ]
 
 [tool.hatch.envs.hatch-test]
-features = ["anthropic", "litellm", "llamaapi", "ollama", "openai", "otel","mistral"]
+features = ["anthropic", "litellm", "llamaapi", "ollama", "openai", "otel", "mistral", "a2a"]
 extra-dependencies = [
     "moto>=5.1.0,<6.0.0",
     "pytest>=8.0.0,<9.0.0",
@@ -135,35 +178,17 @@ extra-args = [
 
 [tool.hatch.envs.dev]
 dev-mode = true
-features = ["dev", "docs", "anthropic", "litellm", "llamaapi", "ollama", "otel","mistral"]
-
-[tool.hatch.envs.a2a]
-dev-mode = true
-features = ["dev", "docs", "anthropic", "litellm", "llamaapi", "ollama", "a2a"]
-
-[tool.hatch.envs.a2a.scripts]
-run = [
-    "pytest{env:HATCH_TEST_ARGS:} tests/multiagent/a2a {args}"
-]
-run-cov = [
-    "pytest{env:HATCH_TEST_ARGS:} tests/multiagent/a2a --cov --cov-config=pyproject.toml {args}"
-]
-lint-check = [
-    "ruff check",
-    "mypy -p src/strands/multiagent/a2a"
-]
+features = ["dev", "docs", "anthropic", "litellm", "llamaapi", "ollama", "otel", "mistral"]
 
 [[tool.hatch.envs.hatch-test.matrix]]
 python = ["3.13", "3.12", "3.11", "3.10"]
 
 [tool.hatch.envs.hatch-test.scripts]
 run = [
-    # excluding due to A2A and OTEL http exporter dependency conflict
-    "pytest{env:HATCH_TEST_ARGS:} {args} --ignore=tests/multiagent/a2a"
+    "pytest{env:HATCH_TEST_ARGS:} {args}"
 ]
 run-cov = [
-    # excluding due to A2A and OTEL http exporter dependency conflict
-    "pytest{env:HATCH_TEST_ARGS:} --cov --cov-config=pyproject.toml {args} --ignore=tests/multiagent/a2a"
+    "pytest{env:HATCH_TEST_ARGS:} --cov --cov-config=pyproject.toml {args}"
 ]
 
 cov-combine = []
@@ -198,10 +223,6 @@ prepare = [
     "hatch run test-lint",
     "hatch test --all"
 ]
-test-a2a = [
-    # required to run manually due to A2A and OTEL http exporter dependency conflict
-    "hatch -e a2a run run {args}"
-]
 
 [tool.mypy]
 python_version = "3.10"

src/strands/agent/agent.py

@@ -15,6 +15,7 @@
 import random
 from concurrent.futures import ThreadPoolExecutor
 from typing import Any, AsyncIterator, Callable, Generator, List, Mapping, Optional, Type, TypeVar, Union, cast
+from uuid import uuid4
 
 from opentelemetry import trace
 from pydantic import BaseModel
@@ -191,6 +192,7 @@ def __init__(
         load_tools_from_directory: bool = True,
         trace_attributes: Optional[Mapping[str, AttributeValue]] = None,
         *,
+        agent_id: Optional[str] = None,
         name: Optional[str] = None,
         description: Optional[str] = None,
         state: Optional[Union[AgentState, dict]] = None,
@@ -226,6 +228,8 @@ def __init__(
             load_tools_from_directory: Whether to load and automatically reload tools in the `./tools/` directory.
                 Defaults to True.
             trace_attributes: Custom trace attributes to apply to the agent's trace span.
+            agent_id: Optional ID for the agent, useful for multi-agent scenarios.
+                If None, a UUID is generated.
             name: name of the Agent
                 Defaults to None.
             description: description of what the Agent does
@@ -240,6 +244,9 @@ def __init__(
         self.messages = messages if messages is not None else []
 
         self.system_prompt = system_prompt
+        self.agent_id = agent_id or str(uuid4())
+        self.name = name
+        self.description = description
 
         # If not provided, create a new PrintingCallbackHandler instance
         # If explicitly set to None, use null_callback_handler
@@ -305,8 +312,6 @@ def __init__(
             self.state = AgentState()
 
         self.tool_caller = Agent.ToolCaller(self)
-        self.name = name
-        self.description = description
 
     @property
     def tool(self) -> ToolCaller:

src/strands/agent/conversation_manager/sliding_window_conversation_manager.py

@@ -75,7 +75,7 @@ def apply_management(self, agent: "Agent") -> None:
 
         if len(messages) <= self.window_size:
             logger.debug(
-                "window_size=<%s>, message_count=<%s> | skipping context reduction", len(messages), self.window_size
+                "message_count=<%s>, window_size=<%s> | skipping context reduction", len(messages), self.window_size
             )
             return
         self.reduce_context(agent)

src/strands/multiagent/__init__.py

@@ -9,5 +9,13 @@
 """
 
 from . import a2a
+from .base import MultiAgentBase, MultiAgentResult
+from .graph import GraphBuilder, GraphResult
 
-__all__ = ["a2a"]
+__all__ = [
+    "a2a",
+    "GraphBuilder",
+    "GraphResult",
+    "MultiAgentBase",
+    "MultiAgentResult",
+]

src/strands/multiagent/base.py

@@ -0,0 +1,73 @@
+"""Multi-Agent Base Class.
+
+Provides minimal foundation for multi-agent patterns (Swarm, Graph).
+"""
+
+import logging
+from abc import ABC, abstractmethod
+from dataclasses import dataclass, field
+from typing import Any, Dict, List, Union
+
+from ..agent import AgentResult
+from ..types.event_loop import Metrics, Usage
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class NodeResult:
+    """Unified result from node execution - handles both Agent and nested MultiAgentBase results."""
+
+    # Core result data - single AgentResult or nested MultiAgentResult
+    results: Union[AgentResult, "MultiAgentResult"]
+
+    # Execution metadata
+    execution_time: float = 0.0
+    status: Any = None
+
+    # 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.results, AgentResult):
+            return [self.results]
+        else:
+            # Flatten nested results from MultiAgentResult
+            flattened = []
+            for nested_node_result in self.results.results.values():
+                flattened.extend(nested_node_result.get_agent_results())
+            return flattened
+
+
+@dataclass
+class MultiAgentResult:
+    """Result from multi-agent execution with accumulated metrics."""
+
+    results: Dict[str, NodeResult]
+    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: float = 0.0
+
+
+class MultiAgentBase(ABC):
+    """Base class for multi-agent helpers.
+
+    This class integrates with existing Strands Agent instances and provides
+    multi-agent orchestration capabilities.
+    """
+
+    @abstractmethod
+    # TODO: for task - multi-modal input (Message), list of messages
+    async def execute(self, task: str) -> MultiAgentResult:
+        """Execute task."""
+        raise NotImplementedError("execute not implemented")
+
+    @abstractmethod
+    # TODO: for task - multi-modal input (Message), list of messages
+    async def resume(self, task: str, state: Any) -> MultiAgentResult:
+        """Resume task from previous state."""
+        raise NotImplementedError("resume not implemented")