mcp-forge/docs/todo.md

MCP-Forge Development TODO

Use uv:

• To run python, use uv run python. To use pytest, use uv run pytest. Using uv run guarantees that we use the correct virtual environment.

• To add packages, use uv add. For development dependencies, use uv add --dev.

Progress tracking: Use this todo.md file to track your progress.

Reference Architecture: See architecture1.md for complete system design.

Development Philosophy:

• Bottom-up approach: build foundational modules first
• Test-driven development: write tests before implementation
• Integration testing: test submodules individually, then integration
• No shortcuts: follow architecture strictly, no simplified POC approaches
• Clear acceptance criteria: prevent circumventing security and architectural decisions
• Build according to Implementation Roadmap (Phases 1-5 in architecture)

---

Progress Summary

Status: Phase 5.2 Complete - 387 tests passing

Completed Phases:

• ✅ Phase 1.1: Configuration Management (35 tests)
  • config/schema.py - Configuration models with validation
  • config/loader.py - YAML loading with environment substitution
• ✅ Phase 1.2: Security & Validation Core (58 tests)
  • security/resource_limits.py - Resource limit parsing and validation
  • security/allowlist.py - Package allowlist/blocklist management
  • security/audit.py - Comprehensive audit logging
• ✅ Phase 1.3: Podman Integration Core (25 tests)
  • podman/client.py - Podman API client wrapper (renamed test_podman_client.py)
  • podman/containers.py - Secure container lifecycle management
• ✅ Phase 2.1: Simple Backend (37 tests)
  • execution/simple/executor.py - Stateless code execution (17 tests)
  • execution/simple/backend.py - Simple backend orchestration (20 tests)
• ✅ Phase 2.2: Jupyter Backend (65 tests)
  • execution/jupyter/kernel.py - IPython kernel management (22 tests)
  • execution/jupyter/sessions.py - Session lifecycle management (24 tests)
  • execution/jupyter/backend.py - Stateful execution orchestration (19 tests)
• ✅ Phase 3.1: Package Management (42 tests)
  • builder/package_validator.py - Package security validation (21 tests)
  • builder/uv_installer.py - UV-based package installer (21 tests)
• ✅ Phase 3.2: Environment Builder (33 tests)
  • builder/image_builder.py - Podman image building (18 tests)
  • ⏭️ builder/security_scanner.py - SKIPPED (optional trivy-based vulnerability scanning)
  • builder/environment_builder.py - Orchestration with rate limiting (15 tests)
• ✅ Phase 4: MCP Tool Integration (40 tests)
  • mcp/client.py - fastmcp-based MCP client wrapper (12 tests)
  • mcp/manager.py - Multi-client manager with collision detection (9 tests)
  • mcp/bridge.py - Unix socket bridge server (8 tests)
  • mcp/injection.py - Python code generator for tool injection (11 tests)
• ✅ Phase 5.1.1: MCP Resource Handlers (13 tests)
  • server/resources.py - MCP resource handlers for discovery and state
• ✅ Phase 5.1.2: MCP Tools (34 tests)
  • server/tools/execute_python.py - Execute Python code with tool injection (15 tests)
  • server/tools/document_state.py - Document session state and variables (8 tests)
  • server/tools/build_environment.py - Build custom environments (11 tests)
• ✅ Phase 5.2: Main MCP Server (5 tests)
  • server/server.py - ForgeServer orchestration with component initialization
  • server/__init__.py - Module exports

Pending:

• Phase 5.3: Integration & End-to-End Testing
• Phase 6: Documentation & Deployment

New Workstream (2026-03-04): R execution environments

• Add language selection to execution tooling (python + r)
• Add simple backend R shell execution path
• Add stateful Jupyter R kernel support (IRkernel)
• Add MCP tool injection generator for R wrappers
• Add tests first for R simple/stateful execution and server dispatch

Test Count: 387 tests passing

Last Updated: 2026-02-06

---

Phase 1: Foundation & Core Infrastructure

1.1 Configuration Management

1.1.1 Configuration Schema Module

Path: src/mcp_forge/config/schema.py

Purpose: Define and validate configuration structures using Pydantic models.

Tests to write first:

• tests/config/test_schema.py
  • Test valid configuration loads successfully
  • Test invalid configuration raises ValidationError
  • Test default values are applied correctly
  • Test environment variable substitution (e.g., ${GITHUB_TOKEN})
  • Test nested configuration validation
  • Test constraint validation (e.g., max_timeout >= default_timeout)

Implementation requirements:

class ServerConfig(BaseModel):
    host: str = "localhost"
    port: int = 3000
    podman_socket: Path
    
    @validator('port')
    def validate_port(cls, v):
        if not 1 <= v <= 65535:
            raise ValueError("Port must be 1-65535")
        return v

class ExecutionConfig(BaseModel):
    default_backend: Literal["simple", "jupyter"] = "simple"
    default_timeout: int = 300
    max_timeout: int = 1800
    default_memory: str = "512m"
    max_memory: str = "2g"
    default_cpu_quota: int = 50000
    max_cpu_quota: int = 100000
    
    @validator('max_timeout')
    def validate_max_timeout(cls, v, values):
        if v < values.get('default_timeout', 0):
            raise ValueError("max_timeout must be >= default_timeout")
        return v

class ImageConfig(BaseModel):
    python_3_11: str = "mcp-forge/python:3.11"
    python_3_12: str = "mcp-forge/python:3.12"
    jupyter: str = "mcp-forge/jupyter:latest"
    auto_pull: bool = True
    pull_interval: int = 86400

class SessionConfig(BaseModel):
    idle_timeout: int = 3600
    max_concurrent: int = 10
    cleanup_interval: int = 300

class VolumeConfig(BaseModel):
    base_path: Path
    session_quota: str = "1g"
    max_session_quota: str = "10g"

class SecurityConfig(BaseModel):
    audit_log: Path
    enforce_resource_limits: bool = True
    allow_network: bool = False

class PackageValidationConfig(BaseModel):
    use_allowlist: bool = True
    allowlist_path: Path
    blocklist_path: Path
    require_approval_patterns: List[str]

class EnvironmentBuilderConfig(BaseModel):
    enabled: bool = True
    uv_cache_path: Path
    max_packages_per_build: int = 50
    max_build_time: int = 600
    max_image_size: int = 2147483648
    max_concurrent_builds: int = 3
    build_rate_limit: dict
    package_validation: PackageValidationConfig
    auto_cleanup: dict
    templates: Dict[str, dict]

class MCPToolConfig(BaseModel):
    command: str
    args: List[str]
    env: Dict[str, str] = {}

class ForgeConfig(BaseModel):
    server: ServerConfig
    execution: ExecutionConfig
    images: ImageConfig
    sessions: SessionConfig
    volumes: VolumeConfig
    security: SecurityConfig
    environment_builder: EnvironmentBuilderConfig
    mcp_tools: Dict[str, MCPToolConfig]

Acceptance criteria:

• All configuration fields have proper type validation
• Cross-field validation works (e.g., max >= default)
• Environment variables are substituted correctly
• Invalid configurations raise clear ValidationError with field path
• No hardcoded values; all configurable
• 100% test coverage on schema validation

---

1.1.2 Configuration Loader Module

Path: src/mcp_forge/config/loader.py

Purpose: Load configuration from YAML files with environment variable substitution.

Tests to write first:

• tests/config/test_loader.py
  • Test load from valid YAML file
  • Test load from non-existent file raises FileNotFoundError
  • Test invalid YAML raises YAMLError
  • Test environment variable substitution in strings
  • Test nested environment variable substitution
  • Test missing environment variable raises clear error
  • Test loading from multiple sources (file + env overrides)
  • Test configuration merging (defaults + file + env)

Implementation requirements:

def substitute_env_vars(value: Any) -> Any:
    """Recursively substitute ${VAR} with environment variables."""
    pass

def load_config(config_path: Optional[Path] = None) -> ForgeConfig:
    """
    Load configuration from YAML file and environment.
    
    Priority: Environment variables > Config file > Defaults
    """
    pass

def load_config_from_dict(config_dict: dict) -> ForgeConfig:
    """Load configuration from dictionary (for testing)."""
    pass

Acceptance criteria:

• Supports YAML configuration files
• Environment variable substitution works recursively
• Missing env vars raise clear errors with variable name
• Configuration priority is respected (env > file > defaults)
• Can load partial configurations (missing sections use defaults)
• No eval() or exec() - only safe string substitution
• 100% test coverage

---

1.2 Security & Validation Core

1.2.1 Resource Limit Parser

Path: src/mcp_forge/security/resource_limits.py

Purpose: Parse and validate resource limit strings (memory, CPU, storage).

Tests to write first:

• tests/security/test_resource_limits.py
  • Test parse_memory_string("512m") → 536870912
  • Test parse_memory_string("2g") → 2147483648
  • Test parse_memory_string("1024k") → 1048576
  • Test invalid format raises ValueError
  • Test negative values raise ValueError
  • Test zero values raise ValueError
  • Test parse_cpu_quota(50000) validates correctly
  • Test CPU quota > 1000000 raises ValueError
  • Test CPU quota < 0 raises ValueError
  • Test storage size parsing (same as memory)

Implementation requirements:

def parse_memory_string(memory: str) -> int:
    """
    Parse memory string to bytes.
    
    Supports: k, m, g suffixes (case-insensitive)
    Examples: "512m" → 536870912, "2g" → 2147483648
    
    Raises:
        ValueError: If format is invalid or value is <= 0
    """
    pass

def parse_cpu_quota(cpu_quota: int) -> int:
    """
    Validate CPU quota value.
    
    CPU quota is in microseconds per 100ms period.
    100000 = 100% of one CPU core
    
    Raises:
        ValueError: If quota <= 0 or > 1000000 (10 cores max)
    """
    pass

def parse_storage_string(storage: str) -> int:
    """Parse storage string to bytes (same as memory)."""
    pass

class ResourceLimits:
    """Resource limits with validation."""
    
    def __init__(
        self,
        memory: str,
        cpu_quota: int,
        storage: Optional[str] = None,
        pids_limit: int = 100,
        timeout: int = 300
    ):
        self.memory_bytes = parse_memory_string(memory)
        self.cpu_quota = parse_cpu_quota(cpu_quota)
        self.storage_bytes = parse_storage_string(storage) if storage else None
        self.pids_limit = pids_limit
        self.timeout = timeout
        
    def to_podman_params(self) -> dict:
        """Convert to Podman container create parameters."""
        pass

