The Ultimate Agentic Systems Cheatsheet

{
  "name": "Research Agent",
  "description": "Performs deep web research and summarization",
  "url": "https://api.example.com/a2a",
  "version": "1.0.0",
  "provider": {
    "organization": "Acme AI",
    "url": "https://acme.ai"
  },
  "capabilities": {
    "streaming": true,
    "pushNotifications": false,
    "stateTransitionHistory": true
  },
  "authentication": {
    "schemes": [{ "type": "oauth2", "flow": "authorizationCode" }]
  },
  "skills": [
    {
      "id": "web-search",
      "name": "Web Search",
      "description": "Search the web for real-time information",
      "tags": ["search", "research", "real-time"],
      "examples": ["Find latest news about X", "Research market trends"]
    },
    {
      "id": "summarize",
      "name": "Summarize",
      "description": "Summarize long documents or article collections",
      "tags": ["summarization", "nlp"],
      "examples": ["Summarize these 5 articles"]
    }
  ],
  "defaultInputModes": ["text/plain", "application/json"],
  "defaultOutputModes": ["text/plain", "application/json"]
}

// Agent Card Authentication Declaration
"authentication": {
  "schemes": [
    {
      "type": "oauth2",
      "flow": "authorizationCode",
      "authorizationUrl": "https://auth.example.com/authorize",
      "tokenUrl": "https://auth.example.com/token",
      "scopes": ["research:read", "research:write"]
    },
    {
      "type": "mtls",
      "caCertificates": ["-----BEGIN CERTIFICATE-----..."]
    }
  ]
}

# ── Minimal A2A Server (FastAPI + a2a-sdk) ──
from fastapi import FastAPI
from a2a.server import A2AServer, task_handler

app = FastAPI()

@a2a_server.task_handler("web-search")
async def handle_search(task: Task) -> Task:
    """Handle incoming search tasks from other agents."""
    query = task.message.parts[0].text
    results = await search_engine.query(query)
    
    # Update task status and return results
    task.status.state = "completed"
    task.message = Message(
        role="agent",
        parts=[TextPart(text=str(results))]
    )
    return task

# Agent Card served at /.well-known/agent.json
a2a_server = A2AServer(
    agent_card=AgentCard(
        name="Search Agent",
        description="Performs web searches",
        skills=[Skill(id="web-search", name="Web Search")]
    )
)
app.mount("/a2a", a2a_server.app)

# ── A2A Client — Sending a Task ──
from a2a.client import A2AClient

client = A2AClient("https://api.example.com/a2a")

# Discover the agent
card = await client.get_agent_card()
print(f"Agent: {card.name} — Skills: {[s.id for s in card.skills]}")

# Send a task
task = await client.send_task(
    message=Message(
        role="user",
        parts=[TextPart(text="Research LLM architectures for 2025")]
    )
)

# Stream the response
async for event in client.stream_task(task.id):
    if event.status:
        print(f"State: {event.status.state}")
    if event.message:
        print(f"Agent: {event.message.parts[0].text}")

// ACP Checkout Request (simplified)
{
  "cart_id": "cart_abc123",
  "items": [
    { "product_id": "prod_001", "quantity": 1, "price": 49.99 },
    { "product_id": "prod_002", "quantity": 2, "price": 19.99 }
  ],
  "discounts": [{ "code": "SUMMER20", "type": "percent", "value": 20 }],
  "payment": {
    "method": "card_on_file",
    "last_four": "4242",
    "authorization": {
      "type": "multi_factor",
      "user_verified": true,
      "verification_method": "biometric",
      "timestamp": "2025-07-15T10:30:00Z"
    }
  },
  "shipping": { "address_id": "addr_home", "method": "express" }
}

# ── MCP + A2A Combined Agent ──
from a2a.client import A2AClient
from mcp import MCPServer

class OrchestratorAgent:
    def __init__(self):
        # MCP tools for this agent's own use
        self.mcp = MCPServer(tools=["web_search", "file_system"])
        # A2A clients for communicating with other agents
        self.researcher = A2AClient("https://research.api/a2a")
        self.writer = A2AClient("https://writer.api/a2a")
    
    async def handle_request(self, user_query: str):
        # Step 1: Send research task via A2A
        research_task = await self.researcher.send_task(
            message=Message(role="user", parts=[
                TextPart(text=f"Research: {user_query}")
            ])
        )
        # Step 2: Get research results
        findings = await research_task.completed()
        # Step 3: Send writing task via A2A
        write_task = await self.writer.send_task(
            message=Message(role="user", parts=[
                TextPart(text=f"Write blog post based on: {findings}")
            ])
        )
        return await write_task.completed()

# ── A2A-Enabled LangGraph Agent ──
from langgraph.prebuilt import create_react_agent
from langgraph_a2a import A2ADeployment

# Define the agent with tools
research_agent = create_react_agent(
    model="claude-sonnet-4-20250514",
    tools=[web_search_tool, scraper_tool, summarizer_tool],
    name="researcher",
    prompt="You are a research specialist. Find accurate, "
           "up-to-date information and provide citations."
)

# Deploy as A2A endpoint
deployment = A2ADeployment(
    agent=research_agent,
    agent_card={
        "name": "Research Agent",
        "description": "Deep web research and fact-checking",
        "url": "https://api.example.com/a2a",
        "version": "1.2.0",
        "capabilities": {
            "streaming": True,
            "pushNotifications": False
        },
        "skills": [
            {
                "id": "web-search",
                "name": "Web Search",
                "description": "Search the web for real-time info"
            },
            {
                "id": "fact-check",
                "name": "Fact Check",
                "description": "Verify claims against sources"
            }
        ]
    }
)

# Run the A2A server
if __name__ == "__main__":
    deployment.run(host="0.0.0.0", port=8000)

# ── Multi-Agent System with LangGraph + A2A ──
from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph_a2a import A2ANode

workflow = StateGraph(MessagesState)

# Add A2A-connected agent nodes
workflow.add_node("researcher", A2ANode(
    endpoint="https://research.api/a2a",
    agent_card_required=True
))
workflow.add_node("writer", A2ANode(
    endpoint="https://writer.api/a2a"
))
workflow.add_node("reviewer", A2ANode(
    endpoint="https://review.api/a2a"
))

# Define the workflow
workflow.add_edge(START, "researcher")
workflow.add_edge("researcher", "writer")
workflow.add_edge("writer", "reviewer")
workflow.add_conditional_edges("reviewer", 
    lambda state: "writer" if state.needs_revision else END
)

app = workflow.compile(checkpointer=checkpointer)

# ── Google ADK Agent with A2A ──
from google.adk import Agent
from google.adk.tools import mcp_tool
from google.adk.a2a import A2AServer

# Create agent with MCP tools and A2A enabled
agent = Agent(
    name="DataAnalyst",
    model="gemini-2.5-pro",
    description="Analyzes datasets and creates visualizations",
    instruction=(
        "You are a data analysis specialist. "
        "Query databases, analyze trends, "
        "and generate charts/visualizations."
    ),
    tools=[
        mcp_tool("sqlite://data.db"),    # MCP database tool
        mcp_tool("python://repl"),        # MCP Python REPL
    ],
)

# Deploy with A2A — one line!
server = A2AServer(
    agent=agent,
    port=8080,
    auth="oauth2"  # Enable OAuth 2.1
)
server.start()