How the Agent2Agent (A2A) protocol enables seamless AI agent collaboration

cd crewai

# Pin the Python version to 3.12 for this project
uv python pin 3.12

# Create and activate a virtual environment
uv venv
source .venv/bin/activate

# Start the CrewAI agent on port 1000
uv run . --host 0.0.0.0 --port 1000

cd A2A/samples/python/agents/langgraph

uv python pin 3.12

uv venv
source .venv/bin/activate

# Start the LangGraph agent on port 1001
uv run . --host 0.0.0.0 --port 1001

import asyncio
import logging
import os
import uuid
import base64
from typing import Optional, List, Dict

import httpx
from litellm import completion
import weave

weave.init("multi-agent-router")

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

AGENT_CARD_PATH = "/.well-known/agent.json"
HOST = "localhost"
START_PORT = 1000
END_PORT = 1010
MODEL_NAME = "gemini/gemini-2.0-flash"

@weave.op
def create_routing_prompt(history: List[Dict], agents: List[Dict]) -> List[Dict]:
    agent_list = "\n".join(
        f"- {a['card'].get('name', 'Unknown')}: {a['card'].get('description', '')}" for a in agents
    )
    convo = []
    for msg in history:
        role = "User" if msg["role"] == "user" else "Agent"
        convo.append(f"{role}: {msg['text']}")
    convo_str = "\n".join(convo)
    content = (
        "You are an expert router for user requests. "
        "Given the following agents:\n"
        f"{agent_list}\n\n"
        "Below is the full conversation so far. "
        "Pay closest attention to the user's most recent request at the end. "
        "Choose the single best matching agent by exact name to handle the user's request, IN CONTEXT of the conversation.\n\n"
        f"Conversation:\n{convo_str}\n\nAgent name:"
    )
    return [{"role": "user", "content": content}]

@weave.op
def make_agent_instruction(history: List[Dict], agent_card: Dict) -> str:
    convo = "\n".join(
        f"User: {msg['text']}" if msg["role"] == "user" else f"Agent: {msg['text']}"
        for msg in history
    )
    agent_desc = agent_card.get('description', 'No agent description provided.')
    agent_name = agent_card.get('name', '[Agent name missing]')
    return (
        f"You are the agent: {agent_name}.\n"
        f"Your job: {agent_desc}\n\n"
        "Conversation so far:\n"
        f"{convo}\n\n"
        "Based on this conversation, take the user's intended action or answer their last request, using all necessary info from the dialog.\n"
        "Return only the task/instruction that the agent will accept ."
    )

async def fetch_agent_card(host: str, port: int) -> Optional[Dict]:
    url = f"http://{host}:{port}{AGENT_CARD_PATH}"
    try:
        async with httpx.AsyncClient(timeout=3) as client:
            resp = await client.get(url)
            resp.raise_for_status()
            card = resp.json()
            logger.info(f"Agent found: {card.get('name', '[no-name]')} on {host}:{port}")
            return {"host": host, "port": port, "card": card}
    except Exception as e:
        logger.debug(f"No agent at {host}:{port} ({e})")
        return None

async def discover_agents(host: str, start_port: int, end_port: int) -> List[Dict]:
    tasks = [fetch_agent_card(host, port) for port in range(start_port, end_port + 1)]
    results = await asyncio.gather(*tasks)
    return [r for r in results if r]

@weave.op 
async def send_instruction_to_agent(agent: Dict, instruction: str) -> Optional[Dict]:
    url = f"http://{agent['host']}:{agent['port']}/"
    jsonrpc_id = str(uuid.uuid4())
    payload = {
        "jsonrpc": "2.0",
        "id": jsonrpc_id,
        "method": "tasks/send",
        "params": {
            "id": jsonrpc_id,
            "message": {
                "role": "user",
                "parts": [
                    {"type": "text", "text": instruction}
                ]
            },
            "acceptedOutputModes": ["text/plain", "image/png", "image/jpeg"],
        }
    }
    try:
        async with httpx.AsyncClient(timeout=30) as client:
            resp = await client.post(url, json=payload)
            resp.raise_for_status()
            data = resp.json()
            return data
    except Exception as e:
        logger.error(f"Error sending query to agent: {e}")
        return None

