How To Use A2A to Access Multiple Data Stores

This blog is the third of a three part series on Agents and particularly the Agent2Agent (A2A) protocol. It will build on the two previous posts, Introduction to A2A and agent search2A and How to use MCP agents to access multiple data stores and demonstrate how A2A can enrich agent search using a “SuperAgent” and two A2A agents in front of MCP servers..

It will use the scenario from the previous blog post to show how A2A enabled agents can be used to simplify search across multiple, disparate data sources including the Bronto logging platform and how a “SuperAgent” can act autonomously to route and orchestrate requests to other agents, in a simple agentic manner.

The Scenario

The example scenario is what happens when an employee leaves a company and the IT department wants to determine that their details and access has been removed from all systems, such as the HR database and a tool such as a logging service. Traditionally, that would be done by an IT person checking each system having been notified of the employee's departure.

‍

The use of specialized data storage systems, such as Bronto’s high-performance logging platform for searching logs, as part of an ecosystem of collaborating agents, aligns with the need for high throughput and the need to handle heterogeneous systems, as identified in a recent paper from Berkeley on agentic search.
‍

The diagram above outlines the agents and protocols used in the demonstration. We will demonstrate:

Using the MCP servers providing access to the Bronto REST API and the MCP server providing access to the SQLite employee database, as described in the previous blog.
Creating A2A agents for each MCP service by extending the simple A2A agent described in the first post in the series.
Creating a simple A2A “SuperAgent” that will access the A2A agents accessing the data stores
The A2A “SuperAgent” answering requests about the stored user information and using A2A to call the two A2A agents to access data.

Note these examples are given for demonstration purposes only and other factors should also be considered for any other use, e.g. agent discovery could use a curated registry of agents for discovery instead of a simple well known URI as shown in these examples.
‍

Create an A2A Bronto bronto-get-users Agent

This code is shown in Appendix 1 and it uses an MCP server to call the Bronto get_users REST API, which was described in the previous blog. The agent has two main files:

main.py

This specifies the skills and extended skills that will be used in the public agent card and extended agent card (see source code in Appendix 1 for details)
It also sets up a request_handler using the ^{BrontoAgentExecutor()} defined in agent_executor.py
It passes the request handler and agent card information to an A2AStarletteApplication server
Calls ^{uvicorn.run()} to run the server on the specified host and port
‍

Agent_executor.py

This consists of:

an ^__init__() call, where we Initialize the FastMCP Client (at the known URL)
an ^invoke() call, where we use the FastMCP client to call a tool named "get_bronto_users" on the MCP server. In the code, we also ping the MCP server and obtain its tools for demonstration purposes.
class BrontoAgentExecutor(AgentExecutor) which
- runs asynchronously accepting A2A messages using code for request and queue handling similar to that in the A2A samples in the official A2A samples github repository.
- calls ^invoke() or each A2A request, it and that calls the MCP server tool to make a call to the Bronto /users REST API and it returns the result it receives in the text part of the A2A response.

This is a simple agent implementation, for example a more complete implementation would check or summarise the message received.

Create an A2A Employee DB bronto-get-employees Agent

This agent reuses the same code as for the A2A agent above to call the MCP SQL server with the following changes:

__main__.py
- Skills, extended skills, agent card and extended card changed to reflect this agent gets employee information from a database.
- Server updated for Host and port it listens on
Agent_executor.py
- url changed to that for the MCP server to get_employees
- Mcp tool called updated to “get_employees”
  ‍

Create an A2A “SuperAgent”

The source code for this agent is shown in Appendix 2. It acts as an A2A "SuperAgent" (Orchestrator) that routes A2A requests it receives to MCP servers based on a simple check on the query parameter it receives from an A2A client. It returns a response comprised of the response or responses it received. It consists of
‍

main.py

Skills, extended skills, agent card and extended card changed to show that this agent gets users from Bronto and employees from a database.
Server updated for Host and port it listens on
‍

Agent_executor.py

This gathers and checks AgentCards and routes messages accordingly.

