diff --git a/verifiers/v1/harnesses/bash/harness.py b/verifiers/v1/harnesses/bash/harness.py index 92273025cc..4852f0d293 100644 --- a/verifiers/v1/harnesses/bash/harness.py +++ b/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) diff --git a/verifiers/v1/harnesses/browser_use/harness.py b/verifiers/v1/harnesses/browser_use/harness.py index ef86ee7a8f..8e5e48bd2b 100644 --- a/verifiers/v1/harnesses/browser_use/harness.py +++ b/verifiers/v1/harnesses/browser_use/harness.py @@ -1,4 +1,3 @@ -import json from pathlib import Path from typing import Literal @@ -6,13 +5,15 @@ 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) diff --git a/verifiers/v1/harnesses/browser_use/program.py b/verifiers/v1/harnesses/browser_use/program.py index d9e1583c5e..4c536b2d3f 100644 --- a/verifiers/v1/harnesses/browser_use/program.py +++ b/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 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) diff --git a/verifiers/v1/harnesses/minimal/__init__.py b/verifiers/v1/harnesses/minimal/__init__.py new file mode 100644 index 0000000000..86c84607ef --- /dev/null +++ b/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"] diff --git a/verifiers/v1/harnesses/bash/program.py b/verifiers/v1/harnesses/minimal/program.py similarity index 70% rename from verifiers/v1/harnesses/bash/program.py rename to verifiers/v1/harnesses/minimal/program.py index bb6e5bc57f..1ba59d3092 100644 --- a/verifiers/v1/harnesses/bash/program.py +++ b/verifiers/v1/harnesses/minimal/program.py @@ -2,23 +2,42 @@ # requires-python = ">=3.10" # dependencies = ["openai", "mcp==2.0.0", "httpx", "httpx2", "tenacity"] # /// -"""Secrets arrive through argv so local tool subprocesses do not inherit them.""" +"""Shared Null/Bash chat program; secrets use argv so tools do not inherit them.""" import argparse import asyncio import json import subprocess -from contextlib import AsyncExitStack, asynccontextmanager, suppress from pathlib import Path +from typing import TYPE_CHECKING import httpx -from openai import AsyncOpenAI -from tenacity import AsyncRetrying, stop_after_attempt, wait_exponential_jitter +from openai import APIStatusError, AsyncOpenAI + +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 SERPER_URL = "https://google.serper.dev/search" -MCP_CALL_ATTEMPTS = 6 -MCP_TIMEOUT = 600.0 +CONTEXT_OVERFLOW_MARKERS = ( + "context_length_exceeded", + "exceeds the context window", + "reduce the length of the messages", + "maximum context length", + "prompt is too long", + "request_too_large", + "request entity too large", + "exceeds the maximum number of tokens", + "maximum prompt length is", +) + + +def is_context_overflow(error: APIStatusError) -> bool: + details = f"{error} {error.body or ''}".casefold() + return error.status_code in (400, 413) and any( + marker in details for marker in CONTEXT_OVERFLOW_MARKERS + ) BASH_TOOL = { @@ -186,117 +205,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 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(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) - - async def run_tool_hook( client: httpx.AsyncClient, url: str, @@ -326,6 +234,7 @@ def parse_args() -> argparse.Namespace: parser.add_argument("--initial-messages-file", default="") parser.add_argument("--mcp-config", default="") parser.add_argument("--tool-interception-url", default="") + parser.add_argument("--bash", action="store_true") parser.add_argument("--edit", action="store_true") parser.add_argument("--search", action="store_true") parser.add_argument("--serper-key", default="") @@ -340,26 +249,31 @@ async def main() -> None: payload = path.read_bytes() path.unlink() initial = json.loads(payload) - client = AsyncOpenAI(base_url=args.base_url, api_key=args.api_key) + client = AsyncOpenAI( + base_url=args.base_url, + api_key=args.api_key, + timeout=httpx.Timeout(600.0 if args.bash else None, connect=5.0), + ) tool_client = ( httpx.AsyncClient(timeout=httpx.Timeout(None, connect=5.0)) if args.tool_interception_url else None ) config = json.loads(args.mcp_config or "{}") - tools = [BASH_TOOL] - reserved = {"bash"} + tools = [BASH_TOOL] if args.bash else [] + reserved = {"bash"} if args.bash else set() if args.edit: tools.append(EDIT_TOOL) reserved.add("edit") if args.search: tools.append(SEARCH_TOOL) reserved.add("search") - mcp_tools, dispatch, servers = ( - await connect_mcp(config, reserved) - if config.get("mcpServers") - else ([], {}, {}) - ) + if config.get("mcpServers"): + mcp_tools, dispatch, servers = await asyncio.wait_for( + connect_mcp(config, reserved), timeout=None if args.bash else 60 + ) + else: + mcp_tools, dispatch, servers = [], {}, {} tools += mcp_tools messages = ( [{"role": "system", "content": args.system_prompt}] @@ -371,7 +285,13 @@ async def main() -> None: elif args.prompt: messages.append({"role": "user", "content": args.prompt}) while True: - message = await chat(client, args.model, messages, tools) + try: + message = await chat(client, args.model, messages, tools) + except APIStatusError as error: + # Null cannot compact, so context exhaustion ends it with the transcript so far. + if args.bash or not is_context_overflow(error): + raise + return messages.append(message.model_dump(exclude_none=True)) if not message.tool_calls: break @@ -405,7 +325,7 @@ async def main() -> None: content = f"error: tool arguments must be a JSON object, got {type(tool_args).__name__}; resend as an object" elif name in dispatch: content = await call_mcp(servers, dispatch, name, tool_args) - elif name == "bash": + elif name == "bash" and args.bash: content = await asyncio.to_thread( run_bash, tool_args.get("command", "") ) diff --git a/verifiers/v1/harnesses/null/harness.py b/verifiers/v1/harnesses/null/harness.py index bc25e29f5e..2c812ef676 100644 --- a/verifiers/v1/harnesses/null/harness.py +++ b/verifiers/v1/harnesses/null/harness.py @@ -1,16 +1,12 @@ -import json -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() - class NullHarnessConfig(HarnessConfig): pass @@ -37,39 +33,15 @@ async def launch( data: TaskData, ) -> ProgramResult: system_prompt, prompt = self.resolve_prompt(data) - env = {**self.config.resolved_env} - args = [ - f"--base-url={endpoint}", - f"--api-key={secret}", - f"--model={ctx.model}", - ] - if system_prompt: - args.append(f"--system-prompt={system_prompt}") - 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, + system_prompt, + prompt, ) - return await runtime.run_program([*program, *args], env) diff --git a/verifiers/v1/harnesses/null/program.py b/verifiers/v1/harnesses/null/program.py deleted file mode 100644 index adb6ccf002..0000000000 --- a/verifiers/v1/harnesses/null/program.py +++ /dev/null @@ -1,258 +0,0 @@ -# /// script -# requires-python = ">=3.11" -# dependencies = ["openai", "mcp==2.0.0", "httpx", "httpx2", "tenacity"] -# /// -"""The interception endpoint and secret arrive through argv rather than the environment.""" - -import argparse -import asyncio -import json -from contextlib import AsyncExitStack, asynccontextmanager, suppress -from pathlib import Path - -import httpx -from openai import APIStatusError, AsyncOpenAI -from tenacity import AsyncRetrying, stop_after_attempt, wait_exponential_jitter - -MCP_CALL_ATTEMPTS = 6 -MCP_TIMEOUT = 600.0 - -CONTEXT_OVERFLOW_MARKERS = ( - # OpenAI error code "context_length_exceeded"; OpenRouter relays the raw body. - "context_length_exceeded", - # OpenAI Responses/Completions: "Your input exceeds the context window of this model". - "exceeds the context window", - # OpenAI chat: "Input tokens exceed the configured limit of N tokens. Please reduce - # the length of the messages."; Groq words it the same way. - "reduce the length of the messages", - # vLLM: "This model's maximum context length is N tokens"; the renderers pre-flight: - # "Prompt length (N) exceeds maximum context length (M)"; Mistral uses the same words. - "maximum context length", - # Anthropic: "prompt is too long: N tokens > M maximum". - "prompt is too long", - # Anthropic byte-size overflow: HTTP 413 {"type": "request_too_large"}. - "request_too_large", - # HTTP proxies reject an oversized body with 413 "Request Entity Too Large". - "request entity too large", - # Google: "The input token count (N) exceeds the maximum number of tokens allowed (M)". - "exceeds the maximum number of tokens", - # xAI: "This model's maximum prompt length is N but the request contains M tokens". - "maximum prompt length is", -) - - -def is_context_overflow(error: APIStatusError) -> bool: - details = f"{error} {error.body or ''}".casefold() - # An overflow is deterministic: a 400, or a 413 for a byte-size cap. - return error.status_code in (400, 413) and any( - marker in details for marker in CONTEXT_OVERFLOW_MARKERS - ) - - -async def chat( - client: AsyncOpenAI, model: str, messages: list[dict], tools: list[dict] -): - completion = await client.chat.completions.create( - model=model, messages=messages, tools=tools or None - ) - 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) -> tuple[list[dict], dict, dict]: - """Enumerate each configured MCP server's tools (a streamable-HTTP `url`); return (tool schemas, - dispatch mapping advertised name -> (server name, raw tool name), servers mapping name -> spec). - No protocol session is held; a fresh client handles each call. Tools are advertised as - `_`; a server named `""` (TOOL_PREFIX = None) advertises its tools bare, so names - must be unique across the rollout's servers.""" - 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): - full = f"{name}_{tool.name}" if name else tool.name - if full in dispatch: - raise ValueError( - f"duplicate tool name {full!r} across servers; keep qualified names" - ) - 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(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) - parser.add_argument("--api-key", required=True) - parser.add_argument("--model", required=True) - parser.add_argument("--system-prompt", default="") - parser.add_argument("--prompt", default="") - parser.add_argument("--initial-messages-file", default="") - parser.add_argument("--mcp-config", default="") - return parser.parse_args() - - -async def main() -> None: - args = parse_args() - initial = [] - if args.initial_messages_file: - path = Path(args.initial_messages_file) - payload = path.read_bytes() - path.unlink() - initial = json.loads(payload) - client = AsyncOpenAI( - base_url=args.base_url, - api_key=args.api_key, - timeout=httpx.Timeout(None, connect=5.0), - ) - config = json.loads(args.mcp_config or "{}") - if config.get("mcpServers"): - # Bound only tool enumeration; each client is opened and closed within this task. - async with asyncio.timeout(60): - tools, dispatch, servers = await connect_mcp(config) - else: - tools, dispatch, servers = [], {}, {} - messages = ( - [{"role": "system", "content": args.system_prompt}] - if args.system_prompt - else [] - ) - if initial: - messages.extend(initial) - elif args.prompt: - messages.append({"role": "user", "content": args.prompt}) - while True: - try: - message = await chat(client, args.model, messages, tools) - except APIStatusError as error: - # Context exhaustion is a budget limit, not a crash: this harness has no - # compaction, so end the run cleanly with what the conversation holds. - if not is_context_overflow(error): - raise - return - messages.append(message.model_dump(exclude_none=True)) - if not message.tool_calls: - break - for call in message.tool_calls: - name = call.function.name - try: - tool_args = json.loads(call.function.arguments or "{}") - except json.JSONDecodeError as e: - messages.append( - { - "role": "tool", - "tool_call_id": call.id, - "content": f"error: invalid JSON in tool arguments ({e}); resend the call with valid JSON", - } - ) - continue - # Valid JSON can still be a non-object (`[]`, `42`, `null`); the MCP dispatch - # assumes a dict, so reject anything else as a tool error rather than crashing. - if not isinstance(tool_args, dict): - messages.append( - { - "role": "tool", - "tool_call_id": call.id, - "content": f"error: tool arguments must be a JSON object, got {type(tool_args).__name__}; resend as an object", - } - ) - continue - if name in dispatch: - content = await call_mcp(servers, dispatch, name, tool_args) - else: - content = f"error: unknown tool {name!r}" - messages.append( - {"role": "tool", "tool_call_id": call.id, "content": content} - ) - - -if __name__ == "__main__": - asyncio.run(main()) diff --git a/verifiers/v1/harnesses/standalone.py b/verifiers/v1/harnesses/standalone.py new file mode 100644 index 0000000000..7546424735 --- /dev/null +++ b/verifiers/v1/harnesses/standalone.py @@ -0,0 +1,74 @@ +import inspect +import json +from collections.abc import Sequence + +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.mcp import client as mcp_client +from verifiers.v1.runtimes import ProgramResult, Runtime +from verifiers.v1.trace import Trace +from verifiers.v1.types import Messages + +MCP_CLIENT_SOURCE = inspect.getsource(mcp_client) +PEP_723_END = "# ///\n" + + +def inline_mcp_client(program: str) -> str: + """Embed the public client so PEP 723 programs need only their declared packages.""" + metadata, body = program.split(PEP_723_END, 1) + return f"{metadata}{PEP_723_END}{MCP_CLIENT_SOURCE}\n{body}" + + +async def launch_chat_program( + source: str, + config: HarnessConfig, + ctx: ModelContext, + trace: Trace, + runtime: Runtime, + endpoint: str, + secret: str, + mcp_urls: dict[str, str], + system_prompt: str | None, + prompt: str | Messages | None, + *, + extra_args: Sequence[str] = (), + env: dict[str, str] | None = None, + activate: bool = True, +) -> ProgramResult: + """Prepare and run a standalone chat program with the shared wire arguments.""" + args = [ + f"--base-url={endpoint}", + f"--api-key={secret}", + f"--model={ctx.model}", + *extra_args, + ] + if system_prompt: + args.append(f"--system-prompt={system_prompt}") + if mcp_urls: + args.append( + "--mcp-config=" + + json.dumps( + { + "mcpServers": { + name: {"url": url, "timeout": config.tool_timeout} + for name, url in mcp_urls.items() + } + } + ) + ) + if isinstance(prompt, str): + args.append(f"--prompt={prompt}") + elif prompt is not None: + path = f".vf-initial-messages-{trace.id}.json" + await runtime.write( + path, + json.dumps([message_to_wire(message) for message in prompt]).encode(), + ) + args.append(f"--initial-messages-file={path}") + program = await runtime.prepare_uv_script( + source, config.resolved_env, activate=activate + ) + return await runtime.run_program( + [*program, *args], env if env is not None else {**config.resolved_env} + ) diff --git a/verifiers/v1/mcp/__init__.py b/verifiers/v1/mcp/__init__.py index 5b98f22827..70ebd0add2 100644 --- a/verifiers/v1/mcp/__init__.py +++ b/verifiers/v1/mcp/__init__.py @@ -1,3 +1,10 @@ +from verifiers.v1.mcp.client import ( + call_mcp, + connect_mcp, + mcp_client, + mcp_content_to_chat_content, + with_retry, +) from verifiers.v1.mcp.launch import ( SharedToolServer, serve, @@ -13,7 +20,12 @@ "SharedToolsetConfig", "Toolset", "ToolsetConfig", + "call_mcp", + "connect_mcp", + "mcp_client", + "mcp_content_to_chat_content", "serve", "serve_shared", "serve_tools", + "with_retry", ] diff --git a/verifiers/v1/mcp/client.py b/verifiers/v1/mcp/client.py new file mode 100644 index 0000000000..863ddd2803 --- /dev/null +++ b/verifiers/v1/mcp/client.py @@ -0,0 +1,127 @@ +from collections.abc import AsyncIterator, Awaitable, Callable, Sequence +from contextlib import AsyncExitStack, asynccontextmanager, suppress +from typing import Any, TypeVar, cast + +import httpx2 +from mcp import Client +from mcp.client.streamable_http import create_mcp_http_client, streamable_http_client +from tenacity import AsyncRetrying, stop_after_attempt, wait_exponential_jitter + +MCP_CALL_ATTEMPTS = 6 +MCP_TIMEOUT = 600.0 + +T = TypeVar("T") + + +@asynccontextmanager +async def mcp_client(spec: dict[str, Any]) -> AsyncIterator[Client]: + """Open one fresh MCP client in the caller's task. + + The client negotiates the newest protocol and falls back for older servers. + Teardown failures after the body completes are suppressed so closing noise cannot + fail or replay a call whose result is already available. + """ + 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=spec.get("connect_timeout", 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: Callable[[], Awaitable[T]]) -> T: + """Run one client operation with the existing at-least-once retries.""" + 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() + raise RuntimeError("retrying stopped without returning or raising") + + +async def connect_mcp( + config: dict[str, Any], reserved: set[str] | None = None +) -> tuple[ + list[dict[str, Any]], + dict[str, tuple[str, str]], + dict[str, dict[str, Any]], +]: + """Enumerate MCP tools and return their schemas, dispatch map, and servers.""" + tool_schemas: list[dict[str, Any]] = [] + dispatch: dict[str, tuple[str, str]] = {} + servers: dict[str, dict[str, Any]] = {} + reserved = reserved or set() + for name, spec in config.get("mcpServers", {}).items(): + servers[name] = spec + + async def list_tools(spec: dict[str, Any] = spec): + async with mcp_client(spec) as client: + return (await client.list_tools()).tools + + for tool in await with_retry(list_tools): + 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: Sequence[Any], +) -> str | list[dict[str, Any]]: + """Convert MCP content blocks to OpenAI chat tool-result content.""" + 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(cast(str, part["text"]) for part in parts) + return parts + + +async def call_mcp( + servers: dict[str, dict[str, Any]], + dispatch: dict[str, tuple[str, str]], + name: str, + arguments: dict[str, Any], +) -> str | list[dict[str, Any]]: + """Call one MCP tool with a fresh client per retry attempt.""" + 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) diff --git a/verifiers/v1/tasksets/nemo_gym/toolset.py b/verifiers/v1/tasksets/nemo_gym/toolset.py index 533e6170f9..894a9afdbe 100644 --- a/verifiers/v1/tasksets/nemo_gym/toolset.py +++ b/verifiers/v1/tasksets/nemo_gym/toolset.py @@ -1,17 +1,13 @@ """Expose NeMo Gym resource tools through Verifiers MCP.""" -from collections.abc import AsyncIterator -from contextlib import AsyncExitStack, asynccontextmanager, suppress from typing import Any, cast import httpx -from mcp import Client -from mcp.client.streamable_http import create_mcp_http_client, streamable_http_client from mcp.server.mcpserver import Context, MCPServer from mcp.types import CallToolResult, TextContent, Tool from pydantic import Field -from verifiers.v1.mcp import SharedToolsetConfig, Toolset +from verifiers.v1.mcp import SharedToolsetConfig, Toolset, mcp_client from verifiers.v1.state import State @@ -36,22 +32,6 @@ async def post(self, path: str, body: dict[str, Any]) -> httpx.Response: ) as client: return await client.post(f"{self.resources_url}/{path}", json=body) - @asynccontextmanager - async def mcp_client(self) -> AsyncIterator[Client]: - """Open an upstream MCP client using this rollout's credentials.""" - assert self.mcp_url is not None - stack = AsyncExitStack() - try: - http_client = await stack.enter_async_context( - create_mcp_http_client(headers=self.mcp_headers) - ) - http_client.timeout = self.request_timeout - transport = streamable_http_client(self.mcp_url, http_client=http_client) - yield await stack.enter_async_context(Client(transport)) - finally: - with suppress(Exception): - await stack.aclose() - class NeMoGymToolset(Toolset[SharedToolsetConfig, NeMoGymState]): """Bridge rollout-specific Gym tools into the standard V1 MCP boundary.""" @@ -67,7 +47,14 @@ def register(self, mcp: MCPServer) -> None: async def list_tools(self) -> list[Tool]: if self.state.mcp_url is not None: - async with self.state.mcp_client() as client: + async with mcp_client( + { + "url": self.state.mcp_url, + "headers": self.state.mcp_headers, + "timeout": self.state.request_timeout, + "connect_timeout": self.state.request_timeout, + } + ) as client: tools = (await client.list_tools()).tools else: tools = [ @@ -88,7 +75,14 @@ async def call_tool( context: Context | None = None, ) -> CallToolResult: if self.state.mcp_url is not None: - async with self.state.mcp_client() as client: + async with mcp_client( + { + "url": self.state.mcp_url, + "headers": self.state.mcp_headers, + "timeout": self.state.request_timeout, + "connect_timeout": self.state.request_timeout, + } + ) as client: return await client.call_tool(name, arguments) if name not in self.state.direct_tools: