Agent-To-Agent Protocol

Comprehensive training notes covering the A2A protocol from architecture and core concepts through multi-agent orchestration, security, testing, and production deployment.

Table of Contents

• Module 1: A2A Protocol Architecture and Core Concepts
  • Understanding the Problem A2A Solves
  • Exploring Agent Cards and Capability Discovery
  • Walking Through Tasks, Messages, and the A2A Lifecycle
• Module 2: Building A2A Agents Across Frameworks
  • Building an A2A Agent with OpenAI Agents SDK
  • Building an A2A Agent with LangGraph
  • Building an A2A Agent with CrewAI
  • Discovering and Calling All Three Agents with an A2A Client
• Module 3: Orchestration Patterns and Multi-Agent Workflows
  • Implementing the Orchestrator-Worker Pattern
  • Implementing Dynamic Routing and Hierarchical Delegation
  • Handling Failures and Retries
• Module 4: Security, Testing, and Production Deployment
  • Securing Authentication, Authorization, and Trust
  • Testing and Debugging with A2A Inspector
  • Deploying to Production with Agent Stack
• Reference: Key Concepts Cheat Sheet

Module 1: A2A Protocol Architecture and Core Concepts

Understanding the Problem A2A Solves

The Integration Problem in Multi-Agent Systems

Modern AI applications increasingly rely on multiple specialized agents working together. Consider a document processing system as a representative example:

• A contract analyzer built with LangGraph
• A compliance checker built with CrewAI
• A triage agent built with OpenAI’s Agents SDK that decides where incoming documents should go

Each agent works perfectly on its own. The hard problem is getting them to work together.

Without a shared standard, building inter-agent communication means:

1. Custom request-and-response formats — each agent pair needs its own schema.
2. Hard-coded URLs — every agent must know exactly where every other agent lives.
3. Exponential integration points — three agents produce six possible integration paths, every one requiring custom code to write and maintain.
4. Fragile coupling — change one agent’s response format and every caller breaks.
5. No discovery — adding a fourth or fifth agent next week requires updating every other agent that might call it.

This approach simply does not scale.

How A2A Solves the Problem

The Agent-To-Agent (A2A) protocol replaces all that custom glue with a single shared standard:

• Agent cards — each agent publishes a JSON document describing what it can do.
• Unified protocol — every agent speaks the same protocol for sending tasks and receiving results.
• Runtime discovery — any agent can discover any other agent at runtime, without hard-coded URLs or custom payloads.

Wrap each agent once. After that, your LangGraph agent, your CrewAI agent, and your OpenAI Agents SDK agent all communicate through the same protocol. Add a fourth or fifth agent later and the existing agents find them automatically.

graph TD
    subgraph "Without A2A"
        U1[User] --> T1[Triage Agent\nOpenAI SDK]
        T1 -->|custom code| C1[Contract Analyzer\nLangGraph]
        T1 -->|custom code| CC1[Compliance Checker\nCrewAI]
        C1 -->|custom code| CC1
    end

    subgraph "With A2A"
        U2[User] --> Client[A2A Client Agent]
        Client -->|A2A Protocol| T2[Triage Agent\nOpenAI SDK :9999]
        Client -->|A2A Protocol| C2[Contract Analyzer\nLangGraph :9998]
        Client -->|A2A Protocol| CC2[Compliance Checker\nCrewAI :9997]
    end

A2A vs MCP: Two Complementary Protocols

A2A works alongside the Model Context Protocol (MCP), not in place of it. The distinction is directional:

Production systems commonly use both. MCP handles what an agent does (tool invocation), while A2A handles how agents collaborate with each other.

graph LR
    subgraph "Agent Layer (A2A - Horizontal)"
        A1[Agent A] <-->|A2A| A2[Agent B]
        A2 <-->|A2A| A3[Agent C]
    end

    subgraph "Tool Layer (MCP - Vertical)"
        A1 -->|MCP| T1[(Database)]
        A1 -->|MCP| T2[File System]
        A2 -->|MCP| T3[REST API]
        A3 -->|MCP| T4[Search Index]
    end

Protocol History and Industry Backing

• April 2025 — A2A was launched by Google.
• June 2025 — Donated to the Linux Foundation.
• Mid 2025 — IBM’s own Agent Communication Protocol (ACP) merged into A2A, consolidating the industry around a single standard.
• 150+ organizations backed the protocol, including AWS, Microsoft, Salesforce, SAP, and IBM.

This matters because A2A has broad industry commitment and open governance. You are not locking into a single vendor’s ecosystem.

Course Overview: What You Will Build

Across the four modules, the course builds a complete multi-agent pipeline:

1. Three specialized agents, each on a different framework, each wrapped as an A2A-compliant server.
2. An orchestrator that routes tasks to the right agent and coordinates multi-agent workflows.
3. Security, observability, and production deployment of the entire system.

By the end, you have a working multi-agent pipeline where agents built on completely different stacks collaborate through a shared protocol.

Exploring Agent Cards and Capability Discovery

What Is an Agent Card?

Before one agent can delegate a task to another, it needs to know three things:

1. Who is this agent?
2. What can it do?
3. How do I talk to it?

The agent card answers all three. It is a JSON document that every A2A agent publishes at a well-known URL. Any client agent can fetch this card, read the capabilities, and decide whether this agent is the right one for the job.

Agent Card Structure

Below is an example agent card for a Contract Review Agent:

{
  "capabilities": {
    "streaming": false
  },
  "defaultInputModes": ["text/plain"],
  "defaultOutputModes": ["text/plain"],
  "description": "Analyzes contract text for risk clauses",
  "name": "ContractReviewAgent",
  "preferredTransport": "JSONRPC",
  "protocolVersion": "0.3.0",
  "skills": [
    {
      "name": "contract_review",
      "description": "Analyzes contract text for risk clauses, including non-compete, indemnification, termination, auto-renewal, and liability.",
      "examples": [
        "Review this NDA for non-compete clauses",
        "Analyze the following vendor agreement for liability risk"
      ]
    }
  ],
  "url": "http://localhost:9998",
  "version": "1.0.0"
}

Key fields explained:

The Discovery Flow

sequenceDiagram
    participant Client as Client Agent
    participant Registry as Agent Registry
    participant Remote as Remote Agent

    Client->>Registry: Where are the agents?
    Registry-->>Client: List of agent card URLs
    Client->>Remote: GET /.well-known/agent.json
    Remote-->>Client: Agent Card (JSON)
    Client->>Client: Read skills array
    Client->>Client: Check input/output modes
    Client->>Client: Check streaming support
    Client->>Remote: POST /tasks (if match found)

The discovery flow is:

