Skip to content
Foldspace Docs

A2A advanced patterns

Open in ChatGPT Open in Claude

This page covers building A2A server agents beyond the Python quickstart: orchestrating with LangGraph, the TypeScript and Java SDKs, and implementing the protocol natively without an SDK.

LangGraph integration

Section titled “LangGraph integration”

LangGraph provides a graph-based orchestration layer for building stateful, multi-step agents. If you already orchestrate agents with LangGraph, you can expose a compiled graph as an A2A server agent with a thin AgentExecutor bridge — no need to rewrite your agent logic.

New to LangGraph? See the official LangGraph documentation and the langchain-ai/langgraph repository.

Basic LangGraph agent

Section titled “Basic LangGraph agent”

Wrap a LangGraph research agent using the AgentExecutor bridge pattern to expose it as an A2A server.

research_agent.py
import uvicorn
from langgraph.graph import StateGraph, MessagesState, START, END
from langchain_openai import ChatOpenAI


from a2a.server.apps.jsonrpc import A2AStarletteApplication
from a2a.server.request_handlers import DefaultRequestHandler
from a2a.server.agent_execution import AgentExecutor, RequestContext
from a2a.server.events import EventQueue
from a2a.server.tasks import InMemoryTaskStore, TaskUpdater
from a2a.types import (
    AgentCard,
    AgentCapabilities,
    AgentSkill,
    Part,
    TextPart,
)




# --- Step 1: Define the LangGraph agent ---


llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)




def research_node(state: MessagesState):
    response = llm.invoke(state["messages"])
    return {"messages": [response]}




graph_builder = StateGraph(MessagesState)
graph_builder.add_node("researcher", research_node)
graph_builder.add_edge(START, "researcher")
graph_builder.add_edge("researcher", END)
research_graph = graph_builder.compile()




# --- Step 2: Bridge LangGraph to A2A ---


class ResearchAgentExecutor(AgentExecutor):


    def __init__(self, graph):
        self.graph = graph


    async def execute(self, context: RequestContext, event_queue: EventQueue) -> None:
        updater = TaskUpdater(event_queue, context.task_id, context.context_id)
        await updater.start_work()


        user_input = context.get_user_input()
        result = await self.graph.ainvoke(
            {"messages": [{"role": "user", "content": user_input}]}
        )


        response_text = result["messages"][-1].content
        await updater.add_artifact(parts=[Part(root=TextPart(text=response_text))])
        await updater.complete()


    async def cancel(self, context: RequestContext, event_queue: EventQueue) -> None:
        updater = TaskUpdater(event_queue, context.task_id, context.context_id)
        await updater.cancel()




# --- Step 3: Agent Card + Server ---


agent_card = AgentCard(
    name="Research Assistant Agent",
    description="Researches topics and provides detailed, cited answers",
    url="http://localhost:9000/",
    version="1.0.0",
    capabilities=AgentCapabilities(streaming=True),
    default_input_modes=["text/plain"],
    default_output_modes=["text/plain"],
    skills=[
        AgentSkill(
            id="research",
            name="Deep Research",
            description="Researches any topic and provides comprehensive answers",
            tags=["research", "knowledge", "analysis"],
            examples=["What are the key differences between gRPC and REST?"],
        ),
    ],
)


request_handler = DefaultRequestHandler(
    agent_executor=ResearchAgentExecutor(research_graph),
    task_store=InMemoryTaskStore(),
)


server = A2AStarletteApplication(
    agent_card=agent_card,
    http_handler=request_handler,
)


if __name__ == "__main__":
    uvicorn.run(server.build(), host="0.0.0.0", port=9000)

LangGraph with streaming progress

Section titled “LangGraph with streaming progress”

For long-running agents, stream intermediate updates with astream_events so users see progress in real time.

class StreamingResearchExecutor(AgentExecutor):


    def __init__(self, graph):
        self.graph = graph


    async def execute(self, context: RequestContext, event_queue: EventQueue) -> None:
        updater = TaskUpdater(event_queue, context.task_id, context.context_id)
        await updater.start_work()


        user_input = context.get_user_input()


        async for event in self.graph.astream_events(
            {"messages": [{"role": "user", "content": user_input}]},
            version="v2",
        ):
            if event["event"] == "on_chat_model_stream":
                chunk = event["data"]["chunk"].content
                if chunk:
                    await updater.update_status(
                        state="working",
                        message=Part(root=TextPart(text=chunk)),
                    )


        final_result = await self.graph.ainvoke(
            {"messages": [{"role": "user", "content": user_input}]}
        )
        response_text = final_result["messages"][-1].content
        await updater.add_artifact(parts=[Part(root=TextPart(text=response_text))])
        await updater.complete()


    async def cancel(self, context: RequestContext, event_queue: EventQueue) -> None:
        updater = TaskUpdater(event_queue, context.task_id, context.context_id)
        await updater.cancel()

