Skip to content

Architecture

Jump to bottom
github-actions[bot] edited this page Mar 22, 2026 · 20 revisions

Architecture

System Overview

AI Agent (Claude, Cursor, Copilot, etc.)
       │
       │ MCP Protocol (stdio or SSE)
       ▼
┌─────────────────────────────────┐
│   MCP Server (FastMCP)           │
│   ├─ 20+ built-in tools          │
│   ├─ Custom tools (from YAML)    │
│   ├─ Session manager             │
│   ├─ Security validator          │
│   ├─ Job executor                │
│   └─ Result formatter            │
└──────────┬──────────────────────┘
           │
┌──────────▼──────────────────────┐
│   Job Executor                    │
│   ├─ Hybrid sync/async execution  │
│   ├─ Timeout-based promotion      │
│   ├─ Job context injection        │
│   ├─ Stdout/stderr capture        │
│   └─ Result building              │
└──────────┬──────────────────────┘
           │
┌──────────▼──────────────────────┐
│   Engine Pool Manager             │
│   ├─ Elastic scaling (min→max)    │
│   ├─ Health checks                │
│   ├─ Proactive warmup             │
│   ├─ Idle scale-down              │
│   └─ Request queueing             │
└──────────┬──────────────────────┘
           │
┌──────────▼──────────────────────┐
│   MATLAB Engines (2020b+)         │
│   Engine 1 │ Engine 2 │ ... │ N   │
└───────────────────────────────────┘

Component Details

MCP Server (server.py)

The entry point that orchestrates all components. Built on FastMCP framework. Responsibilities:

  • Register all built-in tools (code execution, file I/O, job management, discovery, admin)
  • Load and register custom tools from YAML configuration
  • Manage server lifecycle: startup (pool initialization, session creation), shutdown (graceful drain), and background tasks
  • Route incoming tool calls to implementation modules
  • Handle context from different transports (stdio session IDs vs SSE session IDs)
  • Wire up monitoring collector if enabled
  • Manage lifespan events: startup tasks (directories, engine pool, health checks, cleanup scheduler), shutdown tasks (stop collectors, drain jobs, stop pool)

Engine Pool Manager (pool/manager.py)

Manages a pool of MATLAB engine instances with dynamic scaling:

  • Min/Max Engines: Starts with min_engines (default 2), scales up to max_engines (default 8) under load
  • Acquire/Release: acquire() returns immediately if engines available, otherwise scales up if under limit, or waits if at capacity. release() resets workspace and returns engine to available queue
  • Elastic Scaling: When all engines busy and pool below max, new engines are started on-demand
  • Health Checks: Periodic 1+1 eval on idle engines. Dead engines are replaced. Idle engines beyond min_engines are stopped if idle longer than scale_down_idle_timeout (default 15 min)
  • Request Queueing: Async queue (_available) holds available engines. Blocked acquirers wait on queue

Engine Wrapper (pool/engine.py)

Wraps a single matlab.engine instance with state tracking and lifecycle:

  • Lazy Import: Imports matlab.engine only when needed (allows test mocking)
  • Lifecycle States: STOPPED → STARTING → IDLE ↔ BUSY
  • Startup: Starts engine, applies default paths from config, runs startup commands
  • Execute: Runs code with optional background execution, stdout/stderr capture
  • Workspace Reset: Runs clear all, clear global, clear functions, fclose all, restoredefaultpath, re-adds configured paths, re-runs startup commands
  • Health Check: Trivial 1+1 eval to confirm responsiveness
  • Idle Tracking: Records idle time for scale-down decisions

Job Executor (jobs/executor.py)

Orchestrates the full execution lifecycle with hybrid sync/async support:

  1. Job Creation: Creates job record in tracker with session/code
  2. Engine Acquisition: Acquires engine from pool
  3. Context Injection: Injects job ID and temp directory into MATLAB workspace via __mcp_job_id__, __mcp_temp_dir__
  4. Code Execution:
    • Security validation (checked before execution)
    • Starts background execution with engine.execute(code, background=True)
    • Captures stdout/stderr to StringIO buffers
  5. Sync/Async Promotion:
    • Waits up to sync_timeout seconds (default 30s)
    • If completes: returns result inline with status="completed"
    • If timeout: promotes to async background task (status="pending"), continues monitoring in background
  6. Result Building: Queries workspace, formats output, builds response dict
  7. Error Handling: Marks job failed on exceptions, returns error result
  8. Engine Release: Returns engine to pool for reuse

Job Tracker (jobs/tracker.py)

In-memory store for job metadata and status:

  • Create/get/list jobs
  • Track job state: PENDING, RUNNING, COMPLETED, FAILED, CANCELLED
  • Store execution results, stdout/stderr, error details
  • Prune completed jobs older than job_retention_seconds (default 3600s)
  • Thread-safe with asyncio locks for concurrent access

Session Manager (session/manager.py)

Per-user session isolation with temporary directories:

  • Session Creation: Generates unique session ID (UUID or explicit), creates temp directory
  • Session Storage: Maintains registry of active sessions with lifecycle tracking
  • Cleanup: Removes sessions idle longer than session_timeout (default 3600s), skips sessions with active jobs
  • Default Session: For stdio transport, uses fixed "default" session ID
  • Per-Session Directories: Each session owns temp directory for uploads, data files, generated outputs
  • Workspace Isolation: Can reset workspace between sessions if workspace_isolation=true

Security Validator (security/validator.py)

Pre-execution security checks:

  • Function Blocklist: Scans code for dangerous functions: system, unix, dos, !, eval, feval, evalc, evalin, assignin, perl, python. Strips string literals and comments first to avoid false positives
  • Filename Sanitization: Prevents path traversal in upload filenames (converts to basename)
  • Upload Size Limits: Enforces max_upload_size_mb (default 100 MB)
  • Metrics: Records security events to collector if monitoring enabled

Result Formatter (output/formatter.py)

Structures execution results for MCP responses:

  • Text Formatting: Truncates output longer than max_inline_text_length (default 10,000 chars), saves excess to file
  • Variable Formatting: Summarizes workspace variables (name, type, size, value). Large arrays represented by type/size only (threshold configurable)
  • Success Responses: Builds dict with job_id, output, variables, figures, files, warnings, execution_time
  • Error Responses: Includes error_type, message, matlab_id (for MATLAB exceptions)

Plotly Converter (output/plotly_convert.py, output/plotly_style_mapper.py + matlab_helpers/mcp_extract_props.m)

Converts MATLAB figures to interactive Plotly JSON:

  1. MATLAB-side (mcp_extract_props.m): Extracts raw figure properties (axes, line objects, scatter plots, bar charts, histograms, surfaces, images, text)
  2. Python-side (plotly_style_mapper.py): Maps MATLAB styles to Plotly equivalents:
    • Line styles (solid, dashed, dotted)
    • Markers and marker sizes
    • Colormaps (viridis, jet, etc.)
    • Fonts and font sizes
    • Colors (named, RGB)
    • WebGL mode for large datasets (10,000+ points)
  3. Python-side (plotly_convert.py): Loads saved JSON, builds Plotly figure, generates static PNG, optional thumbnail
  4. Result: Includes Plotly JSON (interactive), PNG (static), thumbnail URL

Monitoring (monitoring/collector.py, monitoring/store.py)

Observability subsystem (when monitoring.enabled=true):

  • Metrics Collector: Records events from all components (engine_scale_up, job_completed, job_failed, session_created, security_violation)
  • Metrics Store: In-memory or persistent storage of metrics time-series
  • Lifespan Integration: Started with server, metrics available via admin endpoints
  • Per-Component Integration: Pool, executor, sessions, security validator all call collector.record_event() if available

Data Flow

Sync Execution

Agent → execute_code("x = magic(3)")
  → Security check (OK)
  → Create job in tracker
  → Acquire engine from pool
  → Inject job context (__mcp_job_id__, __mcp_temp_dir__)
  → Engine.execute("x = magic(3)", background=True) with stdout/stderr capture
  → Wait up to 30s (sync_timeout)
  → Complete in <30s
  → Query workspace variables
  → Format output (text, variables)
  → Mark job completed in tracker
  → Release engine to pool
  → Return {status: "completed", job_id: "...", output: {...}, variables: [...]}

Async Promotion

Agent → execute_code("long_simulation()")
  → Security check (OK)
  → Create job in tracker
  → Acquire engine
  → Inject context
  → Engine.execute (background=True) with stdout/stderr
  → Wait 30s timeout exceeded
  → Spawn background task _wait_for_completion()
  → Return {status: "pending", job_id: "abc123"}
  → Agent polls get_job_status("abc123") → {status: "running", progress: 45%, elapsed: 5.2s}
  → Agent polls get_job_status("abc123") → {status: "running", progress: 90%, elapsed: 12.1s}
  → Background task detects future.result() complete
  → Query workspace, format output, mark completed
  → Release engine
  → Agent calls get_job_result("abc123") → {status: "completed", output: {...}, variables: [...]}

Health Check & Scale-Down

Background task runs_health_checks() every 30s:
  → For each idle engine:
      → Run health check (1+1 eval)
      → If fails: stop engine, remove from pool, start replacement
      → If idle > 15min AND pool > min_engines: stop and remove
  → If utilization > 80%: start proactive warmup engine

Transport Modes

stdio (Default)

  • Single agent session per process invocation
  • Communication via stdin/stdout
  • Uses "default" session ID (fixed)
  • Simplest setup, no network required
  • Suitable for single-user development

SSE (Server-Sent Events)

  • Multiple concurrent agents/sessions
  • HTTP-based, stateful connections
  • Session ID from MCP context (unique per client)
  • Supports remote connections
  • Production: Must run behind reverse proxy with auth (require_proxy_auth: true)
  • Example: Agent 1 creates session "uuid-abc", temp dir created, workspace isolated
  • Example: Agent 2 creates session "uuid-def", separate temp dir, separate workspace

Configuration Reference

Key config sections (from config.py):

  • pool: min_engines, max_engines, scale_down_idle_timeout
  • execution: sync_timeout, temp_dir, max_upload_size_mb
  • sessions: session_timeout, job_retention_seconds, max_sessions
  • output: max_inline_text_length, large_result_threshold
  • security: max_upload_size_mb, require_proxy_auth, blocked functions list
  • monitoring: enabled, store_type (memory/file), metrics_interval
  • workspace: default_paths, startup_commands, isolation mode

Clone this wiki locally