Peyton Green

Posted on May 12

MCP and A2A Together in Python: Tool Calls That Cross Agent Boundaries (Without the Cloud Lock-In)

#python #a2a #mcp #ai

In March, CVE-2025-6514 was published: command injection in mcp-remote, CVSS 9.6, around 500,000 downloads affected.

MCP is in production. Real deployments, real users, real security surface.

The MCP Dev Summit NYC ran April 2–3. Six sessions on authentication. Aaron Parecki — OAuth 2.1 spec author — delivered a talk called "Evolution, Not Revolution: How MCP Is Reshaping OAuth." The consistent message: these protocols are stable, the architecture is settled, and the question now is how to build on them correctly.

A2A joined the Linux Foundation in February with AWS, Cisco, Microsoft, and Salesforce as co-signers. The spec also now includes an official statement: "MCP handles tool/resource integration, A2A handles agent-to-agent coordination — complementary, not competing."

If you're looking at both protocols and asking "do I have to rebuild everything?" — the answer is no. They solve different problems and they're designed to work together in the same stack.

Here's what that looks like in code.

The One Diagram That Settles It

┌──────────────────────────────────────────────────┐
│              Orchestrator Agent                  │
│  "Find me the top-3 cited papers on RAG"         │
├──────────────────────────────────────────────────┤
│  A2A: delegates tasks to specialist agents       │
│  sends task → research_agent (localhost:9998)    │
├──────────────────────────────────────────────────┤
│  MCP: calls tools, reads files, fetches data     │
│  research_agent uses fetch + filesystem tools    │
└──────────────────────────────────────────────────┘

MCP connects your agent to external resources — file systems, databases, APIs, browser automation. It answers "what can this agent access?"

A2A connects your agent to other agents — specialist workers, parallel executors, remote services. It answers "what other agents can this agent delegate to?"

They operate at different abstraction layers. MCP is your tool belt. A2A is your interoperability protocol. You need both.

A quick note on A2A's status: on February 27, 2026, A2A joined the Linux Foundation with AWS, Cisco, Microsoft, and Salesforce as co-signers. It's no longer a Google-only proposal — it's an industry standard under neutral governance. The adoption question is settled.

The notable holdout is OpenAI. A contributor submitted a complete A2A implementation to openai-agents-python and the maintainers declined: "we don't have immediate plans to add A2A support to this SDK." Until OpenAI ships an A2A client, cross-framework interoperability between OpenAI agents and A2A agents will require a wrapper or translation layer. For teams that aren't locked into the OpenAI SDK, this is a non-issue — LangGraph, CrewAI, PydanticAI, and Google ADK all either have or are implementing A2A. But if your orchestrator is built on OpenAI agents, factor in that gap today.

What Each Protocol Handles

Problem	Protocol	Mechanism
My agent needs to read a local file	MCP	Tool call to filesystem server
My agent needs to query a database	MCP	Tool call to database server
My agent needs to call a REST API	MCP	Tool call to fetch/http server
My agent needs to hand off a task to a specialist	A2A	Task submission to an A2A agent
My agent needs to run subtasks in parallel	A2A	Multiple A2A task submissions
My orchestrator needs to coordinate across frameworks	A2A	A2A's standard task schema works across LangGraph, CrewAI, ADK

The confusion comes from the fact that both feel like "ways for an agent to do more things." But the mechanism is completely different.

In MCP, your agent calls a tool and gets a result back synchronously. The tool doesn't have goals, memory, or identity. It's a function.

In A2A, your agent submits a task to another agent that has its own goals, memory, and task lifecycle. The downstream agent can run for seconds, minutes, or longer — and send back streaming updates while it works.

A Concrete Example: Both Protocols in One Flow

Here's an orchestrator that uses A2A to delegate to a research agent, which uses MCP to do its actual work.

The research agent (research_agent.py) — an A2A-compliant server that internally uses httpx to fetch pages (you could swap this for an MCP fetch tool):

# research_agent.py
import asyncio
import uvicorn
import httpx
from a2a.server.apps import A2AStarletteApplication
from a2a.server.agent_execution import AgentExecutor, RequestContext
from a2a.server.events import EventQueue
from a2a.server.request_handlers import DefaultRequestHandler
from a2a.server.tasks import InMemoryTaskStore, TaskUpdater
from a2a.types import AgentCapabilities, AgentCard, AgentSkill
from a2a.utils import new_agent_text_message


class ResearchAgentExecutor(AgentExecutor):
    async def execute(self, context: RequestContext, event_queue: EventQueue) -> None:
        updater = TaskUpdater(event_queue, context.task_id, context.context_id)
        await updater.start_work()

        query = context.get_user_input()

        # MCP layer would go here in a real implementation.
        # This agent calls external tools (fetch, filesystem, search API)
        # via whatever MCP server it has configured. For the demo, we simulate.
        result = f"Research results for: '{query}'\n"
        result += "1. Smith et al. (2023) — RAG survey, 1,241 citations\n"
        result += "2. Lewis et al. (2020) — original RAG paper, 4,800+ citations\n"
        result += "3. Gao et al. (2023) — advanced RAG techniques, 890 citations"

        await event_queue.enqueue_event(new_agent_text_message(result))
        await updater.complete()

    async def cancel(self, context: RequestContext, event_queue: EventQueue) -> None:
        raise NotImplementedError


skill = AgentSkill(
    id="research",
    name="Literature Research",
    description="Searches and summarizes academic papers on a given topic.",
    tags=["research", "papers", "citations"],
    examples=["Find the top-cited papers on RAG", "Summarize recent work on agent memory"],
)

agent_card = AgentCard(
    name="Research Agent",
    description="A specialist agent that researches academic topics and returns citations.",
    url="http://localhost:9998/",
    version="1.0.0",
    default_input_modes=["text"],
    default_output_modes=["text"],
    capabilities=AgentCapabilities(streaming=False),
    skills=[skill],
)

app = A2AStarletteApplication(
    agent_card=agent_card,
    http_handler=DefaultRequestHandler(
        agent_executor=ResearchAgentExecutor(),
        task_store=InMemoryTaskStore(),
    ),
)

if __name__ == "__main__":
    uvicorn.run(app.build(), host="0.0.0.0", port=9998)

The orchestrator (orchestrator.py) — uses A2A to dispatch, would use MCP for its own tool access:

# orchestrator.py
import asyncio
import httpx
from a2a.client import A2ACardResolver, ClientFactory, ClientConfig, create_text_message_object


async def delegate_to_research_agent(query: str) -> str:
    """
    Delegates a research task to the research agent via A2A.
    The research agent handles its own tool access (MCP or direct).
    """
    async with httpx.AsyncClient() as http:
        resolver = A2ACardResolver(httpx_client=http, base_url="http://localhost:9998")
        card = await resolver.get_agent_card()
        print(f"→ Delegating to: {card.name}")

        factory = ClientFactory(config=ClientConfig(httpx_client=http))
        client = factory.create(card)
        message = create_text_message_object(content=query)

        async for event in client.send_message(message):
            if hasattr(event, "parts"):
                for part in event.parts:
                    if hasattr(part.root, "text"):
                        return part.root.text
            elif isinstance(event, tuple):
                task, _ = event
                if task.history:
                    for msg in task.history:
                        if msg.role.value == "agent":
                            for part in msg.parts:
                                if hasattr(part.root, "text"):
                                    return part.root.text
    return "No result"


async def main():
    print("Orchestrator: processing research request")
    print("─" * 50)

    # MCP tools would be used here for tasks the orchestrator handles itself
    # (reading local context, checking a database, querying an API).
    # For tasks requiring specialist knowledge, we delegate via A2A.

    query = "top-3 cited papers on retrieval-augmented generation"
    result = await delegate_to_research_agent(query)

    print(f"\nResult from research agent:\n{result}")


asyncio.run(main())

Run it:

# Terminal 1
python research_agent.py

# Terminal 2
python orchestrator.py

Output:

Orchestrator: processing research request
──────────────────────────────────────────────────
→ Delegating to: Research Agent

Result from research agent:
Research results for: 'top-3 cited papers on retrieval-augmented generation'
1. Smith et al. (2023) — RAG survey, 1,241 citations
2. Lewis et al. (2020) — original RAG paper, 4,800+ citations
3. Gao et al. (2023) — advanced RAG techniques, 890 citations

The key point in the orchestrator: delegate_to_research_agent() doesn't know or care how the research agent gets its data. It might use MCP filesystem tools, a fetch tool, a search API, or a local knowledge base. The A2A interface is agnostic to that. The orchestrator says "here's the task" — the specialist agent handles its own tooling.

When to Reach for Each

Reach for MCP when:

Your agent needs to read a file, query a database, or call an API
The operation is synchronous and can return in milliseconds to a few seconds
You're connecting to infrastructure that doesn't have its own agent identity
You want your agent to have access to tools from a pre-built ecosystem (Anthropic, Zapier, etc.)

Reach for A2A when:

You want to delegate a task to a specialized agent that owns its own execution logic
You need a long-running subtask with progress updates back to the orchestrator
You want your orchestrator to be framework-agnostic (works with LangGraph agents, CrewAI agents, Google ADK agents, or a Python script you wrote this afternoon)
You're building multi-agent pipelines where specialists need to be swappable

The practical test: If the downstream thing is a function (read file, query DB, call API) — it's MCP. If the downstream thing is an agent (with its own goals, state, and decision-making) — it's A2A.

Migrating from MCP-only to MCP + A2A

If you're already using MCP, nothing changes. Your existing MCP tool servers are still useful. You're adding a new layer, not replacing one.

The migration pattern:

Keep your MCP tool servers as-is
Identify specialist capabilities in your current monolithic agent that would benefit from isolation (a web researcher, a code analyzer, a document processor)
Extract those into A2A-compliant specialist agents
Your orchestrator calls the specialists via A2A; the specialists use MCP for their own tool access

Your existing Python automation scripts work here too. An AgentExecutor wrapper is about 15 lines — the same pattern from the A2A quickstart. Each script becomes a callable specialist that any A2A-compatible orchestrator can dispatch.

The Stack in One View

Your Domain
└── Orchestrator agent
    ├── MCP: filesystem, database, APIs (your own tools)
    ├── A2A → Research specialist
    │         └── MCP: fetch, search, knowledge base (research agent's tools)
    ├── A2A → Code analyzer specialist
    │         └── MCP: filesystem, linter, AST tools
    └── A2A → Notification agent
              └── MCP: email, Slack, SMS tools

Each specialist owns its own MCP tooling. The orchestrator coordinates via A2A. The boundaries are clean.

Authentication in an MCP + A2A Stack

Two network boundaries need authentication:

Client → MCP server (your agent calling its tools)
Orchestrator → A2A agent (your orchestrator delegating tasks)

Both protocols converge on the same auth model: OAuth 2.1, with an external authorization server. Your auth infrastructure is reusable across both.

MCP auth (mcp>=1.27,<2)

# research_agent_mcp.py — MCP server exposing RFC 9728 resource metadata
from mcp.server.fastmcp import FastMCP
from mcp.server.auth.middleware.bearer import BearerAuthBackend, BearerAuthProvider
import httpx

app = FastMCP("research-agent")

# RFC 9728: tell clients where to get a token for this resource server.
# The MCP server validates tokens — it does NOT issue them.
@app.custom_route("/.well-known/oauth-protected-resource", methods=["GET"])
async def oauth_resource_metadata(request):
    from starlette.responses import JSONResponse
    return JSONResponse({
        "resource": "https://research-agent.example.com",
        "authorization_servers": ["https://auth.example.com"]
    })

@app.tool()
async def search_papers(query: str) -> str:
    """Search for research papers on the given topic."""
    # Your tool implementation here
    return f"Results for: {query}"

Install: pip install "mcp>=1.27,<2"

The MCP server validates tokens from the external AS. It does not issue them. For the full implementation (token endpoint, PKCE, Dynamic Client Registration), see FastAPI + MCP: Adding Real OAuth 2.1 Auth to Your Python MCP Server.

A2A auth (a2a-sdk==0.3.25)

A2A uses OAuth 2.1 with device code flow (RFC 8628) and PKCE. Implicit and password flows are removed in v1.0. The agent card advertises where clients can get a token.

# research_agent_a2a.py — A2A agent with auth metadata in agent card
from fasta2a import FastA2A

app = FastA2A(
    name="Research Agent",
    description="Searches and summarizes research papers.",
    url="http://localhost:9998/a2a",
    version="1.0.0",
    # Agent card auth surface — points clients to the authorization server
    authentication={
        "schemes": ["Bearer"],
        "credentials": "https://auth.example.com/.well-known/oauth-authorization-server"
    },
    capabilities={"streaming": False, "pushNotifications": False},
    defaultInputModes=["text"],
    defaultOutputModes=["text"],
    skills=[{"id": "research", "name": "Research", "description": "Searches research papers"}],
)

The combined auth flow

Authorization Server (auth.example.com)
  └── issues tokens for both MCP servers and A2A agents

Orchestrator
  ├── Bearer token → MCP server (tool calls)
  └── Bearer token → A2A agent (task submissions)

MCP server           A2A agent
  └── validates        └── validates
      token via             token via
      RFC 9728              agent card
      discovery             credentials

A single authorization server can protect both protocol layers. The RFC 9728 discovery pattern (/.well-known/oauth-protected-resource) is the same for both.