Holds URLs for the A2A subagents in front of the MCP servers
SuperAgent class
- fetches the AgentCard from the .well-known/agent.json endpoint for those subagents in order to discover the agent’s capabilities and holds the cards in a registry. This is done in method ^{discover_agents()} by using a httpx client and A2ACardResolver to retrieve the AgentCard of all the agents it has a URL for.
- Provides bronto_agents() to retrieve any agent with “bronto” in its name in the AgentCard held in the registry
- uses ^{a2a_get_text_from_response()} to retrieve the text from the A2A response, as the text holds the reply from the subagent
SuperAgentExecutor class
- Instantiates the SuperAgent class and calls ^{discover_agents()} in SuperAgent, when it is invoked
- On receiving a request from an A2A client, it decides which agent to route the request to by passing a name to ^{call_named_agent()} or to call them all using ^{call_all_bronto_agents()}, based on the query parameter it received
- call_named_agent()tries to find the name of the agent in the AgentCard registry and if it finds a matching AgentCard, it will use this AgentCard in creating an A2AClient which it will use to send an A2A message to that A2A subagent and uses a2a_get_text_from_response() to retrieve the text from the response and put that text in its own response
- ^{call_all_bronto_agents()} calls ^{bronto_agents()} to retrieve all the AgentCards with “bronto” in their name and builds and sends a combined response from all the responses it receives

This is a simple “SuperAgent” agent implementation, based on the A2A project patterns in the samples in their samples github repo. A more advanced agent could;

Use more advanced routing to call subagents instead of the simple checks on the query parameter and the name in the AgentCard as used in this example, e.g. call an LLM to consider the full AgentCard(s) in determining whether to use an agent,
call an LLM with a prompt and the responses it has received from the subagents and then the “SuperAgent” would take actions based on the LLM response.
‍

Create an A2A Client

The source code for this client is shown in Appendix 3 and makes a request to the A2A “SuperAgent” to retrieve Bronto user and/or employee information. It consists of:

a simple handler for the input argument to get the query parameter to use
a httpx client and A2ACardResolver to retrieve the AgentCard of the “SuperAgent”
an A2AClient is instantiated with the retrieved AgentCard
creating an A2A message with the query passed in the text part
sending an A2A request to the “SuperAgent” and awaiting the response
for demo purposes, it also sends the request as a streaming message and prints the chunks returned
‍

A2A Demonstration

Once you have the MCP servers running as described in the previous blog, you can run the A2A bronto-get-users and bronto-get-users subagents in the directories holding their source with ^{uv run}.

Similarly, you can run the “SuperAgent” in the directory holding its source with ^{uv run} .

Finally, you can run the A2A client in the directory holding its source with uv run . and a parameter of “GetEmployees” or “GetUsers” or “GetAllUserDetails” and see the “SuperAgent” send a request to the appropriate A2A subagent(s) and in turn onto the MCP Servers and then to Bronto or SQL.

In the previous blog, Claude was configured for the MCP servers and it did not require the use of A2A. The use of A2A here allows for flexibility by allowing use of a separate “SuperAgent” to discover our employee and user services and deciding whether to use them and make requests, using their agent cards.
‍

Summary

This simple demo scenario showed that there was no need to move any data to a new database or data lake to get access to all the data in an AI agent world by using A2A and MCP. In this case the data could stay in the most suitable data storage, i.e. logs in the special purpose Bronto log platform for fast search and low cost storage and tabular HR data in a SQL database, but this data could be accessed by using agents specific to where that data was stored, with an A2A enabled “SuperAgent” orchestrating the requests to get the requested data.

This demonstrated use could be easily extended by using the “SuperAgent” to filter the data and/or aggregate the data it retrieves from its subordinate agents (such as from Bronto) and then send the filtered/aggregated data to an LLM for further analysis. This use of a “SuperAgent” would have benefits in terms of orchestrating the requests and the use of subagents, but the filtering/aggregation of data provides an opportunity to reduce the tokens used by the LLM and avoid it having to handle large volumes of data directly - the importance of managing agent costs is illustrated in this PostHog blog.

