Dev

plugins/communication_protocols/cli/src/utcp_cli/cli_communication_protocol.py

@@ -972,9 +972,19 @@ async def call_tool(self, caller, tool_name: str, tool_args: Dict[str, Any], too
 
     async def call_tool_streaming(self, caller, tool_name: str, tool_args: Dict[str, Any], tool_call_template: CallTemplate) -> AsyncGenerator[Any, None]:
         """REQUIRED
-        Streaming calls are not supported for the CLI protocol.
+        Execute a tool call through the CLI transport streamingly.
 
-        Raises:
-            NotImplementedError: Always, as this functionality is not supported.
+        The CLI protocol does not natively support streaming, so the command is
+        executed to completion and the full result is yielded as a single chunk.
+
+        Args:
+            caller: The UTCP client that is calling this method.
+            tool_name: Name of the tool to call.
+            tool_args: Dictionary of arguments to pass to the tool.
+            tool_call_template: Call template of the tool to call.
+
+        Yields:
+            The complete tool result as a single item.
         """
-        raise NotImplementedError("Streaming is not supported by the CLI communication protocol.")
+        result = await self.call_tool(caller, tool_name, tool_args, tool_call_template)
+        yield result

plugins/communication_protocols/cli/tests/test_cli_communication_protocol.py

@@ -276,6 +276,22 @@ async def test_call_tool_json_output(transport: CliCommunicationProtocol, mock_c
     assert "Echo:" in result["result"] and "Hello" in result["result"]
 
 
+@pytest.mark.asyncio
+async def test_call_tool_streaming_yields_single_chunk(transport: CliCommunicationProtocol, mock_cli_script, python_executable):
+    """Streaming mode should emit the full result as one chunk instead of failing."""
+    call_template = CliCallTemplate(
+        commands=[
+            {"command": f"{python_executable} {mock_cli_script} --message UTCP_ARG_message_UTCP_END"}
+        ]
+    )
+
+    chunks = [chunk async for chunk in transport.call_tool_streaming(None, "echo", {"message": "Hello World"}, call_template)]
+
+    assert len(chunks) == 1
+    assert isinstance(chunks[0], dict)
+    assert "Echo:" in chunks[0]["result"] and "Hello" in chunks[0]["result"]
+
+
 @pytest.mark.asyncio
 async def test_call_tool_math_operation(transport: CliCommunicationProtocol, mock_cli_script, python_executable):
     """Test calling a math tool with numeric arguments."""

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

@@ -0,0 +1,115 @@
+"""Surface the server's error body on failed HTTP calls.
+
+``aiohttp.ClientResponse.raise_for_status()`` raises a ``ClientResponseError``
+whose ``message`` is only the reason phrase ("Forbidden"). Servers put the
+real reason in the response body, typically ``{"error": "..."}``, and that
+was discarded, so a refused call or discovery surfaced as nothing more than a
+status code. Mirrors the TypeScript SDK's ``_normalizeToolError``.
+"""
+import json
+import re
+from typing import Optional
+
+import aiohttp
+
+# Bodies are folded into an exception message; keep pathological ones bounded.
+MAX_DETAIL_CHARS = 2000
+
+# How much of an error body is read at all. The response may come from an
+# attacker-controlled endpoint (discovery URLs are exactly that trust
+# surface), so the read is bounded up front rather than buffered in full and
+# truncated afterwards. Comfortably larger than MAX_DETAIL_CHARS so a JSON
+# body with a long ``error`` field still parses.
+MAX_BODY_READ_BYTES = 64 * 1024
+
+_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.
+
+    When the body is a JSON object, the first of ``error`` / ``message`` /
+    ``detail`` that is present decides: a non-empty string is returned as the
+    reason; anything else (an object, a list, a number) means the server sent a
+    structured error, so the raw JSON is returned to keep that structure
+    visible rather than skipping ahead to a lower-priority generic string.
+    A non-JSON body is returned as-is. Returns ``None`` for an empty body.
+    """
+    body = text.strip()
+    if not body:
+        return None
+    try:
+        data = json.loads(body)
+    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 _clean(value)
+                continue
+            # Structured error: show it rather than a later generic string.
+            return _clean(body)
+    return _clean(body)
+
+
+async def _read_body_bounded(response: aiohttp.ClientResponse, limit: int) -> str:
+    """Read at most ``limit`` bytes of the body and decode them leniently."""
+    chunks = []
+    total = 0
+    async for chunk in response.content.iter_chunked(8192):
+        chunks.append(chunk)
+        total += len(chunk)
+        if total >= limit:
+            break
+    raw = b"".join(chunks)[:limit]
+    # ``charset`` only parses the Content-Type header, but stay defensive: the
+    # 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:
+        return raw.decode(response.charset or "utf-8", errors="replace")
+    except (LookupError, RuntimeError, ValueError):
+        return raw.decode("utf-8", errors="replace")
+
+
+async def raise_for_status_with_body(response: aiohttp.ClientResponse) -> None:
+    """Like ``response.raise_for_status()``, but with the response body in the error.
+
+    On a 4xx/5xx, reads up to ``MAX_BODY_READ_BYTES`` of the body and raises a
+    ``ClientResponseError`` of the same status and headers whose ``message`` is
+    ``"<reason>: <detail>"``. The text that was read is attached as ``body``
+    for callers that want the structure. Does nothing on a 2xx/3xx.
+    """
+    if response.status < 400:
+        return
+    try:
+        text = await _read_body_bounded(response, MAX_BODY_READ_BYTES)
+    except Exception:
+        text = ""
+    detail = error_detail_from_body(text)
+    reason = response.reason or ""
+    message = f"{reason}: {detail}" if detail else reason
+    error = aiohttp.ClientResponseError(
+        response.request_info,
+        response.history,
+        status=response.status,
+        message=message,
+        headers=response.headers,
+    )
+    error.body = text  # type: ignore[attr-defined]
+    raise error

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

@@ -478,3 +478,35 @@ async def safe_request_with_redirects(
     finally:
         if final_response is not None:
             final_response.release()
+
+
+def reject_remote_loopback_tool_urls(
+    discovery_url: str, manual: Any, *, context: str = "manual discovery"
+) -> None:
+    """Reject a remotely-discovered manual that points tool calls at loopback.
+
+    ``ensure_secure_url`` deliberately permits loopback HTTP so local
+    development works. That leaves one gap: a manual fetched from a remote
+    (non-loopback) origin can still declare tool URLs on the agent's own
+    loopback interface, turning tool invocation into a request against a
+    service that only trusts local callers.
+
+    The OpenAPI converter already closes this for specs it converts (a remote
+    spec may not declare a loopback ``servers[0].url``). Hand-written UTCP
+    manuals bypass the converter, so the same rule is applied here to every
+    tool's call-template URL. A manual fetched from loopback (local dev) is
+    exempt, exactly as the converter exempts a local spec.
+    """
+    if is_loopback_url(discovery_url):
+        return
+    for tool in getattr(manual, "tools", None) or []:
+        call_template = getattr(tool, "tool_call_template", None)
+        url = getattr(call_template, "url", None)
+        if isinstance(url, str) and is_loopback_url(url):
+            raise ValueError(
+                f"Security error during {context}: a manual fetched from "
+                f"{discovery_url!r} declares a loopback tool URL ({url!r}) for "
+                f"tool {getattr(tool, 'name', '?')!r}. A remote manual is not "
+                "allowed to redirect tool calls at the agent's own loopback "
+                "interface."
+            )

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

@@ -33,7 +33,8 @@
 from utcp_http.http_call_template import HttpCallTemplate
 from aiohttp import ClientSession, BasicAuth as AiohttpBasicAuth
 from utcp_http.openapi_converter import OpenApiConverter
-from utcp_http._security import ensure_secure_url, safe_request_with_redirects
+from utcp_http._security import ensure_secure_url, safe_request_with_redirects, reject_remote_loopback_tool_urls
+from utcp_http._errors import raise_for_status_with_body
 import logging
 
 logging.basicConfig(
@@ -210,7 +211,7 @@ async def register_manual(self, caller, manual_call_template: CallTemplate) -> R
                         timeout=aiohttp.ClientTimeout(total=10.0),
                         auth_header_names=auth_header_names,
                     ) as response:
-                        response.raise_for_status()  # Raise exception for 4XX/5XX responses
+                        await raise_for_status_with_body(response)  # 4XX/5XX, with the server's body in the message
 
                         # Check content type to determine how to parse the response
                         content_type = response.headers.get('Content-Type', '')
@@ -225,6 +226,7 @@ async def register_manual(self, caller, manual_call_template: CallTemplate) -> R
                         if "utcp_version" in response_data and "tools" in response_data:
                             logger.info(f"Detected UTCP manual from '{manual_call_template.name}'.")
                             utcp_manual = UtcpManualSerializer().validate_dict(response_data)
+                            reject_remote_loopback_tool_urls(manual_call_template.url, utcp_manual)
                         else:
                             logger.info(f"Assuming OpenAPI spec from '{manual_call_template.name}'. Converting to UTCP manual.")
                             converter = OpenApiConverter(response_data, spec_url=manual_call_template.url, call_template_name=manual_call_template.name, auth_tools=manual_call_template.auth_tools)
@@ -359,7 +361,7 @@ async def call_tool(self, caller, tool_name: str, tool_args: Dict[str, Any], too
                     timeout=aiohttp.ClientTimeout(total=30.0),
                     auth_header_names=auth_header_names,
                 ) as response:
-                    response.raise_for_status()
+                    await raise_for_status_with_body(response)
 
                     content_type = response.headers.get('Content-Type', '').lower()
                     if 'application/json' in content_type: