Skip to content

refactor(a2a)!: migrate A2A SDK from v0.3 to v1.1 #3923

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
waitasecant wants to merge 8 commits into
base: main
Choose a base branch
from
Open

refactor(a2a)!: migrate A2A SDK from v0.3 to v1.1 #3923

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
59dd6b3
chore(a2a): Migrate A2A to v1.0.0
waitasecant Aug 18, 2026
a256a62
nit(tests): clarify comments regarding eager AgentCard construction i…
waitasecant Aug 18, 2026
f53e6fa
fix(a2a): address review feedback for a2a-sdk v1.1 migration
waitasecant Aug 28, 2026
1fb4c11
fix(a2a): validate app_kwargs, document int→float, add e2e tests
waitasecant Aug 29, 2026
c271d23
fix(a2a): fall back to task.status.message when task has no artifacts
waitasecant Aug 29, 2026
94348fd
Merge remote-tracking branch 'upstream/main' into migrate/a2a-sdk-to-…
waitasecant Sep 28, 2026
fb915de
fix(a2a): port a2a_client tool and merged tests to a2a-sdk v1
waitasecant Sep 28, 2026
9547776
Merge remote-tracking branch 'upstream/main' into migrate/a2a-sdk-to-…
waitasecant Oct 28, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions strands-py/pyproject.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -68,8 +68,8 @@ docs = [
]

