> ## Documentation Index
> Fetch the complete documentation index at: https://mintlify.com/primeintellect-ai/verifiers/llms.txt
> Use this file to discover all available pages before exploring further.

# CliAgentEnv

> Environment for running custom agent code in sandboxes with API interception

# CliAgentEnv

An environment for executing full agent implementations inside sandboxes, intercepting their API calls to control model interactions.

<Warning>
  CliAgentEnv is experimental and subject to breaking changes. The API may change in future releases.
</Warning>

## Overview

`CliAgentEnv` enables running arbitrary agent code (Python scripts, Node.js apps, etc.) in isolated sandbox containers. It:

* Intercepts the agent's API requests via HTTP proxy
* Translates requests to Verifiers' multi-turn rollout loop
* Manages sandbox lifecycle and resource provisioning
* Monitors agent process completion and timeouts

## Inheritance

```
Environment
└── MultiTurnEnv
    └── CliAgentEnv (with SandboxMixin)
```

## Constructor

```python theme={null}
CliAgentEnv(
    run_command: str,
    interception_port: int = 8765,
    interception_url: str | None = None,
    max_turns: int = -1,
    timeout_seconds: float = 3600.0,
    poll_interval: float = 5.0,
    docker_image: str = "python:3.11-slim",
    start_command: str = "tail -f /dev/null",
    cpu_cores: int = 1,
    memory_gb: int = 2,
    disk_size_gb: int = 5,
    gpu_count: int = 0,
    timeout_minutes: int = 60,
    environment_vars: dict[str, str] | None = None,
    **kwargs
)
```

<ParamField path="run_command" type="str" required>
  Command to start the agent inside the sandbox (e.g., `"python agent.py"`).
</ParamField>

<ParamField path="interception_port" type="int" default="8765">
  Local port for the HTTP interception server.
</ParamField>

<ParamField path="interception_url" type="str | None" default="None">
  Optional external URL for interception. If None, uses Prime Tunnel.
</ParamField>

<ParamField path="max_turns" type="int" default="-1">
  Maximum API calls per rollout. -1 for unlimited.
</ParamField>

<ParamField path="timeout_seconds" type="float" default="3600.0">
  Rollout timeout in seconds.
</ParamField>

<ParamField path="poll_interval" type="float" default="5.0">
  Interval for checking agent completion.
</ParamField>

<ParamField path="docker_image" type="str" default="python:3.11-slim">
  Docker image for the sandbox.
</ParamField>

<ParamField path="start_command" type="str" default="tail -f /dev/null">
  Initial command to keep sandbox alive.
</ParamField>

<ParamField path="cpu_cores" type="int" default="1">
  CPU cores allocated to sandbox.
</ParamField>

<ParamField path="memory_gb" type="int" default="2">
  Memory in GB allocated to sandbox.
</ParamField>

<ParamField path="disk_size_gb" type="int" default="5">
  Disk size in GB for sandbox.
</ParamField>

<ParamField path="gpu_count" type="int" default="0">
  Number of GPUs allocated to sandbox.
</ParamField>

<ParamField path="environment_vars" type="dict[str, str] | None">
  Custom environment variables for the sandbox.
</ParamField>

## How It Works

### 1. Interception Setup

```mermaid theme={null}
sequenceDiagram
    participant Agent in Sandbox
    participant Interception Server
    participant CliAgentEnv
    participant Model API
    
    Agent->>Interception Server: POST /v1/chat/completions
    Interception Server->>CliAgentEnv: Queue request
    CliAgentEnv->>Model API: Forward to actual API
    Model API-->>CliAgentEnv: Response
    CliAgentEnv-->>Interception Server: Deliver response
    Interception Server-->>Agent: Return to agent
```

### 2. Environment Variables

The agent receives:

```bash theme={null}
OPENAI_BASE_URL=<tunnel_url>/rollout/<rollout_id>/v1
OPENAI_MODEL=<model_name>
OPENAI_TIMEOUT=600
OPENAI_REQUEST_TIMEOUT=600
HTTPX_TIMEOUT=600
# Plus any custom vars from environment_vars parameter
```

### 3. Lifecycle

1. **Setup**: Create sandbox, start tunnel, upload assets
2. **Execution**: Launch agent via background job
3. **Interception**: Handle each API request as a turn
4. **Completion**: Detect agent exit or timeout
5. **Cleanup**: Destroy sandbox, stop tunnel

## Example Usage

### Basic Python Agent

```python theme={null}
import verifiers as vf
from pathlib import Path

def load_environment():
    # Dataset with tasks
    dataset = vf.Environment.make_dataset([
        {"question": "Write a function to reverse a string"},
        {"question": "Debug this code: print('hello'"},
    ])
    
    def task_success(completion: vf.Messages) -> float:
        """Reward based on agent completing task."""
        return 1.0 if len(completion) > 0 else 0.0
    
    return vf.CliAgentEnv(
        run_command="python /app/agent.py",
        dataset=dataset,
        rubric=vf.Rubric(task_success),
        docker_image="python:3.11",
        max_turns=10,
        timeout_seconds=300,
    )

# Agent code (agent.py)
"""
import os
from openai import OpenAI

client = OpenAI()  # Uses OPENAI_BASE_URL from env

instruction = os.environ.get("HARBOR_INSTRUCTION_PATH")
if instruction:
    with open(instruction) as f:
        task = f.read()
else:
    task = "Complete the coding task"

response = client.chat.completions.create(
    model="gpt-4",
    messages=[{"role": "user", "content": task}]
)

print(response.choices[0].message.content)
"""
```

