How to Use A2A to Access Multiple Data Stores

Authored by David Tracey

This is Part 3 of a three-part series on the Agent2Agent (A2A) protocol. Part 1 introduced A2A concepts. Part 2 showed how to use MCP servers with Claude to query multiple data sources. This post extends that scenario by replacing Claude as the coordinator with a proper A2A "SuperAgent" that autonomously discovers and orchestrates specialist agents.

The scenario is the same: an employee leaves a company, and IT needs to verify their details have been removed from all systems — an HR database and a logging platform.

---

Architecture Overview

We'll build:

1. An A2A agent wrapping the Bronto MCP server (GetUsersAgent, port 9998)
2. An A2A agent wrapping the SQLite MCP server (GetEmployeesAgent, port 9999)
3. A "SuperAgent" (port 9000) that discovers both agents via their Agent Cards and routes requests
4. An A2A client that sends queries to the SuperAgent

---

Step 1: Create the A2A Bronto Agent

Each A2A agent has two files: __main__.py (Agent Card + server setup) and agent_executor.py (the actual logic).

__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'],
    )

    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],
        supports_authenticated_extended_card=True,
    )

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

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

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

agent_executor.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:
    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()
                mcp_response = await client.call_tool("get_bronto_users")
                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:
        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')

The Employee DB A2A agent uses the same structure — just updating the MCP URL to port 8081, changing the tool name to get_employees, and updating the Agent Card name/description accordingly.

---

Step 2: Create the A2A SuperAgent

The SuperAgent discovers sub-agents via their Agent Cards and routes requests based on the query it receives.

agent_executor.py (key parts)

import httpx
import asyncio
from uuid import uuid4
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 MessageSendParams, SendMessageRequest

# Sub-agent URLs
sub_agents = [
    "http://localhost:9999",  # bronto get users
    "http://localhost:9998"   # get employees db
]

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

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

    async def _discover_agents(self):
        async with httpx.AsyncClient() as httpx_client:
            for url in self.subordinate_urls:
                resolver = A2ACardResolver(httpx_client=httpx_client, base_url=url)
                try:
                    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: {public_card.name}")
                except Exception as e:
                    print(f"Failed to discover agent at {url}: {e}")

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

    def a2a_get_text_from_response(self, response) -> str:
        try:
            return response.root.result.parts[0].root.text
        except (AttributeError, IndexError) as e:
            return f"Could not extract text: {e}"


class SuperAgentExecutor(AgentExecutor):
    def __init__(self):
        self.agent = SuperAgent()

    async def execute(self, context: RequestContext, event_queue: EventQueue) -> None:
        await self.agent.invoke()

        async with httpx.AsyncClient() as client:
            agent_cmd = context.message.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, agent_name: str) -> str:
        target_agent = self.agent.agent_registry.get(agent_name)
        if not target_agent:
            return f"Unable to find '{agent_name}'."

        a2a_client = A2AClient(httpx_client=client, agent_card=target_agent["card"])

        request = SendMessageRequest(
            id=str(uuid4()),
            params=MessageSendParams(message={
                'role': 'user',
                'parts': [{'kind': 'text', 'text': 'bronto_user_info'}],
                'messageId': uuid4().hex,
            })
        )

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

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

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

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

---

Step 3: Create the A2A Client

# client.py
import sys, asyncio, logging
from uuid import uuid4
import httpx
from a2a.client import A2ACardResolver, A2AClient
from a2a.types import MessageSendParams, SendMessageRequest, SendStreamingMessageRequest, Message, TextPart

async def main() -> None:
    base_url = 'http://localhost:9000'
    query_param = sys.argv[1] if len(sys.argv) > 1 else "GetAllUserInfo"

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

        client = A2AClient(httpx_client=httpx_client, agent_card=agent_card)

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

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

        response = await client.send_message(request)
        print(response.model_dump_json(indent=2, exclude_none=True))

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

---

Running the Full Demo

With the MCP servers running (from Part 2), start the A2A agents:

# Terminal 1: Bronto A2A agent (wraps Bronto MCP server)
cd bronto-get-users && uv run .

# Terminal 2: Employee DB A2A agent (wraps SQLite MCP server)
cd bronto-get-employees && uv run .

# Terminal 3: SuperAgent
cd superagent && uv run .

# Terminal 4: A2A Client
uv run . GetEmployees      # routes to Employee DB agent
uv run . GetUsers          # routes to Bronto agent
uv run . GetAllUserDetails # calls all agents and merges responses

The SuperAgent discovers available agents via their Agent Cards, then routes requests accordingly — without any hardcoded knowledge of what each agent does.

---

Summary

This three-part series has demonstrated that:

1. A2A standardizes agent communication — any A2A agent can discover and call any other A2A agent, regardless of who built it
2. MCP and A2A work together — MCP handles tool access, A2A handles agent-to-agent orchestration
3. Data doesn't need to move — logs stay in Bronto (optimized for high-volume fast search), HR data stays in SQL, each accessed by purpose-built agents

A more advanced SuperAgent could pass all Agent Card descriptions to an LLM for smarter routing, or filter/aggregate data from sub-agents before sending it to an LLM — reducing token usage and avoiding LLM hallucination from large raw datasets. The PostHog blog on optimizing agent costs illustrates why this matters.

Whether A2A becomes the universal standard for agent collaboration or the ecosystem remains a patchwork of custom MCP integrations remains to be seen — but Bronto's high-performance logging platform will be a first-class citizen in whichever world emerges.

Explore Bronto's AI Features