> ## 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.

# MCPEnv

> Environment for integrating MCP (Model Context Protocol) servers as tools

# MCPEnv

An environment that exposes MCP server tools to language models using the official MCP SDK.

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

## Overview

`MCPEnv` connects to MCP servers and exposes their tools to the model as callable functions. It manages:

* MCP server lifecycle (connection, tool discovery, cleanup)
* Persistent background event loops for server processes
* Tool call routing and error handling
* Concurrent multi-server support

<Note>
  MCPEnv is designed for **globally available, read-only MCP servers** where the same toolset can be shared across all rollouts. For per-rollout, stateful servers with mutable task-specific data, consider using a custom environment.
</Note>

## Inheritance

```
Environment
└── MultiTurnEnv
    └── ToolEnv
        └── MCPEnv
```

## Constructor

```python theme={null}
MCPEnv(
    mcp_servers: list[MCPServerConfig | dict] = [],
    max_turns: int = 10,
    error_formatter: Callable[[Exception], str] = lambda e: f"Error: {str(e)}",
    **kwargs
)
```

<ParamField path="mcp_servers" type="list[MCPServerConfig | dict]" default="[]" required>
  List of MCP server configurations. Can be `MCPServerConfig` objects or dicts with keys: `name`, `command`, `args`, `env`, `description`.
</ParamField>

<ParamField path="max_turns" type="int" default="10">
  Maximum turns per rollout. Inherited from `ToolEnv`.
</ParamField>

<ParamField path="error_formatter" type="Callable[[Exception], str]">
  Function to format tool execution errors for the model.
</ParamField>

<ParamField path="**kwargs">
  Additional arguments passed to `ToolEnv` (dataset, rubric, system\_prompt, etc.).
</ParamField>

## MCPServerConfig

```python theme={null}
@dataclass
class MCPServerConfig:
    name: str
    command: str
    args: list[str] | None = None
    env: dict[str, str] | None = None
    description: str = ""
```

<ParamField path="name" type="str" required>
  Unique identifier for the server.
</ParamField>

<ParamField path="command" type="str" required>
  Executable command to start the MCP server (e.g., `"uvx"`, `"npx"`, `"python"`).
</ParamField>

<ParamField path="args" type="list[str] | None">
  Command-line arguments for the server.
</ParamField>

<ParamField path="env" type="dict[str, str] | None">
  Environment variables to pass to the server process.
</ParamField>

<ParamField path="description" type="str">
  Human-readable description of the server's purpose.
</ParamField>

## Example Usage

### Basic Setup

```python theme={null}
import verifiers as vf
from verifiers.envs.experimental.mcp_env import MCPServerConfig

def load_environment():
    # Configure MCP servers
    servers = [
        MCPServerConfig(
            name="filesystem",
            command="npx",
            args=["-y", "@modelcontextprotocol/server-filesystem", "/tmp"],
            description="File system operations"
        ),
        MCPServerConfig(
            name="search",
            command="uvx",
            args=["mcp-server-brave-search"],
            env={"BRAVE_API_KEY": "your-key"},
            description="Web search via Brave"
        ),
    ]
    
    # Create dataset
    dataset = vf.Environment.make_dataset([
        {"question": "Search for recent news about AI"},
        {"question": "List files in /tmp"},
    ])
    
    def task_completed(completion: vf.Messages) -> float:
        """Simple completion reward."""
        return 1.0 if len(completion) > 0 else 0.0
    
    return vf.MCPEnv(
        mcp_servers=servers,
        dataset=dataset,
        rubric=vf.Rubric(task_completed),
        max_turns=5,
    )
```

### Using Dict Configs

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

def load_environment():
    servers = [
        {
            "name": "github",
            "command": "npx",
            "args": ["-y", "@modelcontextprotocol/server-github"],
            "env": {"GITHUB_TOKEN": "ghp_..."},
        },
    ]
    
    dataset = vf.Environment.make_dataset([
        {"question": "List issues in repository owner/repo"},
    ])
    
    return vf.MCPEnv(
        mcp_servers=servers,
        dataset=dataset,
        rubric=vf.Rubric(lambda **kw: 1.0),
    )
