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

┌──────────────────────────────────────────────────┐
│              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    │
└──────────────────────────────────────────────────┘

# 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)

# 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())

# Terminal 1
python research_agent.py

# Terminal 2
python orchestrator.py

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

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

# 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}"

# 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"}],
)

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