Spaces:
Build error
Build error
File size: 3,076 Bytes
c8e832f | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 | """Compatibility helpers for OpenEnv and FastMCP runtime drift."""
from __future__ import annotations
import sys
import types
from typing import Any, Optional
def install_openenv_fastmcp_compat() -> None:
"""Patch FastMCP API differences so older OpenEnv builds keep importing."""
try:
import fastmcp # type: ignore
except Exception:
return
try:
if not hasattr(fastmcp, "Client"):
class CompatClient:
"""Minimal async MCP client used for legacy OpenEnv imports."""
def __init__(self, *args: Any, **kwargs: Any) -> None:
self.args = args
self.kwargs = kwargs
async def __aenter__(self) -> "CompatClient":
return self
async def __aexit__(self, exc_type: Any, exc: Any, tb: Any) -> bool:
return False
async def list_tools(self) -> list[Any]:
return []
async def call_tool(self, tool_name: str, arguments: dict[str, Any]) -> Any:
raise RuntimeError(
f"MCP client compatibility mode cannot call tool: {tool_name}"
)
fastmcp.Client = CompatClient # type: ignore[attr-defined]
except Exception:
pass
try:
client_pkg = sys.modules.get("fastmcp.client")
if client_pkg is None:
client_pkg = types.ModuleType("fastmcp.client")
sys.modules["fastmcp.client"] = client_pkg
client_mod = sys.modules.get("fastmcp.client.client")
if client_mod is None:
client_mod = types.ModuleType("fastmcp.client.client")
sys.modules["fastmcp.client.client"] = client_mod
if not hasattr(client_mod, "CallToolResult"):
class CallToolResult:
"""Compatibility container for legacy OpenEnv response handling."""
def __init__(
self,
content: Any = None,
structured_content: Any = None,
meta: Any = None,
data: Any = None,
is_error: bool = False,
) -> None:
self.content = content
self.structured_content = structured_content
self.meta = meta
self.data = data
self.is_error = is_error
client_mod.CallToolResult = CallToolResult
client_pkg.client = client_mod # type: ignore[attr-defined]
except Exception:
pass
install_openenv_fastmcp_compat()
try:
from openenv.core.env_server.http_server import create_app as openenv_create_app
from openenv.core.env_server.interfaces import Environment
from openenv.core.env_server.types import Action, Observation, State
except Exception as exc: # pragma: no cover
raise RuntimeError(f"OpenEnv runtime import failed after compatibility patch: {exc}") from exc
create_app = openenv_create_app
|