```

## Tool Discovery

MCPEnv automatically:

1. Connects to each server via stdio
2. Calls `list_tools()` to discover available tools
3. Wraps each tool in an `MCPToolWrapper` instance
4. Converts MCP tool schemas to `vf.Tool` format
5. Registers tools with the environment

Tools are available to the model immediately after initialization.

## Tool Call Flow

```mermaid theme={null}
sequenceDiagram
    participant Model
    participant MCPEnv
    participant MCPToolWrapper
    participant MCPServer
    
    Model->>MCPEnv: tool_call(name, args)
    MCPEnv->>MCPToolWrapper: __call__(**args)
    MCPToolWrapper->>MCPServer: call_tool(name, args)
    MCPServer-->>MCPToolWrapper: result
    MCPToolWrapper-->>MCPEnv: formatted result
    MCPEnv-->>Model: ToolMessage
```

## Error Handling

Tool execution errors are caught and returned as error messages:

```python theme={null}
# Custom error formatting
def format_error(e: Exception) -> str:
    return f"Tool failed: {type(e).__name__}: {str(e)}"

env = vf.MCPEnv(
    mcp_servers=[...],
    error_formatter=format_error,
)
```

## Lifecycle Management

<Note>
  MCPEnv runs MCP servers in a persistent background event loop that starts during `__init__` and automatically cleans up on exit.
</Note>

### Server Connection

* Servers connect during environment initialization (blocking)
* Connection failures raise immediately
* Tools are registered once servers are ready

### Cleanup

Cleanup is automatic via `atexit` hooks:

```python theme={null}
# Servers are disconnected when:
# 1. Python process exits
# 2. Environment is garbage collected
# 3. Manually via await env.cleanup()

env = vf.MCPEnv(mcp_servers=[...])
# ... use environment ...
await env.cleanup()  # Optional: explicit cleanup
```

## Multi-Server Example

```python theme={null}
import verifiers as vf
from verifiers.envs.experimental.mcp_env import MCPServerConfig

def load_environment():
    servers = [
        MCPServerConfig(
            name="web",
            command="uvx",
            args=["mcp-server-fetch"],
            description="Web page fetching"
        ),
        MCPServerConfig(
            name="memory",
            command="npx",
            args=["-y", "@modelcontextprotocol/server-memory"],
            description="Knowledge graph memory"
        ),
        MCPServerConfig(
            name="postgres",
            command="npx",
            args=["-y", "@modelcontextprotocol/server-postgres"],
            env={"POSTGRES_URL": "postgresql://..."},
            description="Database queries"
        ),
    ]
    
    dataset = vf.Environment.make_dataset([
        {"question": "Research topic X and store findings in memory"},
        {"question": "Query the database for recent entries"},
    ])
    
    return vf.MCPEnv(
        mcp_servers=servers,
        dataset=dataset,
        rubric=vf.Rubric(lambda **kw: 1.0),
        max_turns=15,
    )
```

## Tool Schema Conversion

MCP tools are automatically converted to Verifiers tool format:

```python theme={null}
# MCP tool schema
{
    "name": "read_file",
    "description": "Read file contents",
    "inputSchema": {
        "type": "object",
        "properties": {
            "path": {"type": "string"}
        },
        "required": ["path"]
    }
}

# Converted to vf.Tool
Tool(
    name="read_file",
    description="Read file contents",
    parameters={
        "type": "object",
        "properties": {"path": {"type": "string"}},
        "required": ["path"]
    }
)
```

## Debugging

Enable detailed MCP logging:

```python theme={null}
import logging

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

env = vf.MCPEnv(mcp_servers=[...])
# Logs server connections, tool registrations, and calls
```

## Limitations

* **Global servers only**: Not designed for per-rollout stateful servers
* **Stdio only**: Uses stdio transport (not SSE or other protocols)
* **No streaming**: Tool results are returned as complete strings
* **Single event loop**: All servers share one background event loop

## When to Use

Use MCPEnv when:

* You have existing MCP servers with read-only tools
* Tools can be shared across all rollouts
* You need multi-server tool composition

For stateful, per-rollout tools, use [StatefulToolEnv](/api/stateful-tool-env) instead.

## See Also

* [ToolEnv](/api/tool-env) - Base class for tool-based environments
* [StatefulToolEnv](/api/stateful-tool-env) - Per-rollout stateful tools
* [MCP Documentation](https://modelcontextprotocol.io/) - Official MCP specification