LangGraph with multi-turn (user input required)

Section titled “LangGraph with multi-turn (user input required)”

When your agent needs clarification before proceeding, transition to input-required — the Copilot prompts the user and relays their reply back.

class MultiTurnResearchExecutor(AgentExecutor):


    def __init__(self, graph):
        self.graph = graph


    async def execute(self, context: RequestContext, event_queue: EventQueue) -> None:
        updater = TaskUpdater(event_queue, context.task_id, context.context_id)
        await updater.start_work()


        user_input = context.get_user_input()
        result = await self.graph.ainvoke(
            {"messages": [{"role": "user", "content": user_input}]}
        )


        response_text = result["messages"][-1].content


        if result.get("needs_clarification", False):
            await updater.update_status(
                state="input-required",
                message=Part(root=TextPart(text=response_text)),
            )
        else:
            await updater.add_artifact(parts=[Part(root=TextPart(text=response_text))])
            await updater.complete()


    async def cancel(self, context: RequestContext, event_queue: EventQueue) -> None:
        updater = TaskUpdater(event_queue, context.task_id, context.context_id)
        await updater.cancel()

TypeScript SDK

Section titled “TypeScript SDK”

Install the SDK and Express:

Terminal window
npm install @a2a-js/sdk express

A complete agent implementation demonstrating message handling, custom execution logic, and skill definitions.

agent.ts
import express from 'express';
import { v4 as uuidv4 } from 'uuid';
import { AgentCard, Message } from '@a2a-js/sdk';
import {
  AgentExecutor,
  RequestContext,
  ExecutionEventBus,
  DefaultRequestHandler,
  InMemoryTaskStore,
} from '@a2a-js/sdk/server';
import {
  agentCardHandler,
  jsonRpcHandler,
  restHandler,
} from '@a2a-js/sdk/server/express';


class DocumentAnalysisExecutor implements AgentExecutor {
  async execute(
    requestContext: RequestContext,
    eventBus: ExecutionEventBus
  ): Promise<void> {
    const userMessage = requestContext.message;


    const inputText = userMessage.parts
      .filter((p) => p.kind === 'text')
      .map((p) => p.text)
      .join(' ');


    const analysisResult = await this.analyzeDocument(inputText);


    const responseMessage: Message = {
      kind: 'message',
      messageId: uuidv4(),
      role: 'agent',
      parts: [{ kind: 'text', text: analysisResult }],
      contextId: requestContext.contextId,
    };


    eventBus.publish(responseMessage);
    eventBus.finished();
  }


  cancelTask = async (): Promise<void> => {};


  private async analyzeDocument(text: string): Promise<string> {
    return `Analysis complete. Document contains ${text.split(' ').length} words. Key topics identified.`;
  }
}


const agentCard: AgentCard = {
  name: 'Document Analysis Agent',
  description: 'Analyzes documents for key topics, entities, and summaries',
  url: 'http://localhost:3000',
  version: '1.0.0',
  capabilities: { streaming: true, pushNotifications: false },
  defaultInputModes: ['text/plain', 'application/pdf'],
  defaultOutputModes: ['text/plain', 'application/json'],
  skills: [
    {
      id: 'summarize',
      name: 'Summarize Document',
      description: 'Produces a concise summary of the input document',
      tags: ['nlp', 'summarization'],
      examples: ['Summarize this contract for key obligations'],
    },
    {
      id: 'extract-entities',
      name: 'Extract Entities',
      description: 'Identifies people, organizations, dates, and amounts',
      tags: ['nlp', 'ner', 'extraction'],
      examples: ['Extract all company names and dates from this filing'],
    },
  ],
};


const handler = new DefaultRequestHandler({
  agentExecutor: new DocumentAnalysisExecutor(),
  taskStore: new InMemoryTaskStore(),
});


const app = express();
app.use(express.json());
app.get('/.well-known/agent.json', agentCardHandler(agentCard));
app.post('/jsonrpc', jsonRpcHandler(handler));
app.use('/api', restHandler(handler));