As outlined in the first blog in this series, there are a number of areas where A2A requires further development. This sample code has shown a simple case of orchestrating two agents, but as in the earlier blog Multi-Agent Orchestration would need higher level standardized support to handle complex interactions autonomously. This sample code also showed a simple use of the name in AgentCards to select an agent, but as per earlier blog, there is no agreed/shared vocabulary to define items that may be described over A2A. This could be assisted by asking LLMs to use the provided AgentCard to determine if an agent is suitable, but this would still benefit by the emergence of a shared vocabulary. Challenges also remain in establishing trust between agents, security (beyond authentication), scalability and enterprise readiness.

A2A has, however, standardized agent communication and the samples here have shown it makes it simple to build interoperable agents. Whether A2A will be widely adopted and become the enabler for a rich ecosystem of collaborating AI agents or whether the ecosystem will be custom point solutions using MCP servers remains to be seen…but Bronto’s high performance logging platform will be part of this ecosystem, using the Bronto remote MCP server and/or A2A.
‍

Code Samples for A2A Clients

Note: All code samples here are for demonstration purposes only as the code is not written for efficiency or with comprehensive error handling. For example, the code manipulating requests and responses could be improved using pydantic, but that is not shown here to keep the code similar to what is used in the official A2A Samples github repo.
‍

Appendix 1- A2A Client Code to access Bronto /users API

main.py

import uvicorn
from a2a.server.apps import A2AStarletteApplication
from a2a.server.request_handlers import DefaultRequestHandler
from a2a.server.tasks import InMemoryTaskStore
from a2a.types import (
    AgentCapabilities,
    AgentCard,
    AgentSkill,
)
from agent_executor import BrontoAgentExecutor

if __name__ == '__main__':
    skill = AgentSkill(
        id='bronto_user_info',
        name='GetUsersAgent',
        description='Returns Information on Bronto Users',
        tags=['bronto users'],
        examples=['David', 'Gary'],
    )

    extended_skill = AgentSkill(
        id='mcp_get_user_call',
        name='Gets Users from sample Bronto MCP Server',
        description='Uses Bronto REST API to get all users in an org',
        tags=['Get User', 'MCP Client', 'extended'],
        examples=['MCP Client call', 'Gets Users'],
    )

    # This will be the public-facing agent card
    public_agent_card = AgentCard(
        name='GetUsersAgent',
        description='Uses Bronto REST API to get all users in an org',
        url='http://localhost:9998/',
        version='1.0.0',
        default_input_modes=['text'],
        default_output_modes=['text'],
        capabilities=AgentCapabilities(streaming=True),
        skills=[skill],  # Only the basic skill for the public card
        supports_authenticated_extended_card=True,
    )

    # This will be the authenticated extended agent card
    # It includes the additional 'extended_skill'
    specific_extended_agent_card = public_agent_card.model_copy(
        update={
            'name': 'Bronto Agent - Extended Edition',
            'description': 'The full-featured Bronto agent for authenticated users.',
            'version': '1.0.1',
            # Capabilities and other fields like url, default_input_modes, default_output_modes,
            # supports_authenticated_extended_card are inherited unless specified here.
            'skills': [
                skill,
                extended_skill,
            ],
        }
    )

    request_handler = DefaultRequestHandler(
        agent_executor=BrontoAgentExecutor(),
        task_store=InMemoryTaskStore(),
    )

    server = A2AStarletteApplication(
        agent_card=public_agent_card,
        http_handler=request_handler,
        extended_agent_card=specific_extended_agent_card,
    )

    uvicorn.run(server.build(), host='0.0.0.0', port=9998)

agent_executor.py
‍

agent.py

from a2a.server.agent_execution import AgentExecutor, RequestContext
from a2a.server.events import EventQueue
from a2a.utils import new_agent_text_message
from fastmcp import Client

