| |
|
|
| """Loopback MCP bridge exposing environment tools to a harness (RFC 005).""" |
|
|
| from __future__ import annotations |
|
|
| import socket |
| import threading |
| import time |
| from contextlib import asynccontextmanager |
| from typing import Any, Optional |
|
|
| from .adapter import HarnessError |
|
|
|
|
| async def _source_tools(mcp_server: Any) -> dict[str, Any]: |
| """Return {name: FastMCP tool}, compatible with FastMCP 2.x and 3.x. |
| |
| Async twin of [`~openenv.core.env_server.mcp_environment.get_server_tools`], |
| which wraps the same calls in `run_async_safely`; awaiting directly avoids |
| spawning a thread when we are already on a loop. |
| """ |
| if hasattr(mcp_server, "get_tools"): |
| result = await mcp_server.get_tools() |
| if isinstance(result, dict): |
| return result |
| if hasattr(mcp_server, "list_tools"): |
| return {tool.name: tool for tool in await mcp_server.list_tools()} |
| return {} |
|
|
|
|
| async def build_bridge_server(mcp_server: Any, renames: dict[str, str]) -> Any: |
| """ |
| Return a FastMCP server exposing the env's tools under their injected names. |
| |
| Tool-name conflict resolution can rename an environment tool before it is |
| injected into a harness (`read_file` -> `env_read_file`). The harness then |
| calls the new name, so the bridge has to answer to it: serving the source |
| server unchanged would advertise a name that does not resolve. |
| |
| Args: |
| mcp_server (`FastMCP`): |
| The environment's own tool server. |
| renames (`dict[str, str]`): |
| Mapping of injected name to source name. Empty means no renames. |
| |
| Returns: |
| `FastMCP`: `mcp_server` itself when there is nothing to rename, |
| otherwise a view of it whose tools carry the injected names. |
| """ |
| if not renames: |
| return mcp_server |
|
|
| from fastmcp import Client |
| from fastmcp.server.providers.proxy import FastMCPProxy, StatefulProxyClient |
| from fastmcp.server.transforms import ToolTransform |
| from fastmcp.tools.tool_transform import ToolTransformConfig |
|
|
| source_tools = await _source_tools(mcp_server) |
| missing = sorted(set(renames.values()) - set(source_tools)) |
| if missing: |
| raise HarnessError( |
| "Cannot rename tools that the MCP server does not expose: " |
| + ", ".join(missing) |
| ) |
|
|
| @asynccontextmanager |
| async def lifespan(server): |
| |
| |
| async with Client(mcp_server): |
| yield {} |
|
|
| |
| |
| client = StatefulProxyClient(mcp_server) |
| view = FastMCPProxy( |
| name=f"{getattr(mcp_server, 'name', 'openenv')}-harness-view", |
| client_factory=client.new_stateful, |
| lifespan=lifespan, |
| ) |
| view.add_transform( |
| ToolTransform( |
| { |
| source: ToolTransformConfig(name=served) |
| for served, source in renames.items() |
| } |
| ) |
| ) |
| return view |
|
|
|
|
| class HarnessMCPBridge: |
| """ |
| Serve an in-process FastMCP server over loopback HTTP for harness use. |
| |
| The bridge runs the environment's FastMCP tool server as a |
| streamable-HTTP ASGI app on `127.0.0.1` with an ephemeral port, inside a |
| daemon thread with its own event loop. It carries only the tool surface: |
| the harness subprocess can reach the environment's domain tools but never |
| OpenEnv's orchestration API (`reset`/`step`/`state`), which lives on a |
| different server entirely. This makes the RFC 005 security boundary |
| structural rather than filter-based. |
| |
| Args: |
| mcp_server (`FastMCP`): |
| The environment's FastMCP server to expose. |
| host (`str`, *optional*, defaults to `"127.0.0.1"`): |
| Interface to bind. Keep this loopback-only. |
| |
| Examples: |
| |
| ```python |
| bridge = HarnessMCPBridge(env.mcp_server) |
| url = bridge.start() |
| # pass url to the harness adapter's inject_tools() |
| bridge.stop() |
| ``` |
| """ |
|
|
| def __init__(self, mcp_server: Any, host: str = "127.0.0.1"): |
| self._mcp_server = mcp_server |
| self._host = host |
| self._url: Optional[str] = None |
| self._thread: Optional[threading.Thread] = None |
| self._uvicorn_server: Optional[Any] = None |
| self._startup_error: Optional[BaseException] = None |
|
|
| @property |
| def url(self) -> Optional[str]: |
| """The bridge's MCP endpoint URL, or `None` when not running.""" |
| return self._url |
|
|
| def start(self, timeout_s: float = 10.0) -> str: |
| """ |
| Start serving and return the MCP endpoint URL. |
| |
| Idempotent: returns the existing URL if already running. |
| |
| Args: |
| timeout_s (`float`, *optional*, defaults to `10.0`): |
| Maximum time to wait for the server to come up. |
| |
| Returns: |
| `str` URL of the MCP endpoint, e.g. `"http://127.0.0.1:54321/mcp"`. |
| |
| Raises: |
| [`~openenv.core.harness.adapter.HarnessError`]: |
| If the server fails to start within the timeout. |
| """ |
| if self._thread is not None and self._thread.is_alive(): |
| if self._uvicorn_server.should_exit: |
| raise HarnessError("MCP bridge is still stopping") |
| assert self._url is not None |
| return self._url |
|
|
| import uvicorn |
|
|
| app = self._mcp_server.http_app() |
| sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) |
| sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) |
| sock.bind((self._host, 0)) |
| port = sock.getsockname()[1] |
|
|
| config = uvicorn.Config(app, log_level="warning", lifespan="on") |
| server = uvicorn.Server(config) |
| self._uvicorn_server = server |
| self._startup_error = None |
|
|
| def _serve() -> None: |
| try: |
| server.run(sockets=[sock]) |
| except BaseException as exc: |
| self._startup_error = exc |
| finally: |
| try: |
| sock.close() |
| except OSError: |
| pass |
|
|
| thread = threading.Thread(target=_serve, daemon=True) |
| thread.start() |
| self._thread = thread |
|
|
| deadline = time.monotonic() + timeout_s |
| while not server.started: |
| if self._startup_error is not None or not thread.is_alive(): |
| self._teardown() |
| raise HarnessError(f"MCP bridge failed to start: {self._startup_error}") |
| if time.monotonic() > deadline: |
| self.stop() |
| raise HarnessError(f"MCP bridge did not start within {timeout_s}s") |
| time.sleep(0.01) |
|
|
| self._url = f"http://{self._host}:{port}/mcp" |
| return self._url |
|
|
| def stop(self, timeout_s: float = 5.0) -> None: |
| """ |
| Stop the bridge server. |
| |
| Idempotent: safe to call when the bridge was never started. |
| |
| Args: |
| timeout_s (`float`, *optional*, defaults to `5.0`): |
| Maximum time to wait for the server thread to exit. |
| |
| Raises: |
| [`~openenv.core.harness.adapter.HarnessError`]: |
| If shutdown times out. The server remains tracked so callers |
| can retry `stop()`; `start()` rejects it while still stopping. |
| """ |
| server = self._uvicorn_server |
| thread = self._thread |
| deadline = time.monotonic() + max(timeout_s, 0.0) |
| if server is not None: |
| |
| |
| server.config.timeout_graceful_shutdown = max(timeout_s, 0.0) / 4 |
| server.should_exit = True |
| if thread is not None and thread.is_alive(): |
| thread.join(timeout=max(timeout_s, 0.0) / 2) |
| if thread.is_alive() and server is not None: |
| server.force_exit = True |
| thread.join(timeout=max(0.0, deadline - time.monotonic())) |
| if thread.is_alive(): |
| |
| |
| raise HarnessError(f"MCP bridge did not stop within {timeout_s}s") |
| self._teardown() |
|
|
| def _teardown(self) -> None: |
| self._url = None |
| self._thread = None |
| self._uvicorn_server = None |
|
|
|
|
| __all__ = ["build_bridge_server", "HarnessMCPBridge"] |
|
|
