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

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 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, 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 and base classes in 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), 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).

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

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:

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

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

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

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}