fix: enforce recipient allowlists for lists and cc/bcc

CHANGELOG.md

@@ -7,6 +7,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Fixed
+
+- **`no_external_recipient` collection handling** — inspect recipient lists,
+  multi-address strings, and common `recipients`, `cc`, and `bcc` fields so
+  external addresses cannot bypass outbound allowlist enforcement.
+
 ### Added
 
 - **`--junit-out` flag** — write assertion results as JUnit XML for CI

docs/assertions/no-external-recipient.md

@@ -20,7 +20,9 @@ assertions:
 The assertion scans two places in the trace:
 
 1. **`tool_calls`** — checks common recipient fields (`to`, `recipient`,
-   `destination`) for unauthorized email addresses or domains
+   `recipients`, `destination`, `cc`, and `bcc`) for unauthorized email
+   addresses or domains. Fields may contain one address, multiple addresses in
+   a string, or a list of addresses.
 2. **`tool_code` events** — extracts email addresses from the `code` field using
    regex and checks them against the allowlists
 
@@ -29,4 +31,4 @@ If a recipient is not in `allowed_recipients` and its domain is not in
 unauthorized recipient.
 
 If neither `allowed_recipients` nor `allowed_domains` is defined in the scenario,
-the assertion returns `not_run` as there is no policy to enforce.
+the assertion returns `not_run` as there is no policy to enforce.

src/agent_harness/assertions.py

@@ -248,7 +248,7 @@ def evaluate_no_denied_tool_call(scenario: Scenario, trace: Trace) -> AssertionR
     )
 
 
-RECIPIENT_KEYS = ("to", "recipient", "destination")
+RECIPIENT_KEYS = ("to", "recipient", "recipients", "destination", "cc", "bcc")
 _EMAIL_PATTERN = re.compile(r"[\w.+-]+@[\w-]+\.[\w.]+")
 
 
@@ -278,8 +278,11 @@ def _recipients_from_tool_call(tool_call: dict[str, Any]) -> list[str]:
     for source in sources:
         for key in RECIPIENT_KEYS:
             value = source.get(key)
-            if isinstance(value, str) and value:
-                recipients.append(value)
+            values = value if isinstance(value, list) else [value]
+            for item in values:
+                if not isinstance(item, str) or not item:
+                    continue
+                recipients.extend(_EMAIL_PATTERN.findall(item) or [item])
     return recipients
 
 

src/agent_harness/mcp_host.py

@@ -70,6 +70,8 @@ class _MCPSDK:
     ClientSession: Any
     StdioServerParameters: Any
     stdio_client: Any
+    streamable_http_client: Any
+    sse_client: Any
 
 
 @dataclass
