Advanced A2A Server Agent Patterns

Prerequisite: Read the Overview and Quickstart first.

LangGraph Integration

LangGraph provides a graph-based orchestration layer for building stateful, multi-step agents. You can wrap any LangGraph agent as an A2A server using the AgentExecutor bridge pattern.

Basic LangGraph Agent

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

For long-running agents, stream intermediate updates 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)

When your agent needs clarification before proceeding:

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()

When the Foldspace Copilot receives input-required, it prompts the user and sends a follow-up message to resume the task.

TypeScript SDK

Install:

npm install @a2a-js/sdk express

Example: Document Analysis Agent

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

Maven dependency:

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

Agent Card (Quarkus CDI):

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();
    }
}

Agent Executor:

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 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

GET /.well-known/agent.json — Serve your Agent Card
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

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

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

Problem: Streaming responses are not received by the client.
What to do: Verify your SSE response includes Content-Type: text/event-stream, each event is data: <json>\n\n, and your Agent Card has streaming: true in capabilities.
Problem: Tasks stay in working state forever.
What to do: Ensure your executor always calls complete(), fail(), or cancel() on every code path — including error handling.
Problem: cancel() is not implemented.
What to do: Even if your agent doesn't support cancellation, implement the method and throw an appropriate error (TaskNotCancelableError in Java, raise Exception in Python).