app.listen(3000, () => {
  console.log('Document Analysis Agent running on port 3000');
});

Java SDK

Section titled “Java SDK”

Maven dependency:

<dependency>
    <groupId>org.a2aproject.sdk</groupId>
    <artifactId>a2a-java-sdk-reference-jsonrpc</artifactId>
    <version>1.0.0.Beta1</version>
</dependency>

A CDI producer configures the agent metadata — capabilities, skills, and supported interfaces:

import org.a2aproject.sdk.server.PublicAgentCard;
import org.a2aproject.sdk.spec.AgentCapabilities;
import org.a2aproject.sdk.spec.AgentCard;
import org.a2aproject.sdk.spec.AgentInterface;
import org.a2aproject.sdk.spec.AgentSkill;
import org.a2aproject.sdk.spec.TransportProtocol;


import jakarta.enterprise.context.ApplicationScoped;
import jakarta.enterprise.inject.Produces;
import java.util.Collections;
import java.util.List;


@ApplicationScoped
public class CodeReviewAgentCardProducer {


    private static final String AGENT_URL = "http://localhost:10001";


    @Produces
    @PublicAgentCard
    public AgentCard agentCard() {
        return AgentCard.builder()
                .name("Code Review Agent")
                .description("Reviews code for bugs, security issues, and style violations")
                .supportedInterfaces(List.of(
                        new AgentInterface(TransportProtocol.JSONRPC.asString(), AGENT_URL)))
                .version("1.0.0")
                .capabilities(AgentCapabilities.builder()
                        .streaming(true)
                        .pushNotifications(false)
                        .build())
                .defaultInputModes(Collections.singletonList("text/plain"))
                .defaultOutputModes(List.of("text/plain", "application/json"))
                .skills(List.of(
                        AgentSkill.builder()
                                .id("review-code")
                                .name("Review Code")
                                .description("Analyzes code for bugs, security vulnerabilities, and improvements")
                                .tags(List.of("code-review", "security", "quality"))
                                .examples(List.of("Review this Java class for thread safety issues"))
                                .build(),
                        AgentSkill.builder()
                                .id("suggest-refactor")
                                .name("Suggest Refactoring")
                                .description("Proposes structural improvements and design patterns")
                                .tags(List.of("refactoring", "design-patterns"))
                                .examples(List.of("Suggest how to refactor this service class"))
                                .build()))
                .build();
    }
}

An executor processes requests and manages the task lifecycle, including cancellation:

import org.a2aproject.sdk.server.agentexecution.AgentExecutor;
import org.a2aproject.sdk.server.agentexecution.RequestContext;
import org.a2aproject.sdk.server.tasks.AgentEmitter;
import org.a2aproject.sdk.spec.JSONRPCError;
import org.a2aproject.sdk.spec.Message;
import org.a2aproject.sdk.spec.Part;
import org.a2aproject.sdk.spec.Task;
import org.a2aproject.sdk.spec.TaskNotCancelableError;
import org.a2aproject.sdk.spec.TaskState;
import org.a2aproject.sdk.spec.TextPart;


import jakarta.enterprise.context.ApplicationScoped;
import jakarta.enterprise.inject.Produces;
import jakarta.inject.Inject;
import java.util.List;


@ApplicationScoped
public class CodeReviewExecutorProducer {


    @Inject
    CodeReviewAgent codeReviewAgent;


    @Produces
    public AgentExecutor agentExecutor() {
        return new CodeReviewAgentExecutor(codeReviewAgent);
    }


    private static class CodeReviewAgentExecutor implements AgentExecutor {


        private final CodeReviewAgent codeReviewAgent;


        public CodeReviewAgentExecutor(CodeReviewAgent agent) {
            this.codeReviewAgent = agent;
        }


        @Override
        public void execute(RequestContext context, AgentEmitter agentEmitter) throws JSONRPCError {
            if (context.getTask() == null) {
                agentEmitter.submit();
            }
            agentEmitter.startWork();


            String code = extractTextFromMessage(context.getMessage());
            String review = codeReviewAgent.review(code);


            TextPart responsePart = new TextPart(review);
            agentEmitter.addArtifact(List.of(responsePart));
            agentEmitter.complete();
        }


        @Override
        public void cancel(RequestContext context, AgentEmitter agentEmitter) throws JSONRPCError {
            Task task = context.getTask();
            if (task.getStatus().state() == TaskState.COMPLETED
                    || task.getStatus().state() == TaskState.CANCELED) {
                throw new TaskNotCancelableError();
            }
            agentEmitter.cancel();
        }


