Subscribe post.

Author: bartek-w

src/a2a/client/transports/rest.py

@@ -262,6 +262,7 @@ async def subscribe(
             f'/tasks/{request.id}:subscribe',
             request.tenant,
             context=context,
+            json=MessageToDict(request),
         ):
             yield event
 

src/a2a/compat/v0_3/rest_transport.py

@@ -287,11 +287,17 @@ async def subscribe(
         to POST for next calls but will not retry again.
         """
         subscribe_method = self._subscribe_method_override or 'POST'
+        if subscribe_method == 'POST':
+            json_body = MessageToDict(request, preserving_proto_field_name=True)
+        else:
+            json_body = None
+
         try:
             async for event in self._send_stream_request(
                 subscribe_method,
                 f'/v1/tasks/{request.id}:subscribe',
                 context=context,
+                json=json_body,
             ):
                 yield event
         except A2AClientError as e:

src/a2a/server/request_handlers/rest_handler.py

@@ -159,9 +159,15 @@ async def on_subscribe_to_task(
         Yields:
             JSON serialized objects containing streaming events
         """
-        task_id = request.path_params['id']
+        params = SubscribeToTaskRequest()
+        if request.method == 'POST':
+            body = await request.body()
+            if body:
+                Parse(body, params)
+
+        params.id = request.path_params['id']
         async for event in self.request_handler.on_subscribe_to_task(
-            SubscribeToTaskRequest(id=task_id), context
+            params, context
         ):
             yield MessageToDict(proto_utils.to_stream_response(event))
 

tests/client/transports/__init__.py

@@ -9,6 +9,7 @@
 from httpx_sse import EventSource, ServerSentEvent
 
 from a2a.client import create_text_message_object
+from a2a.client.client import ClientCallContext
 from a2a.client.errors import A2AClientError
 from a2a.client.transports.rest import RestTransport
 from a2a.extensions.common import HTTP_EXTENSION_HEADER
@@ -162,7 +163,6 @@ async def test_send_message_with_timeout_context(
         self, mock_httpx_client: AsyncMock, mock_agent_card: MagicMock
     ):
         """Test that send_message passes context timeout to build_request."""
-        from a2a.client.client import ClientCallContext
 
         client = RestTransport(
             httpx_client=mock_httpx_client,
@@ -258,7 +258,6 @@ async def test_send_message_with_default_extensions(
         mock_response.status_code = 200
         mock_httpx_client.send.return_value = mock_response
 
-        from a2a.client.client import ClientCallContext
 
         context = ClientCallContext(
             service_parameters={
@@ -302,7 +301,6 @@ async def test_send_message_streaming_with_new_extensions(
             mock_event_source
         )
 
-        from a2a.client.client import ClientCallContext
 
         context = ClientCallContext(
             service_parameters={
@@ -404,7 +402,6 @@ async def test_get_card_with_extended_card_support_with_extensions(
 
         request = GetExtendedAgentCardRequest()
 
-        from a2a.client.client import ClientCallContext
 
         context = ClientCallContext(
             service_parameters={HTTP_EXTENSION_HEADER: extensions_str}
@@ -419,7 +416,6 @@ async def test_get_card_with_extended_card_support_with_extensions(
             await client.get_extended_agent_card(request, context=context)
 
         mock_execute_request.assert_called_once()
-        # _execute_request(method, target, tenant, context)
         call_args = mock_execute_request.call_args
         assert (
             call_args[1].get('context') == context or call_args[0][3] == context
@@ -694,7 +690,8 @@ async def test_rest_get_task_prepend_empty_tenant(
     )
     @pytest.mark.asyncio
     @patch('a2a.client.transports.http_helpers.aconnect_sse')
-    async def test_rest_streaming_methods_prepend_tenant(
+    async def test_rest_streaming_methods_prepend_tenant(  # noqa: PLR0913
+
         self,
         mock_aconnect_sse,
         method_name,
@@ -735,10 +732,7 @@ async def empty_aiter():
         args, kwargs = mock_aconnect_sse.call_args
         # method is 2nd positional argument
         assert args[1] == 'POST'
-        if method_name == 'subscribe':
-            assert kwargs.get('json') is None
-        else:
-            assert kwargs.get('json') == json_format.MessageToDict(request_obj)
+        assert kwargs.get('json') == json_format.MessageToDict(request_obj)
 
         # url is 3rd positional argument in aconnect_sse(client, method, url, ...)
         assert args[2] == f'http://agent.example.com/api{expected_path}'

tests/client/transports/test_rest_client.py

@@ -270,7 +270,10 @@ async def test_compat_rest_transport_subscribe_post_works_no_retry(transport):
 
     async def mock_stream(method, path, context=None, json=None):
         assert method == 'POST'
-        assert json is None
+        if method == 'POST':
+            assert json == {'id': 'task-123'}
+        else:
+            assert json is None
         task = Task(id='task-123')
         task.status.message.role = Role.ROLE_AGENT
         yield StreamResponse(task=task)
@@ -298,7 +301,10 @@ async def mock_stream(method, path, context=None, json=None):
         nonlocal call_count
         call_count += 1
         if method == 'POST':
-            assert json is None
+            if method == 'POST':
+                assert json == {'id': 'task-123'}
+            else:
+                assert json is None
             create_405_error()
         if method == 'GET':
             assert json is None
@@ -334,8 +340,8 @@ async def test_compat_rest_transport_subscribe_post_405_get_405_fails(
     async def mock_stream(method, path, context=None, json=None):
         method_count[method] = method_count.get(method, 0) + 1
         if method == 'POST':
-            assert json is None
-        elif method == 'GET':
+            assert json == {'id': 'task-123'}
+        else:
             assert json is None
         # To make it an async generator even when it raises
         if False:
@@ -370,7 +376,10 @@ async def mock_stream(method, path, context=None, json=None):
         nonlocal call_count
         call_count += 1
         assert method == 'POST'
-        assert json is None
+        if method == 'POST':
+            assert json == {'id': 'task-123'}
+        else:
+            assert json is None
         if False:
             yield
         create_500_error()
@@ -435,11 +444,10 @@ async def mock_stream(method, path, context=None, json=None):
         nonlocal call_count
         call_count += 1
         assert method == 'POST'
-        assert json is None
+        assert json == {'id': 'task-123'}
         if False:
             yield
         create_405_error()
-
     transport._send_stream_request = mock_stream
 
     req = SubscribeToTaskRequest(id='task-123')

tests/compat/v0_3/test_rest_transport.py

@@ -1,5 +1,5 @@
-import logging
 import json
+import logging
 
 from typing import Any
 from unittest.mock import MagicMock
@@ -438,7 +438,7 @@ async def mock_stream_response():
 async def test_subscribe_to_task_post(
     streaming_client: AsyncClient, request_handler: MagicMock
 ) -> None:
-    """Test that POST /tasks/{id}:subscribe works."""
+    """Test that POST /tasks/{id}:subscribe works and parses body."""
 
     async def mock_stream_response():
         yield Task(
@@ -449,8 +449,11 @@ async def mock_stream_response():
 
     request_handler.on_subscribe_to_task.return_value = mock_stream_response()
 
+    request = a2a_pb2.SubscribeToTaskRequest(id='task-1')
+
     response = await streaming_client.post(
         '/tasks/task-1:subscribe',
+        json=json_format.MessageToDict(request),
         headers={'Accept': 'text/event-stream'},
     )
 
@@ -592,7 +595,7 @@ def extended_card_modifier(self) -> MagicMock:
             ('/message:send', 'POST', 'on_message_send', {'message': {}}),
             ('/tasks/1:cancel', 'POST', 'on_cancel_task', None),
             ('/tasks/1:subscribe', 'GET', 'on_subscribe_to_task', None),
-            ('/tasks/1:subscribe', 'POST', 'on_subscribe_to_task', None),
+            ('/tasks/1:subscribe', 'POST', 'on_subscribe_to_task', {'id': '1'}),
             ('/tasks/1', 'GET', 'on_get_task', None),
             ('/tasks', 'GET', 'on_list_tasks', None),
             (