Building a Three‑Agent Multi‑Agent System with MCP and A2A

Single LLMs cannot reliably achieve the complex behaviors shown in many Agentic AI demos; a Multi‑Agent System (MAS) is required to separate intent understanding, planning, execution, and result optimization. Anthropic’s Model‑Context‑Protocol (MCP) provides a USB‑C‑like interface that lets agents reliably access data and tools without custom integration code, while Google’s A2A protocol enables heterogeneous agents to discover each other’s capabilities and negotiate tasks over HTTP.

Setting Up Two MCP Servers

Two MCP servers are needed: one wrapping a web‑crawling tool for Google search and page reading, and another wrapping a stock‑retrieval tool for real‑time prices. The article demonstrates two transport options—STDIO MCP (subprocess communication) and SSE MCP (Server‑Sent Events for networked use)—and recommends SSE when multiple servers must run concurrently.

# Example: create MCP toolsets
async def return_sse_mcp_tools_search():
    print("Attempting to connect to MCP server for search and page read...")
    server_params = SseServerParams(url="http://localhost:8080/sse")
    tools, exit_stack = await MCPToolset.from_server(connection_params=server_params)
    print("MCP Toolset created successfully.")
    return tools, exit_stack

async def return_sse_mcp_tools_stocks():
    print("Attempting to connect to MCP server for stock info...")
    server_params = SseServerParams(url="http://localhost:8181/sse")
    tools, exit_stack = await MCPToolset.from_server(connection_params=server_params)
    print("MCP Toolset created successfully.")
    return tools, exit_stack

Building the Hierarchical Agents

Using the FastMCP framework, three agents are instantiated:

stock_analysis_agent : accesses the stock MCP toolset to retrieve prices and trends.

search_agent : uses the search MCP toolset to perform Google queries and read pages.

root_agent (named company_analysis_assistant): coordinates the team, delegating queries to child agents based on simple keyword rules.

# Agent definitions (simplified)
stock_analysis_agent = Agent(
    model=MODEL,
    name="stock_analysis_agent",
    instruction="Analyze stock data and provide insights.",
    description="Handles stock analysis and provides insights, in particular, can get the latest stock price.",
    tools=stocks_tools,
)

search_agent = Agent(
    model=MODEL,
    name="search_agent",
    instruction="Expert googler. Can search anything on google and read pages online.",
    description="Handles search queries and can read pages online.",
    tools=search_tools,
)

root_agent = Agent(
    name="company_analysis_assistant",
    model=MODEL,
    description="Main assistant: Handles requests about stocks and information of companies.",
    instruction=(
        "You are the main Assistant coordinating a team. Your primary responsibilities are providing company and stocks reports and delegating other tasks.
"
        "1. If the user asks about a company, provide a detailed report.
"
        "2. If you need any information about the current stock price, delegate to the stock_analysis_agent.
"
        "3. If you need to search for information, delegate to the search_agent.
"
        "Analyze the user's query and delegate or handle it appropriately. If unsure, ask for clarification. Only use tools or delegate as described."
    ),
    sub_agents=[search_agent, stock_analysis_agent],
    output_key="last_assistant_response",
)

Delegation Logic

The root agent’s instruction contains explicit rules that the runtime follows: direct handling, delegation to the stock agent, or delegation to the search agent. The rules are sent to child agents via the A2A client.

# Delegation rules inside root_agent
"1. If the user asks about a company, provide a detailed report.
"
"2. If you need any information about the current stock price, delegate to the stock_analysis_agent.
"
"3. If you need to search for information, delegate to the search_agent.
"

A2A Protocol Overview

The core A2A concepts are:

AgentCard : digital identity describing capabilities, endpoint, version, and required authentication.

A2AServer : exposes HTTP endpoints that accept A2A requests.

A2AClient : sends requests to an A2AServer.

Task : unit of work with lifecycle states (submitted, in‑progress, needs input, completed, failed, cancelled).

Message and Artifact : communication units composed of Part objects (text, file, data).

A typical A2A workflow consists of Discovery, Initiation, Processing (streaming vs. non‑streaming), Interaction, and Completion.

ADKAgent Implementation

ADKAgent bridges Google’s Agent Development Kit (ADK) with the A2A protocol. It can run in two modes:

Standalone agent that owns its tools.

Host agent that coordinates other agents. The host maintains a list of remote agent URLs and uses the send_task method to invoke them.

