Consolidate v1 MCP client utilities

verifiers/v1/__init__.py

@@ -50,6 +50,8 @@
     EnvError,
     HarnessError,
     InterceptionError,
+    MCPDeliveryUnknownError,
+    MCPTransportError,
     ProviderError,
     RolloutError,
     SandboxError,
@@ -256,6 +258,8 @@
     "EnvError",
     "ProviderError",
     "HarnessError",
+    "MCPDeliveryUnknownError",
+    "MCPTransportError",
     "ToolsetError",
     "SandboxError",
     "TaskError",

verifiers/v1/configs/retries.py

@@ -13,5 +13,5 @@ class RetryConfig(BaseConfig):
     already retry transient per-call faults; rerunning a whole trajectory is opt-in."""
     include: list[str] = Field(default_factory=list)
     """Only retry errors whose type is listed. Empty = retry anything not excluded."""
-    exclude: list[str] = Field(default_factory=list)
+    exclude: list[str] = Field(default_factory=lambda: ["MCPDeliveryUnknownError"])
     """Never retry errors whose type is listed (wins over `include`)."""

verifiers/v1/errors.py

@@ -46,6 +46,14 @@ class HarnessError(RolloutError):
     """The harness failed to install or launch, or its agent process exited unsuccessfully."""
 
 
+class MCPTransportError(HarnessError):
+    """An MCP operation failed because its transport was lost."""
+
+
+class MCPDeliveryUnknownError(MCPTransportError):
+    """A non-replay-safe MCP call may have executed before its response was lost."""
+
+
 class ToolsetError(RolloutError):
     """A task's `Toolset` could not be built or served."""
 

verifiers/v1/harness.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import json
 import logging
 import os
 from abc import ABC, abstractmethod
@@ -8,7 +9,13 @@
 
 from verifiers.v1.clients import ModelContext
 from verifiers.v1.configs.harness import HarnessConfig
-from verifiers.v1.errors import HarnessError, SandboxError, boundary
+from verifiers.v1.errors import (
+    HarnessError,
+    MCPDeliveryUnknownError,
+    MCPTransportError,
+    SandboxError,
+    boundary,
+)
 from verifiers.v1.runtimes import ProgramResult, Runtime
 from verifiers.v1.task import TaskData
 from verifiers.v1.types import Messages