class BrontoAgent:
    """Bronto Agent."""

    def __init__(self, mcp_server_url: str = "http://localhost:8080/sse"):
        self.mcp_client = Client(mcp_server_url)

    async def invoke(self) -> str:
        async with self.mcp_client as client:
            try:
                await client.ping()
                tools = await client.list_tools()
                print(f"------------ tools: {tools}")

                mcp_response = await client.call_tool("get_bronto_users")
                print(f"MCP Response: {mcp_response}")

                return f"A2A Server received: MCP Server replied: {mcp_response}"

            except Exception as e:
                return f"Error calling MCP server: {str(e)}"


class BrontoAgentExecutor(AgentExecutor):

    def __init__(self):
        self.agent = BrontoAgent()

    async def execute(self, context: RequestContext, event_queue: EventQueue) -> None:
        user_msg = context.message if context.message else "No message from the MCP Server"

        result = await self.agent.invoke()
        await event_queue.enqueue_event(new_agent_text_message(result))

    async def cancel(self, context: RequestContext, event_queue: EventQueue) -> None:
        raise Exception('cancel not supported')

Appendix 2 - “SuperAgent” Code

main.py

import uvicorn
from a2a.server.apps import A2AStarletteApplication
from a2a.server.request_handlers import DefaultRequestHandler
from a2a.server.tasks import InMemoryTaskStore
from a2a.types import (
    AgentCapabilities,
    AgentCard,
    AgentSkill,
)
from agent_executor import (
    SuperAgentExecutor  # type: ignore[import-untyped]
)

HOST = "0.0.0.0"
PORT = 9000
URL = f"http://localhost:{PORT}/"

# Define the Super Agent's own identity
if __name__ == '__main__':
    # Define the base skill exposed by the Super Agent
    skill = AgentSkill(
        id='SuperOrchestrator',
        name='Super Agent',
        description='Retrieves Bronto Users and Employees',
        tags=['bronto user and employee info'],
        examples=['David', 'Gary'],
    )

    # Additional skill only available in the extended authenticated card
    extended_skill = AgentSkill(
        id='get_employee_agent_call',
        name='GetEmployeesAgent',
        description='Uses a sample Bronto Agent Server to get employees',
        tags=['Get Employees', 'A2A Client', 'extended'],
        examples=['Get Users', 'Get Employees'],
    )

    # This will be the public-facing agent card
    public_agent_card = AgentCard(
        name='SuperAgent',
        description='Uses Bronto Agents to get users and employees',
        url='http://localhost:9000/',
        version='1.0.0',
        default_input_modes=['text'],
        default_output_modes=['text'],
        capabilities=AgentCapabilities(streaming=True),
        skills=[skill],  # Only the basic skill for the public card
        supports_authenticated_extended_card=True,
    )

    # This will be the authenticated extended agent card
    # It includes the additional 'extended_skill'
    specific_extended_agent_card = public_agent_card.model_copy(
        update={
            'name': 'Bronto Agent - Extended Edition',  # Different name for clarity
            'description': 'The full-featured Bronto agent for authenticated users.',
            'version': '1.0.1',  # Could even be a different version
            # Capabilities and other fields like url, default_input_modes,
            # default_output_modes, supports_authenticated_extended_card are
            # inherited from public_agent_card unless specified here.
            'skills': [
                skill,
                extended_skill,
            ],  # Both skills for the extended card
        }
    )

    request_handler = DefaultRequestHandler(
        agent_executor=SuperAgentExecutor(),
        task_store=InMemoryTaskStore(),
    )

    server = A2AStarletteApplication(
        agent_card=public_agent_card,
        http_handler=request_handler,
        extended_agent_card=specific_extended_agent_card,
    )

    uvicorn.run(server.build(), host=HOST, port=PORT)

‍

agent_executor.py

import json
import httpx
import asyncio
from uuid import uuid4
from typing import List, Dict, Any

from a2a.server.agent_execution import AgentExecutor, RequestContext
from a2a.server.events import EventQueue
from a2a.utils import new_agent_text_message
from a2a.client import A2ACardResolver, A2AClient
from a2a.types import (
    AgentCard,
    MessageSendParams,
    SendMessageRequest,
    SendStreamingMessageRequest,
)
from a2a.utils.constants import (
    AGENT_CARD_WELL_KNOWN_PATH,
    EXTENDED_AGENT_CARD_PATH,
)

