Building Multi-Agent Systems with MCP and A2A — A Practical Integration Guide to Model Context Protocol and Agent-to-Agent Protocol

Agent (MCP Client)
    │
    ├─ MCP Server A (File System)
    ├─ MCP Server B (Database)
    └─ MCP Server C (External API)

# pip install fastmcp httpx
from fastmcp import FastMCP
import httpx
 
mcp = FastMCP("weather-server")
 
@mcp.tool()
async def get_weather(city: str) -> dict:
    """Returns current weather information for a given city name."""
    async with httpx.AsyncClient() as client:
        response = await client.get(
            "https://api.weather.example.com/current",
            params={"q": city, "units": "metric"}
        )
        data = response.json()
        return {
            "city": city,
            "temperature": data["main"]["temp"],
            "description": data["weather"][0]["description"]
        }
 
@mcp.tool()
async def get_forecast(city: str, days: int = 5) -> list[dict]:
    """Returns the weather forecast for a city."""
    async with httpx.AsyncClient() as client:
        response = await client.get(
            "https://api.weather.example.com/forecast",
            params={"q": city, "cnt": days, "units": "metric"}
        )
        return response.json()["list"]
 
if __name__ == "__main__":
    mcp.run()

// npm install @modelcontextprotocol/sdk
import { Client } from "@modelcontextprotocol/sdk/client/index.js";
import { StdioClientTransport } from "@modelcontextprotocol/sdk/client/stdio.js";
 
const transport = new StdioClientTransport({
  command: "python",
  args: ["weather_server.py"]
});
 
const client = new Client({ name: "weather-agent", version: "1.0.0" });
await client.connect(transport);
 
// Dynamically query the available tool list — no hardcoding needed
const tools = await client.listTools();
console.log("Available tools:", tools.tools.map(t => t.name));
 
// Call a tool
const result = await client.callTool({
  name: "get_weather",
  arguments: { city: "Seoul" }
});
console.log("Weather info:", result.content);

{
  "name": "ResearchAgent",
  "version": "1.0.0",
  "description": "An agent responsible for web search and fact-checking",
  "url": "https://research-agent.example.com",
  "capabilities": {
    "streaming": true,
    "pushNotifications": true
  },
  "skills": [
    {
      "id": "fact-check",
      "name": "Fact Check",
      "description": "Verifies the factual accuracy of given information",
      "inputModes": ["text"],
      "outputModes": ["text", "data"]
    },
    {
      "id": "web-search",
      "name": "Web Search",
      "description": "Searches the web for relevant information",
      "inputModes": ["text"],
      "outputModes": ["text", "data"]
    }
  ],
  "authentication": {
    "schemes": ["bearer"]
  }
}

# pip install httpx
import httpx
import json
import uuid
from typing import AsyncIterator
 
async def delegate_task_to_agent(
    agent_url: str,
    task: str,
    context_id: str | None = None
) -> AsyncIterator[dict]:
    """Delegates a task to another agent using the A2A protocol."""
    # uuid4() is safer than hash() — hash() can return different values across runs
    task_id = str(uuid.uuid4())
 
    payload = {
        "jsonrpc": "2.0",
        "method": "tasks/send",
        "params": {
            "id": task_id,
            "contextId": context_id,
            "message": {
                "role": "user",
                "parts": [{"kind": "text", "text": task}]
            }
        },
        "id": 1
    }
 
    # In production, try-except and timeout configuration are essential
    async with httpx.AsyncClient(timeout=60.0) as client:
        async with client.stream(
            "POST",
            f"{agent_url}/a2a",
            json=payload,
            headers={
                "Accept": "text/event-stream",
                "Content-Type": "application/json"
            }
        ) as response:
            async for line in response.aiter_lines():
                if line.startswith("data: "):
                    event = json.loads(line[6:])
                    yield event

# crm_mcp_server.py
# pip install fastmcp
from fastmcp import FastMCP
 
mcp = FastMCP("crm-server")
 
# In a real project, replace the following with a CRM DB client
# (SQLite, PostgreSQL, or a CRM SaaS API, etc.)
class _MockDB:
    async def find_customer(self, customer_id):
        return None  # Replace with a real implementation
 
    async def vector_search(self, query, limit):
        return []
 
    async def create_ticket(self, **kwargs):
        return type("Ticket", (), {"id": "TICKET-001"})()
 
db = _MockDB()
 
@mcp.tool()
async def get_customer_info(customer_id: str) -> dict:
    """Looks up customer information by customer ID."""
    customer = await db.find_customer(customer_id)
    if not customer:
        return {"error": f"Customer {customer_id} not found"}
    return {
        "id": customer.id,
        "name": customer.name,
        "email": customer.email,
        "subscription_tier": customer.tier,
        "open_tickets": customer.open_ticket_count
    }
 
@mcp.tool()
async def search_knowledge_base(query: str, top_k: int = 3) -> list[dict]:
    """Searches the knowledge base for relevant documents."""
    results = await db.vector_search(query, limit=top_k)
    return [
        {
            "title": doc.title,
            "content": doc.content[:500],
            "relevance_score": doc.score
        }
        for doc in results
    ]
 
@mcp.tool()
async def create_ticket(
    customer_id: str,
    subject: str,
    description: str,
    priority: str = "medium"
) -> dict:
    """Creates a new support ticket."""
    ticket = await db.create_ticket(
        customer_id=customer_id,
        subject=subject,
        description=description,
        priority=priority
    )
    return {"ticket_id": ticket.id, "status": "created"}
 
if __name__ == "__main__":
    mcp.run()

# support_agent.py
# pip install anthropic mcp
import anthropic
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
 
async def run_support_agent(customer_id: str, user_message: str):
    server_params = StdioServerParameters(
        command="python",
        args=["crm_mcp_server.py"]
    )
 
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
 
            # Dynamically query the tool list from the MCP server — no hardcoding needed
            tools_result = await session.list_tools()
            tools = [
                {
                    "name": tool.name,
                    "description": tool.description,
                    "input_schema": tool.inputSchema
                }
                for tool in tools_result.tools
            ]
 
            client = anthropic.Anthropic()
            messages = [{"role": "user", "content": user_message}]
 
            # Agent loop: repeat tool calls until end_turn
            while True:
                response = client.messages.create(
                    model="claude-sonnet-4-6",
                    max_tokens=1024,
                    tools=tools,
                    messages=messages
                )
 
                if response.stop_reason == "end_turn":
                    text_content = next(
                        block for block in response.content
                        if block.type == "text"
                    )
                    return text_content.text
 
                tool_uses = [
                    block for block in response.content
                    if block.type == "tool_use"
                ]
 
                messages.append({
                    "role": "assistant",
                    "content": response.content
                })
 
                tool_results = []
                for tool_use in tool_uses:
                    result = await session.call_tool(
                        tool_use.name,
                        arguments=tool_use.input
                    )
                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": tool_use.id,
                        "content": str(result.content)
                    })
 
                messages.append({
                    "role": "user",
                    "content": tool_results
                })

# researcher_agent.py — A2A Server + MCP Client
# pip install fastapi uvicorn anthropic mcp
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
import anthropic
import json
 
app = FastAPI()
 
AGENT_CARD = {
    "name": "ResearchAgent",
    "version": "1.0.0",
    "description": "Handles news article fact-checking and in-depth research",
    "url": "http://localhost:8001",
    "capabilities": {"streaming": True, "pushNotifications": False},
    "skills": [
        {
            "id": "fact-check",
            "name": "Fact Check",
            "description": "Verifies claims against a database",
            "inputModes": ["text"],
            "outputModes": ["text"]
        }
    ]
}
 
@app.get("/.well-known/agent.json")
async def get_agent_card():
    """Agent Card endpoint — the entry point for other agents to discover capabilities"""
    return AGENT_CARD

