fix(agentcore): parse A2A JSON-RPC responses in AgentCore provider

litellm/llms/bedrock/chat/agentcore/transformation.py

@@ -19,6 +19,7 @@
 from litellm.litellm_core_utils.streaming_handler import CustomStreamWrapper
 from litellm.llms.base_llm.chat.transformation import BaseConfig, BaseLLMException
 from litellm.llms.bedrock.base_aws_llm import BaseAWSLLM
+from litellm.llms.a2a.common_utils import extract_text_from_a2a_response
 from litellm.llms.bedrock.common_utils import BedrockError
 from litellm.types.llms.bedrock_agentcore import (
     AgentCoreMessage,
@@ -343,6 +344,7 @@ def _parse_json_response(self, response_json: dict) -> AgentCoreParsedResponse:
         Parse direct JSON response (non-streaming).
 
         Supports multiple agent response schemas:
+        0. {"jsonrpc": "2.0", "result": {"message": {"parts": [...]}}} - A2A JSON-RPC
         1. {"result": {"role": "assistant", "content": [{"text": "..."}]}} - standard AgentCore
         2. {"response": [{"text": "..."}]} - Strands agent format
         3. {"result": "plain text"} or {"response": "plain text"} - simple string
@@ -361,6 +363,18 @@ def _parse_json_response(self, response_json: dict) -> AgentCoreParsedResponse:
                 final_message=None,
             )
 
+        # Strategy 0: A2A JSON-RPC format
+        # {"jsonrpc": "2.0", "result": {"message": {"parts": [{"kind": "text", "text": "..."}]}}}
+        if "jsonrpc" in response_json:
+            content = extract_text_from_a2a_response(response_json)
+            if content:
+                return AgentCoreParsedResponse(
+                    content=content,
+                    usage=None,
+                    final_message=None,
+                )
+            # Fall through to other strategies if A2A extraction returned empty
+
         # Strategy 1: {"result": {"content": [{"text": "..."}]}} - standard AgentCore format
         if "result" in response_json and isinstance(response_json["result"], dict):
             result = response_json["result"]

tests/test_litellm/llms/bedrock/chat/agentcore/test_agentcore_transformation.py

@@ -160,6 +160,67 @@ def test_parse_json_empty_content_in_result(self, config):
         assert parsed["content"] == ""
         assert parsed["final_message"] == response_json["result"]
 
+    def test_parse_json_a2a_jsonrpc_nested_message(self, config):
+        """Strategy 0: A2A JSON-RPC with result.message.parts[] format."""
+        response_json = {
+            "jsonrpc": "2.0",
+            "id": "test_id",
+            "result": {
+                "message": {
+                    "role": "agent",
+                    "parts": [{"kind": "text", "text": "1 + 1 = 2"}],
+                    "messageId": "123",
+                }
+            },
+        }
+        parsed = config._parse_json_response(response_json)
+        assert parsed["content"] == "1 + 1 = 2"
+        assert parsed["usage"] is None
+
+    def test_parse_json_a2a_jsonrpc_direct_parts(self, config):
+        """Strategy 0: A2A JSON-RPC with result.parts[] format (direct message)."""
+        response_json = {
+            "jsonrpc": "2.0",
+            "id": "test_id",
+            "result": {
+                "kind": "message",
+                "parts": [{"kind": "text", "text": "Direct response"}],
+            },
+        }
+        parsed = config._parse_json_response(response_json)
+        assert parsed["content"] == "Direct response"
+        assert parsed["usage"] is None
+
+    def test_parse_json_a2a_jsonrpc_multi_parts(self, config):
+        """Strategy 0: A2A JSON-RPC with multiple text parts concatenated."""
+        response_json = {
+            "jsonrpc": "2.0",
+            "id": "test_id",
+            "result": {
+                "message": {
+                    "role": "agent",
+                    "parts": [
+                        {"kind": "text", "text": "First part"},
+                        {"kind": "text", "text": "Second part"},
+                    ],
+                }
+            },
+        }
+        parsed = config._parse_json_response(response_json)
+        assert parsed["content"] == "First part Second part"
+        assert parsed["usage"] is None
+
+    def test_parse_json_a2a_jsonrpc_empty_falls_through(self, config):
+        """Strategy 0: A2A JSON-RPC with empty result falls through to Strategy 3."""
+        response_json = {
+            "jsonrpc": "2.0",
+            "id": "test_id",
+            "result": "plain text fallback",
+        }
+        parsed = config._parse_json_response(response_json)
+        assert parsed["content"] == "plain text fallback"
+        assert parsed["usage"] is None
+
 
 class TestAgentCoreNonStreamingJsonFormats:
     """Tests for _get_parsed_response with different JSON formats (non-streaming path)."""