# List of URLs for the 2 subordinate agents
sub_agents = [
    "http://localhost:9999",  # bronto get users
    "http://localhost:9998"   # get employees db
]

# ---------------------------
# Core = discover + call + route
# SuperAgent holds agent logic
# SuperAgentExecutor handles A2A protocol
# ---------------------------

class SuperAgent:

    def __init__(self):
        self.subordinate_urls = sub_agents
        self.agent_registry = {}

    async def invoke(self) -> str:
        await self._discover_agents()
        print("----------- Agent Discovery Complete --------------- ")
        return 'Agent Discovery Complete'

    async def _discover_agents(self):
        print("----------- Super Agent Discovering Agents")

        async with httpx.AsyncClient() as httpx_client:
            for url in self.subordinate_urls:
                resolver = A2ACardResolver(
                    httpx_client=httpx_client,
                    base_url=url
                )
                try:
                    # A2A standard: Cards are at /.well-known/agent.json
                    print(f"Attempting to fetch public agent card from: {url}{AGENT_CARD_WELL_KNOWN_PATH}")
                    public_card = (
                        await resolver.get_agent_card()
                    )

                    self.agent_registry[public_card.name] = {
                        "card": public_card,
                        "skills": [s.description for s in public_card.skills]
                    }

                    print(f"Discovered Agent: {public_card.name} with skills: {self.agent_registry[public_card.name]['skills']}")
                except Exception as e:
                    print(f"Failed to discover agent at {url}: {e}")

    def a2a_get_text_from_response(self, response) -> str:
        # Extract the text content
        try:
            message_obj = response.root.result
            first_part = message_obj.parts[0]
            extracted_text = first_part.root.text
            print(f"=============== Received Message : {extracted_text}")
            return extracted_text
        except (AttributeError, IndexError) as e:
            return f"Could not extract text: {e}"

        # Response -> List -> Part -> Root (TextPart) -> text
        if response.parts:
            content = response.parts[0].root.text
            return content

        return "No text content received."

    def print_agent_info(self, name, agent_data):
        print(f"------ Agent Name : {name} ------")
        print(f"       Agent skills: {agent_data['skills']}")

    def bronto_agents(self) -> List[str]:
        return [
            name for name, info in self.agent_registry.items()
            if any("bronto" in (s or "").lower() for s in info['skills'])
        ]


class SuperAgentExecutor(AgentExecutor):

    def __init__(self):
        self.agent = SuperAgent()

    async def execute(
        self,
        context: RequestContext,
        event_queue: EventQueue,
    ) -> None:
        # Extract the message content from the A2A request
        user_query = context.message if context.message else "No message from the A2A Super Agent Server"

        invoke_result = await self.agent.invoke()

        # Simple Routing Logic - a more powerful 'Super Agent'
        # could pass the agent_registry descriptions to an LLM here
        # Use names specified in skills
        # GetUsersAgent or GetEmployeesAgent to call specific agents
        # or GetAllUserInfo to call all known Bronto Agents

        async with httpx.AsyncClient() as client:
            agent_cmd = user_query.parts[0].root.text

            if "GetAllUserDetails" in agent_cmd:
                merged = await self.call_all_bronto_agents(client, event_queue)
                await event_queue.enqueue_event(new_agent_text_message(merged))

            elif "GetUsers" in agent_cmd:
                response = await self.call_named_agent(client, "GetUsersAgent")
                await event_queue.enqueue_event(new_agent_text_message(str(response)))

            elif "GetEmployees" in agent_cmd:
                response = await self.call_named_agent(client, "GetEmployeesAgent")
                await event_queue.enqueue_event(new_agent_text_message(str(response)))

            else:
                await event_queue.enqueue_event(new_agent_text_message(f"Agent not found for {agent_cmd}"))

    async def call_named_agent(self, client: httpx.AsyncClient, agent_name: str) -> str:
        target_agent = self.agent.agent_registry.get(agent_name)
        if not target_agent:
            return f"Unable to find '{agent_name}'."

        # Call the subordinate agent via A2A, using the agent card
        a2a_client = A2AClient(
            httpx_client=client,
            agent_card=target_agent["card"]
        )

        message_payload: dict[str, Any] = {
            'message': {
                'role': 'user',
                'parts': [
                    {'kind': 'text', 'text': 'bronto_user_info'}
                ],
                'messageId': uuid4().hex,
            },
        }

        request = SendMessageRequest(
            id=str(uuid4()),
            params=MessageSendParams(**message_payload)
        )

        response = await a2a_client.send_message(request)
        return self.agent.a2a_get_text_from_response(response)

    async def call_all_bronto_agents(self, client: httpx.AsyncClient, event_queue) -> str:
        names = self.agent.bronto_agents()
        if not names:
            await event_queue.enqueue_event(new_agent_text_message("No Bronto Agents"))
            return ""

        # gather and merge responses from subordinate agents
        results = await asyncio.gather(*(self.call_named_agent(client, n) for n in names))
        merged_responses = "\n\n".join(
            f"--- Report from {n} ---\n{r}" for n, r in zip(names, results)
        )
        return merged_responses

    async def cancel(
        self,
        context: RequestContext,
        event_queue: EventQueue
    ) -> None:
        raise Exception('cancel not supported')

