feat: Fleet client

[dzorlu]

PR: Fleet environments (OpenEnv)

This PR documents and refines the Fleet runtime integration for OpenEnv.

What this enables

• Run OpenEnv environments on Fleet (remote) with no local Docker.
• Keep a strict split between:
  • Orchestration (HTTP): reset / step / state
  • Agent actions (MCP): tools/list + tools/call

What this is not

• This is not the local “Dockerized env server + env container” setup.
• There is no container/provider abstraction here; Fleet hosts the runtime remotely (HTTP env server + MCP service). The client only connects.

Main abstractions

• FleetEnvClient (HTTP): orchestrator handle for reset/step/state.
• FleetMCPTools (MCP): agent handle for listing/calling tools.
  • Unions tools across Fleet’s MCP endpoints (today often api/v1/mcp and mcp)
  • Returns tools in OpenAI “tools” dict format (via convert_tool_format)
  • Routes tool calls to the owning endpoint (cached after discovery)

Quickstart

• Install: pip install "openenv-core[fleet]"
• Set: export FLEET_API_KEY="..."
• Run: python examples/fleet_env_example.py <env_key>

References

• RFC 001: rfcs/001-abstractions.md
• RFC 003: rfcs/003-mcp-support.md

TODOs / known sharp edges

• Endpoint discovery (avoid hardcoding api/v1/mcp vs mcp)
• Reset inconsistencies across some env keys (better errors + compatibility notes)
• Tool-name collision policy across endpoints
• Retries/backoff and clearer “endpoint down” failure modes

[cursor]

Required parameters missing defaults break example and tests

High Severity

from_fleet requires data_key, data_version, and image_type as mandatory positional parameters with no defaults. However, the example in fleet_env_example.py calls FleetEnvClient.from_fleet(api_key=..., env_key=..., ttl_seconds=600) without providing these three arguments, and the tests similarly call from_fleet(api_key="k", env_key="e"). This will raise a TypeError at runtime. These parameters likely need default values (e.g., data_key: Optional[str] = None, data_version: Optional[str] = None, image_type: str = "standard"), since the method body already handles falsy data_key gracefully.

Additional Locations (1)

• examples/fleet_env_example.py#L81-L86

 

[cursor]

Sync from_fleet missing provisioning telemetry event

Medium Severity

The sync from_fleet method logs provisioning completion only via _logger.info but does not emit the fleet_provisioning_completed telemetry event. The async counterpart from_fleet_async correctly calls fleet_info("fleet_provisioning_completed", ...) at the equivalent point. This asymmetry means sync-provisioned instances are invisible in Logfire dashboards for provisioning latency tracking.

 

[cursor]

Test expects success but code raises RuntimeError

Medium Severity

test_computer_use_clears_tools_when_no_computer_tool expects reset_async() to return normally and asserts env._tools_cache == []. However, the production code in task_env.py raises RuntimeError("computer_use modality but no 'computer' tool found...") when no computer tool is found for computer_use modality. The test assertion at line 447 is unreachable — the test either passes vacuously due to mock setup issues or will fail once mocks are corrected.

 

[cursor]

Test asserts wrong tool count after filtering

Medium Severity

test_tool_use_does_not_filter asserts len(env._tools_cache) == 3, expecting all tools to remain for tool_use modality. But the production code at task_env.py lines 340-346 explicitly filters out the "computer" tool for tool_use modality. With mock tools including {"name": "computer", ...}, the filter t.get("name") != "computer" removes it, leaving only 2 tools. The test name and assertion contradict the intended code behavior.

 

[cursor]

PR-specific readme committed to repository

Low Severity

PR_README.md contains PR-specific documentation ("This PR documents and refines…", "TODOs / known sharp edges") that duplicates the PR description. This appears to be temporary PR documentation that shouldn't be merged into the main branch.

 

[cursor]

Unconditional 8-second sleep on every list_tools call

Medium Severity

list_tools() unconditionally sleeps for initial_wait (8 seconds) on every invocation, not just the first. When FleetTaskEnv.reset_async() is called more than once on the same instance (e.g., multi-episode use), each reset pays the 8-second penalty even though MCP services are already running. There's no mechanism to skip the wait after a successful first call.

 

[cursor]

Cursor Bugbot has reviewed your changes and found 1 potential issue.

There are 7 total unresolved issues (including 6 from previous reviews).

Autofix Details

Bugbot Autofix prepared a fix for the issue found in the latest run.