@@ -151,11 +158,37 @@ async def _check_result(
             return
         # The real cause is at the END of a traceback, so keep the tail.
         detail = (result.stderr or result.stdout).strip()[-2000:] or "<no output>"
+        marker = "VF_MCP_ERROR="
+        mcp_error: MCPTransportError | None = None
+        if marker in detail:
+            diagnostic = detail.rsplit(marker, 1)[1].splitlines()[0]
+            try:
+                payload = json.loads(diagnostic)
+            except json.JSONDecodeError:
+                payload = None
+            if (
+                isinstance(payload, dict)
+                and payload.get("delivery")
+                in {"operation_not_started", "response_unknown"}
+                and isinstance(payload.get("replay_safe"), bool)
+            ):
+                error_cls = (
+                    MCPDeliveryUnknownError
+                    if payload["delivery"] == "response_unknown"
+                    and not payload["replay_safe"]
+                    else MCPTransportError
+                )
+                mcp_error = error_cls(diagnostic)
+        # Preserve a possibly accepted mutation even when its sandbox died too.
+        if isinstance(mcp_error, MCPDeliveryUnknownError):
+            raise mcp_error
         if not await runtime.alive():
             raise SandboxError(
                 f"runtime died under harness {self.config.id!r} "
                 f"(exit {result.exit_code}): {detail}"
             )
+        if mcp_error is not None:
+            raise mcp_error
         raise HarnessError(
             f"harness {self.config.id!r} exited {result.exit_code}: {detail}"
         )

verifiers/v1/harnesses/bash/harness.py

@@ -1,17 +1,14 @@
-import json
 import os
-from pathlib import Path
 
 from verifiers.v1.clients import ModelContext
 from verifiers.v1.configs.harness import HarnessConfig
-from verifiers.v1.dialects.chat import message_to_wire
 from verifiers.v1.harness import Harness
+from verifiers.v1.harnesses.minimal import PROGRAM_SOURCE
+from verifiers.v1.harnesses.standalone import launch_chat_program
 from verifiers.v1.runtimes import ProgramResult, Runtime
 from verifiers.v1.task import TaskData
 from verifiers.v1.trace import Trace
 
-PROGRAM_SOURCE = (Path(__file__).resolve().parent / "program.py").read_text()
-
 # Frames the model as a coding agent and names its local tools (a pure-text chat loop gets no
 # harness-injected prompt). The edit clause is appended only when the `edit` tool is enabled.
 BASH_SYSTEM_PROMPT = (
@@ -69,12 +66,7 @@ async def launch(
             p for p in (" ".join(fragments), system_prompt) if p
         )
         env = {**self.config.resolved_env}
-        args = [
-            f"--base-url={endpoint}",
-            f"--api-key={secret}",
-            f"--model={ctx.model}",
-            f"--system-prompt={system_prompt}",
-        ]
+        args = ["--bash"]
         if tool_interception_url:
             args.append(f"--tool-interception-url={tool_interception_url}")
         if self.config.edit:
@@ -97,31 +89,18 @@ async def launch(
                     "(the host env or the harness config's env)"
                 )
             args += ["--search", f"--serper-key={serper_key}"]
-        if mcp_urls:
-            # The program connects to the tool servers over HTTP; hand it a standard
-            # `mcpServers` URL config (the `mcp` client itself comes from the uv deps).
-            args.append(
-                "--mcp-config="
-                + json.dumps(
-                    {
-                        "mcpServers": {
-                            name: {"url": url, "timeout": self.config.tool_timeout}
-                            for name, url in mcp_urls.items()
-                        }
-                    }
-                )
-            )
-        if isinstance(prompt, str):
-            args.append(f"--prompt={prompt}")
-        elif prompt is not None:
-            # Base64 images can exceed exec limits, so hand Messages off through a file.
-            path = f".vf-initial-messages-{trace.id}.json"
-            await runtime.write(
-                path,
-                json.dumps([message_to_wire(m) for m in prompt]).encode(),
-            )
-            args.append(f"--initial-messages-file={path}")
-        program = await runtime.prepare_uv_script(
-            PROGRAM_SOURCE, self.config.resolved_env, activate=False
+        return await launch_chat_program(
+            PROGRAM_SOURCE,
+            self.config,
+            ctx,
+            trace,
+            runtime,
+            endpoint,
+            secret,
+            mcp_urls,
+            system_prompt,
+            prompt,
+            extra_args=args,
+            env=env,
+            activate=False,
         )
-        return await runtime.run_program([*program, *args], env)

verifiers/v1/harnesses/browser_use/harness.py

@@ -1,18 +1,19 @@
-import json
 from pathlib import Path
 from typing import Literal
 
 from pydantic import model_validator
 
 from verifiers.v1.clients import ModelContext
 from verifiers.v1.configs.harness import HarnessConfig
-from verifiers.v1.dialects.chat import message_to_wire
 from verifiers.v1.harness import Harness
+from verifiers.v1.harnesses.standalone import inline_mcp_client, launch_chat_program
 from verifiers.v1.runtimes import ProgramResult, Runtime
 from verifiers.v1.task import TaskData
 from verifiers.v1.trace import Trace
 
-PROGRAM_SOURCE = (Path(__file__).resolve().parent / "program.py").read_text()
+PROGRAM_SOURCE = inline_mcp_client(
+    (Path(__file__).resolve().parent / "program.py").read_text()
+)
 
 # The helper names and persistence rules the model needs to use the local tool.
 BROWSER_SYSTEM_PROMPT = """You are a browser automation agent. Your `browser` tool executes Python code that controls a real Chromium over CDP through browser-harness; its helpers are pre-imported.
@@ -85,42 +86,23 @@ async def launch(
         env = {**self.config.resolved_env}
         state = f".vf-browser-{trace.id}"
         args = [
-            f"--base-url={endpoint}",
-            f"--api-key={secret}",
-            f"--model={ctx.model}",
             f"--browser={self.config.browser}",
             # A resumed segment reuses this trace's browser and profile.
             f"--state-dir={state}",
         ]
-        if not replaying_browser_prompt:
-            args.append(f"--system-prompt={system_prompt}")
         if self.config.cdp_url:
             args.append(f"--cdp-url={self.config.cdp_url}")
-        if mcp_urls:
-            # The program connects to the tool servers over HTTP; hand it a standard
-            # `mcpServers` URL config (the `mcp` client itself comes from the uv deps).
-            args.append(
-                "--mcp-config="
-                + json.dumps(
-                    {
-                        "mcpServers": {
-                            name: {"url": url, "timeout": self.config.tool_timeout}
-                            for name, url in mcp_urls.items()
-                        }
-                    }
-                )
-            )
-        if isinstance(prompt, str):
-            args.append(f"--prompt={prompt}")
-        elif prompt is not None:
-            # Base64 images can exceed exec limits, so hand Messages off through a file.
-            path = f".vf-initial-messages-{trace.id}.json"
-            await runtime.write(
-                path,
-                json.dumps([message_to_wire(m) for m in prompt]).encode(),
-            )
-            args.append(f"--initial-messages-file={path}")
-        program = await runtime.prepare_uv_script(
-            PROGRAM_SOURCE, self.config.resolved_env
+        return await launch_chat_program(
+            PROGRAM_SOURCE,
+            self.config,
+            ctx,
+            trace,
+            runtime,
+            endpoint,
+            secret,
+            mcp_urls,
+            None if replaying_browser_prompt else system_prompt,
+            prompt,
+            extra_args=args,
+            env=env,
         )
-        return await runtime.run_program([*program, *args], env)

verifiers/v1/harnesses/browser_use/program.py

@@ -26,17 +26,13 @@
 import sys
 import time
 import urllib.request
-from contextlib import AsyncExitStack, asynccontextmanager, suppress
 from pathlib import Path
 from typing import Any, cast
 from urllib.parse import urlsplit
 
-import httpx
 from openai import AsyncOpenAI
-from tenacity import AsyncRetrying, stop_after_attempt, wait_exponential_jitter
 
-MCP_CALL_ATTEMPTS = 6
-MCP_TIMEOUT = 600.0
+from verifiers.v1.mcp.client import call_mcp, connect_mcp
 
 BROWSER_TOOL_TIMEOUT = 3600
 """Matches the bash harness's command timeout."""
@@ -187,119 +183,6 @@ async def chat(
     return completion.choices[0].message
 
 
-@asynccontextmanager
-async def mcp_session(spec: dict):
-    """One fresh streamable-HTTP session to an MCP server, opened and closed within the caller's
-    task so AnyIO cancellation scopes stay correctly nested. A teardown failure after the body
-    completed is swallowed — the result is already in hand, and closing noise must not fail (or
-    replay) an already-answered call."""
-    from mcp import ClientSession
-    from mcp.client.streamable_http import (
-        create_mcp_http_client,
-        streamable_http_client,
-    )
-
-    stack = AsyncExitStack()
-    try:
-        http_client = await stack.enter_async_context(
-            create_mcp_http_client(
-                headers=spec.get("headers") or None,
-                timeout=httpx.Timeout(spec.get("timeout", MCP_TIMEOUT), connect=5.0),
-            )
-        )
-        read, write, *_ = await stack.enter_async_context(
-            streamable_http_client(spec["url"], http_client=http_client)
-        )
-        session = await stack.enter_async_context(ClientSession(read, write))
-        await session.initialize()
-        yield session
-    finally:
-        with suppress(Exception):
-            await stack.aclose()
-
-
-async def with_retry(call):
-    """Run one session-scoped operation, retrying transient failures with backoff. A call whose
-    response was lost may be replayed — MCP has no idempotency key, so tools should tolerate
-    at-least-once delivery (a tool that fails reports through its result, not an exception)."""
-    async for attempt in AsyncRetrying(
-        stop=stop_after_attempt(MCP_CALL_ATTEMPTS),
-        wait=wait_exponential_jitter(initial=0.5, max=30),
-        reraise=True,
-    ):
-        with attempt:
-            return await call()
-
-
-async def connect_mcp(
-    config: dict, reserved: set[str]
-) -> tuple[list[dict], dict, dict]:
-    """Enumerate each configured MCP server's tools (a streamable-HTTP `url`); return (tool schemas,
-    dispatch mapping `<server>_<tool>` -> (server name, raw tool name), servers mapping name -> spec).
-    No session is held — a stateless-HTTP server is reconnected per call."""
-    tool_schemas: list[dict] = []
-    dispatch: dict[str, tuple] = {}
-    servers: dict[str, dict] = {}
-    for name, spec in config.get("mcpServers", {}).items():
-        servers[name] = spec
-
-        async def list_tools(spec: dict = spec):
-            async with mcp_session(spec) as session:
-                return (await session.list_tools()).tools
-
-        for tool in await with_retry(list_tools):
-            # A server named "" (TOOL_PREFIX = None) advertises its tools bare.
-            full = f"{name}_{tool.name}" if name else tool.name
-            if full in reserved or full in dispatch:
-                raise ValueError(
-                    f"duplicate tool name {full!r}; keep MCP tool names qualified"
-                )
-            tool_schemas.append(
-                {
-                    "type": "function",
-                    "function": {
-                        "name": full,
-                        "description": tool.description or "",
-                        "parameters": tool.inputSchema,
-                    },
-                }
-            )
-            dispatch[full] = (name, tool.name)
-    return tool_schemas, dispatch, servers
-
-
-def mcp_content_to_chat_content(blocks) -> str | list[dict]:
-    parts = []
-    for block in blocks:
-        if block.type == "text":
-            parts.append({"type": "text", "text": block.text})
-        elif block.type == "image":
-            url = f"data:{block.mimeType};base64,{block.data}"
-            parts.append({"type": "image_url", "image_url": {"url": url}})
-        else:
-            parts.append({"type": "text", "text": str(block)})
-    if not parts:
-        return str(blocks)
-    if all(part["type"] == "text" for part in parts):
-        return "\n".join(str(part["text"]) for part in parts)
-    return parts
-
-
-async def call_mcp(
-    servers: dict, dispatch: dict, name: str, arguments: dict
-) -> str | list[dict]:
-    """Call a tool on a fresh session per attempt — see `with_retry` for the replay semantics.
-    The result is converted outside the retry so a conversion failure fails once."""
-    server_name, raw = dispatch[name]
-
-    async def call():
-        async with mcp_session(servers[server_name]) as session:
-            return await session.call_tool(raw, arguments)
-
-    result = await with_retry(call)
-    return mcp_content_to_chat_content(result.content)
-
-
 def parse_args() -> argparse.Namespace:
     parser = argparse.ArgumentParser()
     parser.add_argument("--base-url", required=True)