Acceptance criteria:

• Parses all standard units: k, m, g (case-insensitive)
• Validates positive values only
• Raises clear ValueError with problematic value
• CPU quota limited to reasonable maximum (10 cores)
• ResourceLimits class enforces all constraints
• to_podman_params() returns valid Podman parameter dict
• 100% test coverage

---

1.2.2 Podman Operation Allowlist

Path: src/mcp_forge/security/allowlist.py

Purpose: Define and enforce allowed Podman operations with parameter validation.

Tests to write first:

• tests/security/test_allowlist.py
  • Test allowed operation with valid params passes
  • Test allowed operation with forbidden params raises SecurityError
  • Test forbidden operation raises SecurityError
  • Test image allowlist enforcement
  • Test required parameters validation
  • Test volume mount path validation
  • Test capability restrictions
  • Test network mode enforcement
  • Test privilege mode always rejected
  • Test session container tracking

Implementation requirements:

class SecurityError(Exception):
    """Raised when security policy is violated."""
    pass

ALLOWED_IMAGES = [
    "mcp-forge/python:3.11",
    "mcp-forge/python:3.12",
    "mcp-forge/jupyter:latest",
    "mcp-forge/custom:*",  # Custom user images
]

FORBIDDEN_CONTAINER_PARAMS = [
    "privileged",
    "cap_add",
    "devices",
    "pid_mode",
    "ipc_mode",
]

REQUIRED_CONTAINER_PARAMS = {
    "network_mode": "none",
    "read_only": True,
    "security_opt": ["no-new-privileges"],
    "user": "1000:1000",
}

ALLOWED_VOLUME_PATTERNS = [
    "/mcp-forge/sessions/{session_id}/*",
    "/mcp-forge/shared/readonly/*",
    "/mcp-forge/uploads/{session_id}/*",
]

FORBIDDEN_MOUNT_PATHS = [
    "/",
    "/etc",
    "/var/run/docker.sock",
    "/var/run/podman/podman.sock",
    "/sys",
    "/proc",
]

class OperationValidator:
    """Validates Podman operations against security policy."""
    
    def __init__(self, config: SecurityConfig):
        self.config = config
        self.session_containers: Set[str] = set()
    
    def validate_container_create(
        self,
        image: str,
        params: dict,
        session_id: Optional[str] = None
    ) -> None:
        """
        Validate container create operation.
        
        Raises:
            SecurityError: If operation violates security policy
        """
        pass
    
    def validate_container_start(self, container_id: str) -> None:
        """Validate container start - must be session container."""
        pass
    
    def validate_container_stop(self, container_id: str) -> None:
        """Validate container stop - must be session container."""
        pass
    
    def validate_container_remove(self, container_id: str) -> None:
        """Validate container remove - must be session container."""
        pass
    
    def validate_volume_mount(self, mount_path: str, session_id: str) -> None:
        """
        Validate volume mount path against allowed patterns.
        
        Raises:
            SecurityError: If path is forbidden or doesn't match allowed patterns
        """
        pass
    
    def validate_image_name(self, image: str) -> None:
        """
        Validate image name against allowlist.
        
        Supports wildcards: mcp-forge/custom:*
        
        Raises:
            SecurityError: If image not in allowlist
        """
        pass
    
    def register_session_container(self, container_id: str) -> None:
        """Register container as belonging to a session."""
        pass
    
    def unregister_session_container(self, container_id: str) -> None:
        """Unregister session container."""
        pass

Acceptance criteria:

• All forbidden operations are rejected
• Required parameters are enforced
• Image allowlist with wildcard support works
• Volume mount validation prevents host path access
• Forbidden mount paths are blocked
• privileged mode always rejected regardless of other params
• Session container tracking prevents operating on non-session containers
• Clear SecurityError messages indicate what rule was violated
• 100% test coverage including edge cases

---

1.2.3 Audit Logger

Path: src/mcp_forge/security/audit.py

Purpose: Structured logging of security-relevant operations.

Tests to write first:

• tests/security/test_audit.py
  • Test log entries written to file
  • Test log entries are valid JSON
  • Test log entries contain required fields
  • Test timestamp format is ISO 8601
  • Test log rotation works
  • Test concurrent logging is thread-safe
  • Test security violations are logged with correct severity
  • Test PII is not logged

Implementation requirements:

from enum import Enum
from typing import Any, Optional
import json
from datetime import datetime
from pathlib import Path
import threading

class AuditEventType(Enum):
    CONTAINER_CREATE = "container.create"
    CONTAINER_START = "container.start"
    CONTAINER_STOP = "container.stop"
    CONTAINER_REMOVE = "container.remove"
    EXECUTION_REQUEST = "execution.request"
    SECURITY_VIOLATION = "security.violation"
    BUILD_REQUEST = "build.request"
    BUILD_COMPLETE = "build.complete"
    SESSION_CREATE = "session.create"
    SESSION_DESTROY = "session.destroy"

class AuditSeverity(Enum):
    INFO = "info"
    WARNING = "warning"
    ERROR = "error"
    CRITICAL = "critical"

class AuditLogger:
    """Thread-safe structured audit logger."""
    
    def __init__(self, log_path: Path):
        self.log_path = log_path
        self.lock = threading.Lock()
        self._ensure_log_file()
    
    def log(
        self,
        event_type: AuditEventType,
        severity: AuditSeverity = AuditSeverity.INFO,
        session_id: Optional[str] = None,
        user_id: Optional[str] = None,
        details: Optional[dict] = None,
        success: bool = True,
        error: Optional[str] = None
    ) -> None:
        """
        Log an audit event.
        
        Event structure:
        {
            "timestamp": "2026-02-06T10:30:00Z",
            "event": "container.create",
            "severity": "info",
            "session_id": "abc123",
            "user_id": "user@example.com",
            "success": true,
            "details": {...},
            "error": null
        }
        
        Must NOT log:
        - Code content (for privacy)
        - Authentication tokens
        - File contents
        - PII beyond user_id
        """
        pass
    
    def log_container_operation(
        self,
        operation: AuditEventType,
        container_id: str,
        image: str,
        session_id: Optional[str] = None,
        resources: Optional[dict] = None,
        success: bool = True,
        error: Optional[str] = None
    ) -> None:
        """Log container operation with standard fields."""
        pass
    
    def log_security_violation(
        self,
        violation_type: str,
        details: dict,
        session_id: Optional[str] = None
    ) -> None:
        """Log security violation at CRITICAL severity."""
        pass
    
    def _ensure_log_file(self) -> None:
        """Ensure log file and directory exist."""
        pass
    
    def _write_log_entry(self, entry: dict) -> None:
        """Thread-safe write of log entry."""
        pass

Acceptance criteria:

• All log entries are valid JSON
• Timestamps are ISO 8601 format
• Log file is created if it doesn't exist
• Concurrent logging is thread-safe (test with threading)
• No PII or sensitive data is logged (code content, tokens, files)
• Only hash of code is logged for execution requests
• Security violations logged at CRITICAL severity
• Log entries contain all required fields
• 100% test coverage

---

1.3 Podman Integration Core

1.3.1 Podman Client Wrapper

Path: src/mcp_forge/podman/client.py

Purpose: Wrap Podman API with security validation and error handling.

Tests to write first:

• tests/podman/test_client.py
  • Test connection to Podman socket succeeds
  • Test connection failure raises clear error
  • Test socket path validation
  • Test socket permissions check
  • Test API version compatibility check
  • Test ping/health check
  • Mock all actual Podman calls (use pytest-mock)

Implementation requirements:

from podman import PodmanClient as BasePodmanClient
from podman.errors import APIError, NotFound
from typing import Optional
from pathlib import Path

class PodmanConnectionError(Exception):
    """Raised when connection to Podman fails."""
    pass

class PodmanClient:
    """
    Wrapper around Podman API with security validation.
    
    All container operations are validated against security policy
    before being sent to Podman.
    """
    
    def __init__(
        self,
        socket_path: Path,
        validator: OperationValidator,
        audit_logger: AuditLogger
    ):
        self.socket_path = socket_path
        self.validator = validator
        self.audit_logger = audit_logger
        self._client: Optional[BasePodmanClient] = None
    
    def connect(self) -> None:
        """
        Connect to Podman socket.
        
        Raises:
            PodmanConnectionError: If connection fails
        """
        pass
    
    def ping(self) -> bool:
        """Test connection to Podman."""
        pass
    
    def disconnect(self) -> None:
        """Disconnect from Podman."""
        pass
    
    def verify_socket_access(self) -> None:
        """
        Verify socket exists and is accessible.
        
        Raises:
            PodmanConnectionError: If socket not accessible
        """
        pass
    
    def check_api_version(self) -> dict:
        """Get Podman API version information."""
        pass
    
    @property
    def client(self) -> BasePodmanClient:
        """Get underlying Podman client (lazy connection)."""
        if self._client is None:
            self.connect()
        return self._client

Acceptance criteria:

• Validates socket path exists before connecting
• Checks socket permissions (must be readable)
• Lazy connection (only connects when needed)
• Ping/health check works
• API version check works
• Clear error messages for connection failures
• Graceful disconnect/cleanup
• All tests use mocked Podman client (no actual Podman needed)
• 100% test coverage

---

1.3.2 Secure Container Manager

Path: src/mcp_forge/podman/containers.py

Purpose: Create, manage, and cleanup containers with security enforcement.

Tests to write first:

• tests/podman/test_containers.py
  • Test create_container with valid params succeeds
  • Test create_container with forbidden params raises SecurityError
  • Test create_container with invalid image raises SecurityError
  • Test create_container enforces required parameters
  • Test resource limits are applied correctly
  • Test volume mounts are validated
  • Test start_container on session container succeeds
  • Test start_container on non-session container raises SecurityError
  • Test stop_container works
  • Test remove_container works
  • Test cleanup orphaned containers
  • All tests use mocked Podman client

Implementation requirements:

from typing import Optional, Dict, List
from datetime import datetime, timedelta