• ✅ Fixed: Test expects ValueError during init but error deferred to reset
  • Added an explicit env_key validation in FleetTaskEnv.init so missing env_key now raises ValueError during construction as the test expects.

Or push these changes by commenting:

@cursor push 19403ebbf6

Preview (19403ebbf6)

diff --git a/src/envs/fleet_env/task_env.py b/src/envs/fleet_env/task_env.py
--- a/src/envs/fleet_env/task_env.py
+++ b/src/envs/fleet_env/task_env.py
@@ -153,6 +153,8 @@
             raise ValueError(
                 "Fleet API key required (pass api_key or set FLEET_API_KEY)"
             )
+        if not self.task.get("env_key"):
+            raise ValueError("Task config missing env_key")
 
         self._step_count = 0
         self._done = False

_{This Bugbot Autofix run was free. To enable autofix for future PRs, go to the Cursor dashboard.}

[cursor]

Test expects ValueError during init but error deferred to reset

Medium Severity

test_build_env_spec_raises_without_env_key expects FleetTaskEnv(task, api_key="test") to raise ValueError("missing env_key") during __init__. However, __init__ never calls _build_env_spec() — provisioning is deferred to _ensure_provisioned() which runs during reset_async(). The __init__ only accesses self.env_key which returns "unknown" via .get("env_key", "unknown"). The pytest.raises block will fail because no exception is raised.

 

[cursor]

Cursor Bugbot has reviewed your changes and found 2 potential issues.

There are 9 total unresolved issues (including 7 from previous reviews).

Autofix Details

Bugbot Autofix prepared fixes for both issues found in the latest run.

• ✅ Fixed: Example crashes with missing required parameters
  • Made FleetEnvClient.from_fleet() and from_fleet_async() accept optional data_key/data_version and default image_type="standard" so the quickstart call no longer raises TypeError.
• ✅ Fixed: Massive duplication between sync and async provisioning methods
  • Extracted shared provisioning helpers for data key spec, retry/error handling, MCP URL/tool construction, and provisioning telemetry so sync and async paths now reuse the same logic.

Or push these changes by commenting:

@cursor push 82f79a59c5

Preview (82f79a59c5)

diff --git a/src/envs/fleet_env/client.py b/src/envs/fleet_env/client.py
--- a/src/envs/fleet_env/client.py
+++ b/src/envs/fleet_env/client.py
@@ -9,6 +9,7 @@
 import asyncio
 import dataclasses
 import logging
+import time
 from typing import Any, Dict, List, Optional, Tuple, Type
 
 try:
@@ -30,6 +31,15 @@
 class FleetEnvClient(HTTPEnvClient[Action, Observation]):
     """Orchestrator-facing client for Fleet-hosted environments (HTTP only)."""
 
+    _MAX_FLEET_MAKE_RETRIES = 3
+    _FLEET_MAKE_RETRY_BASE_DELAY_S = 2.0
+    _TRANSIENT_FLEET_ERROR_HINTS = (
+        "health check",
+        "timeout",
+        "connection",
+        "temporarily",
+    )
+
     def __init__(
         self,
         base_url: str,
@@ -47,14 +57,103 @@
         self._api_key = api_key
         self._mcp_urls = mcp_urls
 
+    @staticmethod
+    def _build_data_key_spec(
+        data_key: Optional[str], data_version: Optional[str]
+    ) -> Optional[str]:
+        # Fleet SDK expects data_key in "key:version" format.
+        if not data_key:
+            return None
+        return f"{data_key}:{data_version}" if data_version else data_key
+
+    @staticmethod
+    def _sdk_image_type(image_type: str) -> Optional[str]:
+        # Fleet SDK expects image_type=None for standard images.
+        return image_type if image_type == "mcp" else None
+
+    @staticmethod
+    def _get_mcp_urls(root: str, image_type: str) -> Tuple[str, ...]:
+        # Pick MCP endpoint based on modality:
+        # - computer_use (image_type="mcp"): aggregator on port 8081
+        # - tool_use: per-env MCP server on port 3003
+        if image_type == "mcp":
+            return (f"{root}api/v1/mcp",)
+        return (f"{root}mcp",)
+
     @classmethod
+    def _retry_delay_on_make_error(
+        cls,
+        *,
+        env_key: str,
+        attempt: int,
+        error: Exception,
+        logger: logging.Logger,
+        client_name: str,
+    ) -> Optional[float]:
+        max_retries = cls._MAX_FLEET_MAKE_RETRIES
+        error_msg = str(error).lower()
+        is_transient = any(
+            hint in error_msg for hint in cls._TRANSIENT_FLEET_ERROR_HINTS
+        )
+        if attempt < max_retries - 1 and is_transient:
+            delay = cls._FLEET_MAKE_RETRY_BASE_DELAY_S * (2**attempt)
+            logger.warning(
+                f"[env={env_key}] {client_name}.make() failed (attempt {attempt + 1}/{max_retries}): {error}. "
+                f"Retrying in {delay:.1f}s..."
+            )
+            fleet_warning(
+                "fleet_make_retry",
+                attempt=attempt + 1,
+                max_retries=max_retries,
+                error_type=type(error).__name__,
+                error_message=str(error),
+                retry_delay_s=delay,
+            )
+            return delay
+
+        logger.error(
+            f"[env={env_key}] {client_name}.make() failed after {attempt + 1} attempt(s): {error}"
+        )
+        fleet_error(
+            "fleet_make_failed",
+            attempt=attempt + 1,
+            max_retries=max_retries,
+            error_type=type(error).__name__,
+            error_message=str(error),
+        )
+        return None
+
+    @staticmethod
+    def _emit_provisioning_completed(elapsed: float, instance_id: Any) -> None:
+        fleet_info(
+            "fleet_provisioning_completed",
+            provisioning_time_s=round(elapsed, 1),
+            instance_id=instance_id,
+        )
+
+    @classmethod
+    def _build_orchestrator_and_tools(
+        cls, env: Any, api_key: str, image_type: str, **kwargs: Any
+    ) -> Tuple["FleetEnvClient", FleetMCPTools]:
+        mcp_urls = cls._get_mcp_urls(root=env.urls.root, image_type=image_type)
+        orch = cls(
+            base_url=env.urls.manager.api,
+            fleet_env_handle=env,
+            api_key=api_key,
+            mcp_urls=mcp_urls,
+            **kwargs,
+        )
+        tools = FleetMCPTools(api_key=api_key, mcp_urls=mcp_urls)
+        return orch, tools
+
+    @classmethod
     def from_fleet(
         cls: Type["FleetEnvClient"],
         api_key: str,
         env_key: str,
-        data_key: str,
-        data_version: str,
-        image_type: str,
+        data_key: Optional[str] = None,
+        data_version: Optional[str] = None,
+        image_type: str = "standard",
         region: Optional[str] = None,
         ttl_seconds: Optional[int] = 3600,
         env_variables: Optional[Dict[str, Any]] = None,
@@ -72,106 +171,56 @@
         # This ensures .close() and other lifecycle methods are synchronous.
         fleet = Fleet(api_key=api_key)
 
-        # Fleet SDK expects data_key in "key:version" format
-        data_key_spec = None
-        if data_key:
-            if data_version:
-                data_key_spec = f"{data_key}:{data_version}"
-            else:
-                data_key_spec = data_key
-
-        import time
-        import logging
-
+        data_key_spec = cls._build_data_key_spec(data_key=data_key, data_version=data_version)
         _logger = logging.getLogger(__name__)
 
         _logger.info(f"Creating Fleet instance: env_key={env_key}, ttl={ttl_seconds}s")
         start = time.time()
 
-        # Retry logic for transient Fleet API failures (e.g., health check failures)
-        max_retries = 3
-        retry_base_delay = 2.0  # seconds
+        # Retry logic for transient Fleet API failures (e.g., health check failures).
+        max_retries = cls._MAX_FLEET_MAKE_RETRIES
         env = None
 
         for attempt in range(max_retries):
             try:
-                # Fleet SDK expects image_type=None for standard images
-                sdk_image_type = image_type if image_type == "mcp" else None
                 env = fleet.make(
                     env_key=env_key,
                     region=region,
                     ttl_seconds=ttl_seconds,
                     env_variables=env_variables,
-                    image_type=sdk_image_type,
+                    image_type=cls._sdk_image_type(image_type),
                     data_key=data_key_spec,
                 )
                 break  # Success
             except Exception as e:
-                error_msg = str(e)
-                # Retry on transient errors (health check failures, timeouts, etc.)
-                is_transient = any(
-                    x in error_msg.lower()
-                    for x in ["health check", "timeout", "connection", "temporarily"]
+                delay = cls._retry_delay_on_make_error(
+                    env_key=env_key,
+                    attempt=attempt,
+                    error=e,
+                    logger=_logger,
+                    client_name="Fleet",
                 )
-                if attempt < max_retries - 1 and is_transient:
-                    delay = retry_base_delay * (2**attempt)
-                    _logger.warning(
-                        f"[env={env_key}] Fleet.make() failed (attempt {attempt + 1}/{max_retries}): {e}. "
-                        f"Retrying in {delay:.1f}s..."
-                    )
-                    fleet_warning(
-                        "fleet_make_retry",
-                        attempt=attempt + 1,
-                        max_retries=max_retries,
-                        error_type=type(e).__name__,
-                        error_message=str(e),
-                        retry_delay_s=delay,
-                    )
-                    time.sleep(delay)
-                else:
-                    _logger.error(
-                        f"[env={env_key}] Fleet.make() failed after {attempt + 1} attempt(s): {e}"
-                    )
-                    fleet_error(
-                        "fleet_make_failed",
-                        attempt=attempt + 1,
-                        max_retries=max_retries,
-                        error_type=type(e).__name__,
-                        error_message=str(e),
-                    )
+                if delay is None:
                     raise
+                time.sleep(delay)
 
         elapsed = time.time() - start
         instance_id = getattr(env, "instance_id", "unknown")
         _logger.info(f"Fleet instance ready in {elapsed:.1f}s: {instance_id}")
+        cls._emit_provisioning_completed(elapsed=elapsed, instance_id=instance_id)
 
-        root = env.urls.root
-        # Pick MCP endpoint based on modality:
-        # - computer_use: aggregator on port 8081 (has computer tool + API tools)
-        # - tool_use: per-env MCP server on port 3003 (API tools only)
-        if image_type == "mcp":
-            mcp_urls = (f"{root}api/v1/mcp",)
-        else:
-            mcp_urls = (f"{root}mcp",)
-
-        orch = cls(
-            base_url=env.urls.manager.api,
-            fleet_env_handle=env,
-            api_key=api_key,
-            mcp_urls=mcp_urls,
-            **kwargs,
+        return cls._build_orchestrator_and_tools(
+            env=env, api_key=api_key, image_type=image_type, **kwargs
         )
-        tools = FleetMCPTools(api_key=api_key, mcp_urls=mcp_urls)
-        return orch, tools
 
     @classmethod
     async def from_fleet_async(
         cls: Type["FleetEnvClient"],
         api_key: str,
         env_key: str,
-        data_key: str,
-        data_version: str,
-        image_type: str,
+        data_key: Optional[str] = None,
+        data_version: Optional[str] = None,
+        image_type: str = "standard",
         region: Optional[str] = None,
         ttl_seconds: Optional[int] = 3600,
         env_variables: Optional[Dict[str, Any]] = None,
@@ -192,17 +241,7 @@
 
         async_fleet = AsyncFleet(api_key=api_key)
 
-        # Fleet SDK expects data_key in "key:version" format
-        data_key_spec = None
-        if data_key:
-            if data_version:
-                data_key_spec = f"{data_key}:{data_version}"
-            else:
-                data_key_spec = data_key
-
-        import time
-        import logging
-
+        data_key_spec = cls._build_data_key_spec(data_key=data_key, data_version=data_version)
         _logger = logging.getLogger(__name__)
 
         _logger.info(
@@ -210,14 +249,10 @@
         )
         start = time.time()
 
-        # Retry logic with async sleep (non-blocking)
-        max_retries = 3
-        retry_base_delay = 2.0  # seconds
+        # Retry logic with async sleep (non-blocking).
+        max_retries = cls._MAX_FLEET_MAKE_RETRIES
         env = None
 
-        # Fleet SDK expects image_type=None for standard images
-        sdk_image_type = image_type if image_type == "mcp" else None
-
         for attempt in range(max_retries):
             try:
                 env = await async_fleet.make(
@@ -225,72 +260,30 @@
                     region=region,
                     ttl_seconds=ttl_seconds,
                     env_variables=env_variables,
-                    image_type=sdk_image_type,
+                    image_type=cls._sdk_image_type(image_type),
                     data_key=data_key_spec,
                 )
                 break  # Success
             except Exception as e:
-                error_msg = str(e)
-                # Retry on transient errors (health check failures, timeouts, etc.)
-                is_transient = any(
-                    x in error_msg.lower()
-                    for x in ["health check", "timeout", "connection", "temporarily"]
+                delay = cls._retry_delay_on_make_error(
+                    env_key=env_key,
+                    attempt=attempt,
+                    error=e,
+                    logger=_logger,
+                    client_name="AsyncFleet",
                 )
-                if attempt < max_retries - 1 and is_transient:
-                    delay = retry_base_delay * (2**attempt)
-                    _logger.warning(
-                        f"[env={env_key}] AsyncFleet.make() failed (attempt {attempt + 1}/{max_retries}): {e}. "
-                        f"Retrying in {delay:.1f}s..."
-                    )
-                    fleet_warning(
-                        "fleet_make_retry",
-                        attempt=attempt + 1,
-                        max_retries=max_retries,
-                        error_type=type(e).__name__,
-                        error_message=str(e),
-                        retry_delay_s=delay,
-                    )
-                    await asyncio.sleep(delay)
-                else:
-                    _logger.error(
-                        f"[env={env_key}] AsyncFleet.make() failed after {attempt + 1} attempt(s): {e}"
-                    )
-                    fleet_error(
-                        "fleet_make_failed",
-                        attempt=attempt + 1,
-                        max_retries=max_retries,
-                        error_type=type(e).__name__,
-                        error_message=str(e),
-                    )
+                if delay is None:
                     raise
+                await asyncio.sleep(delay)
 
         elapsed = time.time() - start
         instance_id = getattr(env, "instance_id", "unknown")
         _logger.info(f"Fleet instance ready (async) in {elapsed:.1f}s: {instance_id}")
-        fleet_info(
-            "fleet_provisioning_completed",
-            provisioning_time_s=round(elapsed, 1),
-            instance_id=instance_id,
-        )
+        cls._emit_provisioning_completed(elapsed=elapsed, instance_id=instance_id)
 
-        root = env.urls.root
-        # Pick MCP endpoint based on modality:
-        # - computer_use (image_type="mcp"): aggregator on port 8081 (has computer tool + API tools)
-        # - tool_use: per-env MCP server on port 3003 (API tools only)
-        if image_type == "mcp":
-            mcp_urls = (f"{root}api/v1/mcp",)
-        else:
-            mcp_urls = (f"{root}mcp",)
-
-        orch = cls(
-            base_url=env.urls.manager.api,
-            fleet_env_handle=env,
-            api_key=api_key,
-            mcp_urls=mcp_urls,
-            **kwargs,
+        return cls._build_orchestrator_and_tools(
+            env=env, api_key=api_key, image_type=image_type, **kwargs
         )
-        tools = FleetMCPTools(api_key=api_key, mcp_urls=mcp_urls)
-        return orch, tools
 
     # ------------------------------------------------------------------
     # Database query methods (delegate to Fleet SDK's SQLiteResource)

_{This Bugbot Autofix run was free. To enable autofix for future PRs, go to the Cursor dashboard.}

[cursor]

Example crashes with missing required parameters

High Severity

The advertised quickstart example calls FleetEnvClient.from_fleet() with only api_key, env_key, and ttl_seconds, but the method signature requires data_key, data_version, and image_type as mandatory positional parameters with no defaults. This will immediately raise a TypeError at runtime, making the documented quickstart non-functional.

Additional Locations (1)

• src/envs/fleet_env/client.py#L50-L62

 

[cursor]

Massive duplication between sync and async provisioning methods

Low Severity

from_fleet and from_fleet_async duplicate ~130 lines of nearly identical logic: retry loops, data_key spec construction, MCP URL selection, and orchestrator/tools creation. The only structural difference is fleet.make() vs await async_fleet.make() and time.sleep() vs await asyncio.sleep(). This already caused the telemetry inconsistency where only the async path emits fleet_provisioning_completed.

Additional Locations (1)

• src/envs/fleet_env/client.py#L166-L293

 

[cursor]

Cursor Bugbot has reviewed your changes and found 1 potential issue.

There are 10 total unresolved issues (including 9 from previous reviews).

^{❌ Bugbot Autofix is OFF. To automatically fix reported issues with cloud agents, enable autofix in the Cursor dashboard.}

^{Reviewed by Cursor Bugbot for commit 725ddba. Configure here.}

[cursor]

Hardcoded API token committed to source code

High Severity

_BROWSER_API_TOKEN_FALLBACK contains a hardcoded UUID credential (bb46d985-763a-47bf-a67a-c2c98ba6e1d9) used as a fallback authentication token for the browser lease API. This token is committed to the public repository and used in Authorization: Bearer headers, exposing it to anyone with repo access.

 

^{Reviewed by Cursor Bugbot for commit 725ddba. Configure here.}