1. A client agent fetches the agent card from the well-known endpoint (/.well-known/agent.json).
2. It reads the skills array and checks whether any listed capability matches the task at hand.
3. It checks defaultInputModes — for example, if you are sending a PDF but the agent only accepts plain text, you convert the document or pick a different agent.
4. It checks streaming support — if streaming is available, the client can subscribe to progress updates as the task runs; otherwise it waits for the final result.
5. If a match is found, it sends a task to that agent’s URL.

This happens at runtime. The client does not need to know at build time which agents exist or where they live — it only needs to know where to find agent cards. In a production system, a registry aggregates cards from all deployed agents.

Viewing a Live Agent Card

Fetching a running agent card from the command line:

curl http://localhost:9999/.well-known/agent.json | python -m json.tool

Example output:

{
  "capabilities": {
    "streaming": false
  },
  "defaultInputModes": ["text/plain"],
  "defaultOutputModes": ["text/plain"],
  "description": "Classifies documents into contract, invoice, or compliance categories",
  "name": "DocumentTriageAgent",
  "preferredTransport": "JSONRPC",
  "protocolVersion": "0.3.0",
  "skills": [
    {
      "description": "Classifies a document description into one of three categories: contract, invoice, or compliance",
      "examples": [
        "This is a vendor services agreement with payment terms and an SLA"
      ],
      "id": "document_triage",
      "name": "Document Triage",
      "tags": ["document", "classification", "triage"]
    }
  ],
  "url": "http://localhost:9999",
  "version": "1.0.0"
}

Walking Through Tasks, Messages, and the A2A Lifecycle

Tasks: The Unit of Work in A2A

Once an agent finds a peer through its agent card, the next step is sending a task. A task is the fundamental unit of work in A2A and has the following structure:

{
  "id": "task-uuid-1234",
  "contextId": "session-uuid-5678",
  "status": "working",
  "artifacts": [],
  "history": [
    {
      "role": "user",
      "messageId": "msg-001",
      "parts": [
        {
          "type": "text",
          "text": "Review this contract for non-compete clauses."
        }
      ]
    }
  ]
}

Task Lifecycle States

Every task moves through a defined set of states:

stateDiagram-v2
    [*] --> submitted
    submitted --> working : Agent picks up task
    working --> completed : Happy path
    working --> input_required : Agent needs more info
    input_required --> working : Client provides context
    working --> failed : Error occurs
    submitted --> canceled : Either side cancels
    working --> canceled : Either side cancels
    completed --> [*]
    failed --> [*]
    canceled --> [*]

The input_required state is what makes A2A fundamentally different from a simple API call. A tool call is fire-and-forget — you send a request and get a response. An A2A task is a conversation: the remote agent is autonomous, it may reason about the request, break it into sub-steps, ask for clarification, and stream partial results.

Messages and Parts

Messages are how context flows between agents. Each message has:

• A role: user or agent
• A messageId: unique identifier
• An array of parts: the building blocks of content

A part can be:

• Plain text — for natural language content
• A file — for binary or document content
• Structured data — for JSON payloads

This means your contract review agent might return a text summary in one part and a structured JSON risk assessment in another part. These messages accumulate over the life of a task.

{
  "role": "agent",
  "messageId": "msg-002",
  "parts": [
    {
      "type": "text",
      "text": "Overall risk rating: MEDIUM"
    },
    {
      "type": "data",
      "data": {
        "overallRisk": "MEDIUM",
        "flaggedClauses": [
          { "section": "4.2", "type": "non-compete", "severity": "HIGH" },
          { "section": "7.1", "type": "indemnification", "severity": "MEDIUM" }
        ]
      }
    }
  ]
}

Streaming and Artifacts

Streaming lets agents send progress updates as a task runs. If an agent declares "streaming": true in its agent card capabilities, clients can subscribe to a stream of events rather than waiting for the final response. This is valuable for long-running tasks.

Artifacts are the final, structured output of a completed task. They differ from messages in that they represent the definitive result — not an intermediate step.

Tool Calling (MCP) vs. Agent Delegation (A2A)

This is a frequently asked question. The distinction is fundamental:

You will use MCP tools inside your agents and A2A between your agents. Knowing when to reach for which protocol is a key skill for designing multi-agent systems.

Module 2: Building A2A Agents Across Frameworks

Overview: Three Agents, Three Frameworks, One Protocol

The three remote agents built in this module:

graph TD
    User -->|request| Client["Client Agent\n(A2A Server + Client)\nPort 9996"]
    Client -->|A2A| Triage["Document Triage Agent\n(OpenAI Agents SDK)\nPort 9999"]
    Client -->|A2A| Contract["Contract Review Agent\n(LangGraph)\nPort 9998"]
    Client -->|A2A| Compliance["Compliance Checker Agent\n(CrewAI)\nPort 9997"]

Every agent exposes its capabilities through its agent card. The client agent discovers the remote agents at startup by fetching their agent cards, then routes incoming requests to the right specialist.

Building an A2A Agent with OpenAI Agents SDK

Architecture: Two-File Separation

The triage agent follows a clean separation of concerns:

1. triage_agent.py — Pure agent logic, no A2A code.
2. a2a_triage_agent.py — A2A wrapper (executor, agent card, server startup).

This separation matters: the protocol layer does not care what framework powers the agent.

File 1: Agent Logic (triage_agent.py)

from agents import Agent, Runner  # OpenAI Agents SDK

agent = Agent(
    name="DocumentTriageAgent",
    model="gpt-4.1-nano",
    instructions="""
        You are a document classification agent.
        Given a description of a document, classify it into exactly one
        of these categories: contract, invoice, or compliance.
        Respond with only the category name in lowercase. No explanation.
    """
)

async def classify(document_description: str) -> str:
    """Classify a document description into one of three categories."""
    result = await Runner.run(agent, document_description)
    return result.final_output

Key points:

• Uses gpt-4.1-nano for efficiency
• Instructions are tightly scoped: one of three outputs, no explanation
• The classify function is the only public interface — pure agent logic

File 2: A2A Wrapper (a2a_triage_agent.py)

from a2a.server import AgentCard, AgentSkill, DefaultRequestHandler, A2AServer
from a2a.server.agent_execution import AgentExecutor, RequestContext
from a2a.server.events import EventQueue
from triage_agent import classify  # Import the agent logic

class TriageAgentExecutor(AgentExecutor):
    """A2A executor that bridges the protocol layer to the agent logic."""

    async def execute(
        self,
        context: RequestContext,
        event_queue: EventQueue,
    ) -> None:
        # Pull the user's text from the A2A context
        user_input = context.get_user_input()

        # Delegate to the agent logic
        result = await classify(user_input)

        # Send the result back through the event queue
        await event_queue.enqueue_event(
            new_agent_text_message(result)
        )

