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

site/src/content/docs/user-guide/concepts/multi-agent/agent-to-agent.mdx

@@ -387,9 +387,9 @@ This feature is currently available in the Python SDK only.
 
 When a tool or hook on the served agent raises an [interrupt](../interrupts.md), the task moves to the A2A `input_required` state and waits. The client answers it, and the task resumes the paused tool exactly where it stopped.
 
-Answering an interrupt is the one flow on this page that needs the raw [`a2a-sdk`](https://github.com/a2aproject/a2a-python) client rather than `A2AAgent`. `A2AAgent` speaks in text, so it raises `ValueError` if you pass it interrupt responses, and it drops the `DataPart` carrying the interrupt ids when it reads the reply. The examples below build A2A messages directly.
+Answering an interrupt is the one flow on this page that needs the raw [`a2a-sdk`](https://github.com/a2aproject/a2a-python) client rather than `A2AAgent`. `A2AAgent` speaks in text, so it raises `ValueError` if you pass it interrupt responses, and it drops the data part carrying the interrupt ids when it reads the reply. The examples below build A2A messages directly.
 
-Each interrupt has a server-generated id, and an answer is bound to the id of the interrupt that raised it. The server advertises the pending interrupts on the `input_required` status message as a `DataPart`, alongside the human-readable `TextPart`:
+Each interrupt has a server-generated id, and an answer is bound to the id of the interrupt that raised it. The server advertises the pending interrupts on the `input_required` status message as a data part, alongside the human-readable text part:
 
 ```json
 {
@@ -406,7 +406,7 @@ Each interrupt has a server-generated id, and an answer is bound to the id of th
 }
 ```
 
-To answer, send a new message on the same `taskId` containing a `DataPart` that echoes the `interruptId` back with the response:
+To answer, send a new message on the same `taskId` containing a data part that echoes the `interruptId` back with the response:
 
 ```json
 {
@@ -420,25 +420,31 @@ To answer, send a new message on the same `taskId` containing a `DataPart` that
 }
 ```
 
-The `response` becomes the return value of the `interrupt()` call that paused the tool. It is any JSON value except `null`, which the server refuses — a null answer would leave the interrupt unsatisfied and re-raise it. `false` and `0` are fine. Answer several interrupts in one message by sending one `DataPart` for each.
+The `response` becomes the return value of the `interrupt()` call that paused the tool. It is any JSON value except `null`, which the server refuses — a null answer would leave the interrupt unsatisfied and re-raise it. `false` and `0` are fine. Answer several interrupts in one message by sending one data part for each.
+
+A2A data parts carry numbers as protobuf `Value`, which has no integer type — a whole-number
+float like `1.0` is indistinguishable from the int `1` on the wire, so the server normalizes any
+whole float back to an int before it reaches `response`. A non-whole float such as `1.5` is
+unaffected.
 
 Reading the ids off the status message and answering them:
 
 ```python
-from a2a.types import DataPart, Part
+from a2a.helpers import new_data_part
+from google.protobuf.json_format import MessageToDict
 
 # The task parked in input_required; read the interrupts it is waiting on.
 pending = next(
-    part.root.data["interrupts"]
+    MessageToDict(part.data)["interrupts"]
     for part in task.status.message.parts
-    if isinstance(part.root, DataPart) and "interrupts" in part.root.data
+    if part.HasField("data") and "interrupts" in MessageToDict(part.data)
 )
 
 # Answer each one on the same taskId.
 answers = [
-    Part(root=DataPart(data={
+    new_data_part({
         "interruptResponse": {"interruptId": item["interruptId"], "response": {"approved": True}}
-    }))
+    })
     for item in pending
 ]
 ```
@@ -453,7 +459,7 @@ The server rejects an answer it cannot bind, before the agent runs, so a refused
 
 A task with a pending interrupt also rejects an ordinary conversational message — answer the interrupt, or cancel the task.
 
-A `DataPart` without an `interruptResponse` key is unaffected and continues to reach the agent as structured data.
+A data part without an `interruptResponse` key is unaffected and continues to reach the agent as structured data.
 
 ### Server Configuration Options
 

strands-py/pyproject.toml

@@ -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.1.0,<2.0.0",
+    "a2a-sdk[sql]>=1.1.0,<2.0.0",
     "uvicorn>=0.34.2,<1.0.0",
     "httpx>=0.28.1,<1.0.0",
     "fastapi>=0.133.0,<1.0.0",

strands-py/src/strands/agent/a2a_agent.py

@@ -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
@@ -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."""
@@ -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:
@@ -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:
@@ -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)
@@ -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.
@@ -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