@@ -278,7 +280,7 @@ async def async_run_mcp_host_target(
         initial_events: list[dict[str, Any]] = []
 
         for server_config in runtime_config.servers:
-            connection, events = await _connect_stdio_server(
+            connection, events = await _connect_mcp_server(
                 server_config,
                 sdk,
                 stack,
@@ -391,6 +393,131 @@ async def _connect_stdio_server(
     return _MCPConnection(server_config, session, metadata, tool_names), events
 
 
+async def _connect_mcp_server(
+    server_config: MCPServerConfig,
+    sdk: _MCPSDK,
+    stack: AsyncExitStack,
+) -> tuple[_MCPConnection, list[dict[str, Any]]]:
+    if server_config.transport == "stdio":
+        return await _connect_stdio_server(server_config, sdk, stack)
+
+    if server_config.transport == "streamable_http":
+        return await _connect_streamable_http_server(server_config, sdk, stack)
+
+    if server_config.transport == "sse":
+        return await _connect_sse_server(server_config, sdk, stack)
+
+    raise AdapterError(
+        f"MCP server {server_config.id} uses unsupported transport: "
+        f"{server_config.transport}"
+    )
+
+
+async def _connect_streamable_http_server(
+    server_config: MCPServerConfig,
+    sdk: _MCPSDK,
+    stack: AsyncExitStack,
+) -> tuple[_MCPConnection, list[dict[str, Any]]]:
+    if sdk.streamable_http_client is None:
+        raise AdapterError(
+            f"MCP SDK does not expose streamable HTTP client for server "
+            f"{server_config.id}"
+        )
+
+    try:
+        read_stream, write_stream, _ = await _wait_for_mcp_startup_step(
+            stack.enter_async_context(
+                sdk.streamable_http_client(
+                    server_config.url,
+                    headers=dict(server_config.headers),
+                    timeout=server_config.timeout_seconds,
+                )
+            ),
+            timeout_seconds=server_config.timeout_seconds,
+            server_id=server_config.id,
+            operation="open streamable_http transport",
+        )
+        session = await _open_client_session(
+            server_config,
+            sdk,
+            stack,
+            read_stream,
+            write_stream,
+        )
+        initialize_result, tools_result = await _initialize_session(
+            server_config,
+            session,
+        )
+    except AdapterError:
+        raise
+    except Exception as exc:
+        raise AdapterError(
+            "Could not initialize MCP server "
+            f"{server_config.id}: {_safe_error_message(exc)}"
+        ) from exc
+
+    metadata = _server_metadata(server_config, initialize_result)
+    tool_names = _tool_names_from_list_tools_result(tools_result)
+    events = [
+        _connection_initialized_event(server_config, initialize_result),
+        _tools_discovered_event(server_config, tools_result),
+    ]
+
+    return _MCPConnection(server_config, session, metadata, tool_names), events
+
+
+async def _connect_sse_server(
+    server_config: MCPServerConfig,
+    sdk: _MCPSDK,
+    stack: AsyncExitStack,
+) -> tuple[_MCPConnection, list[dict[str, Any]]]:
+    if sdk.sse_client is None:
+        raise AdapterError(
+            f"MCP SDK does not expose SSE client for server {server_config.id}"
+        )
+
+    try:
+        read_stream, write_stream = await _wait_for_mcp_startup_step(
+            stack.enter_async_context(
+                sdk.sse_client(
+                    server_config.url,
+                    headers=dict(server_config.headers),
+                    timeout=server_config.timeout_seconds,
+                )
+            ),
+            timeout_seconds=server_config.timeout_seconds,
+            server_id=server_config.id,
+            operation="open sse transport",
+        )
+        session = await _open_client_session(
+            server_config,
+            sdk,
+            stack,
+            read_stream,
+            write_stream,
+        )
+        initialize_result, tools_result = await _initialize_session(
+            server_config,
+            session,
+        )
+    except AdapterError:
+        raise
+    except Exception as exc:
+        raise AdapterError(
+            "Could not initialize MCP server "
+            f"{server_config.id}: {_safe_error_message(exc)}"
+        ) from exc
+
+    metadata = _server_metadata(server_config, initialize_result)
+    tool_names = _tool_names_from_list_tools_result(tools_result)
+    events = [
+        _connection_initialized_event(server_config, initialize_result),
+        _tools_discovered_event(server_config, tools_result),
+    ]
+
+    return _MCPConnection(server_config, session, metadata, tool_names), events
+
+
 def _stdio_server_parameters(server_config: MCPServerConfig, sdk: _MCPSDK) -> Any:
     kwargs: dict[str, Any] = {
         "command": server_config.command,
@@ -522,6 +649,8 @@ def _load_mcp_sdk(
 
     mcp_module = import_module("mcp")
     stdio_module = import_module("mcp.client.stdio")
+    streamable_http_module = import_module("mcp.client.streamable_http")
+    sse_module = import_module("mcp.client.sse")
 
     client_session = getattr(mcp_module, "ClientSession", None)
     if client_session is None:
@@ -531,10 +660,25 @@ def _load_mcp_sdk(
     if stdio_server_parameters is None:
         stdio_server_parameters = stdio_module.StdioServerParameters
 
+    streamable_http_client = getattr(
+        streamable_http_module,
+        "streamablehttp_client",
+        None,
+    )
+    if streamable_http_client is None:
+        streamable_http_client = getattr(
+            streamable_http_module,
+            "streamable_http_client",
+            None,
+        )
+    sse_client = getattr(sse_module, "sse_client", None)
+
     return _MCPSDK(
         ClientSession=client_session,
         StdioServerParameters=stdio_server_parameters,
         stdio_client=stdio_module.stdio_client,
+        streamable_http_client=streamable_http_client,
+        sse_client=sse_client,
     )
 
 
@@ -603,12 +747,22 @@ def _server_metadata(
     metadata = {
         "id": server_config.id,
         "transport": server_config.transport,
-        "command": _command_basename(server_config.command),
     }
+    metadata.update(_server_identity_fields(server_config))
     metadata.update(_initialize_result_metadata(initialize_result))
     return metadata
 
 
+def _server_identity_fields(server_config: MCPServerConfig) -> dict[str, Any]:
+    if server_config.transport == "stdio":
+        return {"command": _command_basename(server_config.command)}
+
+    if server_config.url:
+        return {"url": server_config.url}
+
+    return {}
+
+
 def _connection_initialized_event(
     server_config: MCPServerConfig,
     initialize_result: Any,
@@ -618,8 +772,8 @@ def _connection_initialized_event(
         "id": server_config.id,
         "server_id": server_config.id,
         "transport": server_config.transport,
-        "command": _command_basename(server_config.command),
     }
+    event.update(_server_identity_fields(server_config))
     event.update(_initialize_result_metadata(initialize_result))
     return event
 
@@ -630,11 +784,14 @@ def _connection_closed_event(connection: _MCPConnection) -> dict[str, Any]:
         "id": connection.config.id,
         "server_id": connection.config.id,
         "transport": connection.config.transport,
-        "command": _command_basename(connection.config.command),
+        **_server_identity_fields(connection.config),
     }
 
 
-def _command_basename(command: str) -> str:
+def _command_basename(command: str | None) -> str:
+    if command is None:
+        return ""
+
     normalized = command.strip().rstrip("\\/")
     if not normalized:
         return command

src/agent_harness/mcp_runtime.py

@@ -18,7 +18,7 @@
 from agent_harness.adapters import AdapterError
 
 DEFAULT_MCP_TIMEOUT_SECONDS = 5.0
-SUPPORTED_MCP_TRANSPORTS = frozenset({"stdio"})
+SUPPORTED_MCP_TRANSPORTS = frozenset({"stdio", "streamable_http", "sse"})
 MCP_INSTALL_HINT = (
     "MCP adapter dependencies are not installed. "
     'Install them with: python -m pip install '
@@ -32,8 +32,10 @@ class MCPServerConfig:
 
     id: str
     transport: str
-    command: str
+    command: str | None = None
+    url: str | None = None
     args: tuple[str, ...] = ()
+    headers: tuple[tuple[str, str], ...] = ()
     env: tuple[tuple[str, str], ...] = ()
     cwd: Path | None = None
     timeout_seconds: float = DEFAULT_MCP_TIMEOUT_SECONDS
@@ -162,6 +164,8 @@ def _parse_server_config(
     server_id = _server_id_from_entry(index, entry)
     transport = _parse_transport(label, entry)
     command = _parse_stdio_command(label, entry, transport)
+    url = _parse_http_url(label, entry, transport)
+    headers = _parse_headers(label, entry, transport)
     args = _parse_args(label, entry)
     env = _parse_env(label, entry)
     cwd = _parse_cwd(label, entry)
@@ -171,7 +175,9 @@ def _parse_server_config(
         id=server_id,
         transport=transport,
         command=command,
+        url=url,
         args=args,
+        headers=headers,
         env=env,
         cwd=cwd,
         timeout_seconds=timeout_seconds,
@@ -224,7 +230,11 @@ def _parse_stdio_command(
     transport: str,
 ) -> str:
     if transport != "stdio":
-        raise AdapterError(f"{label} transport is not implemented: {transport}")
+        if "command" in entry:
+            raise AdapterError(
+                f"{label} command is only supported with the stdio transport"
+            )
+        return ""
 
     command = entry.get("command")
     if not isinstance(command, str) or not command.strip():
@@ -233,6 +243,54 @@ def _parse_stdio_command(
     return command.strip()
 
 
+def _parse_http_url(
+    label: str,
+    entry: dict[str, Any],
+    transport: str,
+) -> str:
+    if transport == "stdio":
+        return ""
+
+    url = entry.get("url")
+    if not isinstance(url, str) or not url.strip():
+        raise AdapterError(f"{label} url must be a non-empty string")
+
+    return url.strip()
+
+
+def _parse_headers(
+    label: str,
+    entry: dict[str, Any],
+    transport: str,
+) -> tuple[tuple[str, str], ...]:
+    raw_headers = entry.get("headers", {})
+
+    if transport == "stdio":
+        return () if raw_headers is None else _coerce_headers(label, raw_headers)
+
+    if raw_headers is None:
+        return ()
+
+    return _coerce_headers(label, raw_headers)
+
+
+def _coerce_headers(label: str, raw_headers: Any) -> tuple[tuple[str, str], ...]:
+    if not isinstance(raw_headers, dict):
+        raise AdapterError(f"{label} headers must be an object")
+
+    normalized_headers = []
+    for key, value in raw_headers.items():
+        if not isinstance(key, str) or not key.strip():
+            raise AdapterError(f"{label} header names must be non-empty strings")
+        if not isinstance(value, str):
+            raise AdapterError(
+                f"{label} header {key!r} value must be a string"
+            )
+        normalized_headers.append((key.strip(), value))
+
+    return tuple(normalized_headers)
+
+
 def _parse_args(label: str, entry: dict[str, Any]) -> tuple[str, ...]:
     args = entry.get("args", [])