def main():
    skill = AgentSkill(
        id="document_triage",
        name="Document Triage",
        description="Classifies a document description into one of three categories: contract, invoice, or compliance.",
        tags=["document", "classification", "triage"],
        examples=["This is a vendor services agreement with payment terms and an SLA."]
    )

    agent_card = AgentCard(
        name="DocumentTriageAgent",
        description="Routes incoming documents to the right specialist agent.",
        url="http://localhost:9999",
        defaultInputModes=["text/plain"],
        defaultOutputModes=["text/plain"],
        capabilities=AgentCapabilities(streaming=False),
        skills=[skill],
        version="1.0.0"
    )

    request_handler = DefaultRequestHandler(
        agent_executor=TriageAgentExecutor(),
        task_store=InMemoryTaskStore(),
    )

    server = A2AServer(
        agent_card=agent_card,
        http_handler=request_handler,
        host="0.0.0.0",
        port=9999,
    )

    import uvicorn
    uvicorn.run(server.build(), host="0.0.0.0", port=9999)

if __name__ == "__main__":
    main()

Key components:

Verifying the Agent Card

# Start the server first, then in a separate terminal:
curl http://localhost:9999/.well-known/agent.json | python -m json.tool

Building an A2A Agent with LangGraph

Architecture: LangGraph A2AServer Bridge

LangGraph provides a dedicated bridge package that eliminates the need to write a custom executor class. You pass a compiled graph and an agent card to the A2AServer, and it handles the executor layer internally.

Trade-off: Less control over the execute flow, but much faster setup.

File 1: LangGraph Agent Logic (contract_review_graph.py)

from langgraph.graph import StateGraph, START, END
from langchain_openai import ChatOpenAI
from typing import TypedDict

llm = ChatOpenAI(model="gpt-4.1-nano")

class ContractState(TypedDict):
    contract_text: str
    clauses: list[str]
    risk_assessment: dict

def extract_clauses(state: ContractState) -> ContractState:
    """Node 1: Extract key clauses from contract text."""
    response = llm.invoke(
        f"Extract the key clauses from this contract:\n{state['contract_text']}"
    )
    state["clauses"] = response.content
    return state

def assess_risk(state: ContractState) -> ContractState:
    """Node 2: Assess risk and assign severity rating."""
    response = llm.invoke(
        f"Assess the risk of these contract clauses and assign a severity rating:\n{state['clauses']}"
    )
    state["risk_assessment"] = response.content
    return state

# Build the graph
builder = StateGraph(ContractState)
builder.add_node("extract_clauses", extract_clauses)
builder.add_node("assess_risk", assess_risk)
builder.add_edge(START, "extract_clauses")
builder.add_edge("extract_clauses", "assess_risk")
builder.add_edge("assess_risk", END)

contract_review_graph = builder.compile()

The graph has two nodes:

• extract_clauses — pulls out the key clauses from the contract text
• assess_risk — assesses risk and assigns a severity rating

In production you might have more nodes. Two nodes are sufficient to demonstrate graph flow through the A2A protocol.

graph LR
    START --> extract_clauses
    extract_clauses --> assess_risk
    assess_risk --> END

File 2: A2A Wrapper (a2a_contract_review_agent.py)

from langgraph_a2a import A2AServer  # LangGraph bridge package
from contract_review_graph import contract_review_graph

skill = AgentSkill(
    id="contract_review",
    name="Contract Review",
    description=(
        "Analyzes contract text for risk clauses, extracts key terms, "
        "flags risk areas, and returns a structured risk assessment with "
        "a severity rating and recommendations."
    ),
    examples=["Review this vendor agreement for liability risk"]
)

agent_card = AgentCard(
    name="ContractReviewAgent",
    description="LangGraph-powered contract risk analysis agent.",
    url="http://localhost:9998",
    defaultInputModes=["text/plain"],
    defaultOutputModes=["text/plain"],
    capabilities=AgentCapabilities(streaming=False),
    skills=[skill],
    version="1.0.0"
)

# No custom executor needed — the bridge handles it
server = A2AServer(
    graph=contract_review_graph,
    agent_card=agent_card,
    host="0.0.0.0",
    port=9998
)

server.serve()

Notice what is not here: no AgentExecutor subclass, no RequestContext, no EventQueue. The langgraph_a2a package handles all of that. You just give it a compiled graph and an agent card.

Building an A2A Agent with CrewAI

Architecture: Single-File Agent

The compliance checker combines agent logic and A2A wrapper in one file. This is practical when the agent logic is short enough to read comfortably inline — not a requirement, just a style choice.

The CrewAI Role-Based Design

CrewAI has a distinctive role-based design. To define an agent you need:

1. Role — what this agent is
2. Goal — what this agent does
3. Backstory — the agent’s persona, expertise, and context (guides behavior, tone, and decision-making)

from crewai import Agent, Crew, Task
from a2a.server import AgentCard, AgentSkill, DefaultRequestHandler, A2AServer
from a2a.server.agent_execution import AgentExecutor, RequestContext
from a2a.server.events import EventQueue
import asyncio

# Define the CrewAI agent
compliance_agent = Agent(
    role="Compliance Analyst",
    goal="Evaluate documents against regulatory and policy requirements.",
    backstory=(
        "You are a senior compliance analyst with 15 years of experience "
        "in financial services regulation. You review documents for adherence "
        "to internal policies, industry standards, and government regulations. "
        "You flag non-compliant items and recommend corrective actions."
    ),
    verbose=False
)

def run_compliance_check(document_text: str) -> str:
    """Run the full CrewAI compliance workflow."""
    task = Task(
        description=(
            f"Review the following document for compliance issues. "
            f"Check for regulatory adherence, policy violations, and missing "
            f"required disclosures. Provide a compliance status "
            f"(compliant, non-compliant, or needs-review) and list any findings.\n\n"
            f"{document_text}"
        ),
        agent=compliance_agent,
        expected_output="Compliance status and list of findings"
    )

    crew = Crew(
        agents=[compliance_agent],
        tasks=[task],
        verbose=False
    )

    result = crew.kickoff()
    return result.raw

class ComplianceAgentExecutor(AgentExecutor):
    """A2A executor wrapping the synchronous CrewAI workflow."""

    async def execute(
        self,
        context: RequestContext,
        event_queue: EventQueue,
    ) -> None:
        user_input = context.get_user_input()

        # CrewAI's kickoff() is synchronous, so wrap it in a thread
        # to avoid blocking the async event loop
        result = await asyncio.to_thread(run_compliance_check, user_input)

        await event_queue.enqueue_event(
            new_agent_text_message(result)
        )

The Async Wrapping Pattern

This is a critical pattern for integrating synchronous agent frameworks with A2A:

# WRONG — blocks the async event loop
result = run_compliance_check(user_input)  # synchronous call in async context

