# A2A advanced patterns

LangGraph, TypeScript, Java, and native (no-SDK) implementations of an A2A server agent, plus streaming progress and multi-turn user input.

This page covers building A2A server agents beyond the [Python quickstart](/guides/a2a-quickstart/): orchestrating with [LangGraph](https://www.langchain.com/langgraph), the TypeScript and Java SDKs, and implementing the protocol natively without an SDK.

## LangGraph integration

[LangGraph](https://langchain-ai.github.io/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](https://langchain-ai.github.io/langgraph/) and the [`langchain-ai/langgraph` repository](https://github.com/langchain-ai/langgraph).

### Basic LangGraph agent

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

```python title="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

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

```python
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, transition to `input-required` — the Copilot prompts the user and relays their reply back.

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

Install the SDK and Express:

```bash
npm install @a2a-js/sdk express
```

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

```typescript title="agent.ts"

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:

```xml
<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:

```java
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:

```java
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)

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

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

```python title="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

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

```typescript title="agent.ts"

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

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

- [A2A Protocol Specification](https://a2a-protocol.org/latest/specification/)
- Official SDKs: [Python](https://github.com/a2aproject/a2a-python) · [TypeScript](https://github.com/a2aproject/a2a-js) · [Java](https://github.com/a2aproject/a2a-java)
- [LangGraph documentation](https://langchain-ai.github.io/langgraph/) · [`langchain-ai/langgraph`](https://github.com/langchain-ai/langgraph)
- [Technology Compatibility Kit (TCK)](https://github.com/a2aproject/a2a-tck)
- Next: [Connect your agent to Foldspace →](/guides/a2a-connect/)