        private String extractTextFromMessage(Message message) {
            StringBuilder sb = new StringBuilder();
            for (Part<?> part : message.parts()) {
                if (part instanceof TextPart textPart) {
                    sb.append(textPart.text());
                }
            }
            return sb.toString();
        }
    }
}

Native implementation (without an SDK)

Section titled “Native implementation (without an SDK)”

If you need full control or work in a language without an official SDK, implement the A2A protocol directly over JSON-RPC 2.0.

What you must implement

Section titled “What you must implement”
  1. GET /.well-known/agent.json — serve your Agent Card.
  2. POST / — handle JSON-RPC requests:
    • message/send — accept a message and return a task.
    • tasks/get — return task status.
    • message/stream — SSE stream of updates (for streaming agents).

Python — native with Flask

Section titled “Python — native with Flask”
agent.py
import json
import uuid
from flask import Flask, request, jsonify, Response


app = Flask(__name__)
tasks = {}


AGENT_CARD = {
    "name": "Expense Report Agent",
    "description": "Processes and categorizes expense reports",
    "url": "http://localhost:5000",
    "version": "1.0.0",
    "capabilities": {"streaming": False, "pushNotifications": False},
    "defaultInputModes": ["text/plain", "application/json"],
    "defaultOutputModes": ["application/json"],
    "skills": [
        {
            "id": "categorize-expenses",
            "name": "Categorize Expenses",
            "description": "Categorizes line items into budget categories",
            "tags": ["finance", "expenses"],
            "examples": ["Categorize these 5 expense line items"],
        }
    ],
}




@app.route("/.well-known/agent.json", methods=["GET"])
def agent_card():
    return jsonify(AGENT_CARD)




@app.route("/", methods=["POST"])
def handle_jsonrpc():
    body = request.get_json()
    method = body.get("method")
    params = body.get("params", {})
    request_id = body.get("id")


    if method == "message/send":
        return handle_send_message(params, request_id)
    elif method == "tasks/get":
        return handle_get_task(params, request_id)
    else:
        return jsonify({
            "jsonrpc": "2.0",
            "id": request_id,
            "error": {"code": -32601, "message": f"Method not found: {method}"},
        })




def handle_send_message(params, request_id):
    message = params.get("message", {})
    context_id = params.get("metadata", {}).get("contextId", str(uuid.uuid4()))


    parts = message.get("parts", [])
    user_text = " ".join(p["text"] for p in parts if p.get("kind") == "text")


    result = categorize_expenses(user_text)


    task_id = str(uuid.uuid4())
    task = {
        "kind": "task",
        "id": task_id,
        "contextId": context_id,
        "status": {
            "state": "completed",
            "message": {
                "role": "agent",
                "parts": [{"kind": "data", "data": result}],
            },
        },
        "artifacts": [
            {
                "artifactId": str(uuid.uuid4()),
                "parts": [{"kind": "data", "data": result}],
                "lastChunk": True,
            }
        ],
    }
    tasks[task_id] = task
    return jsonify({"jsonrpc": "2.0", "id": request_id, "result": task})




def handle_get_task(params, request_id):
    task_id = params.get("id")
    task = tasks.get(task_id)
    if not task:
        return jsonify({
            "jsonrpc": "2.0",
            "id": request_id,
            "error": {"code": -32602, "message": "Task not found"},
        })
    return jsonify({"jsonrpc": "2.0", "id": request_id, "result": task})




def categorize_expenses(text: str) -> dict:
    return {
        "categories": [
            {"item": "Client dinner", "category": "Meals & Entertainment", "amount": 127.50},
            {"item": "Uber to airport", "category": "Transportation", "amount": 45.00},
            {"item": "Conference ticket", "category": "Professional Development", "amount": 599.00},
        ],
        "total": 771.50,
    }




if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5000)

TypeScript — native with Express and SSE streaming

Section titled “TypeScript — native with Express and SSE streaming”

Demonstrates SSE streaming via the message/stream method to push updates asynchronously.

agent.ts
import express from 'express';
import { v4 as uuidv4 } from 'uuid';


const app = express();
app.use(express.json());


const tasks: Record<string, any> = {};