def handle_agent_response(response_json: Dict):
    result = response_json.get("result", {})
    artifacts = result.get("artifacts", [])
    if not artifacts:
        logger.warning("No artifacts in agent response.")
        print("Agent response has no content.")
        return

    for artifact in artifacts:
        parts = artifact.get("parts", [])
        for idx, part in enumerate(parts):
            ptype = part.get("type")
            if ptype == "text":
                text = part.get("text", "")
                print(f"\n[Agent Text]:\n{text}")
            elif ptype in ("image/png", "image/jpeg", "image/jpg"):
                b64data = part.get("text") or part.get("bytes") or ""
                if not b64data:
                    logger.warning(f"Image artifact part missing data")
                    continue
                try:
                    image_bytes = base64.b64decode(b64data)
                    ext = "png" if "png" in ptype else "jpg"
                    filename = f"agent_image_{uuid.uuid4().hex}.{ext}"
                    with open(filename, "wb") as f:
                        f.write(image_bytes)
                    print(f"\n[Agent Image]: saved to file '{filename}'")
                except Exception as e:
                    logger.error(f"Failed to decode or save image: {e}")
            elif ptype == "file":
                file_info = part.get("file", {})
                b64data = file_info.get("bytes")
                mime = file_info.get("mimeType", "")
                if not b64data:
                    logger.warning(f"'file' artifact missing 'bytes' data")
                    continue
                try:
                    image_bytes = base64.b64decode(b64data)
                    ext = "bin"
                    if "png" in mime:
                        ext = "png"
                    elif "jpeg" in mime or "jpg" in mime:
                        ext = "jpg"
                    filename = f"agent_file_{uuid.uuid4().hex}.{ext}"
                    with open(filename, "wb") as f:
                        f.write(image_bytes)
                    print(f"\n[Agent File]: saved to file '{filename}' (mime-type: {mime})")
                except Exception as e:
                    logger.error(f"Failed to decode or save file artifact: {e}")
            else:
                logger.info(f"Unknown artifact type: {ptype}")

async def main():
    api_key = os.getenv("GEMINI_API_KEY")
    if not api_key:
        logger.error("Please set GEMINI_API_KEY environment variable.")
        return

    print(f"Discovering agents on {HOST} ports {START_PORT}-{END_PORT} ...")
    agents = await discover_agents(HOST, START_PORT, END_PORT)
    if not agents:
        logger.error("No agents discovered.")
        return

    print(f"Discovered {len(agents)} agents:")
    for agent in agents:
        print(f" - {agent['card'].get('name')} @ {agent['host']}:{agent['port']}")

    history = []
    query = input("\nEnter your query: ").strip()
    history.append({"role": "user", "text": query})

    print("\nType your questions to chat. Type 'exit' or 'quit' to end the session.\n")

    while True:
        prompt_messages = create_routing_prompt(history, agents)
        response = completion(
            model=MODEL_NAME,
            messages=prompt_messages,
            max_tokens=16,
            temperature=0.0,
        )
        agent_name = response.get("choices", [{}])[0].get("message", {}).get("content", "").strip()
        matched_agent = None
        for agent in agents:
            if agent["card"].get("name", "").lower() == agent_name.lower():
                matched_agent = agent
                break
        if not matched_agent:
            print("No matching agent found for the name, defaulting to first agent.")
            matched_agent = agents[0]
        print(f"\nQuery routed to agent: {matched_agent['card'].get('name')} at {matched_agent['host']}:{matched_agent['port']}")

        agent_instruction = make_agent_instruction(history, matched_agent["card"])
        agent_response = await send_instruction_to_agent(matched_agent, agent_instruction)

        if not agent_response:
            print("Agent did not return a valid response.")
            user_query = input("You: ").strip()
            if user_query.lower() in {"exit", "quit"}:
                print("Exiting chat.")
                break
            history.append({"role": "user", "text": user_query})
            continue

        result = agent_response.get("result", {})
        state = result.get("status", {}).get("state", "")

        if state == "input-required":
            print("* need more input from user *", flush=True)
            msg = result.get("status", {}).get("message", {})
            parts = msg.get("parts", [])
            agent_utterance = ""
            for part in parts:
                if part.get("type") == "text":
                    agent_utterance = part.get("text")
                    print("Agent:", agent_utterance)
            if agent_utterance:
                history.append({"role": "agent", "text": agent_utterance})
            user_query = input("You: ").strip()
            if user_query.lower() in {"exit", "quit"}:
                print("Exiting chat.")
                break
            history.append({"role": "user", "text": user_query})
            continue

        handle_agent_response(agent_response)
        artifacts = result.get("artifacts", [])
        if artifacts:
            texts = []
            for artifact in artifacts:
                for part in artifact.get("parts", []):
                    if part.get("type") == "text":
                        texts.append(part.get("text", ""))
            if texts:
                full_agent_text = "\n".join(texts)
                history.append({"role": "agent", "text": full_agent_text})

        user_query = input("You: ").strip()
        if user_query.lower() in {"exit", "quit"}:
            print("Exiting chat.")
            break
        history.append({"role": "user", "text": user_query})

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