# CORRECT — offloads to a thread pool
result = await asyncio.to_thread(run_compliance_check, user_input)

CrewAI’s kickoff() function is synchronous. The A2A executor is async. Always wrap synchronous agent framework calls in asyncio.to_thread() to keep the event loop from blocking.

Server Startup (Port 9997)

def main():
    skill = AgentSkill(
        id="compliance_check",
        name="Compliance Check",
        description=(
            "Reviews documents for regulatory adherence, policy violations, "
            "and missing required disclosures. Returns a compliance status "
            "and list of findings."
        ),
        tags=["compliance", "regulatory", "gdpr", "policy"],
        examples=["Check this vendor contract for GDPR compliance."]
    )

    agent_card = AgentCard(
        name="ComplianceCheckerAgent",
        description="CrewAI-powered compliance analysis agent.",
        url="http://localhost:9997",
        defaultInputModes=["text/plain"],
        defaultOutputModes=["text/plain"],
        capabilities=AgentCapabilities(streaming=False),
        skills=[skill],
        version="1.0.0"
    )

    server = A2AServer(
        agent_card=agent_card,
        http_handler=DefaultRequestHandler(
            agent_executor=ComplianceAgentExecutor(),
            task_store=InMemoryTaskStore(),
        ),
        host="0.0.0.0",
        port=9997,
    )

    import uvicorn
    uvicorn.run(server.build(), host="0.0.0.0", port=9997)

At this point, three agents are running on three different ports, using three different frameworks, but all speaking the same A2A protocol.

Discovering and Calling All Three Agents with an A2A Client

The Client Agent’s Dual Role

The client agent is simultaneously:

• A server to the user (has its own agent card, runs on port 9996)
• A client to the three remote agents (discovers and delegates to them)

The user does not know about the remote agents. They only know about the client agent. Routing happens behind the scenes.

graph TD
    User -->|sends request| Client["Client Agent\nPort 9996\n(A2A Server + A2A Client)"]
    Client -->|"1. Fetch agent card"| Triage["DocumentTriageAgent\n:9999"]
    Client -->|"1. Fetch agent card"| Contract["ContractReviewAgent\n:9998"]
    Client -->|"1. Fetch agent card"| Compliance["ComplianceCheckerAgent\n:9997"]
    Client -->|"2. Route & delegate task"| Triage
    Client -->|"2. Route & delegate task"| Contract
    Client -->|"2. Route & delegate task"| Compliance
    Triage -->|response| Client
    Contract -->|response| Client
    Compliance -->|response| Client
    Client -->|combined result| User

File 1: Client Agent (a2a_client_agent.py)

Key functions explained:

Agent Discovery at Startup

REMOTE_AGENT_REGISTRY = [
    "http://localhost:9999",  # Triage
    "http://localhost:9998",  # Contract Review
    "http://localhost:9997",  # Compliance Checker
]

discovered_agents = {}

async def discover_remote_agents():
    """Fetch agent cards from all remote agents at startup."""
    async with httpx.AsyncClient() as client:
        for url in REMOTE_AGENT_REGISTRY:
            card_url = f"{url}/.well-known/agent.json"
            response = await client.get(card_url)
            card = response.json()
            discovered_agents[url] = card
            print(f"Discovered: {card['name']}")

When the client agent starts, it reaches out to every remote agent and reads their agent card. Now it knows who is available and what each one can do.

LLM-Based Routing

def build_routing_prompt(user_request: str) -> str:
    """Build a prompt that lets an LLM pick the best agent."""
    agent_list = "\n".join([
        f"{i}. {card['name']}: {card['skills'][0]['description']}"
        for i, (url, card) in enumerate(discovered_agents.items())
    ])
    return (
        f"Available agents:\n{agent_list}\n\n"
        f"User request: {user_request}\n\n"
        "Which agent number (0-indexed) should handle this request? "
        "Respond with only the number."
    )

The routing decision uses the skill descriptions from the agent cards. The LLM reads those descriptions alongside the user’s request and picks the best match by returning a number.

Task Delegation

async def execute(self, context: RequestContext, event_queue: EventQueue) -> None:
    user_input = context.get_user_input()

    # Update task status to working
    await event_queue.enqueue_event(
        new_task_status_update(TaskState.working, "Routing your request...")
    )

    # Determine which agent should handle this
    routing_decision = await route_request(user_input)
    target_url, target_card = list(discovered_agents.items())[routing_decision]

    await event_queue.enqueue_event(
        new_task_status_update(
            TaskState.working,
            f"Delegating to {target_card['name']}..."
        )
    )

    # Send task to the selected remote agent via A2A
    async with httpx.AsyncClient() as http_client:
        a2a_client = A2AClient(http_client=http_client, url=target_url)
        response = await a2a_client.send_message(
            message=new_user_text_message(user_input)
        )

    # Collect and return the result
    result = extract_text_from_response(response)

    await event_queue.enqueue_event(
        new_agent_artifact(result)
    )
    await event_queue.enqueue_event(
        new_task_status_update(TaskState.completed)
    )

File 2: User Script (user.py)

import httpx
from a2a.client import A2AClient

async def main():
    async with httpx.AsyncClient() as http_client:
        # Connect only to the client agent — the user doesn't know about remote agents
        client = A2AClient(http_client=http_client, url="http://localhost:9996")

        requests = [
            # Request 1: Classification → routed to DocumentTriageAgent
            """Classify this document: a vendor services agreement with payment terms,
               an SLA, and a non-compete clause.""",

            # Request 2: Contract risk → routed to ContractReviewAgent
            """Review this contract for risk:
               Section 4.2 Non-Compete: the contractor shall not engage in any
               competing business for 24 months after termination.
               Section 7.1 Indemnification: the contractor agrees to indemnify
               the company from any claims arising from performance of services.""",

            # Request 3: GDPR compliance → routed to ComplianceCheckerAgent
            """Check this for GDPR compliance:
               The vendor will process personal data of EU residents, including
               names and emails. Data will be stored on US-based servers.
               No data protection officer is specified."""
        ]

        for req in requests:
            response = await client.send_message(
                message=new_user_text_message(req)
            )
            print(response)

This user script does not import OpenAI, LangGraph, or CrewAI. It does not even know those frameworks exist. It connects to one endpoint, sends requests, and gets results.

Sample Output

Request 1 (Classification):
  → Routed to: DocumentTriageAgent
  → Result: contract

Request 2 (Contract Risk):
  → Routed to: ContractReviewAgent
  → Result:
    Overall Risk: MEDIUM
    Flagged Clauses:
      - Section 4.2 (Non-Compete): HIGH — contractor shall not engage in
        competing business for 24 months after termination.
        Recommendation: Narrow the non-compete to specific activities.
      - Section 7.1 (Indemnification): MEDIUM — contractor indemnifies company
        from any claims.

Request 3 (GDPR Compliance):
  → Routed to: ComplianceCheckerAgent
  → Result:
    Compliance Status: NON-COMPLIANT
    Findings:
      - Missing required GDPR roles: controller/processor clarity not established.
        The document states the vendor will process personal data but does not
        clarify whether the vendor acts as a processor or a joint controller.

Behind the scenes: three frameworks, one protocol. The user never needed to know about any of that.

Module 3: Orchestration Patterns and Multi-Agent Workflows

Implementing the Orchestrator-Worker Pattern

Why the Simple Client Is Not Enough

In Module 2, the client agent picked one remote agent for each request. That works for simple queries, but real-world document processing often requires multiple agents working on the same request.

A user might say: “Review this contract and check it for regulatory compliance.” That is two jobs:

1. The contract review agent handles the first task.
2. The compliance agent handles the second task.

The orchestrator must break the request apart, send each piece to the right agent, and combine the results into a single response.

The Three Steps of the Orchestrator-Worker Pattern

flowchart TD
    A[User Request] --> B[Step 1: Decompose\nLLM breaks request into subtasks]
    B --> C{Subtask 1}
    B --> D{Subtask 2}
    B --> E{Subtask N}
    C --> F[Step 2: Delegate\nSend each subtask to the right agent]
    D --> F
    E --> F
    F --> G[Agent 1 Result]
    F --> H[Agent 2 Result]
    F --> I[Agent N Result]
    G --> J[Step 3: Aggregate\nCombine results with labels]
    H --> J
    I --> J
    J --> K[Combined Response to User]

1. Decompose the request into subtasks
2. Delegate each subtask to the right worker agent
3. Aggregate the results into a single response

Key Difference: Decomposition Prompt

The orchestrator agent has a new function that the simple client agent did not: build_decomposition_prompt.

def build_decomposition_prompt(user_request: str) -> str:
    """Build a prompt for LLM-based task decomposition."""
    agent_descriptions = "\n".join([
        f"Agent {i} — {card['name']}: {card['skills'][0]['description']}"
        for i, (url, card) in enumerate(discovered_agents.items())
    ])

    return (
        "You are a task decomposition agent. Given a user request, break it "
        "into one or more subtasks. Each subtask should be handled by one of "
        "the available agents.\n\n"
        f"Available agents:\n{agent_descriptions}\n\n"
        f"User request: {user_request}\n\n"
        "Return a JSON array of subtasks. Each subtask has:\n"
        "  - 'text': the text to send to the agent\n"
        "  - 'agent_index': the index of the agent to handle it\n"
        "Example: [{\"text\": \"classify this\", \"agent_index\": 0}]"
    )

Orchestrator Execute Flow

async def execute(self, context: RequestContext, event_queue: EventQueue) -> None:
    user_input = context.get_user_input()

    # Step 1: Decompose using LLM
    decomposition_response = await decomposition_agent.run(
        build_decomposition_prompt(user_input)
    )
    subtasks = json.loads(decomposition_response)

    # Step 2: Delegate each subtask to the right agent
    results = []
    for subtask in subtasks:
        agent_url = list(discovered_agents.keys())[subtask["agent_index"]]
        agent_name = discovered_agents[agent_url]["name"]

        result = await call_remote_agent(agent_url, subtask["text"])
        results.append(f"=== {agent_name} ===\n{result}")

    # Step 3: Aggregate results with clear labels
    combined = "\n\n".join(results)

    await event_queue.enqueue_event(new_agent_artifact(combined))
    await event_queue.enqueue_event(new_task_status_update(TaskState.completed))

Non-Determinism Warning

LLMs are non-deterministic. The same request might produce a different decomposition plan on the next run. For example, “review this contract for risk clauses and check it for GDPR compliance” might produce two subtasks on one run and only one subtask on another run, because the LLM decided that one agent could handle everything.

Mitigation: Make your decomposition prompt very specific about when to use each agent. Include explicit routing rules:

routing_rules = """
Routing rules:
- Use the compliance agent ONLY for regulatory/GDPR/legal compliance questions.
- Use the contract review agent for risk clause identification and analysis.
- Use the triage agent ONLY for document type classification.
- If the request mentions both risk and compliance, create TWO subtasks.
"""

Sample Multi-Agent Output

Request: "Review this contract for risk clauses and check it for GDPR compliance."

Decomposed into 2 subtasks:
  Subtask 1 → ContractReviewAgent
  Subtask 2 → ComplianceCheckerAgent

=== ContractReviewAgent ===
Overall Risk: MEDIUM
Flagged: Section 4.2 Non-Compete (HIGH), Section 7.1 Indemnification (MEDIUM)

=== ComplianceCheckerAgent ===
Compliance Status: NON-COMPLIANT
Findings: Missing controller/processor clarity under GDPR Article 28

Implementing Dynamic Routing and Hierarchical Delegation

What Is Hierarchical Delegation?

Dynamic routing uses metadata on agent cards to make smarter decisions. Hierarchical delegation goes a step further: a worker agent can itself act as a client to another agent.

In the previous orchestrator-worker pattern, the orchestrator coordinates all agent calls. In hierarchical delegation:

1. The orchestrator asks the contract review agent to analyze a contract.
2. The contract review agent runs its LangGraph analysis.
3. The contract review agent autonomously decides the findings contain compliance-related content.
4. The contract review agent delegates directly to the compliance agent.
5. The contract review agent combines both results and returns everything to the orchestrator.

The orchestrator only made one call, but two agents did work. The sub-agent handled the coordination autonomously.

sequenceDiagram
    participant Orch as Orchestrator
    participant CR as ContractReviewAgent :9998
    participant Comp as ComplianceCheckerAgent :9997

    Orch->>CR: Review contract for risk and GDPR compliance
    CR->>CR: Run LangGraph (extract_clauses + assess_risk)
    CR->>CR: Scan results for compliance keywords
    Note over CR: Found: "gdpr", "data protection"
    CR->>Comp: Review these findings for compliance issues
    Comp-->>CR: NON-COMPLIANT — missing controller/processor clarity
    CR->>CR: Combine risk assessment + compliance findings
    CR-->>Orch: Combined: MEDIUM risk + NON-COMPLIANT

Updating the Contract Review Agent for Hierarchical Delegation

The LangGraph graph itself is unchanged. The new behavior is added in the executor, after the graph runs.

# Compliance keywords that trigger delegation
COMPLIANCE_KEYWORDS = [
    "gdpr", "regulatory", "compliance", "data protection",
    "privacy", "regulation", "pii", "personal data"
]