‍

Appendix 3 - Code for A2A Client

client.py

import logging
import sys
from typing import Any
from uuid import uuid4

import httpx

from a2a.client import A2ACardResolver, A2AClient
from a2a.types import (
    AgentCard,
    MessageSendParams,
    SendMessageRequest,
    SendStreamingMessageRequest,
    Message,
    TextPart,
)
from a2a.utils.constants import (
    AGENT_CARD_WELL_KNOWN_PATH,
    EXTENDED_AGENT_CARD_PATH,
)

async def main() -> None:
    logging.basicConfig(level=logging.INFO)
    logger = logging.getLogger(__name__)

    base_url = 'http://localhost:9000'

    # read query to make
    if len(sys.argv) > 1:
        query_param = sys.argv[1]
        print(f"The Query is for: {query_param}")
    else:
        query_param = "GetAllUserInfo"
        print("No query provided - defaulting to GetAllUserInfo")

    async with httpx.AsyncClient() as httpx_client:
        resolver = A2ACardResolver(
            httpx_client=httpx_client,
            base_url=base_url,
        )

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

        try:
            logger.info(
                f'Attempting to fetch public agent card from: {base_url}{AGENT_CARD_WELL_KNOWN_PATH}'
            )
            _public_card = (
                await resolver.get_agent_card()
            )  # Fetches from default public path

            logger.info('Successfully fetched public agent card:')
            logger.info(
                _public_card.model_dump_json(indent=2, exclude_none=True)
            )

            final_agent_card_to_use = _public_card
            logger.info(
                '\nUsing PUBLIC agent card for client initialization (default).'
            )

        except Exception as e:
            logger.error(
                f'Critical error fetching public agent card: {e}', exc_info=True
            )
            raise RuntimeError(
                'Failed to fetch the public agent card. Cannot continue.'
            ) from e

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

        user_message = Message(
            role='user',
            parts=[TextPart(text=query_param)],
            message_id=str(uuid4())
        )

        params = MessageSendParams(message=user_message)
        request = SendMessageRequest(
            id=str(uuid4()),
            params=params
        )

        try:
            logger.info("Sending message...")
            response = await client.send_message(request)
            print("--- Response ---")
            print(response.model_dump_json(indent=2, exclude_none=True))
        except Exception as e:
            logger.error(f"Error during send_message: {e}")

        print("\n--- Starting Stream ---")
        streaming_request = SendStreamingMessageRequest(
            id=str(uuid4()),
            params=params
        )

        try:
            async for chunk in client.send_message_streaming(streaming_request):
                print(chunk.model_dump_json(exclude_none=True))
        except Exception as e:
            logger.error(f"Error during streaming: {e}")

if __name__ == '__main__':
    import asyncio
    asyncio.run(main())

Find out more?

Our AI features

Customer Success Story

Compare to others

Book a Demo