test(e2e): route external agent card fetch through the typed transport

Adds get_external to e2e_http.py for absolute third-party GETs (no proxy base url or auth, same Result classification) and rewires fetch_agent_card through it, dropping the urllib.request escape hatch. Creates tests/code_coverage_tests/check_e2e_no_raw_requests.py, the checker tests/e2e/CLAUDE.md already referenced, and wires it into the code-quality workflow so raw HTTP client imports outside the transport fail CI; pre-existing uses (root conftest liveness probe, claude_code version resolver) are grandfathered and exception-type-only imports stay allowed.

.github/workflows/test-code-quality.yml

@@ -115,6 +115,9 @@ jobs:
       - name: check_fastuuid_usage
         run: uv run --no-sync python ./tests/code_coverage_tests/check_fastuuid_usage.py
 
+      - name: check_e2e_no_raw_requests
+        run: uv run --no-sync python ./tests/code_coverage_tests/check_e2e_no_raw_requests.py
+
       - name: memory_test
         run: uv run --no-sync python ./tests/code_coverage_tests/memory_test.py
 

tests/code_coverage_tests/check_e2e_no_raw_requests.py

@@ -0,0 +1,81 @@
+"""tests/e2e routes every HTTP call through the typed transport (e2e_http.py), so
+raw HTTP client imports (requests, urllib.request, httpx, aiohttp, http.client) are
+banned in suite code. Importing requests' exception types for catching is fine
+anywhere; a small allowlist grandfathers the files that legitimately make raw calls
+(the transport itself, the root conftest liveness probe, and the claude_code version
+resolver's constant registry URL fetch). Referenced by tests/e2e/CLAUDE.md."""
+
+from __future__ import annotations
+
+import ast
+import sys
+from pathlib import Path
+
+E2E_DIR = Path(__file__).resolve().parents[1] / "e2e"
+
+BANNED_MODULES = ("requests", "urllib.request", "http.client", "httpx", "aiohttp")
+
+ALLOWED_RAW_CLIENT_FILES = {
+    "e2e_http.py": ("requests",),
+    "conftest.py": ("requests",),
+    "claude_code/pr_gate_version_resolver.py": ("urllib.request",),
+}
+
+EXCEPTION_ONLY_NAMES = frozenset({"RequestException", "ConnectionError", "Timeout", "HTTPError"})
+
+
+def _is_banned(module: str) -> bool:
+    return any(module == banned or module.startswith(banned + ".") for banned in BANNED_MODULES)
+
+
+def _banned_imports(tree: ast.Module) -> tuple[tuple[str, int], ...]:
+    plain = tuple(
+        (alias.name, node.lineno)
+        for node in ast.walk(tree)
+        if isinstance(node, ast.Import)
+        for alias in node.names
+        if _is_banned(alias.name)
+    )
+    from_imports = tuple(
+        (node.module, node.lineno)
+        for node in ast.walk(tree)
+        if isinstance(node, ast.ImportFrom)
+        and node.module is not None
+        and _is_banned(node.module)
+        and not all(alias.name in EXCEPTION_ONLY_NAMES for alias in node.names)
+    )
+    return plain + from_imports
+
+
+def _violations_in(path: Path) -> tuple[str, ...]:
+    relative = path.relative_to(E2E_DIR).as_posix()
+    allowed = ALLOWED_RAW_CLIENT_FILES.get(relative, ())
+    tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
+    return tuple(
+        f"tests/e2e/{relative}:{lineno}: raw HTTP client import '{module}'"
+        for module, lineno in _banned_imports(tree)
+        if module not in allowed
+    )
+
+
+def main() -> int:
+    violations = tuple(
+        violation
+        for path in sorted(E2E_DIR.rglob("*.py"))
+        for violation in _violations_in(path)
+    )
+    for violation in violations:
+        print(violation)
+    if violations:
+        print(
+            f"\n{len(violations)} raw HTTP client import(s) in tests/e2e. "
+            "Route the call through tests/e2e/e2e_http.py (get_external for absolute "
+            "third-party URLs) so it gets the typed Result handling."
+        )
+        return 1
+    print("tests/e2e raw HTTP client check passed")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

tests/e2e/a2a/a2a_client.py

@@ -11,13 +11,12 @@ here because only this suite uses them.
 
 from __future__ import annotations
 
-import urllib.request
 import warnings
 from dataclasses import dataclass
 
 from pydantic import BaseModel, ConfigDict, Field
 
-from e2e_http import NoBody, Result, is_ok
+from e2e_http import NoBody, Result, get_external, is_ok
 from proxy_client import ProxyClient
 
 
@@ -286,11 +285,8 @@ def build_a2a_client(proxy: ProxyClient) -> A2AClient:
     return A2AClient(proxy=proxy)
 
 
-def fetch_agent_card(url: str, *, timeout: float = 20.0) -> UpstreamAgentCard:
+def fetch_agent_card(url: str, *, timeout: float = 20.0) -> Result[UpstreamAgentCard]:
     """Fetch a live A2A agent card from its /.well-known endpoint and parse it into the
     registration model, so a test can register a real published card verbatim rather
     than a hand-rolled one."""
-    request = urllib.request.Request(url, headers={"Accept": "application/json"})
-    with urllib.request.urlopen(request, timeout=timeout) as response:  # noqa: S310  # pyright: ignore[reportAny]  # fixed https well-known url; typeshed types urlopen as Any
-        payload: bytes = response.read()  # pyright: ignore[reportAny]  # typeshed types urlopen as Any
-    return UpstreamAgentCard.model_validate_json(payload)
+    return get_external(url, response_type=UpstreamAgentCard, timeout=timeout)

tests/e2e/a2a/test_a2a_agent_e2e.py

@@ -106,7 +106,7 @@ class TestA2AAgentLifecycle:
 
     @pytest.mark.covers("other.a2a.message_send.real_world_agent_replies")
     def test_real_world_agent_replies_to_property_query(self, client: A2AClient, resources: ResourceManager, scoped_key: str) -> None:
-        upstream = fetch_agent_card(MOVEHOME_AGENT_CARD_URL).model_copy(update={"url": MOVEHOME_ORIGIN})
+        upstream = unwrap(fetch_agent_card(MOVEHOME_AGENT_CARD_URL)).model_copy(update={"url": MOVEHOME_ORIGIN})
         assert upstream.protocol_version == "0.3.0"
         marker = unique_marker()
         body = AgentRegisterBody(agent_name=f"e2e-a2a-real-{marker}", agent_card_params=upstream)

tests/e2e/e2e_http.py

@@ -244,6 +244,26 @@ def get[R: BaseModel](
     return _classify(resp, response_type)
 
 
+def get_external[R: BaseModel](
+    url: str,
+    *,
+    response_type: type[R],
+    timeout: float = 30.0,
+) -> Result[R]:
+    """GET an absolute URL outside the proxy (e.g. a public /.well-known document).
+    Unlike the transport wrappers there is no proxy base url and no proxy auth; the
+    response still gets the same tagged-union classification as every other call."""
+    try:
+        resp = requests.get(
+            url,
+            headers={"Accept": "application/json"},
+            timeout=timeout,
+        )
+    except requests.RequestException as exc:
+        return NetworkError(message=str(exc))
+    return _classify(resp, response_type)
+
+
 def delete[R: BaseModel](
     url: URL,
     *,