def needs_compliance_review(findings_text: str) -> bool:
    """Check if LangGraph results contain compliance-related content."""
    text_lower = findings_text.lower()
    return any(keyword in text_lower for keyword in COMPLIANCE_KEYWORDS)

async def delegate_to_compliance(findings_text: str) -> str:
    """Call the compliance agent as an A2A client."""
    async with httpx.AsyncClient() as http_client:
        compliance_client = A2AClient(
            http_client=http_client,
            url="http://localhost:9997"  # ComplianceCheckerAgent
        )
        response = await compliance_client.send_message(
            message=new_user_text_message(
                f"Review the following contract risk findings for compliance issues. "
                f"Flag any regulatory concerns and recommend corrective actions.\n\n"
                f"{findings_text}"
            )
        )
    return extract_text_from_response(response)

Executor with Hierarchical Delegation

async def execute(self, context: RequestContext, event_queue: EventQueue) -> None:
    contract_text = context.get_user_input()

    # Run the LangGraph analysis
    graph_result = await contract_review_graph.ainvoke(
        {"contract_text": contract_text}
    )
    risk_findings = graph_result["risk_assessment"]

    # Autonomously decide whether to call the compliance agent
    compliance_findings = None
    if needs_compliance_review(str(risk_findings)):
        compliance_findings = await delegate_to_compliance(str(risk_findings))

    # Combine results
    combined = f"CONTRACT RISK ASSESSMENT\n\n{risk_findings}"
    if compliance_findings:
        combined += f"\n\nCOMPLIANCE REVIEW (via hierarchical delegation)\n\n{compliance_findings}"

    await event_queue.enqueue_event(new_agent_artifact(combined))
    await event_queue.enqueue_event(new_task_status_update(TaskState.completed))

The contract review agent is now:

• A server on port 9998 (receives tasks from the orchestrator)
• A client to port 9997 (delegates to the compliance agent when needed)

This is the essence of hierarchical delegation: agents calling agents, each making their own routing decisions.

Handling Failures and Retries

Three Failure Modes to Plan For

In production, agents go down, networks drop, and LLM calls take longer than expected. Your orchestrator needs to handle all of this without losing work that already succeeded.

Retry with Exponential Backoff

MAX_RETRIES = 3      # Number of retry attempts
BACKOFF_BASE = 1.0   # Starting delay in seconds (1s → 2s → 4s)

async def call_remote_agent(agent_url: str, task_text: str) -> str | None:
    """Invoke a remote agent with retry and exponential backoff."""
    for attempt in range(MAX_RETRIES):
        try:
            async with httpx.AsyncClient(timeout=30.0) as http_client:
                a2a_client = A2AClient(http_client=http_client, url=agent_url)
                response = await a2a_client.send_message(
                    message=new_user_text_message(task_text)
                )
                return extract_text_from_response(response)

        except (httpx.ConnectError, httpx.TimeoutException) as e:
            if attempt < MAX_RETRIES - 1:
                wait = BACKOFF_BASE * (2 ** attempt)  # 1s, 2s, 4s
                print(f"Attempt {attempt + 1} failed: {e}. Retrying in {wait}s...")
                await asyncio.sleep(wait)
            else:
                print(f"All {MAX_RETRIES} attempts failed for {agent_url}")
                return None  # Graceful degradation — return None, don't crash

Key design decisions:

• timeout=30.0 — if the agent does not respond within 30 seconds, the request fails and enters the retry loop.
• Exponential backoff — 1s, 2s, 4s delays between retries, reducing thundering herd pressure.
• Returns None on full failure — the orchestrator does not crash; it records the failure and continues with subtasks that succeeded.

Graceful Degradation in the Aggregator

async def execute(self, context: RequestContext, event_queue: EventQueue) -> None:
    user_input = context.get_user_input()
    subtasks = await decompose(user_input)

    results = []
    for subtask in subtasks:
        agent_url = get_agent_url(subtask["agent_index"])
        agent_name = get_agent_name(subtask["agent_index"])

        result = await call_remote_agent(agent_url, subtask["text"])

        if result is not None:
            results.append(f"=== {agent_name} ===\n{result}")
        else:
            # Record the failure but do NOT discard results that succeeded
            results.append(f"=== {agent_name} ===\n[UNAVAILABLE after {MAX_RETRIES} attempts]")

    # Always return whatever we have
    combined = "\n\n".join(results)
    await event_queue.enqueue_event(new_agent_artifact(combined))
    await event_queue.enqueue_event(new_task_status_update(TaskState.completed))

Key principle: Never lose work that already succeeded. If you have partial results, return them with a clear note about which agent was unavailable.

Behavior with a Down Agent

When the compliance agent (port 9997) is stopped mid-run:

Request 1:
  → ContractReviewAgent: [full risk assessment returned]
  → ComplianceCheckerAgent: [UNAVAILABLE after 3 attempts]

Request 2:
  → DocumentTriageAgent: contract

Request 3:
  → ContractReviewAgent: [full risk assessment returned]
  → ComplianceCheckerAgent: [UNAVAILABLE after 3 attempts]

The orchestrator does not crash. In production, you would also log the failure, trigger an alert, and let the user know. The system remains functional even when individual agents are offline.

Module 4: Security, Testing, and Production Deployment

Securing Authentication, Authorization, and Trust

Security Fundamentals for A2A

Every agent is a network endpoint. Treat it that way. Security in A2A follows the same principles as any service-to-service communication.

Authentication starts at the agent card. The card declares what auth schemes the agent supports:

• Bearer tokens
• API keys
• OAuth

A client reads the card, sees the auth requirement, and attaches the right credentials before sending any requests.

Authorization is about scope. When the orchestrator delegates a subtask to a worker agent, it should pass a scoped token that limits what that worker can do. Do not pass an admin token to a worker that only needs read access.

Trust is the hardest part. Treat any agent outside your direct control as a potentially untrusted entity.

Three Threats to Plan For

Declaring Security in the Agent Card

agent_card = AgentCard(
    name="DocumentTriageAgent",
    # ... other fields ...
    security_schemes={
        "bearerAuth": {
            "type": "http",
            "scheme": "bearer"
        }
    },
    security=[{"bearerAuth": []}]
)

Any client that reads this card knows to include an Authorization: Bearer <token> header with every request (except the well-known agent card endpoint, which is public).

Bearer Token Middleware

VALID_TOKENS = {"demo-token-2026", "prod-token-abc123"}

class BearerAuthMiddleware:
    """ASGI middleware that validates bearer tokens on every request."""

    def __init__(self, app, public_paths: list[str] = None):
        self.app = app
        self.public_paths = public_paths or ["/.well-known/agent.json"]

    async def __call__(self, scope, receive, send):
        if scope["type"] == "http":
            path = scope["path"]

            # Public paths skip auth (agent card is always accessible for discovery)
            if path not in self.public_paths:
                headers = dict(scope.get("headers", []))
                auth_header = headers.get(b"authorization", b"").decode()

                if not auth_header.startswith("Bearer "):
                    await send_401(send)
                    return

                token = auth_header[len("Bearer "):]
                if token not in VALID_TOKENS:
                    await send_401(send)
                    return

        await self.app(scope, receive, send)

Key design: the /.well-known/agent.json endpoint is always public so clients can discover your agents. All other endpoints require a valid token.

Audit Logging

import json
import datetime

def audit_log(
    caller_token: str,
    user_input: str,
    result: str,
    task_id: str
) -> None:
    """Log every interaction for traceability."""
    entry = {
        "timestamp": datetime.datetime.utcnow().isoformat(),
        "task_id": task_id,
        "caller": caller_token[:8] + "...",  # Partial token for privacy
        "input_preview": user_input[:100],
        "result_preview": result[:100],
    }
    with open("audit.log", "a") as f:
        f.write(json.dumps(entry) + "\n")

The audit log captures:

• Timestamp of every interaction
• Who called it (partial token for privacy)
• What they asked (the task input)
• The result (or failure reason)

Testing Authentication

# Fetch the agent card — always public, no auth needed
curl http://localhost:9999/.well-known/agent.json

# Attempt a request WITHOUT auth — should return 401
curl -X POST http://localhost:9999/ \
  -H "Content-Type: application/json" \
  -d '{"jsonrpc":"2.0","method":"message/send","params":{"message":{"parts":[{"type":"text","text":"hello"}]}},"id":1}'
# → 401 Unauthorized: missing or invalid Authorization header

# Attempt a request WITH auth — should succeed
curl -X POST http://localhost:9999/ \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer demo-token-2026" \
  -d '{"jsonrpc":"2.0","method":"message/send","params":{"message":{"parts":[{"type":"text","text":"This is a vendor services agreement with SLA"}]}},"id":1}'
# → 200 OK: {"result": "contract"}

Reading the Audit Log

cat audit.log

Sample output:

{"timestamp": "2026-06-28T10:05:12.345Z", "task_id": "task-001", "caller": "demo-tok...", "input_preview": "This is a vendor services agreement with SLA", "result_preview": "contract"}
{"timestamp": "2026-06-28T10:05:01.123Z", "task_id": "task-000", "caller": "MISSING", "input_preview": "hello", "result_preview": "ERROR: 401 Unauthorized"}

Every interaction is recorded — failed auth attempts and successful ones alike.

Testing and Debugging with A2A Inspector

The A2A Inspector

The A2A Inspector is a web-based tool from the A2A project. Point it at an agent and it:

• Validates the agent card
• Provides an interactive chat interface
• Shows raw protocol messages in a debug console

Setup:

git clone https://github.com/a2aproject/a2a-inspector.git
cd a2a-inspector
docker build -t a2a-inspector .
docker run -d --network host --name a2a-inspector a2a-inspector
# Inspector runs at http://localhost:8080

Using the inspector:

1. Open http://localhost:8080 in your browser.
2. Enter the agent URL in the “Inspector URL” field (e.g., http://localhost:9999).
3. Click Connect — the inspector fetches and displays the agent card with validation results (green = valid, red = problem).
4. Use the Chat Interface to send test messages interactively.
5. Open the Debug Console to see raw JSON-RPC messages flowing in both directions.

The Debug Console is invaluable for diagnosing issues: you can see exactly what is being sent and received, identify malformed messages, and confirm that the agent is responding correctly.

Three Levels of A2A Agent Testing

graph TD
    L1["Level 1: Agent Card Conformance\n✓ Has name, description, version\n✓ Has at least one skill\n✓ Interface URL points to live server"]
    L2["Level 2: Task Lifecycle Testing\n✓ submitted → working → completed\n✓ Produces artifact with expected content\n✓ Handles empty input without crashing"]
    L3["Level 3: Protocol Compliance Testing\n✓ JSON-RPC messages are well-formed\n✓ Correct error codes returned\n✓ Streaming works if declared"]

    L1 --> L2 --> L3

Level 1: Agent Card Conformance

• Does the card have a name, description, and version?
• Does it have at least one skill?
• Does the interface URL point to where the agent is actually running?
• A malformed card means no client can discover your agent.

Level 2: Task Lifecycle Testing

• Does the agent move from submitted → working → completed?
• Does it produce an artifact with the expected content?
• What happens with empty input? Does it handle edge cases without crashing?

Level 3: Protocol Compliance Testing

• Are JSON-RPC messages well-formed?
• Does the agent return proper error codes?
• Does streaming work if declared in the capabilities?

Automated Tests with pytest (test_protocol.py)

import pytest
import httpx
import asyncio

BASE_URL = "http://localhost:9999"

class TestAgentCardConformance:
    """Level 1: Validate the agent card structure."""

    def test_agent_card_reachable(self):
        response = httpx.get(f"{BASE_URL}/.well-known/agent.json")
        assert response.status_code == 200

    def test_agent_card_has_name(self):
        card = httpx.get(f"{BASE_URL}/.well-known/agent.json").json()
        assert "name" in card
        assert len(card["name"]) > 0

    def test_agent_card_has_version(self):
        card = httpx.get(f"{BASE_URL}/.well-known/agent.json").json()
        assert "version" in card

    def test_agent_card_has_skills(self):
        card = httpx.get(f"{BASE_URL}/.well-known/agent.json").json()
        assert "skills" in card
        assert len(card["skills"]) >= 1

    def test_agent_card_skill_has_description(self):
        card = httpx.get(f"{BASE_URL}/.well-known/agent.json").json()
        for skill in card["skills"]:
            assert "description" in skill

class TestTaskLifecycle:
    """Level 2: Validate the task state machine."""

    def test_send_message_returns_200(self):
        payload = {
            "jsonrpc": "2.0",
            "method": "message/send",
            "params": {
                "message": {
                    "parts": [{"type": "text", "text": "vendor services agreement"}]
                }
            },
            "id": 1
        }
        response = httpx.post(BASE_URL, json=payload)
        assert response.status_code == 200

    def test_task_reaches_completed_state(self):
        # ... send message and poll for task state ...
        task = send_and_wait(BASE_URL, "test document")
        assert task["status"]["state"] == "completed"

    def test_task_produces_artifact(self):
        task = send_and_wait(BASE_URL, "vendor agreement")
        assert len(task.get("artifacts", [])) > 0

    def test_artifact_has_content(self):
        task = send_and_wait(BASE_URL, "vendor agreement")
        artifact = task["artifacts"][0]
        assert len(artifact.get("parts", [])) > 0

class TestErrorHandling:
    """Level 3: Validate error handling and protocol compliance."""

    def test_empty_input_does_not_crash(self):
        payload = {
            "jsonrpc": "2.0",
            "method": "message/send",
            "params": {
                "message": {"parts": [{"type": "text", "text": ""}]}
            },
            "id": 2
        }
        response = httpx.post(BASE_URL, json=payload)
        # Should return 200 even for empty input — not crash with 500
        assert response.status_code in (200, 400)
        assert response.status_code != 500

    def test_malformed_request_returns_error_code(self):
        payload = {"jsonrpc": "2.0", "method": "nonexistent/method", "id": 3}
        response = httpx.post(BASE_URL, json=payload)
        data = response.json()
        assert "error" in data

Run all tests:

pytest test_protocol.py -v

All 11 tests should pass (green). These tests run in any continuous integration pipeline. Every time you update an agent, verify it still speaks the protocol correctly.

Deploying to Production with Agent Stack

What Is Agent Stack?

Agent Stack is an open infrastructure from the Linux Foundation for turning agents into running services. You write the agent logic; Agent Stack provides everything else:

• Agent runtime — container management and scheduling
• LLM routing — across 15+ providers
• Vector storage — for RAG and memory
• File handling — for document workflows
• Secrets management — for API keys and tokens
• Deployment tooling — CLI-based deploy commands

Agent Stack is framework-agnostic. Your LangGraph agent, your CrewAI agent, your custom Python agent — all deploy the same way. Every deployed agent is automatically exposed as A2A-compatible.

Quickstart documentation: https://agentstack.beeai.dev/stable/introduction/quickstart

Wrapping an Agent with the Agent Stack SDK

from agentstack import Server, RunContext, LLMExtension
from a2a.types import Message
from triage_agent import classify  # Existing agent logic, unchanged

# Lines 27-42 are identical to the Module 2 triage agent
# Only the wrapping changes

server = Server()  # Agent Stack server — handles A2A protocol plumbing

@server.agent(name="TriageAgent")
async def triage_wrapper(
    message: Message,           # Incoming A2A message
    context: RunContext,        # Platform context (metadata, state)
    llm: LLMExtension,          # LLM extension for credential management
) -> str:
    """Agent Stack wrapper for the document triage agent."""
    user_text = message.parts[0].text
    return await classify(user_text)

if __name__ == "__main__":
    server.run()

The @server.agent decorator:

• Binds the function to the Agent Stack platform
• Registers it under the name "TriageAgent"
• Automatically exposes it as an A2A-compatible endpoint

Local Development with Agent Stack CLI

# Install Agent Stack (one-line install from the quickstart page)
# (See https://agentstack.beeai.dev/stable/introduction/quickstart)

# Register and run the agent locally
uv run server.py
# → Attempting to register agent to the Agent Stack platform...
# → Agent registered successfully.

# List registered agents
agentstack list
# → TriageAgent (running)

# Send a test message
agentstack run TriageAgent --prompt "This is a vendor services agreement with an SLA."
# → contract

In local development mode, the agent runs on your machine and Agent Stack routes to it.

Production Deployment

For production, two additional steps:

# Step 1: Push your code to a GitHub repository
git push origin main

# Step 2: Deploy via Agent Stack CLI
agentstack add https://github.com/your-org/your-agent-repo

Agent Stack then:

1. Clones the repository
2. Builds a Docker image from your Dockerfile
3. Starts the container inside a managed runtime
4. Registers the agent with the platform

Production Dockerfile

FROM python:3.11-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt

COPY . .

EXPOSE 9999
CMD ["python", "server.py"]

Production Readiness Checklist

graph TD
    A["Production Readiness Checklist"]
    A --> B["Authentication\non every endpoint"]
    A --> C["Audit Logging\nfor traceability"]
    A --> D["Protocol Tests\nin CI pipeline"]
    A --> E["Containerized Deployment\nwith health monitoring"]
    A --> F["Versioning Strategy\nbump version field on agent card"]
    A --> G["Retry + Graceful Degradation\nin all orchestrators"]

Agent Versioning

agent_card = AgentCard(
    name="ContractReviewAgent",
    version="1.2.0",  # Bump this on every release
    # ...
)

For rolling deployments:

1. Deploy version 1.2.0 alongside version 1.1.0 (different port or host).
2. Gradually shift traffic to 1.2.0 (via the registry or load balancer).
3. Decommission 1.1.0 once 1.2.0 is stable.

Existing workflows that discovered 1.1.0 keep working until they are updated to discover 1.2.0.

Reference: Key Concepts Cheat Sheet

A2A Protocol Quick Reference

Task States Reference

submitted → working → completed
                   ↘ input_required → working (loop)
                   ↘ failed
         → canceled (from any state)

Port Assignments (Course Example)

Framework Integration Patterns

Common Command Patterns

# Fetch agent card
curl http://localhost:{PORT}/.well-known/agent.json | python -m json.tool

# Send a task (no auth)
curl -X POST http://localhost:{PORT}/ \
  -H "Content-Type: application/json" \
  -d '{"jsonrpc":"2.0","method":"message/send","params":{"message":{"parts":[{"type":"text","text":"YOUR TEXT"}]}},"id":1}'

# Send a task (with bearer auth)
curl -X POST http://localhost:{PORT}/ \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer YOUR-TOKEN" \
  -d '{"jsonrpc":"2.0","method":"message/send","params":{"message":{"parts":[{"type":"text","text":"YOUR TEXT"}]}},"id":1}'

# Run protocol tests
pytest test_protocol.py -v

# A2A Inspector setup
git clone https://github.com/a2aproject/a2a-inspector.git
cd a2a-inspector
docker build -t a2a-inspector .
docker run -d --network host --name a2a-inspector a2a-inspector

# Agent Stack local run
uv run server.py

# Agent Stack list agents
agentstack list

# Agent Stack invoke agent
agentstack run AgentName --prompt "your prompt"

# Agent Stack production deploy
agentstack add https://github.com/your-org/your-repo

MCP vs A2A Decision Guide

Use MCP when:
  ✓ Your agent needs to call a specific function (query DB, read file, call API)
  ✓ You want a single round-trip, structured input/output
  ✓ The resource doesn't have autonomous reasoning

Use A2A when:
  ✓ You're delegating to another agent that will reason about the request
  ✓ The interaction might require multiple steps or clarifying questions
  ✓ You want runtime discovery (don't know which agents exist at build time)
  ✓ You're building a system where agents from different teams/orgs collaborate

Source Repository

Course code: https://github.com/keshawillz/ps-a2a-course

Search Terms

agent-to-agent · protocol · ai · agents · orchestration · artificial · intelligence · generative · agent · a2a · card · delegation · production · architecture · client · hierarchical · stack · testing · langgraph · mcp · multi-agent · pattern · patterns · reference