Pre-release fixes: reuse the MCP client across calls; error bodies on streaming calls

plugins/communication_protocols/http/src/utcp_http/_errors.py

@@ -7,6 +7,7 @@
 status code. Mirrors the TypeScript SDK's ``_normalizeToolError``.
 """
 import json
+import re
 from typing import Optional
 
 import aiohttp
@@ -23,6 +24,16 @@
 
 _DETAIL_KEYS = ("error", "message", "detail")
 
+# Control characters (newlines, ANSI escape introducers, NUL) are collapsed so
+# server-controlled text folded into an exception message or a log line cannot
+# forge extra log records or terminal escape sequences.
+# C0 and C1 control ranges: C1 (U+0080..U+009F) carries escape introducers too.
+_CONTROL_CHARS = re.compile(r"[\x00-\x1f\x7f-\x9f]+")
+
+
+def _clean(text: str) -> str:
+    return _CONTROL_CHARS.sub(" ", text).strip()[:MAX_DETAIL_CHARS]
+
 
 def error_detail_from_body(text: str) -> Optional[str]:
     """Extract a human-readable reason from an error response body.
@@ -39,20 +50,22 @@ def error_detail_from_body(text: str) -> Optional[str]:
         return None
     try:
         data = json.loads(body)
-    except ValueError:
-        return body[:MAX_DETAIL_CHARS]
+    except (ValueError, RecursionError):
+        # RecursionError: a deeply nested body ("[[[[...") within the read cap
+        # can exceed the parser's recursion limit; it is still just text.
+        return _clean(body)
     if isinstance(data, dict):
         for key in _DETAIL_KEYS:
             if key not in data or data[key] is None:
                 continue
             value = data[key]
             if isinstance(value, str):
                 if value.strip():
-                    return value.strip()[:MAX_DETAIL_CHARS]
+                    return _clean(value)
                 continue
             # Structured error: show it rather than a later generic string.
-            return body[:MAX_DETAIL_CHARS]
-    return body[:MAX_DETAIL_CHARS]
+            return _clean(body)
+    return _clean(body)
 
 
 async def _read_body_bounded(response: aiohttp.ClientResponse, limit: int) -> str:
@@ -69,11 +82,9 @@ async def _read_body_bounded(response: aiohttp.ClientResponse, limit: int) -> st
     # body was read directly, so aiohttp's own buffered-body machinery must not
     # be relied on, and an unknown charset name must not lose the detail.
     try:
-        encoding = response.charset or "utf-8"
-        raw.decode(encoding, errors="replace")
+        return raw.decode(response.charset or "utf-8", errors="replace")
     except (LookupError, RuntimeError, ValueError):
-        encoding = "utf-8"
-    return raw.decode(encoding, errors="replace")
+        return raw.decode("utf-8", errors="replace")
 
 
 async def raise_for_status_with_body(response: aiohttp.ClientResponse) -> None:

plugins/communication_protocols/http/src/utcp_http/sse_communication_protocol.py

@@ -252,16 +252,21 @@ async def call_tool_streaming(self, caller, tool_name: str, tool_args: Dict[str,
         data = body_content if "application/json" not in content_type else None
         json_data = body_content if "application/json" in content_type else None
 
-        reconnect = bool(tool_call_template.reconnect)
+        # Never re-send a request body: a reconnect re-issues the request, and for
+        # a POST that would re-execute a possibly non-idempotent tool.
+        reconnect = bool(tool_call_template.reconnect) and body_content is None
+        if tool_call_template.reconnect and body_content is not None:
+            logger.info(f"Reconnection is disabled for '{tool_call_template.name}' because the call sends a request body.")
         retry_delay_ms = tool_call_template.retry_timeout
         last_event_id: Optional[str] = None
         reconnect_attempts = 0
         provider_name = tool_call_template.name
 
         while True:
             attempt_headers = dict(request_headers)
-            if last_event_id is not None:
-                # Let the server resume from where we left off (SSE spec).
+            if last_event_id:
+                # Let the server resume from where we left off (SSE spec). An empty
+                # last event ID means "none": the header is not sent.
                 attempt_headers["Last-Event-ID"] = last_event_id
 
             session = aiohttp.ClientSession()
@@ -292,7 +297,22 @@ async def call_tool_streaming(self, caller, tool_name: str, tool_args: Dict[str,
                             f"handshakes; update the call template to point at "
                             f"the final URL directly."
                         )
-                    response.raise_for_status()
+                    # The error-body read is bounded like the handshake, so a server that
+                    # answers 4xx/5xx and then stalls cannot hang the call either.
+                    await asyncio.wait_for(raise_for_status_with_body(response), timeout=self.HANDSHAKE_TIMEOUT_SECONDS)
+                    # Anything but an event stream would be parsed into silence: a
+                    # JSON error document, say, yields zero events and a "successful" call.
+                    content_type = response.headers.get("Content-Type", "")
+                    # Compare the media type exactly (parameters such as charset allowed),
+                    # so "text/event-stream-invalid" does not pass a substring check.
+                    media_type = content_type.split(";", 1)[0].strip().lower()
+                    if media_type != "text/event-stream":
+                        response.release()
+                        raise SseProtocolError(
+                            f"Expected a text/event-stream response but got {content_type or 'no Content-Type'!r}"
+                        )
+                except SseProtocolError:
+                    raise
                 except Exception as e:
                     if reconnect_attempts == 0:
                         # The initial handshake failing (refused, timed out, non-2xx) is a
@@ -315,13 +335,16 @@ async def call_tool_streaming(self, caller, tool_name: str, tool_args: Dict[str,
 
                 try:
                     async for event in self._iter_sse_events(response):
-                        if event.get("id") is not None:
+                        # Per the SSE spec an id containing NUL is ignored, and an empty id
+                        # resets the last event ID.
+                        if event.get("id") is not None and "\x00" not in event["id"]:
                             last_event_id = event["id"]
                         if event.get("retry") is not None:
                             retry_delay_ms = event["retry"]
                         if "data" not in event:
                             continue
-                        if tool_call_template.event_type and event.get("event") != tool_call_template.event_type:
+                        # An event block without an ``event:`` field has the type "message".
+                        if tool_call_template.event_type and (event.get("event") or "message") != tool_call_template.event_type:
                             continue
                         yield self._parse_event_data(event["data"])
                     # The server ended the stream cleanly: the tool call is complete.
@@ -375,10 +398,11 @@ def flush(event_string: str):
                 elif field == 'id':
                     current_event['id'] = value
                 elif field == 'retry':
-                    try:
+                    # Spec: only a value made of ASCII digits sets the reconnection time.
+                    # Anything longer than 18 digits is absurd (and would be capped anyway);
+                    # bounding the length keeps the conversion cheap whatever the interpreter.
+                    if value.isascii() and value.isdigit() and len(value) <= 18:
                         current_event['retry'] = int(value)
-                    except ValueError:
-                        pass
             if data_lines:
                 current_event['data'] = '\n'.join(data_lines)
             return current_event or None
@@ -413,13 +437,25 @@ def normalise(text: str) -> str:
                     f"SSE event exceeded {self.MAX_EVENT_BUFFER_CHARS} characters without a blank-line delimiter"
                 )
 
-        # Flush a trailing event that was not terminated by a blank line.
+        # At end of stream, a held-back CR is a real line terminator and may
+        # complete the closing blank line of the last event. Dispatch whatever is
+        # fully delimited; per spec, an event still incomplete after that (no
+        # final blank line) is discarded.
         buffer += normalise(decoder.decode(b"", final=True))
         if pending_cr:
             buffer += "\n"
-        event = flush(buffer)
-        if event is not None:
-            yield event
+            pending_cr = False
+        while "\n\n" in buffer:
+            event_string, buffer = buffer.split("\n\n", 1)
+            event = flush(event_string)
+            if event is not None:
+                yield event
+        # The residual (discarded) buffer is still subject to the cap, so an
+        # over-limit malformed stream fails the same way at end of stream.
+        if len(buffer) > self.MAX_EVENT_BUFFER_CHARS:
+            raise SseProtocolError(
+                f"SSE event exceeded {self.MAX_EVENT_BUFFER_CHARS} characters without a blank-line delimiter"
+            )
 
     @staticmethod
     def _parse_event_data(data: str) -> Any:

plugins/communication_protocols/http/src/utcp_http/streamable_http_communication_protocol.py

@@ -293,7 +293,7 @@ async def call_tool_streaming(self, caller, tool_name: str, tool_args: Dict[str,
                     f"followed during streaming handshakes; update the "
                     f"call template to point at the final URL directly."
                 )
-            response.raise_for_status()
+            await raise_for_status_with_body(response)
 
             async for chunk in self._process_http_stream(response, tool_call_template.chunk_size, tool_call_template.name):
                 yield chunk
@@ -310,7 +310,7 @@ async def call_tool_streaming(self, caller, tool_name: str, tool_args: Dict[str,
     async def _process_http_stream(self, response: ClientResponse, chunk_size: Optional[int], provider_name: str) -> AsyncIterator[Any]:
         """Process the HTTP stream and yield chunks based on content type."""
         try:
-            content_type = response.headers.get('Content-Type', '')
+            content_type = response.headers.get('Content-Type', '').lower()
 
             if 'application/x-ndjson' in content_type:
                 async for line in response.content:

plugins/communication_protocols/http/tests/test_http_communication_protocol.py

@@ -174,6 +174,12 @@ async def forbidden_bad_charset_handler(request):
         return web.Response(status=403, body=b'{"error": "odd charset"}', content_type="application/json", charset="x-unknown-charset")
 
     app.router.add_route('*', '/forbidden-no-charset', forbidden_no_charset_handler)
+
+    # A body nested deeper than the JSON parser's recursion limit.
+    async def forbidden_deep_handler(request):
+        return web.Response(status=503, body=b"[" * 20000, content_type="application/json")
+
+    app.router.add_route('*', '/forbidden-deep', forbidden_deep_handler)
     app.router.add_route('*', '/forbidden-bad-charset', forbidden_bad_charset_handler)
 
     return app
@@ -866,3 +872,18 @@ async def test_error_body_is_surfaced_without_a_usable_charset(http_transport, a
     with pytest.raises(aiohttp.ClientResponseError) as excinfo:
         await http_transport.call_tool(None, "t.tool", {}, call_template)
     assert expected in excinfo.value.message
+
+
+@pytest.mark.asyncio
+async def test_deeply_nested_error_body_still_raises_client_response_error(http_transport, aiohttp_client, app):
+    """A body that overflows the JSON parser's recursion limit must not escape as RecursionError."""
+    client = await aiohttp_client(app)
+    call_template = HttpCallTemplate(name="t", url=f"http://localhost:{client.port}/forbidden-deep", http_method="POST")
+    with pytest.raises(aiohttp.ClientResponseError) as excinfo:
+        await http_transport.call_tool(None, "t.tool", {}, call_template)
+    assert excinfo.value.status == 503
+
+
+def test_error_detail_collapses_control_characters():
+    from utcp_http._errors import error_detail_from_body
+    assert error_detail_from_body('{"error": "line one\\nline two\\u001b[31m"}') == "line one line two [31m"