Subscribe post fix v2.

Author: bartek-w

src/a2a/client/transports/rest.py

@@ -262,7 +262,6 @@ 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

@@ -64,13 +64,14 @@ def __init__(
         httpx_client: httpx.AsyncClient,
         agent_card: AgentCard | None,
         url: str,
+        subscribe_method_override: str | None = None,
     ):
         """Initializes the CompatRestTransport."""
         self.url = url.removesuffix('/')
         self.httpx_client = httpx_client
         self.agent_card = agent_card
-        self._subscribe_method = 'POST'
-        self._subscribe_retry_attempted = False
+        self._subscribe_method_override = subscribe_method_override
+        self._subscribe_auto_method_override = subscribe_method_override is None
 
     async def send_message(
         self,
@@ -285,17 +286,12 @@ async def subscribe(
         on this transport instance. If both fail with 405, it will default back
         to POST for next calls but will not retry again.
         """
-        if self._subscribe_method == 'POST':
-            json_body = MessageToDict(request, preserving_proto_field_name=True)
-        else:
-            json_body = None
-
+        subscribe_method = self._subscribe_method_override or 'POST'
         try:
             async for event in self._send_stream_request(
-                self._subscribe_method,
+                subscribe_method,
                 f'/v1/tasks/{request.id}:subscribe',
                 context=context,
-                json=json_body,
             ):
                 yield event
         except A2AClientError as e:
@@ -305,12 +301,13 @@ async def subscribe(
                 isinstance(cause, httpx.HTTPStatusError)
                 and cause.response.status_code == httpx.codes.METHOD_NOT_ALLOWED
             ):
-                if self._subscribe_retry_attempted:
-                    self._subscribe_method = 'POST'
+                if self._subscribe_method_override:
+                    if self._subscribe_auto_method_override:
+                        self._subscribe_auto_method_override = False
+                        self._subscribe_method_override = 'POST'
                     raise
                 else:
-                    self._subscribe_method = 'GET'
-                    self._subscribe_retry_attempted = True
+                    self._subscribe_method_override = 'GET'
                     async for event in self.subscribe(request, context=context):
                         yield event
             else:

src/a2a/server/request_handlers/rest_handler.py

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

tests/client/transports/test_rest_client.py

@@ -735,7 +735,10 @@ async def empty_aiter():
         args, kwargs = mock_aconnect_sse.call_args
         # method is 2nd positional argument
         assert args[1] == 'POST'
-        assert kwargs.get('json') == json_format.MessageToDict(request_obj)
+        if method_name == 'subscribe':
+            assert kwargs.get('json') is None
+        else:
+            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/compat/v0_3/test_rest_transport.py

@@ -270,7 +270,7 @@ 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 == {'id': 'task-123'}
+        assert json is None
         task = Task(id='task-123')
         task.status.message.role = Role.ROLE_AGENT
         yield StreamResponse(task=task)
@@ -284,8 +284,7 @@ async def mock_stream(method, path, context=None, json=None):
     expected_task = Task(id='task-123')
     expected_task.status.message.role = Role.ROLE_AGENT
     assert events[0] == StreamResponse(task=expected_task)
-    assert transport._subscribe_method == 'POST'
-    assert transport._subscribe_retry_attempted is False
+    assert transport._subscribe_method_override is None
 
 
 @pytest.mark.asyncio
@@ -299,7 +298,7 @@ async def mock_stream(method, path, context=None, json=None):
         nonlocal call_count
         call_count += 1
         if method == 'POST':
-            assert json == {'id': 'task-123'}
+            assert json is None
             create_405_error()
         if method == 'GET':
             assert json is None
@@ -314,29 +313,28 @@ async def mock_stream(method, path, context=None, json=None):
 
     assert len(events) == 1
     assert call_count == 2
-    assert transport._subscribe_method == 'GET'
-    assert transport._subscribe_retry_attempted is True
+    assert transport._subscribe_method_override == 'GET'
 
     # Second call should use GET directly
     call_count = 0
     events = [event async for event in transport.subscribe(req)]
     assert len(events) == 1
     assert call_count == 1  # Only GET called
-    assert transport._subscribe_method == 'GET'
+    assert transport._subscribe_method_override == 'GET'
 
 
 @pytest.mark.asyncio
 async def test_compat_rest_transport_subscribe_post_405_get_405_fails(
     transport,
 ):
     """Scenario: POST return 405, retry GET, return 405 - error. Second call is just POST."""
-    call_count = 0
+
+    method_count = {}
 
     async def mock_stream(method, path, context=None, json=None):
-        nonlocal call_count
-        call_count += 1
+        method_count[method] = method_count.get(method, 0) + 1
         if method == 'POST':
-            assert json == {'id': 'task-123'}
+            assert json is None
         elif method == 'GET':
             assert json is None
         # To make it an async generator even when it raises
@@ -351,16 +349,16 @@ async def mock_stream(method, path, context=None, json=None):
         [event async for event in transport.subscribe(req)]
 
     assert '405' in str(exc_info.value)
-    assert call_count == 2  # Tried POST then GET
-    assert transport._subscribe_method == 'POST'
-    assert transport._subscribe_retry_attempted is True
+    assert transport._subscribe_method_override == 'POST'
+    assert method_count == {'POST': 1, 'GET': 1}
+    assert transport._subscribe_auto_method_override is False
 
     # Second call should try POST directly and fail without retry
-    call_count = 0
     with pytest.raises(A2AClientError):
         [event async for event in transport.subscribe(req)]
-    assert call_count == 1
-    assert transport._subscribe_method == 'POST'
+    assert transport._subscribe_auto_method_override is False
+    assert transport._subscribe_method_override == 'POST'
+    assert method_count == {'POST': 2, 'GET': 1}
 
 
 @pytest.mark.asyncio
@@ -372,7 +370,7 @@ async def mock_stream(method, path, context=None, json=None):
         nonlocal call_count
         call_count += 1
         assert method == 'POST'
-        assert json == {'id': 'task-123'}
+        assert json is None
         if False:
             yield
         create_500_error()
@@ -385,8 +383,71 @@ async def mock_stream(method, path, context=None, json=None):
 
     assert '500' in str(exc_info.value)
     assert call_count == 1  # No retry on 500
-    assert transport._subscribe_method == 'POST'
-    assert transport._subscribe_retry_attempted is False
+    assert transport._subscribe_method_override is None
+
+
+@pytest.mark.asyncio
+async def test_compat_rest_transport_subscribe_method_override_avoids_retry_get(
+    mock_httpx_client, agent_card
+):
+    """Scenario: Init with GET override, server returns 405, no automatic retry."""
+    transport = CompatRestTransport(
+        httpx_client=mock_httpx_client,
+        agent_card=agent_card,
+        url='http://example.com',
+        subscribe_method_override='GET',
+    )
+    call_count = 0
+
+    async def mock_stream(method, path, context=None, json=None):
+        nonlocal call_count
+        call_count += 1
+        assert method == 'GET'
+        assert json is None
+        if False:
+            yield
+        create_405_error()
+
+    transport._send_stream_request = mock_stream
+
+    req = SubscribeToTaskRequest(id='task-123')
+    with pytest.raises(A2AClientError) as exc_info:
+        [event async for event in transport.subscribe(req)]
+
+    assert '405' in str(exc_info.value)
+    assert call_count == 1
+
+
+@pytest.mark.asyncio
+async def test_compat_rest_transport_subscribe_method_override_avoids_retry_post(
+    mock_httpx_client, agent_card
+):
+    """Scenario: Init with POST override, server returns 405, no automatic retry."""
+    transport = CompatRestTransport(
+        httpx_client=mock_httpx_client,
+        agent_card=agent_card,
+        url='http://example.com',
+        subscribe_method_override='POST',
+    )
+    call_count = 0
+
+    async def mock_stream(method, path, context=None, json=None):
+        nonlocal call_count
+        call_count += 1
+        assert method == 'POST'
+        assert json is None
+        if False:
+            yield
+        create_405_error()
+
+    transport._send_stream_request = mock_stream
+
+    req = SubscribeToTaskRequest(id='task-123')
+    with pytest.raises(A2AClientError) as exc_info:
+        [event async for event in transport.subscribe(req)]
+
+    assert '405' in str(exc_info.value)
+    assert call_count == 1
 
 
 def test_compat_rest_transport_handle_http_error(transport):

tests/server/apps/rest/test_rest_fastapi_app.py

@@ -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 and parses body."""
+    """Test that POST /tasks/{id}:subscribe works."""
 
     async def mock_stream_response():
         yield Task(
@@ -449,11 +449,8 @@ 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'},
     )
 
@@ -595,7 +592,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', {'id': '1'}),
+            ('/tasks/1:subscribe', 'POST', 'on_subscribe_to_task', None),
             ('/tasks/1', 'GET', 'on_get_task', None),
             ('/tasks', 'GET', 'on_list_tasks', None),
             (