Skip to main content

Documentation Index

Fetch the complete documentation index at: https://docs.hud.ai/llms.txt

Use this file to discover all available pages before exploring further.

MCPServer is the base class for building MCP-compatible servers that work with any MCP client. It extends FastMCP with Docker-friendly features.

​
Why MCP?

Traditional agent frameworks couple agents tightly to specific environments. MCP decouples them:

Without MCP

  • Agent code hardcoded for each environment
  • No standardization across tools
  • Difficult to swap agents or environments

With MCP

  • Any agent works with any environment
  • Standard protocol for all interactions
  • Easy to swap components
MCP standardizes agent-environment communication through JSON-RPC messages. Agents call tools exposed by servers and receive structured responses.

​
MCPServer

from hud.server import MCPServer
Enhanced FastMCP server with Docker-friendly features for building HUD environments. Constructor Parameters:
ParameterTypeDescriptionDefault
namestrServer name for MCP handshakeRequired
instructionsstrServer instructions/descriptionNone
**fastmcp_kwargsAnyAdditional FastMCP parameters-
Key Features:
  1. SIGTERM handling - Graceful shutdown in containers via custom runner
  2. Initialize decorator - Async setup during MCP initialize request (stdout is temporarily redirected to stderr during initialization to avoid corrupting MCP output)
  3. Shutdown decorator - Runs only on SIGTERM (container termination), not on hot‑reload/SIGINT
  4. Enhanced add_tool() - Automatically handles BaseTool instances and raw FastMCP Tool objects
  5. Tool decorator passthrough - @mcp.tool returns the original function for easy composition
  6. FastMCP inheritance - All FastMCP methods available (mount, resource, tool)

​
Decorators

​
@initialize

Run async setup during MCP initialize request: 
mcp = MCPServer(name="my-env")

@mcp.initialize
async def setup_environment(ctx):
    """
    Initialize environment resources.
    
    Args:
        ctx: RequestContext with:
            - ctx.meta: Client metadata dict
            - ctx.session: MCP ServerSession
    """
    # Access metadata from agent (if provided)
    if ctx.meta:
        progress_token = ctx.meta.get("progressToken")
        display_width = ctx.meta.get("display_width", 1920)
        display_height = ctx.meta.get("display_height", 1080)
        
        # Send progress notifications
        if progress_token:
            await ctx.session.send_progress_notification(
                progress_token=progress_token,
                progress=50,
                total=100,
                message="Initializing environment..."
            )

​
@shutdown

Run cleanup on SIGTERM (container termination only): 
@mcp.shutdown
async def cleanup():
    """Clean up resources on shutdown."""
    if browser_provider:
        browser_provider.close()
    logger.info("Cleanup complete")

​
Tool Registration

Three ways to register tools: 
# 1. Decorator for simple functions
@mcp.tool()
async def my_tool(param: str) -> dict:
    return {"result": param}

# 2. Add BaseTool instances
from hud.tools import BashTool
bash = BashTool()
mcp.add_tool(bash)  # Automatically uses bash.mcp internally

# 3. Add non-BaseTool instances directly
from custom import PlaywrightTool
playwright = PlaywrightTool()
mcp.add_tool(playwright)  # Added as-is

​
Hub Pattern (mount)

Use BaseHub for organized tool namespaces: 
from hud.tools import BaseHub

# Create hub
setup_hub = BaseHub("setup")

# Add internal tools (hidden from agents)
@setup_hub.tool("board")
async def setup_board(size: int = 4):
    game = setup_hub.env
    game.reset(size=size)
    return [TextContent(text=f"{size}x{size} board initialized")]

# Mount hub on server
mcp.mount(setup_hub)

# Agents call via dispatcher: setup(name="board", arguments={"size": 4})

​
Resources

Expose metadata via MCP resources: 
@mcp.resource("telemetry://live")
async def get_telemetry():
    """Expose live telemetry data."""
    return {
        "provider": os.getenv("BROWSER_PROVIDER"),
        "status": "running" if browser_provider else "stopped",
        "live_url": browser_provider.get_live_view_url() if browser_provider else None,
        "timestamp": datetime.now().isoformat()
    }

​
Running the Server

if __name__ == "__main__":
    # Run with SIGTERM handling (stdio by default)
    mcp.run()

    # Or use development transports (HTTP/SSE)
    mcp.run(transport="http", port=8765)
    mcp.run(transport="sse", port=8080)
When using HTTP/SSE, HUD development helper endpoints are available:
  • GET /hud – overview
  • GET /hud/tools – list tools with schemas
  • GET /hud/resources – list resources
  • GET /hud/prompts – list prompts

​
Real Environment Examples

​
Minimal Environment

# src/hud_controller/server.py
from hud.server import MCPServer
from mcp.types import TextContent

mcp = MCPServer(name="counter-env")
counter = {"value": 0}

@mcp.tool()
async def setup(start_value: int = 0):
    """Initialize counter."""
    counter["value"] = start_value
    return {"status": "ready", "counter": counter["value"]}

@mcp.tool()
async def increment():
    """Increment counter."""
    counter["value"] += 1
    return [TextContent(text=f"Counter: {counter['value']}", type="text")]

@mcp.tool()
async def evaluate(target: int):
    """Check if target reached."""
    from hud.tools.types import EvaluationResult
    return EvaluationResult(
        reward=1.0 if counter["value"] >= target else 0.0,
        done=counter["value"] >= target
    )

if __name__ == "__main__":
    mcp.run()

​
text_2048 Environment

From environments/text_2048/src/hud_controller/server.py: 
from hud.server import MCPServer
from .game import Game2048
from .tools import MoveTool
from .setup import setup as setup_hub
from .evaluate import evaluate as evaluate_hub

mcp = MCPServer(name="text-2048")
game = None

@mcp.initialize
async def initialize_environment(ctx):
    global game
    
    # Progress notifications
    progress_token = getattr(ctx.meta, "progressToken", None) if ctx.meta else None
    
    async def send_progress(progress: int, message: str):
        if progress_token:
            await ctx.session.send_progress_notification(
                progress_token=progress_token,
                progress=progress,
                total=100,
                message=message
            )
    
    await send_progress(0, "Starting 2048 game environment...")
    
    # Create game
    game = Game2048()
    game.reset()
    
    await send_progress(50, "Setting up game board...")
    
    # Set game on hubs
    setup_hub.env = game
    evaluate_hub.env = game
    
    # Mount hubs
    mcp.mount(setup_hub)
    mcp.mount(evaluate_hub)
    
    await send_progress(70, "Configuring tools...")
    
    # Add move tool
    mcp.add_tool(MoveTool(env=game))
    
    await send_progress(100, "2048 environment ready")

​
remote_browser Environment

From environments/remote_browser/src/hud_controller/server.py: 
from hud.server import MCPServer
from hud.tools.computer import HudComputerTool, AnthropicComputerTool, OpenAIComputerTool
from .tools import PlaywrightToolWithMemory, BrowserExecutor
from .setup import setup as setup_hub
from .evaluate import evaluate as evaluate_hub
from .providers import get_provider

mcp = MCPServer(
    name="HUD Remote Browser Environment",
    instructions="""Remote browser automation environment..."""
)

# Global state
browser_provider = None
playwright_tool = None

@mcp.resource("telemetry://live")
async def get_telemetry_resource():
    """MCP resource with live browser status."""
    return {
        "provider": os.getenv("BROWSER_PROVIDER", "unknown"),
        "status": "running" if browser_provider else "stopped",
        "live_url": browser_provider.get_live_view_url() if browser_provider else None,
        "cdp_url": browser_provider.cdp_url if browser_provider else None
    }

@mcp.initialize
async def initialize_environment(ctx):
    global browser_provider, playwright_tool
    
    # Get metadata
    metadata = ctx.meta
    progress_token = metadata.get("progressToken", None)
    
    # Initialize provider
    provider_name = os.getenv("BROWSER_PROVIDER")
    provider_class = get_provider(provider_name)
    browser_provider = provider_class(config)
    
    # Launch browser
    cdp_url = await browser_provider.launch()
    
    # Create playwright tool
    playwright_tool = PlaywrightToolWithMemory(cdp_url=cdp_url)
    await playwright_tool._ensure_browser()
    
    # Add playwright tool (not a BaseTool, added directly)
    mcp.add_tool(playwright_tool)
    
    # Create computer tools
    executor = BrowserExecutor(playwright_tool)
    tool_kwargs = {"executor": executor}
    
    # Add display dimensions from metadata
    if metadata:
        width = metadata.get("display_width")
        height = metadata.get("display_height")
        if width and height:
            tool_kwargs["width"] = width
            tool_kwargs["height"] = height
    
    # Add computer tools (all are BaseTool subclasses)
    mcp.add_tool(HudComputerTool(**tool_kwargs))
    mcp.add_tool(AnthropicComputerTool(**tool_kwargs))
    mcp.add_tool(OpenAIComputerTool(**tool_kwargs))
    
    # Mount hubs
    setup_hub.env = playwright_tool
    evaluate_hub.env = playwright_tool
    mcp.mount(setup_hub)
    mcp.mount(evaluate_hub)