const AGENT_CARD = {
  name: 'Invoice Processing Agent',
  description: 'Extracts structured data from invoices and receipts',
  url: 'http://localhost:4000',
  version: '1.0.0',
  capabilities: { streaming: true, pushNotifications: false },
  defaultInputModes: ['text/plain', 'application/pdf'],
  defaultOutputModes: ['application/json'],
  skills: [
    {
      id: 'extract-invoice',
      name: 'Extract Invoice Data',
      description: 'Parses invoices and returns structured line items',
      tags: ['finance', 'ocr', 'extraction'],
      examples: ['Extract all line items from this invoice'],
    },
  ],
};


app.get('/.well-known/agent.json', (req, res) => {
  res.json(AGENT_CARD);
});


app.post('/', (req, res) => {
  const { method, params, id: requestId } = req.body;


  switch (method) {
    case 'message/send':
      return handleSendMessage(params, requestId, res);
    case 'message/stream':
      return handleStreamMessage(params, requestId, res);
    case 'tasks/get':
      return handleGetTask(params, requestId, res);
    default:
      return res.json({
        jsonrpc: '2.0',
        id: requestId,
        error: { code: -32601, message: `Method not found: ${method}` },
      });
  }
});


function handleSendMessage(params: any, requestId: string, res: express.Response) {
  const userText = params.message.parts
    .filter((p: any) => p.kind === 'text')
    .map((p: any) => p.text)
    .join(' ');


  const taskId = uuidv4();
  const result = processInvoice(userText);


  const task = {
    kind: 'task',
    id: taskId,
    contextId: params.metadata?.contextId || uuidv4(),
    status: { state: 'completed' },
    artifacts: [{ artifactId: uuidv4(), parts: [{ kind: 'data', data: result }], lastChunk: true }],
  };


  tasks[taskId] = task;
  res.json({ jsonrpc: '2.0', id: requestId, result: task });
}


function handleStreamMessage(params: any, requestId: string, res: express.Response) {
  res.setHeader('Content-Type', 'text/event-stream');
  res.setHeader('Cache-Control', 'no-cache');
  res.setHeader('Connection', 'keep-alive');


  const taskId = uuidv4();
  const contextId = params.metadata?.contextId || uuidv4();


  // Emit working status
  res.write(`data: ${JSON.stringify({
    jsonrpc: '2.0',
    method: 'tasks/statusUpdate',
    params: { taskId, contextId, status: { state: 'working' } },
  })}\n\n`);


  setTimeout(() => {
    const result = processInvoice(
      params.message.parts.filter((p: any) => p.kind === 'text').map((p: any) => p.text).join(' ')
    );


    // Emit artifact
    res.write(`data: ${JSON.stringify({
      jsonrpc: '2.0',
      method: 'tasks/artifactUpdate',
      params: {
        taskId,
        contextId,
        artifact: { artifactId: uuidv4(), parts: [{ kind: 'data', data: result }], lastChunk: true },
      },
    })}\n\n`);


    // Emit completed status
    res.write(`data: ${JSON.stringify({
      jsonrpc: '2.0',
      method: 'tasks/statusUpdate',
      params: { taskId, contextId, status: { state: 'completed' } },
    })}\n\n`);


    res.end();
  }, 1000);
}


function handleGetTask(params: any, requestId: string, res: express.Response) {
  const task = tasks[params.id];
  if (!task) {
    return res.json({
      jsonrpc: '2.0',
      id: requestId,
      error: { code: -32602, message: 'Task not found' },
    });
  }
  res.json({ jsonrpc: '2.0', id: requestId, result: task });
}


function processInvoice(text: string) {
  return {
    vendor: 'Acme Corp',
    invoiceNumber: 'INV-2026-0847',
    lineItems: [
      { description: 'Consulting services', quantity: 40, unitPrice: 150, total: 6000 },
      { description: 'Travel expenses', quantity: 1, unitPrice: 1200, total: 1200 },
    ],
    subtotal: 7200,
    tax: 648,
    total: 7848,
  };
}


app.listen(4000, () => {
  console.log('Invoice Processing Agent running on port 4000');
});

Troubleshooting

Section titled “Troubleshooting”
  • Streaming responses aren’t received by the client — confirm capabilities.streaming is true in your Agent Card and that you emit SSE events on message/stream.
  • Tasks stay in working indefinitely — ensure every path calls complete(), fail(), or cancel().
  • Cancellation isn’t honored — implement cancel() and guard against canceling already-completed/canceled tasks.

Further reading

Section titled “Further reading”