### With Custom Asset Upload

```python theme={null}
import verifiers as vf
from pathlib import Path

class CustomAgentEnv(vf.CliAgentEnv):
    async def post_sandbox_setup(self, state: vf.State) -> None:
        """Upload agent code and dependencies after sandbox creation."""
        sandbox_id = state["sandbox_id"]
        
        # Upload agent script
        await self.sandbox_client.upload_file(
            sandbox_id,
            "/app/agent.py",
            "./agent_code/main.py"
        )
        
        # Upload requirements
        await self.sandbox_client.upload_file(
            sandbox_id,
            "/app/requirements.txt",
            "./agent_code/requirements.txt"
        )
        
        # Install dependencies
        await self.sandbox_client.execute_command(
            sandbox_id,
            "pip install -r requirements.txt",
            working_dir="/app"
        )

def load_environment():
    dataset = vf.Environment.make_dataset([{"question": "Task 1"}])
    
    return CustomAgentEnv(
        run_command="python /app/agent.py",
        dataset=dataset,
        rubric=vf.Rubric(lambda **kw: 1.0),
    )
```

### Per-Task Docker Images

```python theme={null}
import verifiers as vf

class MultiImageEnv(vf.CliAgentEnv):
    async def get_docker_image(self, state: vf.State) -> str:
        """Select Docker image based on task."""
        task_type = state.get("info", {}).get("language")
        
        if task_type == "python":
            return "python:3.11"
        elif task_type == "node":
            return "node:20-slim"
        else:
            return self.docker_image  # Fallback

def load_environment():
    dataset = vf.Environment.make_dataset([
        {"question": "Python task", "info": {"language": "python"}},
        {"question": "Node task", "info": {"language": "node"}},
    ])
    
    return MultiImageEnv(
        run_command="python /app/agent.py",
        dataset=dataset,
        rubric=vf.Rubric(lambda **kw: 1.0),
    )
```

## Key Methods

### build\_env\_vars

```python theme={null}
async def build_env_vars(self, state: vf.State) -> dict[str, str]
```

Build environment variables for the sandbox. Override to add custom variables:

```python theme={null}
class CustomEnv(vf.CliAgentEnv):
    async def build_env_vars(self, state: vf.State) -> dict[str, str]:
        env_vars = await super().build_env_vars(state)
        env_vars["CUSTOM_VAR"] = "value"
        env_vars["TASK_ID"] = state.get("task", "")
        return env_vars
```

### post\_sandbox\_setup

```python theme={null}
async def post_sandbox_setup(self, state: vf.State) -> None
```

Hook called after sandbox creation but before agent starts. Use for file uploads, dependency installation, etc.

### post\_rollout

```python theme={null}
async def post_rollout(self, state: vf.State) -> None
```

Hook called after agent completes but before sandbox destruction. Use for extracting artifacts or computing rewards that require sandbox access:

```python theme={null}
class ArtifactEnv(vf.CliAgentEnv):
    async def post_rollout(self, state: vf.State):
        sandbox_id = state["sandbox_id"]
        
        # Download agent output
        result = await self.sandbox_client.execute_command(
            sandbox_id,
            "cat /app/output.json",
            working_dir="/app"
        )
        
        state["agent_output"] = result.stdout
```

## State Keys

CliAgentEnv adds these state keys:

<ParamField path="rollout_id" type="str">
  Unique identifier for this rollout.
</ParamField>

<ParamField path="sandbox_id" type="str">
  Prime Sandbox ID.
</ParamField>

<ParamField path="interception_base_url" type="str">
  Full interception URL with rollout ID.
</ParamField>

<ParamField path="agent_completed" type="bool">
  Whether the agent process has finished.
</ParamField>

<ParamField path="agent_exit_code" type="int">
  Agent process exit code.
</ParamField>

<ParamField path="agent_stdout" type="str">
  Captured stdout from agent.
</ParamField>

<ParamField path="agent_stderr" type="str">
  Captured stderr from agent.
</ParamField>

<ParamField path="agent_timed_out" type="bool">
  Whether agent exceeded timeout.
</ParamField>

## Stop Conditions

Rollout stops when:

1. Agent process exits (detected via background job polling)
2. `timeout_seconds` is exceeded
3. `max_turns` is reached

## Error Handling

* **Sandbox creation failure**: Raises `SandboxCreationError`
* **Tunnel failure**: Raises `TunnelError` with frpc logs
* **Agent timeout**: Sets `state["agent_timed_out"] = True`
* **Infrastructure errors**: Sets `state["error"]` to `InfraError` instance

## Debugging

Enable detailed logging:

```python theme={null}
import logging

logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger("verifiers.envs.experimental.cli_agent_env")
logger.setLevel(logging.DEBUG)

# Also enable httpx logs for tunnel debugging
import os
os.environ["HTTPX_LOG_LEVEL"] = "DEBUG"
```

## Limitations

* **Streaming**: Agent must use non-streaming API calls (streaming synthesis is WIP)
* **Tool calls**: Agent can use tools, but schemas are normalized to OpenAI format
* **Timeouts**: Long-running agents may hit sandbox timeout limits

## See Also

* [HarborEnv](/api/experimental/harbor-env) - Specialized for Harbor benchmark tasks
* [SandboxEnv](/api/sandbox-env) - Base sandbox environment
* [MultiTurnEnv](/api/multi-turn-env) - Parent multi-turn environment