# How the A2A Protocol Enables Agent-to-Agent Communication in AgentScope

> Discover how AgentScope's A2A protocol facilitates agent-to-agent communication. Learn about its three-layer architecture for seamless message streaming and translation.

- Repository: [AgentScope-AI/agentscope](https://github.com/agentscope-ai/agentscope)
- Tags: internals
- Published: 2026-03-09

---

**The A2A protocol in AgentScope enables an `A2AAgent` to communicate with remote agents through a three-layer architecture—agent abstraction, message formatting, and client resolution—that streams messages via the A2A standard while translating between AgentScope's native `Msg` objects and A2A message models.**

The A2A (Agent-to-Agent) protocol implementation in the AgentScope framework allows local agents to delegate tasks to remote services that expose a compatible A2A server. According to the agentscope-ai/agentscope source code, this capability is realized through tightly-coupled layers that handle message observation, protocol translation, and HTTP streaming.

## The Three-Layer A2A Architecture in AgentScope

AgentScope implements the A2A protocol through three distinct layers that work together to bridge local agent execution with remote A2A endpoints:

- **Agent layer** – The `A2AAgent` class in [`src/agentscope/agent/_a2a_agent.py`](https://github.com/agentscope-ai/agentscope/blob/main/src/agentscope/agent/_a2a_agent.py) exposes a high-level interface that behaves like any other AgentScope agent. It manages message observation, lifecycle state, and streaming responses.

- **Formatter layer** – Located in [`src/agentscope/formatter/_a2a_formatter.py`](https://github.com/agentscope-ai/agentscope/blob/main/src/agentscope/formatter/_a2a_formatter.py), this layer translates between AgentScope's internal `Msg` objects and the A2A message model (`Message`, `Task`, `Part`). It merges multiple `Msg` instances into a single A2A `Message` because the protocol accepts only one request per call.

- **Resolver/client layer** – The `WellKnownAgentCardResolver` in [`src/agentscope/a2a/_well_known_resolver.py`](https://github.com/agentscope-ai/agentscope/blob/main/src/agentscope/a2a/_well_known_resolver.py) and base classes in [`src/agentscope/a2a/_base.py`](https://github.com/agentscope-ai/agentscope/blob/main/src/agentscope/a2a/_base.py) handle discovery of remote agent capabilities and creation of concrete A2A clients via `ClientFactory`.

## How A2AAgent Handles Remote Communication

The core workflow occurs within the `reply` method of `A2AAgent` (lines 15‑54 of [`src/agentscope/agent/_a2a_agent.py`](https://github.com/agentscope-ai/agentscope/blob/main/src/agentscope/agent/_a2a_agent.py)), which executes six distinct phases:

### Step 1: Observing and Merging Messages

When an agent needs to maintain context across turns, it calls `observe()` to store `Msg` objects in the private list `self._observed_msgs`. Upon invoking `reply()`, the agent automatically concatenates these observed messages with the explicit input message (`self._observed_msgs` + `msg`), creating a merged conversation history for the remote agent.

### Step 2: Formatting AgentScope Messages to A2A

The merged message list passes through `self.formatter.format`, which constructs a single A2A `Message` with role=`user`. Each `ContentBlock` converts to an appropriate `Part` type:

- **Text and thinking blocks** become `TextPart` objects
- **Media blocks** (image, video, audio) become `FilePart` with either `FileWithUri` (URL source) or `FileWithBytes` (base64 source)
- **Tool usage blocks** become `DataPart` preserving the raw payload

Unsupported block types trigger a logger error (lines 33‑38 of [`_a2a_formatter.py`](https://github.com/agentscope-ai/agentscope/blob/main/_a2a_formatter.py)).

### Step 3: Client Resolution and Streaming

An A2A client is created lazily via `self._a2a_client_factory.create(self.agent_card)`, encapsulating transport configuration including httpx async client settings, streaming flags, and retry policies. The client then calls `send_message(a2a_message)` to stream responses, which may yield either a plain `A2AMessage` or a `(Task, ...)` tuple.

### Step 4: State Cleanup and Response Handling

For plain messages, the formatter converts them back via `format_a2a_message` (lines 61‑84), mapping A2A roles (`Role.user`, `Role.agent`) to AgentScope roles (`"user"`, `"assistant"`). For tasks, `format_a2a_task` extracts status and artifacts. Finally, `self._observed_msgs.clear()` removes cached observations (line 54), and the reconstructed `Msg` returns to the caller.

## Message Formatting Between AgentScope and A2A

The formatter handles bidirectional translation with specific type mappings. When converting to A2A, the formatter guesses MIME types for URLs via Python's `mimetypes` module when not explicitly provided. When converting from A2A back to AgentScope, the resulting `Msg` contains a reconstructed content list preserving the original structure of the remote agent's response.

The A2A protocol strictly accepts only one message per request, so the formatter merges multi-turn conversations automatically. This design means the remote agent receives the full context in a single payload rather than maintaining server-side session state.

## Resolving Remote Agent Cards

Before communication begins, `WellKnownAgentCardResolver` in [`src/agentscope/a2a/_well_known_resolver.py`](https://github.com/agentscope-ai/agentscope/blob/main/src/agentscope/a2a/_well_known_resolver.py) builds the full URL from a base host and optional path, validates it, and uses `a2a.client.A2ACardResolver` to fetch the JSON **AgentCard**. This card supplies the endpoint URL, version string, and capability metadata required by the `ClientFactory` to instantiate the correct client type.

## Current Limitations of the A2A Implementation

The current A2A protocol implementation in AgentScope carries three specific constraints:

- **Single-request per call** – The A2A server only accepts one message per invocation, requiring multi-turn conversations to be merged client-side (handled automatically by the formatter).

- **No structured output support** – The `reply` method deliberately rejects a `structured_model` argument (lines 6‑11), returning tasks as plain text messages or artifact blocks rather than parsed Pydantic models.

- **Observed-message clearing** – The agent clears `self._observed_msgs` after each reply, meaning context must be re-observed if needed for subsequent turns unless the caller maintains external state.

## Practical Implementation Examples

Create an A2A agent from a well-known endpoint and send messages:

```python
from agentscope.agent import A2AAgent
from agentscope.message import Msg
from a2a.types import AgentCard, AgentCapabilities

card = AgentCard(
    name="RemoteHelper",
    url="http://localhost:8000",
    description="A remote chat-bot",
    version="1.0.0",
    capabilities=AgentCapabilities(),
    default_input_modes=["text/plain"],
    default_output_modes=["text/plain"],
    skills=[],
)

agent = A2AAgent(card)

# Observe context that should persist for the next turn

await agent.observe(Msg(name="user", content="Remember my last request.", role="user"))

# Send request; observed messages merge automatically

response = await agent(
    Msg(name="user", content="Now answer my question.", role="user")
)

print("Remote response:", response.content)

```

Handle task-based responses with artifacts:

```python

# Assuming a client factory configured for task responses

response = await agent(Msg(name="user", content="Process the data.", role="user"))

# Iterate through content blocks containing task status and artifacts

for block in response.get_content_blocks():
    print(block)  # e.g., {"type": "text", "text": "Task completed"}

```

Handle conversation interruptions:

```python

# Store interrupt message for next reply continuation

interrupt_msg = await A2AAgent.handle_interrupt()

# The message is stored in _observed_msgs for seamless recovery

```

## Summary

- **Three-layer architecture** – The A2A protocol implementation combines the `A2AAgent` class, message formatter, and card resolver to enable seamless remote delegation.
- **Automatic message merging** – The agent concatenates observed messages with new inputs to satisfy the A2A single-request constraint.
- **Bidirectional translation** – The formatter converts between AgentScope `Msg` objects and A2A `Message`/`Task` models, handling text, media, and tool content blocks.
- **Dynamic endpoint discovery** – `WellKnownAgentCardResolver` fetches remote capability metadata via the A2A AgentCard standard.
- **Stateless design** – Observed messages clear after each reply, requiring explicit re-observation for multi-turn context maintenance.

## Frequently Asked Questions

### What is the primary class for implementing A2A protocol communication in AgentScope?

The **`A2AAgent`** class in [`src/agentscope/agent/_a2a_agent.py`](https://github.com/agentscope-ai/agentscope/blob/main/src/agentscope/agent/_a2a_agent.py) serves as the primary entry point. It extends AgentScope's base agent interface while internally managing an A2A client factory and message formatter to communicate with remote A2A servers.

### How does AgentScope handle conversation history when using the A2A protocol?

AgentScope merges observed messages (stored via `agent.observe()`) with the current input message into a single A2A `Message` before transmission. Because the A2A protocol accepts only one request per call, the formatter concatenates the content blocks automatically, though it clears the observation cache after each `reply()` invocation.

### What content types can be transmitted via AgentScope's A2A implementation?

The formatter supports **text**, **thinking**, **file** (images, video, audio via URL or base64 bytes), and **tool_use** content blocks. These map to A2A `TextPart`, `FilePart`, and `DataPart` objects respectively. Unsupported types trigger errors in [`src/agentscope/formatter/_a2a_formatter.py`](https://github.com/agentscope-ai/agentscope/blob/main/src/agentscope/formatter/_a2a_formatter.py).

### Why can't I use structured output models with A2AAgent?

The current implementation explicitly rejects the `structured_model` parameter (lines 6‑11 of [`_a2a_agent.py`](https://github.com/agentscope-ai/agentscope/blob/main/_a2a_agent.py)) because the A2A protocol returns responses as plain messages or task artifacts rather than structured JSON schemas compatible with Pydantic models. Tasks are rendered as text content blocks or artifact data instead.