Initial A2A Integration #218

jer96 · 2025-06-13T21:10:32Z

Description

This PR seeks to integrate Google's A2A protocol into Strands agents

Related Issues

Documentation PR

N/A

Type of Change

New feature

Limitations and Next Steps

This is the most basic integration with A2A. A Strands agent is wrapped in an A2AAgentClass which currently only supports messages.
Currently there is a dependency conflict with opentelemetry-exporter-otlp-proto-http and a2a-sdk dependencies. a2a-sdk relies on protobuf>=6.3.1, but opentelemetry-exporter-otlp-proto-http requires <6.0. This warranted the dependency changes to pyproject.toml and implementation changes to tracer.py. I plan on documenting the conflict and submitting issues to both the opentelemetry-exporter-otlp-proto-http and the a2a-sdk repo.
Unit tests for the A2A integration are currently not run because of the dependency conflict mentioned above. I will investigate how to run unit tests for two separate environments in the interim.
Native Strands tools and MCP tool integration
Resolve dependency and build papercuts
Documentation

Testing

unit tests
manual testing

sample server: a2a_server.py

import logging
import sys

from strands import Agent
from strands.multiagent.a2a import A2AAgent

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
    handlers=[logging.StreamHandler(sys.stdout)],
    force=True,
)

# Log that we're starting
logging.info("Starting A2A server with root logger")

strands_agent = Agent(model="us.anthropic.claude-3-haiku-20240307-v1:0", callback_handler=None)
strands_a2a_agent = A2AAgent(agent=strands_agent, name="Hello World Agent", description="Just a hello world agent")
strands_a2a_agent.serve()

sample client a2a_client.py

import logging
from typing import Any
from uuid import uuid4

import httpx
from a2a.client import A2ACardResolver, A2AClient
from a2a.types import (
    AgentCard,
    MessageSendParams,
    SendMessageRequest,
)

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
PUBLIC_AGENT_CARD_PATH = "/.well-known/agent.json"
BASE_URL = "http://localhost:9000"


async def main() -> None:
    async with httpx.AsyncClient() as httpx_client:
        # Initialize A2ACardResolver
        resolver = A2ACardResolver(
            httpx_client=httpx_client,
            base_url=BASE_URL,
        )

        # Fetch Public Agent Card and Initialize Client
        agent_card: AgentCard | None = None

        try:
            logger.info("Attempting to fetch public agent card from: {} {}", BASE_URL, PUBLIC_AGENT_CARD_PATH)
            agent_card = await resolver.get_agent_card()  # Fetches from default public path
            logger.info("Successfully fetched public agent card:")
            logger.info(agent_card.model_dump_json(indent=2, exclude_none=True))
        except Exception as e:
            logger.exception("Critical error fetching public agent card")
            raise RuntimeError("Failed to fetch the public agent card. Cannot continue.") from e

        client = A2AClient(httpx_client=httpx_client, agent_card=agent_card)
        logger.info("A2AClient initialized.")

        send_message_payload: dict[str, Any] = {
            "message": {
                "role": "user",
                "parts": [{"kind": "text", "text": "how much is 10 USD in INR?"}],
                "messageId": uuid4().hex,
            },
        }
        request = SendMessageRequest(id=str(uuid4()), params=MessageSendParams(**send_message_payload))

        response = await client.send_message(request)
        print(response.model_dump(mode="json", exclude_none=True))


if __name__ == "__main__":
    import asyncio

    asyncio.run(main())

Run python a2a_server.py and python a2a_client.py in separate terminal panes and observe output.

Standard Testing

hatch fmt --linter
hatch fmt --formatter
hatch test --all
Verify that the changes do not break functionality or introduce warnings in consuming repositories: agents-docs, agents-tools, agents-cli

Checklist

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

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

mikegc-aws · 2025-06-14T20:58:50Z

Thanks for this. Let's get the changes to pyproject.toml that give the feature options for otel in it's own PR so we can get this deployed ASAP and unblock A2A. Then follow up with the implementation stuff? Thoughts?

jer96 · 2025-06-16T13:06:12Z

@mikegc-aws yes great callout, i'll extract that into a separate PR and ship that first.

zastrowm · 2025-06-18T20:57:18Z

pyproject.toml

@@ -108,7 +108,7 @@ format-fix = [
 ]
 lint-check = [
    "ruff check",
-    "mypy -p src"
+    "mypy -p src --exclude src/strands/multiagent"


🥺 What's this needed for?

Oh, is this because of the dependency conflict stuff? If so, can you add comments throughout the file indicating that?

Also, do we have a tracking issue for the dependency issue. I think the long-term fix is to remove the http exporter dependency from strands?

zastrowm · 2025-06-18T20:59:56Z

src/strands/agent/agent.py

@@ -216,6 +216,9 @@ def __init__(
        record_direct_tool_call: bool = True,
        load_tools_from_directory: bool = True,
        trace_attributes: Optional[Mapping[str, AttributeValue]] = None,
+        *,
+        name: str | None = None,


Can we switch to Optional[str] for consistency elsewhere? (I like this syntax more TBH, but we use Optional everywhere else - issue to switch the library over maybe?)

zastrowm · 2025-06-18T21:01:11Z