class ContainerConfig:
    """Container configuration with security defaults."""
    
    def __init__(
        self,
        image: str,
        command: Optional[List[str]] = None,
        environment: Optional[Dict[str, str]] = None,
        volumes: Optional[Dict[str, dict]] = None,
        resource_limits: Optional[ResourceLimits] = None,
        working_dir: str = "/workspace",
        user: str = "1000:1000"
    ):
        self.image = image
        self.command = command or []
        self.environment = environment or {}
        self.volumes = volumes or {}
        self.resource_limits = resource_limits
        self.working_dir = working_dir
        self.user = user
    
    def to_podman_params(self) -> dict:
        """
        Convert to Podman container create parameters.
        
        Ensures all security requirements are included:
        - network_mode: none
        - read_only: True
        - security_opt: ["no-new-privileges"]
        - resource limits
        """
        pass

class SecureContainerManager:
    """Manages container lifecycle with security enforcement."""
    
    def __init__(
        self,
        podman_client: PodmanClient,
        validator: OperationValidator,
        audit_logger: AuditLogger
    ):
        self.podman = podman_client
        self.validator = validator
        self.audit_logger = audit_logger
    
    def create_container(
        self,
        config: ContainerConfig,
        session_id: Optional[str] = None,
        name: Optional[str] = None
    ) -> str:
        """
        Create a container with security validation.
        
        Returns:
            Container ID
        
        Raises:
            SecurityError: If configuration violates security policy
            PodmanError: If container creation fails
        """
        pass
    
    def start_container(self, container_id: str) -> None:
        """
        Start a container.
        
        Raises:
            SecurityError: If container is not a session container
        """
        pass
    
    def stop_container(
        self,
        container_id: str,
        timeout: int = 10
    ) -> None:
        """Stop a container."""
        pass
    
    def remove_container(
        self,
        container_id: str,
        force: bool = False
    ) -> None:
        """Remove a container."""
        pass
    
    def get_container_logs(
        self,
        container_id: str,
        tail: int = 100
    ) -> tuple[str, str]:
        """
        Get container stdout and stderr logs.
        
        Returns:
            (stdout, stderr)
        """
        pass
    
    def wait_for_container(
        self,
        container_id: str,
        timeout: int = 300
    ) -> int:
        """
        Wait for container to exit.
        
        Returns:
            Exit code
        
        Raises:
            TimeoutError: If container doesn't exit within timeout
        """
        pass
    
    def cleanup_old_containers(
        self,
        max_age: timedelta = timedelta(hours=24)
    ) -> int:
        """
        Cleanup containers older than max_age.
        
        Returns:
            Number of containers removed
        """
        pass

Acceptance criteria:

• All container operations validated before execution
• Security parameters (read_only, network_mode, etc.) enforced
• Resource limits applied correctly
• Volume mounts validated against allowlist
• Only session containers can be started/stopped/removed
• Container creation logs to audit log
• Clear error messages for security violations
• Cleanup function removes old containers safely
• All operations are idempotent where possible
• All tests use mocked Podman (no real containers)
• 100% test coverage

---

Phase 2: Execution Backends

2.1 Simple Backend (Stateless Execution)

2.1.1 Code Executor Module

Path: src/mcp_forge/execution/simple/executor.py

Purpose: Execute Python code in stateless containers.

Tests to write first:

• tests/execution/simple/test_executor.py
  • Test execute simple Python code returns result
  • Test execute code with stdout capture
  • Test execute code with stderr capture
  • Test execute code timeout enforcement
  • Test execute code with exception handling
  • Test execute code with syntax error returns clear error
  • Test execute code with runtime error returns clear error
  • Test result serialization (JSON-compatible types)
  • Test large output handling
  • All tests use mocked containers

Implementation requirements:

from typing import Any, Optional, Dict
from dataclasses import dataclass
import json

@dataclass
class ExecutionResult:
    """Result of code execution."""
    success: bool
    stdout: str
    stderr: str
    result: Optional[Any]
    execution_time: float
    exit_code: int
    error: Optional[str] = None
    
    def to_dict(self) -> dict:
        """Convert to dictionary for JSON serialization."""
        pass

class CodeExecutor:
    """Executes Python code in isolated containers."""
    
    def __init__(
        self,
        container_manager: SecureContainerManager,
        image: str,
        resource_limits: ResourceLimits
    ):
        self.container_manager = container_manager
        self.image = image
        self.resource_limits = resource_limits
    
    def execute(
        self,
        code: str,
        timeout: Optional[int] = None
    ) -> ExecutionResult:
        """
        Execute Python code in a fresh container.
        
        Process:
        1. Create container with code
        2. Start container
        3. Wait for completion (with timeout)
        4. Capture stdout/stderr
        5. Extract result from last expression
        6. Cleanup container
        
        Args:
            code: Python code to execute
            timeout: Maximum execution time in seconds
        
        Returns:
            ExecutionResult with stdout, stderr, result, and timing
        """
        pass
    
    def _prepare_code(self, code: str) -> str:
        """
        Wrap code to capture result and handle errors.
        
        Wraps code in try/except and captures:
        - Last expression result
        - Exceptions with traceback
        - Execution metadata
        
        Returns wrapped code that outputs JSON to stdout.
        """
        pass
    
    def _parse_output(self, stdout: str) -> tuple[Any, Optional[str]]:
        """
        Parse execution output to extract result and error.
        
        Returns:
            (result, error_message)
        """
        pass

Acceptance criteria:

• Executes code in fresh container each time
• Captures stdout and stderr separately
• Returns result of last expression
• Handles syntax errors gracefully
• Handles runtime errors with traceback
• Enforces timeout strictly
• Cleans up container after execution (even on error)
• Result must be JSON-serializable
• Large output doesn't cause issues
• Code wrapping preserves line numbers for errors
• All tests use mocked containers
• 100% test coverage

---

2.1.2 Simple Backend Implementation

Path: src/mcp_forge/execution/simple/backend.py

Purpose: Simple backend orchestrating code execution.

Tests to write first:

• tests/execution/simple/test_backend.py
  • Test execute without MCP tools
  • Test execute with resource limit override
  • Test execute with custom image
  • Test execute with volume mounts
  • Test execute respects configuration defaults
  • Test execute validates resource limits against max
  • Test multiple concurrent executions
  • All tests use mocked components

Implementation requirements:

class SimpleBackend:
    """Stateless code execution backend."""
    
    def __init__(
        self,
        config: ForgeConfig,
        container_manager: SecureContainerManager,
        audit_logger: AuditLogger
    ):
        self.config = config
        self.container_manager = container_manager
        self.audit_logger = audit_logger
    
    def execute(
        self,
        code: str,
        timeout: Optional[int] = None,
        memory: Optional[str] = None,
        cpu_quota: Optional[int] = None,
        custom_image: Optional[str] = None,
        volumes: Optional[Dict[str, dict]] = None
    ) -> ExecutionResult:
        """
        Execute Python code in stateless container.
        
        Args:
            code: Python code to execute
            timeout: Max execution time (uses config default if None)
            memory: Memory limit (uses config default if None)
            cpu_quota: CPU quota (uses config default if None)
            custom_image: Custom image name (uses config default if None)
            volumes: Volume mounts
        
        Returns:
            ExecutionResult
        
        Raises:
            ValueError: If limits exceed configured maximums
        """
        pass
    
    def _validate_limits(
        self,
        timeout: int,
        memory: str,
        cpu_quota: int
    ) -> None:
        """
        Validate resource limits against configuration maximums.
        
        Raises:
            ValueError: If any limit exceeds maximum
        """
        pass
    
    def _get_image(self, custom_image: Optional[str]) -> str:
        """Get image name, defaulting to configured image."""
        pass

Acceptance criteria:

• Uses configuration defaults for unspecified limits
• Validates limits against configured maximums
• Supports custom images
• Supports volume mounts
• Logs execution to audit log (hash of code, not content)
• Handles concurrent executions safely
• Clear error for limit violations
• All tests use mocked components
• 100% test coverage

---

2.2 Jupyter Backend (Stateful Execution)

2.2.1 Jupyter Kernel Manager

Path: src/mcp_forge/execution/jupyter/kernel.py

Purpose: Manage IPython kernel lifecycle and communication.

Tests to write first:

• tests/execution/jupyter/test_kernel.py
  • Test start kernel in container
  • Test execute code in kernel
  • Test kernel namespace persistence
  • Test kernel shutdown
  • Test kernel timeout/restart
  • Test kernel error handling
  • Test introspection (list variables)
  • Test variable type/size inspection
  • All tests use mocked ZMQ and containers

Implementation requirements:

Architecture:

• jupyter-client runs in MCP-Forge server (host) - manages ZMQ connections
• ipykernel runs inside Podman container - actual kernel process
• 1:1 mapping: One container per session, one kernel per container
• No shared variables between sessions (separate namespaces)
• Optional shared volumes for file-based data exchange

from jupyter_client import KernelManager, BlockingKernelClient
from typing import Any, Dict, List, Optional
import zmq
import json
import tempfile

@dataclass
class KernelInfo:
    """Information about running kernel."""
    kernel_id: str
    container_id: str
    session_id: str
    connection_info: Dict[str, Any]  # ZMQ ports and keys
    started_at: datetime
    last_activity: datetime
    client: Optional[BlockingKernelClient] = None

class JupyterKernelManager:
    """
    Manages IPython kernels in containers via jupyter-client.
    
    Architecture:
    - This class runs on host (MCP-Forge server process)
    - Creates one container per session with ipykernel running inside
    - Connects to kernel via ZMQ protocol (jupyter-client)
    - Communicates with kernel using Jupyter message protocol
    """
    
    def __init__(
        self,
        container_manager: SecureContainerManager,
        image: str,
        resource_limits: ResourceLimits
    ):
        self.container_manager = container_manager
        self.image = image  # Image with ipykernel installed
        self.resource_limits = resource_limits
        self.kernels: Dict[str, KernelInfo] = {}
    
    def start_kernel(
        self,
        session_id: str,
        volumes: Optional[Dict[str, dict]] = None
    ) -> str:
        """
        Start IPython kernel in dedicated container.
        
        Process:
        1. Generate ZMQ connection info (ports, keys)
        2. Create container with ipykernel
        3. Start ipykernel process with connection file
        4. Wait for kernel to be ready
        5. Connect jupyter-client to kernel via ZMQ
        6. Verify kernel is responsive
        
        Returns:
            kernel_id: Unique identifier for this kernel
        """
        pass
    
    def execute_code(
        self,
        kernel_id: str,
        code: str,
        timeout: int = 300
    ) -> ExecutionResult:
        """
        Execute code in kernel via ZMQ.
        
        Uses jupyter-client to:
        1. Send execute_request message
        2. Receive stream (stdout/stderr) messages
        3. Receive execute_result/display_data messages
        4. Collect and parse all output
        
        Returns ExecutionResult with stdout, stderr, result
        """
        pass
    
    def shutdown_kernel(self, kernel_id: str) -> None:
        """
        Shutdown kernel and cleanup container.
        
        1. Send shutdown_request via ZMQ
        2. Wait for kernel shutdown
        3. Stop and remove container
        """
        pass
    
    def inspect_namespace(self, kernel_id: str) -> List[str]:
        """
        Get list of variables in kernel namespace.
        
        Executes introspection code:
        [var for var in dir() if not var.startswith('_')]
        """
        pass
    
    def get_variable_info(
        self,
        kernel_id: str,
        variable_name: str
    ) -> Dict[str, Any]:
        """
        Get detailed information about a variable.
        
        Executes introspection code to get:
        - type(var).__name__
        - sys.getsizeof(var) if available
        - var.shape if hasattr(var, 'shape')
        - repr(var)[:100]
        
        Returns dict with type, size, shape, repr
        """
        pass
    
    def restart_kernel(self, kernel_id: str) -> None:
        """
        Restart kernel (namespace reset, container kept).
        
        Sends restart_request via ZMQ.
        """
        pass
    
    def cleanup_idle_kernels(
        self,
        idle_timeout: timedelta
    ) -> int:
        """
        Cleanup kernels idle longer than timeout.
        
        Returns:
            Number of kernels cleaned up
        """
        pass

Acceptance criteria:

• jupyter-client dependency in server (host), ipykernel in container image
• Kernel starts successfully in dedicated container (1 per session)
• ZMQ connection established correctly (ports exposed from container)
• Code execution works via Jupyter message protocol
• Namespace persists between executions within same session
• Each session has completely isolated namespace
• Variable introspection works
• Variable info includes type, size, shape
• Kernel shutdown cleans up container
• Idle kernel cleanup works
• Kernel restart clears namespace but keeps container
• Kernel restart works
• Handles kernel crashes gracefully
• All tests use mocked ZMQ and containers
• 100% test coverage

---

2.2.2 Session Manager

Path: src/mcp_forge/execution/jupyter/sessions.py

Purpose: Manage stateful execution sessions with state documentation.

Tests to write first:

• tests/execution/jupyter/test_sessions.py
  • Test create session
  • Test execute in session
  • Test session state persistence
  • Test document state
  • Test retrieve documented state
  • Test session cleanup
  • Test session timeout
  • Test max concurrent sessions enforcement
  • Test session isolation
  • All tests use mocked kernel manager

Implementation requirements:

from typing import Dict, Optional
from datetime import datetime, timedelta
from dataclasses import dataclass, field

@dataclass
class SessionState:
    """Documented state for a session."""
    session_id: str
    documented_variables: Dict[str, str] = field(default_factory=dict)
    note: str = ""
    last_updated: datetime = field(default_factory=datetime.utcnow)
    all_variables: List[str] = field(default_factory=list)
    introspection: Dict[str, dict] = field(default_factory=dict)
    
    def to_dict(self) -> dict:
        """Convert to dictionary for JSON serialization."""
        pass

class Session:
    """Stateful execution session."""
    
    def __init__(
        self,
        session_id: str,
        kernel_id: str,
        created_at: datetime,
        resource_limits: ResourceLimits
    ):
        self.session_id = session_id
        self.kernel_id = kernel_id
        self.created_at = created_at
        self.last_activity = created_at
        self.resource_limits = resource_limits
        self.state = SessionState(session_id=session_id)
    
    def update_activity(self) -> None:
        """Update last activity timestamp."""
        pass
    
    def is_idle(self, timeout: timedelta) -> bool:
        """Check if session is idle beyond timeout."""
        pass

class SessionManager:
    """Manages stateful execution sessions."""
    
    def __init__(
        self,
        config: SessionConfig,
        kernel_manager: JupyterKernelManager,
        audit_logger: AuditLogger
    ):
        self.config = config
        self.kernel_manager = kernel_manager
        self.audit_logger = audit_logger
        self.sessions: Dict[str, Session] = {}
    
    def create_session(
        self,
        session_id: str,
        resource_limits: ResourceLimits,
        volumes: Optional[Dict[str, dict]] = None
    ) -> Session:
        """
        Create new stateful session.
        
        Raises:
            ValueError: If session_id already exists
            RuntimeError: If max concurrent sessions exceeded
        """
        pass
    
    def get_session(self, session_id: str) -> Session:
        """
        Get session by ID.
        
        Raises:
            KeyError: If session doesn't exist
        """
        pass
    
    def execute_in_session(
        self,
        session_id: str,
        code: str,
        timeout: int = 300
    ) -> ExecutionResult:
        """Execute code in session kernel."""
        pass
    
    def document_state(
        self,
        session_id: str,
        variables: Dict[str, str],
        note: str = "",
        clear: bool = False
    ) -> None:
        """
        Document important variables in session.
        
        Updates session.state with variable descriptions.
        Also runs introspection to get current variable info.
        """
        pass
    
    def get_session_state(self, session_id: str) -> SessionState:
        """Get documented state for session."""
        pass
    
    def destroy_session(self, session_id: str) -> None:
        """Destroy session and cleanup kernel."""
        pass
    
    def cleanup_idle_sessions(self) -> int:
        """
        Cleanup sessions idle beyond configured timeout.
        
        Returns:
            Number of sessions cleaned up
        """
        pass
    
    def _enforce_max_concurrent(self) -> None:
        """
        Enforce max concurrent sessions limit.
        
        Raises:
            RuntimeError: If at max concurrent sessions
        """
        pass

Acceptance criteria:

• Sessions are created with unique IDs
• Max concurrent sessions enforced
• Session state persists between executions
• Variable documentation works
• State introspection captures type/size info
• Idle sessions cleaned up automatically
• Session destruction cleans up kernel
• Session isolation verified (can't access other sessions)
• Activity tracking works
• All tests use mocked components
• 100% test coverage

---

2.2.3 Jupyter Backend Implementation

Path: src/mcp_forge/execution/jupyter/backend.py

Purpose: Jupyter backend orchestrating stateful execution.

Tests to write first:

• tests/execution/jupyter/test_backend.py
  • Test execute creates session if needed
  • Test execute reuses existing session
  • Test execute updates session activity
  • Test session state documentation
  • Test session cleanup
  • All tests use mocked session manager

Implementation requirements:

class JupyterBackend:
    """Stateful code execution backend using Jupyter kernels."""
    
    def __init__(
        self,
        config: ForgeConfig,
        container_manager: SecureContainerManager,
        audit_logger: AuditLogger
    ):
        self.config = config
        self.container_manager = container_manager
        self.audit_logger = audit_logger
        
        # Initialize kernel manager and session manager
        kernel_manager = JupyterKernelManager(
            container_manager=container_manager,
            image=config.images.jupyter,
            resource_limits=self._default_resource_limits()
        )
        
        self.session_manager = SessionManager(
            config=config.sessions,
            kernel_manager=kernel_manager,
            audit_logger=audit_logger
        )
    
    def execute(
        self,
        code: str,
        session_id: str,
        timeout: Optional[int] = None,
        memory: Optional[str] = None,
        cpu_quota: Optional[int] = None,
        custom_image: Optional[str] = None,
        volumes: Optional[Dict[str, dict]] = None
    ) -> ExecutionResult:
        """
        Execute code in stateful session.
        
        Creates session if it doesn't exist.
        Reuses existing session if it exists.
        """
        pass
    
    def document_state(
        self,
        session_id: str,
        variables: Dict[str, str],
        note: str = "",
        clear: bool = False
    ) -> dict:
        """Document session state."""
        pass
    
    def get_session_state(self, session_id: str) -> SessionState:
        """Get session state."""
        pass
    
    def destroy_session(self, session_id: str) -> None:
        """Destroy session."""
        pass
    
    def list_sessions(self) -> List[dict]:
        """List all active sessions with metadata."""
        pass
    
    def _default_resource_limits(self) -> ResourceLimits:
        """Get default resource limits from config."""
        pass

Acceptance criteria:

• Auto-creates session on first execute
• Reuses session on subsequent executes
• Updates session activity timestamp
• Integrates with session manager correctly
• State documentation works
• Session listing works
• All tests use mocked components
• 100% test coverage

---

Phase 3: Custom Environment Builder

3.1 Package Management

3.1.1 Package Validator

Path: src/mcp_forge/builder/package_validator.py

Purpose: Validate package names against allowlist/blocklist.

Tests to write first:

• tests/builder/test_package_validator.py
  • Test allowlisted package passes
  • Test blocklisted package raises SecurityError
  • Test package requiring approval raises ApprovalRequiredError
  • Test wildcard pattern matching
  • Test version specifier parsing
  • Test max packages enforcement
  • Test load allowlist from file
  • Test load blocklist from file

Implementation requirements:

import re
from typing import List, Set, Pattern
from pathlib import Path

class ApprovalRequiredError(Exception):
    """Raised when package requires manual approval."""
    pass

class PackageValidator:
    """Validates package names against security policy."""
    
    def __init__(self, config: PackageValidationConfig):
        self.config = config
        self.allowlist: Set[str] = self._load_allowlist()
        self.blocklist: Set[str] = self._load_blocklist()
        self.approval_patterns: List[Pattern] = self._compile_patterns()
    
    def validate_packages(
        self,
        packages: List[str],
        max_packages: Optional[int] = None
    ) -> None:
        """
        Validate list of package specifications.
        
        Args:
            packages: List of package specs (e.g., ["numpy>=1.24", "pandas"])
            max_packages: Maximum number of packages allowed
        
        Raises:
            ValueError: If too many packages
            SecurityError: If package is blocklisted
            ApprovalRequiredError: If package requires approval
        """
        pass
    
    def validate_package(self, package_spec: str) -> None:
        """
        Validate single package specification.
        
        Extracts package name from spec (handles >=, ==, <=, etc.)
        Checks against blocklist, allowlist, and approval patterns.
        """
        pass
    
    def extract_package_name(self, package_spec: str) -> str:
        """
        Extract package name from specification.
        
        Examples:
            "numpy>=1.24.0" → "numpy"
            "requests==2.28.0" → "requests"
            "pandas" → "pandas"
        """
        pass
    
    def _load_allowlist(self) -> Set[str]:
        """Load allowlist from file."""
        pass
    
    def _load_blocklist(self) -> Set[str]:
        """Load blocklist from file."""
        pass
    
    def _compile_patterns(self) -> List[Pattern]:
        """Compile approval patterns to regex."""
        pass
    
    def _matches_pattern(self, package_name: str, pattern: Pattern) -> bool:
        """Check if package matches approval pattern."""
        pass

Acceptance criteria:

• Correctly parses package specifications with version operators
• Blocklisted packages rejected immediately
• Allowlisted packages pass validation
• Packages matching approval patterns raise ApprovalRequiredError
• Max packages limit enforced
• Allowlist/blocklist loaded from files
• Wildcard patterns work (e.g., crypto)
• Clear error messages indicate which package failed
• 100% test coverage

---

3.1.2 UV Package Installer

Path: src/mcp_forge/builder/uv_installer.py

Purpose: Install packages using UV with caching.

Tests to write first:

• tests/builder/test_uv_installer.py
  • Test generate requirements file
  • Test generate Containerfile
  • Test UV installation script
  • Test cache directory handling
  • Test build with cache hit
  • Test build with cache miss
  • All tests use mocked file operations

Implementation requirements:

from typing import List, Optional
from pathlib import Path
import tempfile
import shutil

class UVInstaller:
    """Manages UV-based package installation in containers."""
    
    def __init__(self, cache_path: Path):
        self.cache_path = cache_path
        self._ensure_cache_dir()
    
    def generate_containerfile(
        self,
        base_image: str,
        packages: List[str],
        python_version: str = "3.11"
    ) -> str:
        """
        Generate Containerfile for building custom environment.
        
        Containerfile structure:
        1. Base image
        2. Install UV (cached layer)
        3. Create user
        4. Copy requirements.txt (cache-friendly)
        5. Install packages with UV
        6. Copy MCP tools
        7. Set working directory
        """
        pass
    
    def generate_requirements(self, packages: List[str]) -> str:
        """
        Generate requirements.txt content.
        
        One package per line with version specifiers preserved.
        """
        pass
    
    def create_build_context(
        self,
        base_image: str,
        packages: List[str],
        python_version: str = "3.11"
    ) -> Path:
        """
        Create temporary build context directory.
        
        Contains:
        - Containerfile
        - requirements.txt
        - mcp_tools.py (if exists)
        
        Returns:
            Path to build context directory (caller must cleanup)
        """
        pass
    
    def _ensure_cache_dir(self) -> None:
        """Ensure UV cache directory exists."""
        pass
    
    def _get_cache_volume_mount(self) -> dict:
        """Get volume mount configuration for UV cache."""
        pass

Acceptance criteria:

• Generates valid Containerfile
• Containerfile uses multi-stage caching
• Requirements.txt has one package per line
• UV installation layer is cached
• Cache directory mounted correctly during build
• Build context is temporary and isolated
• All required files included in context
• 100% test coverage

---

3.2 Environment Building

3.2.1 Image Builder

Path: src/mcp_forge/builder/image_builder.py

Purpose: Build container images using Podman with security scanning.

Tests to write first:

• tests/builder/test_image_builder.py
  • Test build image from context
  • Test build with cache
  • Test build timeout enforcement
  • Test image size validation
  • Test image tagging
  • Test build failure handling
  • All tests use mocked Podman client

Implementation requirements:

from typing import Optional, List
from datetime import datetime
import hashlib

@dataclass
class BuildResult:
    """Result of image build."""
    success: bool
    image_name: str
    image_id: str
    build_time: float
    size_bytes: int
    cache_hit: bool
    installed_packages: List[str]
    error: Optional[str] = None
    
    def to_dict(self) -> dict:
        """Convert to dictionary."""
        pass

class ImageBuilder:
    """Builds container images with security validation."""
    
    def __init__(
        self,
        podman_client: PodmanClient,
        config: EnvironmentBuilderConfig,
        audit_logger: AuditLogger
    ):
        self.podman = podman_client
        self.config = config
        self.audit_logger = audit_logger
    
    def build_image(
        self,
        name: str,
        build_context: Path,
        base_image: str,
        packages: List[str],
        timeout: Optional[int] = None
    ) -> BuildResult:
        """
        Build container image from build context.
        
        Process:
        1. Validate build context
        2. Generate image tag
        3. Build image with Podman
        4. Validate image size
        5. Tag image
        6. Cleanup build artifacts
        
        Args:
            name: Environment name (user-provided)
            build_context: Path to build context directory
            base_image: Base image to build from
            packages: List of packages being installed
            timeout: Build timeout (uses config default if None)
        
        Returns:
            BuildResult
        
        Raises:
            ValueError: If timeout exceeds max
            RuntimeError: If build fails
        """
        pass
    
    def _generate_tag(self, name: str) -> str:
        """
        Generate image tag.
        
        Format: mcp-forge/custom:{name}
        Validates name is alphanumeric + hyphens only.
        """
        pass
    
    def _validate_image_size(self, image_id: str) -> int:
        """
        Validate image size against maximum.
        
        Returns:
            Size in bytes
        
        Raises:
            ValueError: If image exceeds max size
        """
        pass
    
    def _extract_installed_packages(self, image_id: str) -> List[str]:
        """
        Extract list of installed packages from image.
        
        Runs: pip list --format=json in container
        """
        pass
    
    def _calculate_cache_hash(self, packages: List[str]) -> str:
        """Calculate hash of package list for cache key."""
        pass

Acceptance criteria:

• Builds image successfully with Podman
• Enforces build timeout
• Validates image size against maximum
• Tags image correctly (mcp-forge/custom:name)
• Extracts installed packages list
• Logs build to audit log
• Handles build failures gracefully
• Name validation (alphanumeric + hyphens only)
• All tests use mocked Podman
• 100% test coverage

---

3.2.2 Security Scanner

Path: src/mcp_forge/builder/security_scanner.py

Purpose: Scan images for security vulnerabilities (optional, uses trivy if available).

Tests to write first:

• tests/builder/test_security_scanner.py
  • Test scan with trivy available
  • Test scan with trivy unavailable (skip)
  • Test parse trivy JSON output
  • Test vulnerability severity classification
  • Test critical vulnerabilities rejected
  • All tests use mocked subprocess

Implementation requirements:

from typing import Optional, List
from dataclasses import dataclass
from enum import Enum
import subprocess
import json

class VulnerabilitySeverity(Enum):
    UNKNOWN = "UNKNOWN"
    LOW = "LOW"
    MEDIUM = "MEDIUM"
    HIGH = "HIGH"
    CRITICAL = "CRITICAL"

@dataclass
class Vulnerability:
    """Security vulnerability."""
    cve_id: str
    severity: VulnerabilitySeverity
    package: str
    fixed_version: Optional[str]
    description: str

@dataclass
class ScanResult:
    """Result of security scan."""
    scanned: bool
    total_vulns: int
    critical_vulns: int
    high_vulns: int
    medium_vulns: int
    low_vulns: int
    vulnerabilities: List[Vulnerability]
    
    def has_critical(self) -> bool:
        """Check if scan found critical vulnerabilities."""
        return self.critical_vulns > 0

class SecurityScanner:
    """Scans container images for vulnerabilities."""
    
    def __init__(self, enabled: bool = True):
        self.enabled = enabled
        self._trivy_available: Optional[bool] = None
    
    def scan_image(self, image_id: str) -> ScanResult:
        """
        Scan image for vulnerabilities.
        
        Uses trivy if available, otherwise returns empty result.
        
        Args:
            image_id: Image ID or tag to scan
        
        Returns:
            ScanResult with vulnerabilities found
        """
        pass
    
    def is_trivy_available(self) -> bool:
        """Check if trivy is installed and available."""
        pass
    
    def _run_trivy_scan(self, image_id: str) -> dict:
        """
        Run trivy scan and return JSON results.
        
        Command: trivy image --format json {image_id}
        """
        pass
    
    def _parse_trivy_output(self, trivy_json: dict) -> ScanResult:
        """Parse trivy JSON output to ScanResult."""
        pass

Acceptance criteria:

• Detects if trivy is available
• Skips scan gracefully if trivy not available
• Runs trivy correctly if available
• Parses trivy JSON output correctly
• Classifies vulnerabilities by severity
• Returns accurate counts
• All tests use mocked subprocess
• 100% test coverage

---

3.2.3 Environment Builder

Path: src/mcp_forge/builder/environment_builder.py

Purpose: Orchestrate custom environment building with all validations.

Tests to write first:

• tests/builder/test_environment_builder.py
  • Test build with valid packages
  • Test build with blocklisted package fails
  • Test build with too many packages fails
  • Test build with critical vulns fails
  • Test build timeout enforcement
  • Test build rate limiting
  • Test concurrent build limiting
  • Test template expansion
  • All tests use mocked components

Implementation requirements:

from typing import Optional, Dict, List
from datetime import datetime, timedelta
import threading

class BuildRateLimiter:
    """Rate limiter for build requests."""
    
    def __init__(self, max_requests: int, period_seconds: int):
        self.max_requests = max_requests
        self.period_seconds = period_seconds
        self.requests: Dict[str, List[datetime]] = {}
        self.lock = threading.Lock()
    
    def check_rate_limit(self, user_id: str) -> None:
        """
        Check if user is within rate limit.
        
        Raises:
            RuntimeError: If rate limit exceeded
        """
        pass
    
    def _cleanup_old_requests(self, user_id: str) -> None:
        """Remove requests older than period."""
        pass

class EnvironmentBuilder:
    """Builds custom Python environments with security validation."""
    
    def __init__(
        self,
        config: EnvironmentBuilderConfig,
        podman_client: PodmanClient,
        audit_logger: AuditLogger
    ):
        self.config = config
        self.podman = podman_client
        self.audit_logger = audit_logger
        
        # Initialize sub-components
        self.package_validator = PackageValidator(
            config.package_validation
        )
        self.uv_installer = UVInstaller(config.uv_cache_path)
        self.image_builder = ImageBuilder(
            podman_client, config, audit_logger
        )
        self.security_scanner = SecurityScanner(enabled=True)
        
        # Rate limiting and concurrency control
        self.rate_limiter = BuildRateLimiter(
            max_requests=config.build_rate_limit['requests'],
            period_seconds=config.build_rate_limit['period']
        )
        self.active_builds: Set[str] = set()
        self.active_builds_lock = threading.Lock()
    
    def build_custom_environment(
        self,
        name: str,
        packages: List[str],
        base_image: str = "python:3.11-slim",
        python_version: str = "3.11",
        description: str = "",
        user_id: str = "default"
    ) -> BuildResult:
        """
        Build custom environment with packages.
        
        Process:
        1. Check rate limit
        2. Check concurrent builds limit
        3. Validate package count
        4. Validate package names (allowlist/blocklist)
        5. Generate build context with UV
        6. Build image
        7. Scan for vulnerabilities
        8. Validate scan results
        9. Tag and register environment
        10. Cleanup build context
        
        Args:
            name: Environment name (alphanumeric + hyphens)
            packages: List of package specifications
            base_image: Base image to build from
            python_version: Python version
            description: Optional description
            user_id: User ID for rate limiting
        
        Returns:
            BuildResult
        
        Raises:
            ValueError: If validation fails
            SecurityError: If security check fails
            RuntimeError: If rate limit or concurrency exceeded
        """
        pass
    
    def build_from_template(
        self,
        template_name: str,
        additional_packages: Optional[List[str]] = None,
        name: Optional[str] = None,
        user_id: str = "default"
    ) -> BuildResult:
        """
        Build environment from template.
        
        Expands template packages and adds additional packages.
        """
        pass
    
    def list_templates(self) -> Dict[str, dict]:
        """List available templates."""
        return self.config.templates
    
    def _check_concurrent_builds(self) -> None:
        """
        Check concurrent builds limit.
        
        Raises:
            RuntimeError: If at max concurrent builds
        """
        pass
    
    def _register_build_start(self, name: str) -> None:
        """Register build as started."""
        pass
    
    def _register_build_complete(self, name: str) -> None:
        """Register build as completed."""
        pass

Acceptance criteria:

• Complete validation pipeline works
• Rate limiting enforced per user
• Concurrent builds limited
• Package validation runs first (fail fast)
• UV build context generated correctly
• Image building works
• Security scanning runs (if available)
• Critical vulnerabilities rejected
• Build context cleaned up even on failure
• Templates expand correctly
• All validations logged to audit log
• All tests use mocked components
• 100% test coverage

---

Phase 4: MCP Tool Integration

IMPORTANT: MCP Client Library Choice

This project uses fastmcp (https://gofastmcp.com) for MCP client operations instead of the standard mcp SDK. fastmcp provides:

• Modern, production-ready client implementation
• Automatic structured data deserialization
• Multiple transport options (STDIO, HTTP, SSE)
• Session persistence and caching
• Better error handling and connection management

Note: Example code in this document may show standard mcp SDK patterns for illustration. The actual implementation uses fastmcp APIs.

---

4.1 MCP Client Management

4.1.1 MCP Client Wrapper

Path: src/mcp_forge/mcp/client.py

Purpose: Connect to and manage MCP server clients using fastmcp.

Tests written:

• tests/mcp/test_client.py (12 tests)
  • Test connect to MCP server
  • Test connect with environment variables
  • Test disconnect from server
  • Test list available tools
  • Test list tools when not connected
  • Test get tool schema
  • Test get tool schema for unknown tool
  • Test call tool successfully
  • Test call tool when not connected
  • Test tool call failure handling
  • Test connection failure
  • Test reconnection after disconnect

Implementation complete:

from fastmcp import Client
from fastmcp.client.transports import StdioTransport

class MCPClientWrapper:
    """Wrapper for fastmcp Client connection."""
    
    def __init__(name, command, args, env)
    async def connect() -> None
    async def disconnect() -> None
    def is_connected() -> bool
    async def list_tools() -> List[str]
    async def get_tool_schema(tool_name) -> dict
    async def call_tool(tool_name, arguments) -> Any

Key features:

• Uses fastmcp for MCP client connections
• STDIO transport with environment variable support
• Automatic structured data deserialization via result.data
• Tools caching for performance
• Comprehensive error handling
• All tests use mocked fastmcp client

Acceptance criteria:

• ✅ Connects to MCP server successfully using fastmcp
• ✅ Lists available tools correctly
• ✅ Gets tool schemas correctly
• ✅ Calls tools with proper argument passing
• ✅ Returns structured results (result.data when available)
• ✅ Handles connection failures gracefully
• ✅ Disconnects cleanly
• ✅ All tests use mocked fastmcp client
• ✅ 100% test coverage (12 tests passing)

---

4.1.2 MCP Client Manager

Path: src/mcp_forge/mcp/manager.py

Purpose: Manage multiple MCP client connections.

Tests written:

• tests/mcp/test_manager.py (9 tests)
  • Test initialize clients from config with correct parameters
  • Test get client for tool returns correct client
  • Test get client for unknown tool raises KeyError
  • Test list all tools across all clients
  • Test detect tool name collision raises ValueError
  • Test get tool schema routes to correct client
  • Test call tool routes to correct client with arguments
  • Test shutdown all clients cleanly
  • Test manager before initialization raises RuntimeError

Implementation complete:

class MCPClientManager:
    """Manages multiple MCP client connections."""
    
    def __init__(config: Dict[str, Dict[str, Any]])
    async def initialize() -> None
    async def get_client_for_tool(tool_name) -> MCPClientWrapper
    async def list_all_tools() -> List[str]
    async def get_tool_schema(tool_name) -> dict
    async def call_tool(tool_name, arguments) -> Any
    async def shutdown() -> None
    def _check_initialized() -> None
    async def _build_tool_mapping() -> None

Key features:

• Initializes multiple MCPClientWrapper instances from config dictionary
• Builds comprehensive tool-to-client mapping for routing
• Detects and rejects tool name collisions with detailed error messages
• Routes tool calls to the correct client automatically
• Aggregates tools from all clients
• Clean shutdown of all client connections
• Initialization state checking with clear error messages
• All tests use mocked fastmcp clients

Acceptance criteria:

• ✅ Initializes all clients from config
• ✅ Builds tool-to-client mapping
• ✅ Detects and rejects tool name collisions
• ✅ Routes tool calls to correct client
• ✅ Lists all tools across clients
• ✅ Shuts down all clients cleanly
• ✅ All tests use mocked clients
• ✅ 100% test coverage (9 tests passing)

---

4.2 MCP Tool Injection

4.2.1 Tool Bridge Server

Path: src/mcp_forge/mcp/bridge.py

Purpose: Unix socket server that forwards tool calls from container to MCP clients.

Tests written:

• tests/mcp/test_bridge.py (8 tests)
  • Test bridge server starts and stops cleanly
  • Test receive and forward tool call to client manager
  • Test handle tool call error from client
  • Test handle invalid JSON in request
  • Test handle missing 'tool' field in request
  • Test concurrent requests handling
  • Test audit logging includes tool name only (not params)
  • Test socket cleanup on error

Implementation complete:

class ToolBridgeServer:
    """Unix socket server for MCP tool calls from containers."""
    
    def __init__(socket_path, client_manager, audit_logger)
    def start() -> None
    def stop() -> None
    def _run_server() -> None
    def _handle_connection(conn) -> None
    async def _call_tool_async(tool_name, params) -> dict
    def _send_error(conn, error) -> None

Key features:

• Unix socket server with AF_UNIX/SOCK_STREAM
• Background thread with graceful shutdown
• JSON protocol: {"tool": "name", "params": {...}}
• Async tool forwarding to MCPClientManager
• Concurrent request handling via separate threads
• Error responses with JSON format
• Audit logging (tool name only, parameters excluded for security)
• Automatic socket cleanup on shutdown

Acceptance criteria:

• ✅ Server starts and listens on Unix socket
• ✅ Accepts connections from containers
• ✅ Parses JSON tool call requests
• ✅ Forwards to correct MCP client
• ✅ Returns results as JSON
• ✅ Handles errors gracefully
• ✅ Supports concurrent requests
• ✅ Cleans up socket on shutdown
• ✅ Logs tool calls to audit log (tool name only, not params)
• ✅ All tests use mocked sockets and clients
• ✅ 100% test coverage (8 tests passing)

---

4.2.2 Tool Injection Generator

Path: src/mcp_forge/mcp/injection.py

Purpose: Generate Python code to inject MCP tools into container namespace.

Tests written:

• tests/mcp/test_injection.py (11 tests)
  • Test generated code is valid Python (AST parse)
  • Test generated code includes bridge client function
  • Test generated code includes tool wrapper functions
  • Test function signatures match tool schemas
  • Test generated functions have docstrings with descriptions
  • Test generated functions call bridge (_mcp_call)
  • Test type hints generated from JSON schema types
  • Test required vs optional parameters
  • Test generated code has necessary imports
  • Test empty tool list handling
  • Test custom socket path usage

Implementation complete:

class ToolInjectionGenerator:
    """Generates Python code to inject MCP tools into container."""
    
    def __init__(client_manager)
    async def generate_injection_code(tool_names, bridge_socket_path) -> str
    def _generate_imports() -> str
    def _generate_bridge_client(socket_path) -> str
    async def _generate_tool_function(tool_name, tool_schema) -> str
    def _extract_parameters(schema) -> List[Tuple]
    def _json_type_to_python(json_type) -> str

Key features:

• Generates valid, executable Python code
• Bridge client uses Unix socket to communicate with server
• Tool functions with proper signatures from schemas
• Type hints: string→str, integer→int, number→float, boolean→bool, array→list, object→dict
• Required parameters (no default) vs optional (default=None)
• Docstrings with tool descriptions and parameter docs
• Each function calls _mcp_call(tool_name, **kwargs)
• Socket path configurable for different deployment scenarios
• Comprehensive imports (socket, json, typing.Any)

Acceptance criteria:

• ✅ Generated code is valid Python

• ✅ Function signatures match tool schemas

• ✅ Type hints included from schema

• ✅ Docstrings include tool descriptions

• ✅ Bridge client code included

• ✅ Generated code can be executed without errors

• ✅ All required parameters are non-optional

• ✅ Optional parameters have defaults

• ✅ 100% test coverage (11 tests passing) Generate Python code that provides MCP tools as functions.

    Generated code includes:
    1. Bridge client to communicate with MCP bridge server
    2. Wrapper function for each tool with proper signature
    3. Docstrings from tool schemas
    4. Type hints from tool schemas
  
    Args:
        tool_names: List of MCP tool names to inject
        bridge_socket_path: Path to bridge socket in container
  
    Returns:
        Python code as string
    """
    pass
  

  def _generate_bridge_client(self, socket_path: str) -> str: """Generate code for bridge client communication.""" return ''' import socket import json from typing import Any

def _mcp_call(tool_name: str, **kwargs) -> Any: """Internal: Call MCP tool via bridge.""" s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) s.connect('{socket_path}') request = json.dumps({{'tool': tool_name, 'params': kwargs}}) s.sendall(request.encode('utf-8'))

# Receive response
response_data = b''
while True:
    chunk = s.recv(4096)
    if not chunk:
        break
    response_data += chunk

s.close()
response = json.loads(response_data.decode('utf-8'))

if not response.get('success'):
    raise RuntimeError(f"Tool call failed: {{response.get('error')}}")

return response.get('result')

'''.format(socket_path=socket_path)

async def _generate_tool_function(
    self,
    tool_name: str,
    tool_schema: dict
) -> str:
    """
    Generate wrapper function for a single tool.
    
    Includes:
    - Function signature from schema parameters
    - Type hints
    - Docstring with description and parameters
    - Call to _mcp_call()
    """
    pass

def _extract_parameters(self, schema: dict) -> List[tuple]:
    """
    Extract parameter definitions from tool schema.
    
    Returns:
        List of (name, type_hint, required, default) tuples
    """
    pass

def _json_type_to_python(self, json_type: str) -> str:
    """Convert JSON schema type to Python type hint."""
    type_map = {
        "string": "str",
        "integer": "int",
        "number": "float",
        "boolean": "bool",
        "array": "list",
        "object": "dict",
    }
    return type_map.get(json_type, "Any")

**Acceptance criteria:**
- [ ] Generated code is valid Python
- [ ] Function signatures match tool schemas
- [ ] Type hints included from schema
- [ ] Docstrings include tool descriptions
- [ ] Bridge client code included
- [ ] Generated code can be executed without errors
- [ ] All required parameters are non-optional
- [ ] Optional parameters have defaults
- [ ] 100% test coverage

---

## Phase 5: MCP Server Implementation

### 5.1 MCP Resources

#### 5.1.1 Resource Handlers
**Path:** `src/mcp_forge/server/resources.py`

**Purpose:** Implement MCP resource handlers for discovery and state.

**Tests to write first:**
- `tests/server/test_resources.py`
  - Test tools/available resource
  - Test sessions/{id}/state resource
  - Test sessions/{id}/variables resource
  - Test environments/list resource
  - Test environment/info resource
  - Test resource not found handling
  - All tests use mocked backends

**Implementation requirements:**

*Note: The example code below uses standard mcp.server patterns for illustration. Phase 5 server implementation may use fastmcp server features or standard mcp server depending on requirements.*

```python
from mcp.server import Server
from mcp.types import Resource, TextContent
from typing import Optional, List
import json

class ResourceHandler:
    """Handles MCP resource requests."""
    
    def __init__(
        self,
        client_manager: MCPClientManager,
        session_manager: SessionManager,
        environment_builder: EnvironmentBuilder,
        config: ForgeConfig
    ):
        self.client_manager = client_manager
        self.session_manager = session_manager
        self.environment_builder = environment_builder
        self.config = config
    
    async def handle_resource(self, uri: str) -> Resource:
        """
        Handle resource request based on URI.
        
        Supported URIs:
        - mcp://forge/tools/available
        - mcp://forge/sessions/{id}/state
        - mcp://forge/sessions/{id}/variables
        - mcp://forge/environments/list
        - mcp://forge/environment/info
        
        Raises:
            ValueError: If URI not recognized
        """
        pass
    
    async def _handle_tools_available(self) -> Resource:
        """Return list of available MCP tools."""
        tools = await self.client_manager.list_all_tools()
        return Resource(
            uri="mcp://forge/tools/available",
            mimeType="application/json",
            text=json.dumps(tools)
        )
    
    async def _handle_session_state(self, session_id: str) -> Resource:
        """Return documented state for session."""
        state = self.session_manager.get_session_state(session_id)
        return Resource(
            uri=f"mcp://forge/sessions/{session_id}/state",
            mimeType="application/json",
            text=json.dumps(state.to_dict())
        )
    
    async def _handle_session_variables(self, session_id: str) -> Resource:
        """Return list of variables in session."""
        state = self.session_manager.get_session_state(session_id)
        return Resource(
            uri=f"mcp://forge/sessions/{session_id}/variables",
            mimeType="application/json",
            text=json.dumps(state.all_variables)
        )
    
    async def _handle_environments_list(self) -> Resource:
        """Return list of custom environments and templates."""
        pass
    
    async def _handle_environment_info(self) -> Resource:
        """Return environment information."""
        pass
    
    def _parse_uri(self, uri: str) -> tuple:
        """
        Parse URI into components.
        
        Returns:
            (resource_type, parameters)
        """
        pass

Acceptance criteria:

• All resource URIs handled correctly
• Returns proper Resource objects
• JSON content is valid
• Session resources validate session exists
• Clear errors for invalid URIs
• Clear errors for non-existent sessions
• All tests use mocked components
• 100% test coverage

---

5.2 MCP Tools

5.2.1 Execute Python Tool

Path: src/mcp_forge/server/tools/execute_python.py

Purpose: Implement execute_python MCP tool.

Tests to write first:

• tests/server/tools/test_execute_python.py
  • Test execute with simple backend
  • Test execute with jupyter backend
  • Test execute with MCP tools
  • Test execute with custom image
  • Test execute with environment template
  • Test parameter validation
  • Test resource limit validation
  • All tests use mocked backends

Implementation requirements:

Note: The example code below uses standard mcp.types patterns for illustration. Phase 5 server implementation may use fastmcp server features or standard mcp server depending on requirements.

from mcp.types import Tool, TextContent
from typing import Optional, List, Dict, Any
import json

class ExecutePythonTool:
    """MCP tool for executing Python code."""
    
    def __init__(
        self,
        simple_backend: SimpleBackend,
        jupyter_backend: JupyterBackend,
        client_manager: MCPClientManager,
        bridge_server: ToolBridgeServer,
        injection_generator: ToolInjectionGenerator,
        config: ForgeConfig
    ):
        self.simple_backend = simple_backend
        self.jupyter_backend = jupyter_backend
        self.client_manager = client_manager
        self.bridge_server = bridge_server
        self.injection_generator = injection_generator
        self.config = config
    
    def get_tool_definition(self) -> Tool:
        """Return MCP tool definition."""
        return Tool(
            name="execute_python",
            description="Execute Python code in isolated container with MCP tools available",
            inputSchema={
                "type": "object",
                "properties": {
                    "code": {
                        "type": "string",
                        "description": "Python code to execute"
                    },
                    "mcp_tools": {
                        "type": "array",
                        "items": {"type": "string"},
                        "description": "List of MCP tool names to inject"
                    },
                    "session_id": {
                        "type": "string",
                        "description": "Session ID for stateful execution (null = stateless)"
                    },
                    "backend": {
                        "type": "string",
                        "enum": ["simple", "jupyter"],
                        "description": "Backend to use"
                    },
                    "timeout": {
                        "type": "integer",
                        "description": "Max execution time in seconds"
                    },
                    "custom_image": {
                        "type": "string",
                        "description": "Custom environment name"
                    },
                    "environment": {
                        "type": "string",
                        "description": "Template environment name"
                    }
                },
                "required": ["code"]
            }
        )
    
    async def execute(self, arguments: dict) -> List[TextContent]:
        """
        Execute Python code with MCP tool injection.
        
        Process:
        1. Validate arguments
        2. Determine backend (simple vs jupyter)
        3. Generate tool injection code if mcp_tools specified
        4. Inject tools into container
        5. Execute code
        6. Return result
        """
        pass
    
    def _validate_arguments(self, arguments: dict) -> None:
        """Validate tool arguments."""
        pass
    
    async def _inject_tools(
        self,
        container_id: str,
        tool_names: List[str]
    ) -> None:
        """
        Inject MCP tools into container.
        
        Generates injection code and writes to container filesystem.
        Mounts bridge socket into container.
        """
        pass
    
    def _select_backend(
        self,
        session_id: Optional[str],
        backend: Optional[str]
    ) -> str:
        """Determine which backend to use."""
        if session_id is not None:
            return "jupyter"
        return backend or self.config.execution.default_backend

Acceptance criteria:

• Tool definition matches architecture spec
• Validates all arguments
• Routes to correct backend
• Injects MCP tools correctly
• Mounts bridge socket
• Returns proper MCP response format
• Handles errors gracefully
• All tests use mocked components
• 100% test coverage

---

5.2.2 Document State Tool

Path: src/mcp_forge/server/tools/document_state.py

Purpose: Implement document_state MCP tool.

Tests to write first:

• tests/server/tools/test_document_state.py
  • Test document variables
  • Test document with note
  • Test clear existing documentation
  • Test session validation
  • All tests use mocked session manager

Implementation requirements:

class DocumentStateTool:
    """MCP tool for documenting session state."""
    
    def __init__(self, session_manager: SessionManager):
        self.session_manager = session_manager
    
    def get_tool_definition(self) -> Tool:
        """Return MCP tool definition."""
        pass
    
    async def execute(self, arguments: dict) -> List[TextContent]:
        """
        Document session state.
        
        Updates session's documented variables and note.
        """
        pass

Acceptance criteria:

• Tool definition matches architecture spec
• Documents variables correctly
• Updates session state
• Clear option works
• Validates session exists
• All tests use mocked components
• 100% test coverage

---

5.2.3 Build Custom Environment Tool

Path: src/mcp_forge/server/tools/build_environment.py

Purpose: Implement build_custom_environment MCP tool.

Tests to write first:

• tests/server/tools/test_build_environment.py
  • Test build with valid packages
  • Test build with template
  • Test validation errors
  • Test rate limiting
  • All tests use mocked builder

Implementation requirements:

class BuildEnvironmentTool:
    """MCP tool for building custom environments."""
    
    def __init__(
        self,
        environment_builder: EnvironmentBuilder,
        audit_logger: AuditLogger
    ):
        self.builder = environment_builder
        self.audit_logger = audit_logger
    
    def get_tool_definition(self) -> Tool:
        """Return MCP tool definition."""
        pass
    
    async def execute(self, arguments: dict) -> List[TextContent]:
        """
        Build custom environment.
        
        Validates packages, builds image, scans for vulnerabilities.
        Returns build result with image name and installed packages.
        """
        pass

Acceptance criteria:

• Tool definition matches architecture spec
• Validates all arguments
• Calls builder correctly
• Returns build result
• Logs to audit log
• All tests use mocked components
• 100% test coverage

---

5.3 MCP Server

Note: Use FastMCP for the server implementation. For documentation, see

• https://gofastmcp.com/servers/server
• https://gofastmcp.com/servers/tools
• https://gofastmcp.com/servers/resources
• https://gofastmcp.com/servers/context

5.3.1 Server Implementation

Path: src/mcp_forge/server/server.py

Purpose: Main MCP server implementation.

Tests to write first:

• tests/server/test_server.py
  • Test server initialization
  • Test tool registration
  • Test resource registration
  • Test server startup
  • Test server shutdown
  • All tests use mocked components

Implementation requirements:

from mcp.server import Server
from mcp.server.stdio import stdio_server
import asyncio

class ForgeServer:
    """MCP-Forge server implementation."""
    
    def __init__(self, config: ForgeConfig):
        self.config = config
        self.server = Server("mcp-forge")
        
        # Initialize all components
        self._init_security()
        self._init_podman()
        self._init_mcp_clients()
        self._init_backends()
        self._init_builder()
        self._init_tools()
        self._init_resources()
    
    def _init_security(self) -> None:
        """Initialize security components."""
        self.audit_logger = AuditLogger(self.config.security.audit_log)
        self.operation_validator = OperationValidator(self.config.security)
    
    def _init_podman(self) -> None:
        """Initialize Podman client and container manager."""
        self.podman_client = PodmanClient(
            socket_path=self.config.server.podman_socket,
            validator=self.operation_validator,
            audit_logger=self.audit_logger
        )
        self.container_manager = SecureContainerManager(
            podman_client=self.podman_client,
            validator=self.operation_validator,
            audit_logger=self.audit_logger
        )
    
    def _init_mcp_clients(self) -> None:
        """Initialize MCP client manager and bridge server."""
        pass
    
    def _init_backends(self) -> None:
        """Initialize execution backends."""
        pass
    
    def _init_builder(self) -> None:
        """Initialize environment builder."""
        pass
    
    def _init_tools(self) -> None:
        """Register MCP tools."""
        pass
    
    def _init_resources(self) -> None:
        """Register MCP resources."""
        pass
    
    async def run(self) -> None:
        """Run server on stdio."""
        async with stdio_server() as (read_stream, write_stream):
            await self.server.run(
                read_stream,
                write_stream,
                self.server.create_initialization_options()
            )
    
    async def shutdown(self) -> None:
        """Shutdown server and cleanup."""
        pass

Acceptance criteria:

• All components initialized correctly
• Tools registered with server
• Resources registered with server
• Server runs on stdio
• Graceful shutdown works
• All tests use mocked components
• Integration test verifies full stack
• 100% test coverage

---

Phase 6: Testing & Integration

6.1 Integration Tests

6.1.1 End-to-End Tests

Path: tests/integration/test_e2e.py

Purpose: Test complete workflows end-to-end.

Tests to write:

• Test simple code execution workflow
• Test stateful session workflow
• Test custom environment build and use
• Test MCP tool injection and usage
• Test session state documentation and retrieval
• Test concurrent executions
• Test resource cleanup

Requirements:

• Use real Podman (in CI/CD environment)
• Use real MCP servers (mocked external services)
• Test actual container creation and execution
• Verify security controls are enforced
• Test error handling and recovery

---

6.1.2 Performance Tests

Path: tests/performance/test_performance.py

Purpose: Test performance characteristics.

Tests to write:

• Test container startup time
• Test code execution overhead
• Test concurrent execution scalability
• Test session creation overhead
• Test environment build time
• Test cache effectiveness

---

6.2 Security Testing

6.2.1 Security Tests

Path: tests/security/test_security.py

Purpose: Verify security controls.

Tests to write:

• Test privileged mode is rejected
• Test forbidden operations are blocked
• Test volume mount restrictions
• Test resource limits are enforced
• Test network isolation
• Test package blocklist enforcement
• Test container escape prevention
• Test audit logging completeness

---

Phase 7: Documentation & Deployment

7.1 Documentation

7.1.1 User Documentation

• README.md with:
  • Project overview and features
  • Quick start guide
  • Basic usage examples
  • Link to full documentation
• Installation guide:
  • System requirements
  • Podman setup (rootless)
  • MCP-Forge installation
  • Configuration setup
  • Verification steps
• Configuration reference:
  • Complete configuration schema
  • All options explained
  • Environment variable reference
  • Security configuration guide
• User guide:
  • Using execute_python tool
  • Stateful vs stateless execution
  • Building custom environments
  • MCP tool integration
  • Session management
  • Best practices

7.1.2 API Documentation

• MCP Protocol documentation:
  • Resources (URIs, response formats)
  • Tools (parameters, return values)
• Python API documentation:
  • All public classes and functions
  • Usage examples
  • Type signatures

7.1.3 Security Documentation

• Security model overview
• Threat model
• Security best practices:
  • Podman configuration
  • Network isolation
  • Volume mount security
  • Package validation
  • Audit logging
  • Resource limits
• Security audit checklist
• Incident response guide

7.1.4 Operations Documentation

• Deployment guide
• Monitoring guide
• Backup and restore
• Troubleshooting guide:
  • Common errors
  • Diagnostic procedures
  • Log analysis
  • Performance tuning
• Upgrade guide

7.1.5 Developer Documentation

• Development setup
• Architecture overview (link to architecture1.md)
• Contributing guide
• Testing guide
• Code style guide
• Release process

7.2 Deployment

7.2.1 Container Images

• Build mcp-forge/python:3.11 base image
• Build mcp-forge/python:3.12 base image
• Build mcp-forge/jupyter:latest image
• Pre-install common packages in base images
• Include MCP bridge client (mcp_tools.py)
• Include startup scripts

7.2.2 Deployment Configurations

• Docker Compose configuration for production
• Kubernetes deployment manifests (optional)
• systemd service file
• Example mcp-forge.yaml configurations:
  • Development (minimal)
  • Production (full security)
  • High-concurrency setup
• Environment variable templates

7.2.3 Monitoring & Observability

• Prometheus metrics exporter (optional)
• Health check endpoints
• Log aggregation setup (e.g., Loki)
• Grafana dashboard examples
• Alert rules for:
  • High memory usage
  • Container creation failures
  • Security violations
  • Build failures

7.2.4 Operations

• Backup and restore procedures
• Session data management
• Image cleanup scripts
• Cache management (UV cache, image layers)
• Log rotation config
• Update procedures

---

Development Guidelines

Code Quality Standards

1. Type Hints: All functions must have type hints
2. Docstrings: All public functions/classes must have docstrings
3. Error Handling: Use specific exception types
4. Logging: Use structured logging (JSON)
5. Testing: Minimum 90% code coverage
6. Linting: Pass mypy, ruff, black

Test Requirements

1. Unit Tests: Test individual functions/classes in isolation
2. Integration Tests: Test component integration
3. Mocking: Use pytest-mock for external dependencies
4. Fixtures: Use pytest fixtures for common test setup
5. Parametrization: Use pytest.mark.parametrize for multiple cases

Security Requirements

1. No eval/exec: Never use eval() or exec()
2. Input Validation: Validate all user inputs
3. Resource Limits: Always enforce resource limits
4. Audit Logging: Log all security-relevant operations
5. Least Privilege: Run with minimum required privileges

---

Progress Tracking

Use this checklist to track completion:

• Phase 1: Foundation & Core Infrastructure
  • 1.1 Configuration Management
  • 1.2 Security & Validation Core
  • 1.3 Podman Integration Core
• Phase 2: Execution Backends
  • 2.1 Simple Backend
  • 2.2 Jupyter Backend
• Phase 3: Custom Environment Builder
  • 3.1 Package Management
  • 3.2 Environment Building
• Phase 4: MCP Tool Integration
  • 4.1 MCP Client Management
  • 4.2 MCP Tool Injection
• Phase 5: MCP Server Implementation
  • 5.1 MCP Resources
  • 5.2 MCP Tools
  • 5.3 MCP Server
• Phase 6: Testing & Integration
  • 6.1 Integration Tests
  • 6.2 Security Testing
• Phase 7: Documentation & Deployment

---

Notes for Coding Agent

1. Always write tests first - No implementation without tests
2. Follow the architecture - Don't take shortcuts or simplify (see architecture1.md)
3. Use mocking extensively - Don't require real Podman/MCP for unit tests
4. Validate everything - Security is paramount
5. Document as you go - Docstrings and comments are required
6. Test coverage matters - Aim for 100% on critical security code
7. Error messages are UX - Make them clear and actionable
8. Log appropriately - Audit log for security, debug log for troubleshooting
9. No hardcoded values - Everything should be configurable
10. Think about concurrency - Use locks where needed
11. Security first - pip install is NEVER allowed during execution; only through build_custom_environment
12. UV-based builds - All package installation must use UV for speed and security
13. Allowlist enforcement - Image allowlist, volume patterns, operation restrictions are non-negotiable
14. Rootless Podman - All container operations must work in rootless mode
15. Audit everything - All security-relevant operations must be logged

Implementation Roadmap Alignment

This TODO follows the Implementation Roadmap from architecture1.md:

• Phase 1 (MVP): Sections 1.1-1.3, 2.1, 4.2, 5.3 (basic execution)
• Phase 2 (Stateful): Section 2.2, additional resources in 5.1
• Phase 3 (Custom Environments): Section 3.1-3.2, tool in 5.2.3
• Phase 4 (Advanced Features): Enhanced monitoring, templates, rate limiting
• Phase 5 (Production): Phase 6-7 testing, deployment, documentation

Start with Phase 1 components before moving to subsequent phases.