# Host agent example
host_agent = ADKAgent(
    model=MODEL,
    name="host_agent",
    description="",
    tools=[],
    is_host_agent=True,
    remote_agent_addresses=[
        "http://localhost:11000/google_search_agent",
        "http://localhost:10000/stock_agent",
    ],
)

Agent Card Definition

Each service generates an AgentCard. The stock agent example includes skill metadata, JSON schemas for input and output, and flags for streaming and push notifications.

stock_agent_card = generate_agent_card(
    agent_name="stock_analyzer",
    agent_description="Analyzes stock market data and provides insights",
    agent_url="http://localhost:8080",
    agent_version="1.0.0",
    can_stream=True,
    can_push_notifications=True,
    skills=[
        AgentSkill(
            name="stock_analysis",
            description="Analyzes stock prices and market trends",
            input_schema={"type": "object", "properties": {"symbol": {"type": "string"}, "timeframe": {"type": "string"}}},
            output_schema={"type": "object", "properties": {"price": {"type": "number"}, "trend": {"type": "string"}}},
        ),
    ],
)

Running the A2A Servers

Separate server processes are launched for the host, search, and stock agents. Each server is configured with host, port, endpoint, the generated AgentCard, and an AgentTaskManager. Asynchronous startup uses await server.astart().

# Host server startup (simplified)
async def run_agent():
    AGENT_NAME = "host_agent"
    PORT = 12000
    HOST = "0.0.0.0"
    AGENT_URL = f"http://{HOST}:{PORT}"
    AGENT_CARD = generate_agent_card(...)
    host_agent = ADKAgent(...)
    task_manager = generate_agent_task_manager(agent=host_agent)
    server = A2AServer(host=HOST, port=PORT, endpoint="/host_agent", agent_card=AGENT_CARD, task_manager=task_manager)
    print(f"Starting {AGENT_NAME} A2A Server on {AGENT_URL}")
    await server.astart()

Similar scripts are provided for the Google search agent (port 11000) and the stock report agent (port 10000), each loading its MCP toolset and registering its own AgentCard.

Execution Flow

Create a session and services.

Instantiate the three agents and the root coordinator.

Initialize MCP toolsets for search and stock data.

Build a Runner with the root agent and run it asynchronously.

The root agent receives the user query, decides whether to handle it directly or delegate, and forwards the task via A2A.

Child agents fetch data through their MCP connections and return results.

The root agent aggregates the final response and prints it.

All MCP connections are gracefully closed.

A2A Framework Core Components

Key classes:

AgentCard : unique identifier, description, endpoint, version, skill list, streaming and push‑notification capabilities, authentication requirements.

A2AServer : HTTP server that processes incoming A2A requests, manages task lifecycles, supports both immediate JSON responses and SSE streaming.

A2AClient : sends task initiation requests and handles responses.

Task : tracks state transitions and maintains context.

Message and Artifact : encapsulate textual, file, or structured data payloads.

Part : atomic unit inside messages/artifacts (TextPart, FilePart, DataPart).

Typical A2A sequence:

Discovery – client retrieves an agent’s AgentCard to learn its capabilities.

Initiation – client creates a new task with a unique ID and an initial message.

Processing – server may stream intermediate updates via SSE or return a final result synchronously.

Interaction – if additional input is required, the client sends follow‑up messages using the same task ID.

Completion – task reaches a terminal state (completed, failed, or cancelled).

ADKAgent Modes

Standalone agent: possesses its own tools and executes tasks directly.

Host agent: acts as a coordinator, delegating tasks to remote agents identified by their URLs.

Resulting Agents

Host Agent

async def run_agent():
    AGENT_NAME = "host_agent"
    AGENT_DESCRIPTION = "An agent orchestrates the decomposition of the user request into tasks that can be performed by the child agents."
    PORT = 12000
    HOST = "0.0.0.0"
    AGENT_URL = f"http://{HOST}:{PORT}"
    AGENT_VERSION = "1.0.0"
    MODEL = 'gemini-2.5-pro-preview-03-25'
    AGENT_SKILLS = [
        AgentSkill(id="COORDINATE_AGENT_TASKS", name="coordinate_tasks", description="coordinate tasks between agents."),
    ]
    list_urls = [
        "http://localhost:11000/google_search_agent",
        "http://localhost:10000/stock_agent",
    ]
    AGENT_CARD = generate_agent_card(
        agent_name=AGENT_NAME,
        agent_description=AGENT_DESCRIPTION,
        agent_url=AGENT_URL,
        agent_version=AGENT_VERSION,
        can_stream=False,
        can_push_notifications=False,
        can_state_transition_history=True,
        default_input_modes=["text"],
        default_output_modes=["text"],
        skills=AGENT_SKILLS,
    )
    host_agent = ADKAgent(
        model=MODEL,
        name="host_agent",
        description="",
        tools=[],
        instructions="",
        is_host_agent=True,
        remote_agent_addresses=list_urls,
    )
    task_manager = generate_agent_task_manager(agent=host_agent)
    server = A2AServer(host=HOST, port=PORT, endpoint="/host_agent", agent_card=AGENT_CARD, task_manager=task_manager)
    print(f"Starting {AGENT_NAME} A2A Server on {AGENT_URL}")
    await server.astart()

Search Agent

async def run_agent():
    AGENT_NAME = "google_search_agent"
    AGENT_DESCRIPTION = "An agent that handles search queries and can read pages online."
    HOST = "0.0.0.0"
    PORT = 11000
    AGENT_URL = f"http://{HOST}:{PORT}"
    AGENT_VERSION = "1.0.0"
    MODEL = 'gemini-2.5-pro-preview-03-25'
    AGENT_SKILLS = [
        AgentSkill(id="GOOGLE_SEARCH", name="google_search", description="Handles search queries and can read pages online."),
    ]
    AGENT_CARD = generate_agent_card(
        agent_name=AGENT_NAME,
        agent_description=AGENT_DESCRIPTION,
        agent_url=AGENT_URL,
        agent_version=AGENT_VERSION,
        can_stream=False,
        can_push_notifications=False,
        can_state_transition_history=True,
        default_input_modes=["text"],
        default_output_modes=["text"],
        skills=AGENT_SKILLS,
    )
    gsearch_tools, g_search_exit_stack = await return_sse_mcp_tools_search()
    google_search_agent = ADKAgent(
        model=MODEL,
        name="google_search_agent",
        description="Handles search queries and can read pages online.",
        tools=gsearch_tools,
        instructions=("You are an expert googler. Can search anything on google and read pages online."),
    )
    task_manager = generate_agent_task_manager(agent=google_search_agent)
    server = A2AServer(host=HOST, port=PORT, endpoint="/google_search_agent", agent_card=AGENT_CARD, task_manager=task_manager)
    print(f"Starting {AGENT_NAME} A2A Server on {AGENT_URL}")
    await server.astart()

Stock Agent

async def run_agent():
    AGENT_NAME = "stock_report_agent"
    AGENT_DESCRIPTION = "An agent that provides US stock prices and info."
    PORT = 10000
    HOST = "0.0.0.0"
    AGENT_URL = f"http://{HOST}:{PORT}"
    AGENT_VERSION = "1.0.0"
    MODEL = 'gemini-2.5-pro-preview-03-25'
    AGENT_SKILLS = [
        AgentSkill(id="SKILL_STOCK_REPORT", name="stock_report", description="Provides stock prices and info."),
    ]
    AGENT_CARD = generate_agent_card(
        agent_name=AGENT_NAME,
        agent_description=AGENT_DESCRIPTION,
        agent_url=AGENT_URL,
        agent_version=AGENT_VERSION,
        can_stream=False,
        can_push_notifications=False,
        can_state_transition_history=True,
        default_input_modes=["text"],
        default_output_modes=["text"],
        skills=AGENT_SKILLS,
    )
    stocks_tools, stocks_exit_stack = await return_sse_mcp_tools_stocks()
    stock_analysis_agent = ADKAgent(
        model=MODEL,
        name="stock_analysis_agent",
        description="Handles stock analysis and provides insights, in particular, can get the latest stock price.",
        tools=stocks_tools,
        instructions=(
            "Analyze stock data and provide insights. You can also get the latest stock price.
"
            "If the user asks about a company, the stock prices for the said company.
"
            "If the user asks about a stock, provide the latest stock price and any other relevant information.
"
            "You can get only the latest stock price for US companies."
        ),
    )
    task_manager = generate_agent_task_manager(agent=stock_analysis_agent)
    server = A2AServer(host=HOST, port=PORT, endpoint="/stock_agent", agent_card=AGENT_CARD, task_manager=task_manager)
    print(f"Starting {AGENT_NAME} A2A Server on {AGENT_URL}")
    await server.astart()

All source code is available at https://github.com/Tsadoq/a2a-mcp-tutorial. Additional useful libraries include Google’s ADK (https://google.github.io/adk-docs/), the mcp-agent repository (https://github.com/lastmile-ai/mcp-agent), and collections of sample MCP servers (https://github.com/punkpeye/awesome-mcp-servers).