fix(a2a): advertise streaming on agent_registry agent cards

agent_registry called build_agent_card without capabilities or streaming,
so constructed cards advertised streaming:false. Pass streaming=True on
that convenience path and document that streaming= is ignored when
capabilities is provided.

Fixes #6778

Author: lllakshit

src/google/adk/a2a/_compat.py

@@ -485,6 +485,10 @@ def build_agent_card(
          transport is ``preferredTransport``.
   1.x:   ``AgentCard`` is a proto message — RPC URL lives in
          ``supported_interfaces[i].url`` (with ``protocol_binding``).
+
+  ``streaming`` is the no-capabilities convenience path: it is applied only
+  when ``capabilities`` is omitted. A passed ``capabilities`` object is used
+  as-is, so ``streaming`` has no effect on that call.
   """
 
   def _as_dict(obj: Any) -> Any:

src/google/adk/integrations/agent_registry/agent_registry.py

@@ -637,6 +637,8 @@ def get_remote_a2a_agent(
       )
 
     binding = protocol_binding or _compat.TP_HTTP_JSON
+    # Registry metadata has no capabilities object. Pass streaming=True so
+    # constructed cards do not advertise streaming:false.
     agent_card = _compat.build_agent_card(
         name=name,
         description=description,
@@ -647,6 +649,7 @@ def get_remote_a2a_agent(
         skills=skills,
         default_input_modes=["text"],
         default_output_modes=["text"],
+        streaming=True,
     )
 
     return RemoteA2aAgent(

tests/unittests/a2a/test_compat.py

@@ -94,6 +94,43 @@ def _build_card(**overrides):
   return _compat.build_agent_card(**kwargs)
 
 
+def _registry_shaped_card(**overrides):
+  """Call ``build_agent_card`` without a capabilities object."""
+  kwargs = dict(
+      name='x',
+      description='d',
+      version='1',
+      url='http://h/a',
+      protocol_binding=getattr(
+          _compat.TP_HTTP_JSON, 'value', _compat.TP_HTTP_JSON
+      ),
+      skills=[],
+      default_input_modes=['text'],
+      default_output_modes=['text'],
+  )
+  kwargs.update(overrides)
+  return _compat.build_agent_card(**kwargs)
+
+
+def test_build_agent_card_registry_shape_defaults_streaming_false():
+  """Omitting capabilities still defaults streaming to false on the helper."""
+  assert _registry_shaped_card().capabilities.streaming is False
+
+
+def test_build_agent_card_streaming_true_without_capabilities():
+  """streaming= is honoured when capabilities is omitted."""
+  assert _registry_shaped_card(streaming=True).capabilities.streaming is True
+
+
+def test_build_agent_card_capabilities_object_ignores_streaming_flag():
+  """A passed capabilities object is used as-is; streaming= does not compose."""
+  card = _registry_shaped_card(
+      streaming=False,
+      capabilities=AgentCapabilities(streaming=True),
+  )
+  assert card.capabilities.streaming is True
+
+
 @v03_only
 def test_build_agent_card_strips_trailing_slash_from_url():
   # The RPC URL is concatenated with paths by callers, so the card must not
@@ -113,6 +150,12 @@ def test_build_agent_card_with_protocol_version_keeps_caller_value():
   assert _build_card(protocol_version='0.2.9').protocol_version == '0.2.9'
 
 
+@v03_only
+def test_build_agent_card_omitted_streaming_defaults_to_false():
+  """Callers that omit capabilities must pass streaming=True to advertise it."""
+  assert _build_card().capabilities.streaming is False
+
+
 @pytest.mark.parametrize('streaming', [True, False])
 @v03_only
 def test_build_agent_card_default_capabilities_follow_streaming_flag(streaming):

tests/unittests/integrations/agent_registry/test_agent_registry.py

@@ -583,6 +583,29 @@ def test_get_remote_a2a_agent_defaults(self, registry):
         agent._agent_card, "1.0" if _compat.IS_A2A_V1 else "0.3.0"
     )
 
+  def test_get_remote_a2a_agent_advertises_streaming(self, registry):
+    """Constructed registry cards advertise streaming when no card is supplied."""
+    mock_response = MagicMock()
+    mock_response.json.return_value = {
+        "displayName": "TestAgent",
+        "description": "Test Desc",
+        "version": "1.0",
+        "protocols": [{
+            "type": _ProtocolType.A2A_AGENT,
+            "interfaces": [{
+                "url": "https://my-agent.com",
+            }],
+        }],
+    }
+    mock_response.raise_for_status = MagicMock()
+    registry._session.get.return_value = mock_response
+
+    registry._credentials.token = "token"
+    registry._credentials.refresh = MagicMock()
+
+    agent = registry.get_remote_a2a_agent("test-agent")
+    assert agent._agent_card.capabilities.streaming is True
+
   def test_get_remote_a2a_agent_with_card(self, registry):
     mock_response = MagicMock()
     mock_response.json.return_value = {