Initial A2A server Integration

[jer96]

Description

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

Related Issues

• [FEATURE] Agent to Agent Communication (A2A) #24
• [Bug]: OTEL exporter dependency conflict a2aproject/a2a-python#198

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", , name="Hello World Agent", description="Just a hello world agent", callback_handler=None)
strands_a2a_agent = A2AAgent(agent=strands_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]

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]

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