a2a = [
"a2a-sdk>=0.3.0,<0.4.0",
"a2a-sdk[sql]>=0.3.0,<0.4.0",
"a2a-sdk>=1.0.0,<2.0.0",

@arielnabavian arielnabavian Aug 27, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

concern — this floor can't import. Two of the new imports do not exist at 1.0.0.

  • from a2a.server.routes import add_a2a_routes_to_fastapi (server.py:14). This lands in a2a/server/routes/fastapi_routes.py in 1.1.0. At 1.0.0, 1.0.1, 1.0.2 and 1.0.3, routes/__init__.py exports only create_agent_card_routes, create_jsonrpc_routes, create_rest_routes and the two context builders.
  • from a2a.helpers import new_data_part, ..., new_text_part (executor.py:21). Both are defined in proto_helpers.py from 1.0.2, but only re-exported from the a2a.helpers package in 1.0.3. At 1.0.2 the package __init__ never imports them, so from a2a.helpers import new_text_part still raises.

So any resolver that picks 1.0.0–1.0.3 fails at import strands.multiagent.a2a with ImportError. Your environment presumably has 1.1.x installed, which is why hatch test is green.

Please bump both entries to >=1.1.0,<2.0.0. The PR and issue titles are worth renaming too — this is a migration to 1.1, not 1.0.

@waitasecant waitasecant Aug 28, 2026 •
edited
Loading

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks, I totally missed it because after adding this constrain, the dependency manager installed the latest version 1.1.2 so CI never caught it. I'll bump to >=1.1.0,<2.0.0 would ensure the missing functions being present.

I'll rename the issue and PR title and remove any false claim in the description.

"a2a-sdk[sql]>=1.0.0,<2.0.0",

@strandly-the-agent strandly-the-agent Aug 27, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🔴 The declared floor doesn't import — a2a-sdk>=1.0.0 is not a version this code runs on.

add_a2a_routes_to_fastapi (imported in server.py) first appears in 1.1.0; new_text_part/new_data_part (imported in executor.py) are only re-exported from a2a.helpers in 1.0.3. Verified against the real wheels:

1.0.0–1.0.2   a2a.helpers: missing   | a2a.server.routes: missing
1.0.3         helpers OK             | still missing
1.1.0+        both present

Any install whose resolver lands on 1.0.x — an existing a2a-sdk~=1.0 constraint, an older lockfile, pip backtracking — gets an ImportError importing strands.multiagent.a2a or the client-only strands.agent.a2a_agent; fresh installs resolve 1.1.2, so CI never exercises the floor.

Suggestion: raise both bounds. Cost: a published lower bound can't be raised later without that itself being a breaking dependency change, so this is cheapest before release.

Suggested change
"a2a-sdk>=1.0.0,<2.0.0",
"a2a-sdk[sql]>=1.0.0,<2.0.0",
"a2a-sdk>=1.1.0,<2.0.0",
"a2a-sdk[sql]>=1.1.0,<2.0.0",

@waitasecant waitasecant Aug 29, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Addressed as suggested.

"uvicorn>=0.34.2,<1.0.0",
"httpx>=0.28.1,<1.0.0",
"fastapi>=0.133.0,<1.0.0",
Expand Down
88 changes: 17 additions & 71 deletions strands-py/src/strands/agent/a2a_agent.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,14 +15,10 @@

import httpx
from a2a.client import A2ACardResolver, ClientConfig, ClientFactory
from a2a.types import AgentCard, Message, TaskArtifactUpdateEvent, TaskStatusUpdateEvent
from a2a.types import AgentCard, SendMessageRequest

from .._async import run_async
from ..multiagent.a2a._converters import (
_STATE_TO_STOP_REASON,
convert_input_to_message,
convert_response_to_agent_result,
)
from ..multiagent.a2a._converters import convert_input_to_message, convert_responses_to_agent_result
from ..types._events import AgentResultEvent
from ..types.a2a import A2AResponse, A2AStreamEvent
from ..types.agent import AgentInput
Expand All @@ -33,13 +29,6 @@

_DEFAULT_TIMEOUT = 300

# A2A task states that indicate the response stream is complete.
# Derived from the canonical _STATE_TO_STOP_REASON mapping in _converters.
# Terminal states (end_turn) mean no more events; input states (interrupt) mean execution is paused.
_TERMINAL_STATES = {state for state, reason in _STATE_TO_STOP_REASON.items() if reason == "end_turn"}
_INPUT_STATES = {state for state, reason in _STATE_TO_STOP_REASON.items() if reason == "interrupt"}
_COMPLETE_STATES = _TERMINAL_STATES | _INPUT_STATES


class A2AAgent(AgentBase):
"""Client wrapper for remote A2A agents."""
Expand Down Expand Up @@ -157,9 +146,9 @@ async def stream_async(

Yields:
An async iterator that yields events. Each event is a dictionary:
- A2AStreamEvent: {"type": "a2a_stream", "event": <A2A object>}
where the A2A object can be a Message, or a tuple of
(Task, TaskStatusUpdateEvent) or (Task, TaskArtifactUpdateEvent).
- A2AStreamEvent: {"type": "a2a_stream", "event": <StreamResponse>}
where the StreamResponse carries exactly one of a task, message,
status_update, or artifact_update.
- AgentResultEvent: {"result": AgentResult} - always emitted last.

Raises:
Expand All @@ -174,20 +163,14 @@ async def stream_async(
print(f"Final result: {event['result'].message}")
```
"""
last_event = None
last_complete_event = None

async for event in self._send_message(prompt):
last_event = event
if self._is_complete_event(event):
last_complete_event = event
yield A2AStreamEvent(event)
responses: list[A2AResponse] = []

# Use the last complete event if available, otherwise fall back to last event
final_event = last_complete_event or last_event
async for response in self._send_message(prompt):
responses.append(response)
yield A2AStreamEvent(response)

if final_event is not None:
result = convert_response_to_agent_result(final_event)
if responses:
result = convert_responses_to_agent_result(responses)
yield AgentResultEvent(result)

async def get_agent_card(self) -> AgentCard:
Expand All @@ -214,11 +197,11 @@ async def get_agent_card(self) -> AgentCard:
self._agent_card = await resolver.get_agent_card()

# Populate name from card if not set
if self.name is None and self._agent_card.name is not None:
if self.name is None and self._agent_card.name:
self.name = self._agent_card.name

# Populate description from card if not set
if self.description is None and self._agent_card.description is not None:
if self.description is None and self._agent_card.description:
self.description = self._agent_card.description

logger.debug("agent=<%s>, endpoint=<%s> | discovered agent card", self.name, self.endpoint)
Expand Down Expand Up @@ -258,7 +241,7 @@ async def _send_message(self, prompt: AgentInput) -> AsyncIterator[A2AResponse]:
prompt: Input to send to the agent.

Yields:
A2A response events.
A2A StreamResponse events.

Raises:
ValueError: If prompt is None.
Expand All @@ -267,46 +250,9 @@ async def _send_message(self, prompt: AgentInput) -> AsyncIterator[A2AResponse]:
raise ValueError("prompt is required for A2AAgent")

message = convert_input_to_message(prompt)
request = SendMessageRequest(message=message)
logger.debug("agent=<%s>, endpoint=<%s> | sending message", self.name, self.endpoint)

async with self._get_a2a_client() as client:
async for event in client.send_message(message):
yield event

def _is_complete_event(self, event: A2AResponse) -> bool:
"""Check if an A2A event represents a complete response.

Recognizes all terminal states (completed, failed, canceled, rejected)
and pausing states (input_required, auth_required) as complete events.

Args:
event: A2A event.

Returns:
True if the event represents a complete response.
"""
# Direct Message is always complete
if isinstance(event, Message):
return True

# Handle tuple responses (Task, UpdateEvent | None)
if isinstance(event, tuple) and len(event) == 2:
task, update_event = event

# Initial task response (no update event)
if update_event is None:
return True

# Artifact update with last_chunk flag
if isinstance(update_event, TaskArtifactUpdateEvent):
if hasattr(update_event, "last_chunk") and update_event.last_chunk is not None:
return update_event.last_chunk
return False

# Status update - check for terminal or pausing states
if isinstance(update_event, TaskStatusUpdateEvent):
if update_event.status and hasattr(update_event.status, "state"):
state = update_event.status.state
return state in _COMPLETE_STATES

return False
async for response in client.send_message(request):
yield response
141 changes: 77 additions & 64 deletions strands-py/src/strands/multiagent/a2a/_converters.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,10 +1,11 @@
"""Conversion functions between Strands and A2A types."""

from collections.abc import Sequence
from typing import cast
from uuid import uuid4

from a2a.types import Message as A2AMessage
from a2a.types import Part, Role, TaskArtifactUpdateEvent, TaskState, TaskStatusUpdateEvent, TextPart
from a2a.types import Part, Role, TaskState

from ...agent.agent_result import AgentResult
from ...telemetry.metrics import EventLoopMetrics
Expand All @@ -15,15 +16,25 @@

# Mapping from A2A TaskState to Strands stop_reason
_STATE_TO_STOP_REASON: dict[TaskState, StopReason] = {
TaskState.completed: "end_turn",
TaskState.failed: "end_turn",
TaskState.canceled: "end_turn",
TaskState.rejected: "end_turn",
TaskState.input_required: "interrupt",
TaskState.auth_required: "interrupt",
TaskState.TASK_STATE_COMPLETED: "end_turn",
TaskState.TASK_STATE_FAILED: "end_turn",
TaskState.TASK_STATE_CANCELED: "end_turn",
TaskState.TASK_STATE_REJECTED: "end_turn",
TaskState.TASK_STATE_INPUT_REQUIRED: "interrupt",
TaskState.TASK_STATE_AUTH_REQUIRED: "interrupt",
}


def _task_state_to_str(task_state: TaskState) -> str:
"""Render a TaskState as the kebab-case wire string used by the A2A v0.3 JSON spec.

v1.0's protobuf enum names are SCREAMING_SNAKE_CASE (e.g. ``TASK_STATE_INPUT_REQUIRED``);
this keeps ``AgentResult.state["a2a_task_state"]`` stable for existing Strands callers.
"""
name: str = TaskState.Name(task_state) # type: ignore[attr-defined]
return name.removeprefix("TASK_STATE_").lower().replace("_", "-")


def convert_input_to_message(prompt: AgentInput) -> A2AMessage:
"""Convert AgentInput to A2A Message.

Expand All @@ -40,9 +51,8 @@ def convert_input_to_message(prompt: AgentInput) -> A2AMessage:

if isinstance(prompt, str):
return A2AMessage(
kind="message",
role=Role.user,
parts=[Part(TextPart(kind="text", text=prompt))],
role=Role.ROLE_USER,
parts=[Part(text=prompt)],
message_id=message_id,
)

Expand All @@ -57,16 +67,14 @@ def convert_input_to_message(prompt: AgentInput) -> A2AMessage:
content = cast(list[ContentBlock], msg.get("content", []))
parts = convert_content_blocks_to_parts(content)
return A2AMessage(
kind="message",
role=Role.user,
role=Role.ROLE_USER,
parts=parts,
message_id=message_id,
)
else:
parts = convert_content_blocks_to_parts(cast(list[ContentBlock], prompt))
return A2AMessage(
kind="message",
role=Role.user,
role=Role.ROLE_USER,
parts=parts,
message_id=message_id,
)
Expand All @@ -86,29 +94,45 @@ def convert_content_blocks_to_parts(content_blocks: list[ContentBlock]) -> list[
parts = []
for block in content_blocks:
if "text" in block:
parts.append(Part(TextPart(kind="text", text=block["text"])))
parts.append(Part(text=block["text"]))
return parts


def _extract_task_state(response: A2AResponse) -> TaskState | None:
"""Extract the task state from an A2A response.
"""Extract the task state carried by a single A2A StreamResponse, if any.

Args:
response: A2A response (either A2AMessage or tuple of task and update event).
response: A single StreamResponse from the A2A event stream.

Returns:
The TaskState if available, None otherwise.
The TaskState if this response carries one (a ``task`` or ``status_update``), else None.
"""
if isinstance(response, tuple) and len(response) == 2:
_task, update_event = response
if isinstance(update_event, TaskStatusUpdateEvent):
if update_event.status and hasattr(update_event.status, "state"):
return update_event.status.state
if response.HasField("status_update"):
return response.status_update.status.state
if response.HasField("task"):
return response.task.status.state
return None


def convert_response_to_agent_result(response: A2AResponse) -> AgentResult:
"""Convert A2A response to AgentResult.
def _parts_to_content(parts: Sequence[Part]) -> list[ContentBlock]:
"""Convert a sequence of A2A text Parts into Strands ContentBlocks, dropping non-text parts."""
return [{"text": part.text} for part in parts if part.HasField("text")]


def convert_responses_to_agent_result(responses: Sequence[A2AResponse]) -> AgentResult:
"""Convert the full sequence of A2A StreamResponse events from one call into an AgentResult.

A2A v1.0 streams flat ``StreamResponse`` events (``task`` | ``message`` | ``status_update`` |
``artifact_update``) rather than pairing a cumulative Task with each update, so the final
content is reconstructed across the whole stream:
- ``artifact_update`` parts accumulate: a compliant-streaming server sends incremental deltas,
while a non-compliant server that sends the full text once still works, as that is then the
only chunk.
- ``status_update`` messages are progress narration on non-compliant-streaming servers and
would duplicate the artifact content above, so they are only used when no artifact content
was found (e.g. a status-only terminal event such as a rejection or failure message).
- a bare ``task`` or ``message`` response (no separate update events) supplies content
directly.

Maps A2A task lifecycle states to appropriate Strands stop_reasons:
- completed → end_turn
Expand All @@ -119,61 +143,50 @@ def convert_response_to_agent_result(response: A2AResponse) -> AgentResult:
- auth_required → interrupt (agent needs authentication)

Args:
response: A2A response (either A2AMessage or tuple of task and update event).
responses: All StreamResponse events observed for one ``send_message`` call, in order.

Returns:
AgentResult with extracted content and metadata.
"""
content: list[ContentBlock] = []
task_state = _extract_task_state(response)
artifact_content: list[ContentBlock] = []
status_message_content: list[ContentBlock] = []
task_content: list[ContentBlock] = []
message_content: list[ContentBlock] = []
task_state: TaskState | None = None

for response in responses:
state = _extract_task_state(response)
if state is not None:
task_state = state

if response.HasField("artifact_update"):
artifact_content.extend(_parts_to_content(response.artifact_update.artifact.parts))

@strandly-the-agent strandly-the-agent Aug 27, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟡 artifact_update.append is ignored, so a server that re-sends cumulative artifacts duplicates its whole answer.

server re-sends the full artifact each turn (append=False, i.e. replace):
  content=[{'text': 'Hel'}, {'text': 'Hello'}, {'text': 'Hello world'}]

append=False means replace in the A2A schema, but every update unconditionally extends. Reached by A2AAgent against any third-party A2A server that streams cumulative rather than delta artifacts — interop with non-Strands peers is the client wrapper's whole purpose. The docstring's assumption ("a non-compliant server … sends the full text once") holds for A2AServer, not for arbitrary peers; last_chunk is likewise unused.

Suggestion: key accumulation on artifact.artifact_id and reset that entry when append is false. Cost: a dict keyed by artifact id instead of a flat list — which also handles multi-artifact responses, and pairs naturally with folding accumulation into the stream_async loop (removing its full-stream buffer).

@waitasecant waitasecant Aug 29, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fixed as part of the same _ResponseAccumulator change.

elif response.HasField("status_update"):
if response.status_update.status.HasField("message"):
status_message_content = _parts_to_content(response.status_update.status.message.parts)
elif response.HasField("task"):
task_content = [
content for artifact in response.task.artifacts for content in _parts_to_content(artifact.parts)
]
elif response.HasField("message"):
message_content = _parts_to_content(response.message.parts)

content = artifact_content or status_message_content or task_content or message_content

@arielnabavian arielnabavian Aug 27, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

concern — input_required loses the one thing the caller needs.

The docstring above treats status messages as duplicate narration. True for a terminal completed; not true for input_required / auth_required, where the status message is the payload. _handle_interrupt_result puts "what input is needed" in a text part and the interrupt ids in a data part, then calls updater.requires_input(...).

With enable_a2a_compliant_streaming=True the stream is:

  1. task (submitted)
  2. artifact_update × N — the agent's text deltas
  3. status_update TASK_STATE_INPUT_REQUIRED, carrying the question

artifact_content is non-empty from step 2, so or short-circuits and step 3's question is dropped. The caller gets stop_reason="interrupt" with content that never says what is wanted. Any agent that emits text before hitting an interrupt lands here — which is the normal case, since interrupts are raised from tools.

0.3 did not have this problem: _is_complete_event selected the last complete event, which was the input_required status update, so the question was the content.

Appending instead of falling back when the final state is an interrupt would restore it:

content = artifact_content or task_content or message_content
if _STATE_TO_STOP_REASON.get(task_state) == "interrupt" or not content:
    content = content + status_message_content

On tests: test_artifact_update_then_status_update_does_not_duplicate covers artifact + empty terminal status. The uncovered combination is artifact + input_required-carrying-a-message, which is precisely what the interrupt feature emits. Worth adding whichever way you resolve the behaviour.

@waitasecant waitasecant Aug 28, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Will make a combined fix for this as well as #3923 (comment)

@strandly-the-agent strandly-the-agent Aug 27, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🔴 A terminal status message is silently dropped whenever any artifact content exists — the caller gets stop_reason="interrupt" with no prompt text.

[artifact("partial answer"), status_update(INPUT_REQUIRED, msg="…need approval")]
  → stop_reason='interrupt'  content=[{'text': 'partial answer'}]      # prompt gone
[status_update alone]
  → content=[{'text': 'Agent requires input: …need approval'}]         # control

Reached from A2AAgent.invoke_async() (and graph nodes) against A2AServer(enable_a2a_compliant_streaming=True) whose agent interrupts or fails: requires_input/failed put the actionable text in the terminal status message after artifact deltas have streamed, and the server warns users on every request to turn that flag on. v0.3 surfaced this text. The rationale in the docstring above holds for legacy WORKING narration but not for a terminal message.

Suggestion: capture a status message whose status.state is in _STATE_TO_STOP_REASON into its own accumulator and append it here, leaving non-terminal messages as today's fallback. A blanket precedence swap is not the fix — it drops the streamed answer instead. Cost: one extra local, plus the missing test (nothing covers artifact + status with a message — that mutation survives the suite).

@waitasecant waitasecant Aug 29, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Replaced the flat artifact_content or status_message_content or task_content or message_content chain with a _ResponseAccumulator that tracks artifact content and terminal-status-message content separately.

Added a test covering artifact deltas + a terminal input_required message together, plus an end-to-end integ test (test_a2a_agent_interrupt_round_trip) that asserts the interrupt question text survives through A2AAgent.

stop_reason: StopReason = _STATE_TO_STOP_REASON.get(task_state, "end_turn") if task_state else "end_turn"

if isinstance(response, tuple) and len(response) == 2:
task, update_event = response

# Handle artifact updates
if isinstance(update_event, TaskArtifactUpdateEvent):
if update_event.artifact and hasattr(update_event.artifact, "parts") and update_event.artifact.parts:
for part in update_event.artifact.parts:
if hasattr(part, "root") and hasattr(part.root, "text"):
content.append({"text": part.root.text})
# Handle status updates with messages
elif isinstance(update_event, TaskStatusUpdateEvent):
if (
update_event.status
and hasattr(update_event.status, "message")
and update_event.status.message
and update_event.status.message.parts
):
for part in update_event.status.message.parts:
if hasattr(part, "root") and hasattr(part.root, "text"):
content.append({"text": part.root.text})

# Use task.artifacts when no content was extracted from the event
if not content and task and hasattr(task, "artifacts") and task.artifacts is not None:
for artifact in task.artifacts:
if hasattr(artifact, "parts") and artifact.parts:
for part in artifact.parts:
if hasattr(part, "root") and hasattr(part.root, "text"):
content.append({"text": part.root.text})
elif isinstance(response, A2AMessage):
for part in response.parts:
if hasattr(part, "root") and hasattr(part.root, "text"):
content.append({"text": part.root.text})

message: Message = {
"role": "assistant",
"content": content,
}

# Build state dict with A2A metadata
state: dict[str, str] = {}
state_dict: dict[str, str] = {}
if task_state is not None:
state["a2a_task_state"] = task_state.value
state_dict["a2a_task_state"] = _task_state_to_str(task_state)

return AgentResult(
stop_reason=stop_reason,
message=message,
metrics=EventLoopMetrics(),
state=state,
state=state_dict,
)
Loading
Loading