Consolidate v1 MCP client utilities

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

@@ -27,16 +27,15 @@
 import sys
 import time
 import urllib.request
-from contextlib import AsyncExitStack, asynccontextmanager, suppress
 from pathlib import Path
-from typing import Any, cast
+from typing import TYPE_CHECKING, Any, cast
 from urllib.parse import urlsplit
 
 from openai import AsyncOpenAI
-from tenacity import AsyncRetrying, stop_after_attempt, wait_exponential_jitter
 
-MCP_CALL_ATTEMPTS = 6
-MCP_TIMEOUT = 600.0
+if TYPE_CHECKING:
+    # The harness bundles this module into the generated script before execution.
+    from verifiers.v1.mcp.client import call_mcp, connect_mcp  # noqa: TC004
 
 BROWSER_TOOL_TIMEOUT = 3600
 """Matches the bash harness's command timeout."""
@@ -187,117 +186,6 @@ async def chat(
     return completion.choices[0].message
 
 
-@asynccontextmanager
-async def mcp_client(spec: dict):
-    """One fresh MCP client, negotiating the newest protocol and falling back for old servers.
-
-    It is opened and closed within the caller's task so AnyIO cancellation scopes stay correctly
-    nested. Closing noise after a completed body cannot fail or replay an answered call.
-    """
-    import httpx2
-    from mcp import Client
-    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=httpx2.Timeout(spec.get("timeout", MCP_TIMEOUT), connect=5.0),
-            )
-        )
-        transport = streamable_http_client(spec["url"], http_client=http_client)
-        yield await stack.enter_async_context(Client(transport))
-    finally:
-        with suppress(Exception):
-            await stack.aclose()
-
-
-async def with_retry(call):
-    """Run one client 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 protocol session is held; a fresh client handles each 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_client(spec) as client:
-                return (await client.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.input_schema,
-                    },
-                }
-            )
-            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.mime_type};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 with a fresh client per attempt — see `with_retry` for 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_client(servers[server_name]) as client:
-            return await client.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)

verifiers/v1/harnesses/minimal/__init__.py

@@ -0,0 +1,9 @@
+from pathlib import Path
+
+from verifiers.v1.harnesses.standalone import inline_mcp_client
+
+PROGRAM_SOURCE = inline_mcp_client(
+    (Path(__file__).resolve().parent / "program.py").read_text()
+)
+
+__all__ = ["PROGRAM_SOURCE"]