@app.post("/a2a")
async def handle_a2a_request(request: Request):
    body = await request.json()
    task_text = body["params"]["message"]["parts"][0]["text"]
 
    async def generate_response():
        # — MCP client role: connect to the news DB MCP server —
        server_params = StdioServerParameters(
            command="python",
            args=["news_db_mcp_server.py"]
        )
 
        async with stdio_client(server_params) as (read, write):
            async with ClientSession(read, write) as session:
                await session.initialize()
                tools_result = await session.list_tools()
                tools = [
                    {
                        "name": t.name,
                        "description": t.description,
                        "input_schema": t.inputSchema
                    }
                    for t in tools_result.tools
                ]
 
                client = anthropic.Anthropic()
                messages = [{"role": "user", "content": task_text}]
 
                # Stream task start status to the Reporter
                yield f"data: {json.dumps({'status': 'working', 'message': 'Fact-check started'})}\n\n"
 
                while True:
                    response = client.messages.create(
                        model="claude-sonnet-4-6",
                        max_tokens=1024,
                        tools=tools,
                        messages=messages
                    )
 
                    if response.stop_reason == "end_turn":
                        result_text = next(
                            b.text for b in response.content
                            if b.type == "text"
                        )
                        # Stream the final result to the Reporter
                        yield f"data: {json.dumps({'status': 'completed', 'result': result_text})}\n\n"
                        break
 
                    # — Handle MCP tool calls —
                    tool_uses = [b for b in response.content if b.type == "tool_use"]
                    messages.append({"role": "assistant", "content": response.content})
 
                    tool_results = []
                    for tool_use in tool_uses:
                        yield f"data: {json.dumps({'status': 'working', 'message': f'Querying {tool_use.name}'})}\n\n"
                        result = await session.call_tool(tool_use.name, arguments=tool_use.input)
                        tool_results.append({
                            "type": "tool_result",
                            "tool_use_id": tool_use.id,
                            "content": str(result.content)
                        })
 
                    messages.append({"role": "user", "content": tool_results})
 
    return StreamingResponse(generate_response(), media_type="text/event-stream")

# reporter_agent.py
# pip install httpx
import httpx
import json
import uuid
 
async def write_article_with_fact_check(topic: str) -> str:
    """Delegates fact-checking needed during article writing to the Researcher."""
 
    # 1. Read the Researcher's Agent Card to confirm capabilities
    async with httpx.AsyncClient() as client:
        card_response = await client.get(
            "http://localhost:8001/.well-known/agent.json"
        )
        agent_card = card_response.json()
        print(f"Collaborating agent: {agent_card['name']} — {agent_card['description']}")
 
    # 2. Delegate the fact-check task via A2A
    fact_check_request = {
        "jsonrpc": "2.0",
        "method": "tasks/send",
        "params": {
            "id": str(uuid.uuid4()),  # uuid4 is safe without collisions
            "message": {
                "role": "user",
                "parts": [{
                    "kind": "text",
                    "text": f"Please verify the key facts on the following topic: {topic}"
                }]
            }
        },
        "id": 1
    }
 
    verified_facts = ""  # The result received from SSE is a string
    async with httpx.AsyncClient(timeout=60.0) as client:
        async with client.stream(
            "POST",
            "http://localhost:8001/a2a",
            json=fact_check_request,
            headers={"Accept": "text/event-stream"}
        ) as response:
            async for line in response.aiter_lines():
                if line.startswith("data: "):
                    event = json.loads(line[6:])
                    if event["status"] == "completed":
                        verified_facts = event["result"]
                    elif event["status"] == "working":
                        print(f"  In progress: {event['message']}")
 
    # 3. Write the article based on the verified facts
    return f"[Article Based on Verified Information]\n\n{verified_facts}"

User request
    │
    ▼
Reporter Agent (MCP Client + A2A Client)
    │
    ├── MCP ──► CRM MCP Server
    │               └── CRM Database
    │
    ├── MCP ──► KnowledgeBase MCP Server
    │               └── Vector DB
    │
    └── A2A ──► Researcher Agent (A2A Server + MCP Client)
                    │
                    └── MCP ──► NewsDB MCP Server
                                    └── Elasticsearch