@mcp.shutdown
async def shutdown_environment():
    """Cleanup browser resources."""
    global browser_provider
    if browser_provider:
        browser_provider.close()
    browser_provider = None

​
Standard Structure

​
Directory Layout

my-environment/
├── Dockerfile
├── pyproject.toml
├── controller/                 # MCP controller (stdio)
│   ├── __init__.py             # mcp = MCPServer()
│   ├── __main__.py             # python -m controller → mcp.run()
│   ├── hooks.py                # @mcp.initialize / @mcp.shutdown
│   └── tools.py                # @mcp.tool(...)
└── environment/                # Optional backend (HTTP/IPC)
    └── server.py               # e.g., FastAPI app

​
Dockerfile

FROM python:3.11-slim

WORKDIR /app

# Copy and install
COPY pyproject.toml ./
COPY controller/ ./controller/
COPY environment/ ./environment/
RUN pip install --no-cache-dir -e .

ENV ENV_SERVER_PORT=8005

# Start optional backend, then MCP controller on stdio
CMD ["sh", "-c", "uvicorn environment.server:app --host 0.0.0.0 --port $ENV_SERVER_PORT --log-level warning & python -m controller"]

​
Hub Module Pattern

Example from text_2048: 
# src/hud_controller/setup/__init__.py
from hud.tools.base import BaseHub

setup = BaseHub("setup")

# Import all setup functions to register them
from . import board

__all__ = ["setup"]

# src/hud_controller/setup/board.py
from . import setup

@setup.tool("board")
async def setup_board(board_size: int = 4):
    """Initialize game board."""
    game = setup.env  # Access environment from hub
    game.reset(size=board_size)
    return [TextContent(text=f"{board_size}x{board_size} game initialized")]

​
Key Concepts

​
Environment State

Three patterns for managing state:
  1. Global variables (simple environments): 
    game = None

@mcp.initialize
async def initialize_environment(ctx):
    global game
    game = Game2048()
  2. Context class (complex environments): 
    class EnvironmentContext:
    def __init__(self):
        self.browser = None
        self.page = None

env = EnvironmentContext()
  3. Hub env attribute (for tool access): 
    setup_hub.env = game  # Tools access via hub.env

​
Tool Lifecycle

  1. Setup tools - Hidden from agents, prepare environment state
  2. Interaction tools - Available to agents for control
  3. Evaluate tools - Hidden from agents, score performance

​
Progress Notifications

Send progress updates during long-running operations: 
async def send_progress(progress: int, message: str):
    if progress_token:
        await ctx.session.send_progress_notification(
            progress_token=progress_token,
            progress=progress,
            total=100,
            message=message
        )
Progress notifications follow the MCP progress specification. The progressToken comes from the client’s request metadata.

​
Metadata Access

Agent metadata flows through initialization: 
@mcp.initialize
async def initialize_environment(ctx):
    # From agent's metadata class variable
    width = ctx.meta.get("display_width", 1920) if ctx.meta else 1920
    height = ctx.meta.get("display_height", 1080) if ctx.meta else 1080

​
Testing

# CLI testing
hud debug my-env:latest
hud analyze my-env:latest

# Python testing
async def test():
    from hud import Environment
    
    env = Environment("test").connect_mcp_config({
        "env": {"command": "docker", "args": ["run", "-i", "my-env"]}
    })
    
    async with env:
        tools = await env.list_tools()
        result = await env.call_tool("setup", value=0)

​
See Also

  • Environments - Environment class (client-side)
  • Tools - Tool implementation reference
  • Evals - Running evaluations