Initial A2A server Integration (strands-agents#218)

Author: jer96

pyproject.toml

@@ -108,7 +108,8 @@ format-fix = [
 ]
 lint-check = [
     "ruff check",
-    "mypy -p src"
+    # excluding due to A2A and OTEL http exporter dependency conflict
+    "mypy -p src --exclude src/strands/multiagent"
 ]
 lint-fix = [
     "ruff check --fix"
@@ -137,17 +138,29 @@ features = ["dev", "docs", "anthropic", "litellm", "llamaapi", "ollama", "otel"]
 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"
+]
 
 [[tool.hatch.envs.hatch-test.matrix]]
 python = ["3.13", "3.12", "3.11", "3.10"]
 
-
 [tool.hatch.envs.hatch-test.scripts]
 run = [
-    "pytest{env:HATCH_TEST_ARGS:} {args}"
+    # excluding due to A2A and OTEL http exporter dependency conflict
+    "pytest{env:HATCH_TEST_ARGS:} {args} --ignore=tests/multiagent/a2a"
 ]
 run-cov = [
-    "pytest{env:HATCH_TEST_ARGS:} --cov --cov-config=pyproject.toml {args}"
+    # 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"
 ]
 
 cov-combine = []
@@ -181,6 +194,10 @@ prepare = [
     "hatch fmt --formatter",
     "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

@@ -220,6 +220,9 @@ def __init__(
         record_direct_tool_call: bool = True,
         load_tools_from_directory: bool = True,
         trace_attributes: Optional[Mapping[str, AttributeValue]] = None,
+        *,
+        name: Optional[str] = None,
+        description: Optional[str] = None,
     ):
         """Initialize the Agent with the specified configuration.
 
@@ -252,6 +255,10 @@ 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.
+            name: name of the Agent
+                Defaults to None.
+            description: description of what the Agent does
+                Defaults to None.
 
         Raises:
             ValueError: If max_parallel_tools is less than 1.
@@ -313,6 +320,8 @@ def __init__(
         self.tracer = get_tracer()
         self.trace_span: Optional[trace.Span] = None
         self.tool_caller = Agent.ToolCaller(self)
+        self.name = name
+        self.description = description
 
     @property
     def tool(self) -> ToolCaller:

src/strands/multiagent/__init__.py

@@ -0,0 +1,13 @@
+"""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.
+"""
+
+from . import a2a
+
+__all__ = ["a2a"]

src/strands/multiagent/a2a/__init__.py

@@ -0,0 +1,14 @@
+"""Agent-to-Agent (A2A) communication protocol implementation for Strands Agents.
+
+This module provides classes and utilities for enabling Strands Agents to communicate
+with other agents using the Agent-to-Agent (A2A) protocol.
+
+Docs: https://google-a2a.github.io/A2A/latest/
+
+Classes:
+    A2AAgent: A wrapper that adapts a Strands Agent to be A2A-compatible.
+"""
+
+from .agent import A2AAgent
+
+__all__ = ["A2AAgent"]

src/strands/multiagent/a2a/agent.py

@@ -0,0 +1,149 @@
+"""A2A-compatible wrapper for Strands Agent.
+
+This module provides the A2AAgent class, which adapts a Strands Agent to the A2A protocol,
+allowing it to be used in A2A-compatible systems.
+"""
+
+import logging
+from typing import Any, Literal
+
+import uvicorn
+from a2a.server.apps import A2AFastAPIApplication, A2AStarletteApplication
+from a2a.server.request_handlers import DefaultRequestHandler
+from a2a.server.tasks import InMemoryTaskStore
+from a2a.types import AgentCapabilities, AgentCard, AgentSkill
+from fastapi import FastAPI
+from starlette.applications import Starlette
+
+from ...agent.agent import Agent as SAAgent
+from .executor import StrandsA2AExecutor
+
+logger = logging.getLogger(__name__)
+
+
+class A2AAgent:
+    """A2A-compatible wrapper for Strands Agent."""
+
+    def __init__(
+        self,
+        agent: SAAgent,
+        *,
+        # AgentCard
+        host: str = "0.0.0.0",
+        port: int = 9000,
+        version: str = "0.0.1",
+    ):
+        """Initialize an A2A-compatible agent from a Strands agent.
+
+        Args:
+            agent: The Strands Agent to wrap with A2A compatibility.
+            name: The name of the agent, used in the AgentCard.
+            description: A description of the agent's capabilities, used in the AgentCard.
+            host: The hostname or IP address to bind the A2A server to. Defaults to "0.0.0.0".
+            port: The port to bind the A2A server to. Defaults to 9000.
+            version: The version of the agent. Defaults to "0.0.1".
+        """
+        self.host = host
+        self.port = port
+        self.http_url = f"http://{self.host}:{self.port}/"
+        self.version = version
+        self.strands_agent = agent
+        self.name = self.strands_agent.name
+        self.description = self.strands_agent.description
+        # TODO: enable configurable capabilities and request handler
+        self.capabilities = AgentCapabilities()
+        self.request_handler = DefaultRequestHandler(
+            agent_executor=StrandsA2AExecutor(self.strands_agent),
+            task_store=InMemoryTaskStore(),
+        )
+        logger.info("Strands' integration with A2A is experimental. Be aware of frequent breaking changes.")
+
+    @property
+    def public_agent_card(self) -> AgentCard:
+        """Get the public AgentCard for this agent.
+
+        The AgentCard contains metadata about the agent, including its name,
+        description, URL, version, skills, and capabilities. This information
+        is used by other agents and systems to discover and interact with this agent.
+
+        Returns:
+            AgentCard: The public agent card containing metadata about this agent.
+
+        Raises:
+            ValueError: If name or description is None or empty.
+        """
+        if not self.name:
+            raise ValueError("A2A agent name cannot be None or empty")
+        if not self.description:
+            raise ValueError("A2A agent description cannot be None or empty")
+
+        return AgentCard(
+            name=self.name,
+            description=self.description,
+            url=self.http_url,
+            version=self.version,
+            skills=self.agent_skills,
+            defaultInputModes=["text"],
+            defaultOutputModes=["text"],
+            capabilities=self.capabilities,
+        )
+
+    @property
+    def agent_skills(self) -> list[AgentSkill]:
+        """Get the list of skills this agent provides.
+
+        Skills represent specific capabilities that the agent can perform.
+        Strands agent tools are adapted to A2A skills.
+
+        Returns:
+            list[AgentSkill]: A list of skills this agent provides.
+        """
+        # TODO: translate Strands tools (native & MCP) to skills
+        return []
+
+    def to_starlette_app(self) -> Starlette:
+        """Create a Starlette application for serving this agent via HTTP.
+
+        This method creates a Starlette application that can be used to serve
+        the agent via HTTP using the A2A protocol.
+
+        Returns:
+            Starlette: A Starlette application configured to serve this agent.
+        """
+        return A2AStarletteApplication(agent_card=self.public_agent_card, http_handler=self.request_handler).build()
+
+    def to_fastapi_app(self) -> FastAPI:
+        """Create a FastAPI application for serving this agent via HTTP.
+
+        This method creates a FastAPI application that can be used to serve
+        the agent via HTTP using the A2A protocol.
+
+        Returns:
+            FastAPI: A FastAPI application configured to serve this agent.
+        """
+        return A2AFastAPIApplication(agent_card=self.public_agent_card, http_handler=self.request_handler).build()
+
+    def serve(self, app_type: Literal["fastapi", "starlette"] = "starlette", **kwargs: Any) -> None:
+        """Start the A2A server with the specified application type.
+
+        This method starts an HTTP server that exposes the agent via the A2A protocol.
+        The server can be implemented using either FastAPI or Starlette, depending on
+        the specified app_type.
+
+        Args:
+            app_type: The type of application to serve, either "fastapi" or "starlette".
+                Defaults to "starlette".
+            **kwargs: Additional keyword arguments to pass to uvicorn.run.
+        """
+        try:
+            logger.info("Starting Strands A2A server...")
+            if app_type == "fastapi":
+                uvicorn.run(self.to_fastapi_app(), host=self.host, port=self.port, **kwargs)
+            else:
+                uvicorn.run(self.to_starlette_app(), host=self.host, port=self.port, **kwargs)
+        except KeyboardInterrupt:
+            logger.warning("Strands A2A server shutdown requested (KeyboardInterrupt).")
+        except Exception:
+            logger.exception("Strands A2A server encountered exception.")
+        finally:
+            logger.info("Strands A2A server has shutdown.")

src/strands/multiagent/a2a/executor.py

@@ -0,0 +1,67 @@
+"""Strands Agent executor for the A2A protocol.
+
+This module provides the StrandsA2AExecutor class, which adapts a Strands Agent
+to be used as an executor in the A2A protocol. It handles the execution of agent
+requests and the conversion of Strands Agent responses to A2A events.
+"""
+
+import logging
+
+from a2a.server.agent_execution import AgentExecutor, RequestContext
+from a2a.server.events import EventQueue
+from a2a.types import UnsupportedOperationError
+from a2a.utils import new_agent_text_message
+from a2a.utils.errors import ServerError
+
+from ...agent.agent import Agent as SAAgent
+from ...agent.agent_result import AgentResult as SAAgentResult
+
+log = logging.getLogger(__name__)
+
+
+class StrandsA2AExecutor(AgentExecutor):
+    """Executor that adapts a Strands Agent to the A2A protocol."""
+
+    def __init__(self, agent: SAAgent):
+        """Initialize a StrandsA2AExecutor.
+
+        Args:
+            agent: The Strands Agent to adapt to the A2A protocol.
+        """
+        self.agent = agent
+
+    async def execute(
+        self,
+        context: RequestContext,
+        event_queue: EventQueue,
+    ) -> None:
+        """Execute a request using the Strands Agent and send the response as A2A events.
+
+        This method executes the user's input using the Strands Agent and converts
+        the agent's response to A2A events, which are then sent to the event queue.
+
+        Args:
+            context: The A2A request context, containing the user's input and other metadata.
+            event_queue: The A2A event queue, used to send response events.
+        """
+        result: SAAgentResult = self.agent(context.get_user_input())
+        if result.message and "content" in result.message:
+            for content_block in result.message["content"]:
+                if "text" in content_block:
+                    await event_queue.enqueue_event(new_agent_text_message(content_block["text"]))
+
+    async def cancel(self, context: RequestContext, event_queue: EventQueue) -> None:
+        """Cancel an ongoing execution.
+
+        This method is called when a request is cancelled. Currently, cancellation
+        is not supported, so this method raises an UnsupportedOperationError.
+
+        Args:
+            context: The A2A request context.
+            event_queue: The A2A event queue.
+
+        Raises:
+            ServerError: Always raised with an UnsupportedOperationError, as cancellation
+                is not currently supported.
+        """
+        raise ServerError(error=UnsupportedOperationError())

tests/multiagent/__init__.py

@@ -0,0 +1 @@
+"""Tests for the multiagent module."""

tests/multiagent/a2a/__init__.py

@@ -0,0 +1 @@
+"""Tests for the A2A module."""

tests/multiagent/a2a/conftest.py

@@ -0,0 +1,41 @@
+"""Common fixtures for A2A module tests."""
+
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+from a2a.server.agent_execution import RequestContext
+from a2a.server.events import EventQueue
+
+from strands.agent.agent import Agent as SAAgent
+from strands.agent.agent_result import AgentResult as SAAgentResult
+
+
+@pytest.fixture
+def mock_strands_agent():
+    """Create a mock Strands Agent for testing."""
+    agent = MagicMock(spec=SAAgent)
+    agent.name = "Test Agent"
+    agent.description = "A test agent for unit testing"
+
+    # Setup default response
+    mock_result = MagicMock(spec=SAAgentResult)
+    mock_result.message = {"content": [{"text": "Test response"}]}
+    agent.return_value = mock_result
+
+    return agent
+
+
+@pytest.fixture
+def mock_request_context():
+    """Create a mock RequestContext for testing."""
+    context = MagicMock(spec=RequestContext)
+    context.get_user_input.return_value = "Test input"
+    return context
+
+
+@pytest.fixture
+def mock_event_queue():
+    """Create a mock EventQueue for testing."""
+    queue = MagicMock(spec=EventQueue)
+    queue.enqueue_event = AsyncMock()
+    return queue