Skip to content

API Reference

Jump to bottom
github-actions[bot] edited this page Apr 3, 2026 · 10 revisions

API Reference

Overview

The MATLAB MCP Server provides a comprehensive API for executing MATLAB code, managing sessions, discovering toolboxes, and monitoring server health via the Model Context Protocol (MCP). This reference documents all public APIs, classes, and functions organized by module.

Architecture

classDiagram
    class FastMCPServer {
        +create_server() MatlabMCPServer
        +run(transport, host, port)
    }
    
    class MatlabMCPServer {
        +engine_pool EnginePoolManager
        +job_tracker JobTracker
        +job_executor JobExecutor
        +session_manager SessionManager
        +security_validator SecurityValidator
        +metrics_collector MetricsCollector
        +result_formatter ResultFormatter
    }
    
    class EnginePoolManager {
        +acquire_engine() MatlabEngineWrapper
        +release_engine(engine)
        +get_status() dict
        +scale_up()
        +scale_down()
    }
    
    class MatlabEngineWrapper {
        +state EngineState
        +eval(code) str
        +start()
        +stop()
        +is_alive() bool
    }
    
    class JobExecutor {
        +execute(code, session_id) dict
        +execute_async(code, session_id) Job
    }
    
    class JobTracker {
        +create_job(code, session_id) Job
        +get_job(job_id) Job
        +list_jobs(session_id) list
        +mark_completed(job_id, result)
    }
    
    class SessionManager {
        +create_session() Session
        +get_session(session_id) Session
        +destroy_session(session_id)
    }
    
    class SecurityValidator {
        +check_code(code)
        +sanitize_filename(filename) str
    }
    
    class ResultFormatter {
        +format_text(text) str
        +format_variables(variables) dict
        +build_success_response(output, variables, figures) dict
    }
    
    class MetricsCollector {
        +record_job_completed(job_id, duration)
        +record_job_failed(job_id)
        +get_metrics() dict
    }
    
    FastMCPServer --> MatlabMCPServer
    MatlabMCPServer --> EnginePoolManager
    MatlabMCPServer --> JobExecutor
    MatlabMCPServer --> JobTracker
    MatlabMCPServer --> SessionManager
    MatlabMCPServer --> SecurityValidator
    MatlabMCPServer --> ResultFormatter
    MatlabMCPServer --> MetricsCollector
    JobExecutor --> EnginePoolManager
    JobExecutor --> JobTracker
Loading

Server Module (server.py)

create_server(config: AppConfig) → FastMCP

Creates and initializes the MCP server instance with all subsystems.

Parameters:

  • config (AppConfig): Application configuration object

Returns:

  • FastMCP: Configured FastMCP server ready for transport initialization

Raises:

  • ConfigError: If configuration is invalid

Example:

from matlab_mcp.config import AppConfig
from matlab_mcp.server import create_server

config = AppConfig.from_file("config.yaml")
mcp = create_server(config)
await mcp.run(transport="http", host="127.0.0.1", port=8765)

main()

CLI entry point for running the server with command-line arguments.

Arguments:

  • --transport {stdio,sse,http,streamablehttp}: Transport protocol (default: stdio)
  • --config CONFIG_PATH: YAML configuration file path
  • --inspect: Run in inspect mode (no MATLAB engines, skips pool initialization)
  • --generate-token: Generate and print a new authentication token
  • --host HOST: Bind address (default: 127.0.0.1)
  • --port PORT: Bind port (default: 8765)

Returns:

  • int: Exit code (0 = success)

Example:

python -m matlab_mcp --transport streamablehttp --config config.yaml --host 127.0.0.1 --port 8765
python -m matlab_mcp --inspect  # For testing without MATLAB
python -m matlab_mcp --generate-token  # Print 64-char hex token

Configuration Module (config.py)

AppConfig

Class: pydantic.BaseModel

Root configuration object containing all subsystem settings.

Fields:

  • server (ServerConfig): Server transport and binding
  • pool (PoolConfig): Engine pool sizing and health checks
  • execution (ExecutionConfig): Job execution timeouts
  • workspace (WorkspaceConfig): MATLAB workspace defaults
  • security (SecurityConfig): Code validation and file restrictions
  • hitl (HITLConfig): Human-in-the-loop approval gates
  • tools (ToolsConfig): Toolbox discovery and custom tools
  • output (OutputConfig): Result formatting (Plotly, thumbnails)
  • sessions (SessionsConfig): Session management (timeouts, limits)
  • monitoring (MonitoringConfig): Metrics collection and storage

Methods:

from_file(path: str) → AppConfig

Load configuration from YAML file with environment variable overrides.

Parameters:

  • path (str): Path to config.yaml

Returns:

  • AppConfig: Loaded configuration

Raises:

  • FileNotFoundError: If YAML file not found
  • ValueError: If YAML is invalid or configuration values fail validation

Example:

config = AppConfig.from_file("config.yaml")

from_defaults() → AppConfig

Create configuration with all defaults (no file loading).

Returns:

  • AppConfig: Default configuration

Example:

config = AppConfig.from_defaults()

ServerConfig

Class: pydantic.BaseModel

Server transport and binding configuration.

Fields:

  • host (str): Bind address (default: "127.0.0.1")
  • port (int): Bind port (default: 8765)
  • transport (Literal["stdio", "sse", "http", "streamablehttp"]): Protocol (default: "stdio")
  • log_level (str): Logging level (default: "INFO")
  • stateless_http (bool): For streamablehttp: treat each request as stateless (default: False)

Environment Variable Overrides:

  • MATLAB_MCP_HOST
  • MATLAB_MCP_PORT
  • MATLAB_MCP_TRANSPORT
  • MATLAB_MCP_LOG_LEVEL
  • MATLAB_MCP_STATELESS_HTTP

PoolConfig

Class: pydantic.BaseModel

MATLAB engine pool sizing and lifecycle.

Fields:

  • min_engines (int): Minimum engines to keep warm (default: 2)
  • max_engines (int): Maximum engines to create (default: 10)
  • startup_timeout_s (int): Timeout for engine startup (default: 30)
  • health_check_interval_s (int): Interval between health checks (default: 60)
  • health_check_timeout_s (int): Timeout per health check (default: 5)
  • scale_down_threshold_s (int): Idle time before scale-down (default: 300)

Environment Variable Overrides:

  • MATLAB_MCP_POOL_MIN_ENGINES
  • MATLAB_MCP_POOL_MAX_ENGINES

ExecutionConfig

Class: pydantic.BaseModel

Job execution and timeout settings.

Fields:

  • sync_timeout_s (float): Max duration to block on code execution (default: 10.0)
  • job_timeout_s (float): Max duration for async jobs (default: 3600.0)
  • temp_dir (str): Base temp directory for session workspaces (default: auto via tempfile.gettempdir())

SecurityConfig

Class: pydantic.BaseModel

Code safety validation and file restrictions.

Fields:

  • blocked_functions_enabled (bool): Enable function blocklist (default: True)
  • blocked_functions (List[str]): MATLAB functions to deny (default: ["system", "unix", "dos", "eval", "feval"])
  • max_upload_size_mb (int): Maximum file upload size (default: 100)
  • require_proxy_auth (bool): Require Proxy-Authorization header for SSE (default: False)

Environment Variable Overrides:

  • MATLAB_MCP_SECURITY_BLOCKED_FUNCTIONS (comma-separated list)

HITLConfig

Class: pydantic.BaseModel

Human-in-the-loop approval gate configuration.

Fields:

  • enabled (bool): Master switch for HITL prompts (default: False)
  • protected_functions (List[str]): MATLAB functions requiring approval (default: [])
  • protect_file_ops (bool): Gate file uploads and deletes (default: False)
  • all_execute (bool): Require approval for ALL code execution (default: False)

Environment Variable Overrides:

  • MATLAB_MCP_HITL_ENABLED
  • MATLAB_MCP_HITL_PROTECTED_FUNCTIONS (comma-separated)
  • MATLAB_MCP_HITL_PROTECT_FILE_OPS
  • MATLAB_MCP_HITL_ALL_EXECUTE

Example:

# Enable HITL for specific functions
config.hitl.enabled = True
config.hitl.protected_functions = ["system", "eval"]

# Or via environment:
# MATLAB_MCP_HITL_ENABLED=true
# MATLAB_MCP_HITL_PROTECTED_FUNCTIONS=system,eval

MonitoringConfig

Class: pydantic.BaseModel

Metrics collection and storage.

Fields:

  • enabled (bool): Enable metrics collection (default: True)
  • db_path (str): SQLite database path (default: "monitoring/metrics.db")
  • sample_interval_s (int): Sampling interval (default: 60)
  • retention_days (int): How long to keep metrics (default: 7)
  • dashboard_port (int): HTTP port for dashboard (default: 8766)

Pool Module (pool/)

EnginePoolManager

Class: Manages MATLAB engine lifecycle and pooling.

Constructor:

EnginePoolManager(
    config: PoolConfig,
    startup_callback: Optional[Callable] = None,
    event_collector: Optional[MetricsCollector] = None
)

Parameters:

  • config (PoolConfig): Pool configuration
  • startup_callback (Optional[Callable]): Called after each engine starts
  • event_collector (Optional[MetricsCollector]): For recording events

Methods:

async initialize() → None

Start minimum engines and perform initial health checks.

Raises:

  • RuntimeError: If minimum engines cannot start

Example:

pool = EnginePoolManager(config.pool)
await pool.initialize()

async acquire_engine() → MatlabEngineWrapper

Acquire an available engine, blocking until one is free.

Returns:

  • MatlabEngineWrapper: Available MATLAB engine

Raises:

  • TimeoutError: If no engine available after timeout

Example:

engine = await pool.acquire_engine()
result = engine.eval("1+1")
await pool.release_engine(engine)

async release_engine(engine: MatlabEngineWrapper) → None

Return an engine to the pool.

Parameters:

  • engine (MatlabEngineWrapper): Engine to release

def get_status() → dict

Get current pool status.

Returns:

  • dict: {"total": int, "available": int, "busy": int, "max": int}

Example:

status = pool.get_status()
print(f"Pool: {status['busy']}/{status['total']} engines busy")

async shutdown() → None

Stop all engines and clean up resources.

Example:

await pool.shutdown()

MatlabEngineWrapper

Class: Wraps a single MATLAB engine with lifecycle management.

Constructor:

MatlabEngineWrapper(
    config: PoolConfig,
    startup_callback: Optional[Callable] = None
)

Properties:

state: EngineState

Current engine state (STOPPED, STARTING, IDLE, BUSY).

async start() → None

Start the MATLAB engine.

Raises:

  • RuntimeError: If MATLAB Engine API unavailable

async stop() → None

Gracefully stop the engine.

def is_alive() → bool

Check if engine is running.

Returns:

  • bool: True if IDLE or BUSY

async eval(code: str, nargout: int = 1, background: bool = False) → Any

Execute MATLAB code synchronously or asynchronously.

Parameters:

  • code (str): MATLAB code to execute
  • nargout (int): Expected output arguments (default: 1)
  • background (bool): Run asynchronously (default: False)

Returns:

  • For synchronous: Result value or tuple of values
  • For asynchronous: Future object (call .result() to wait)

Raises:

  • MatlabExecutionError: If MATLAB code raises an error

Example:

# Synchronous
result = engine.eval("1 + 2")  # Returns 3

# Asynchronous
future = engine.eval("pause(10); disp('done')", background=True)
result = future.result()  # Blocks until ready

async reset_workspace() → None

Clear all workspace variables and restore defaults.

Jobs Module (jobs/)

JobExecutor

Class: Orchestrates MATLAB code execution with job tracking.

Constructor:

JobExecutor(
    pool: EnginePoolManager,
    tracker: JobTracker,
    config: ExecutionConfig,
    security_validator: SecurityValidator,
    result_formatter: ResultFormatter,
    session_manager: SessionManager,
    event_collector: Optional[MetricsCollector] = None
)

Methods:

async execute(code: str, session_id: str, job_id: Optional[str] = None) → dict

Execute MATLAB code (synchronously or promoting to async).

Parameters:

  • code (str): MATLAB code to execute
  • session_id (str): Session ID for workspace isolation
  • job_id (Optional[str]): Job ID (auto-generated if None)

Returns:

  • dict: Success response with keys:
    • output (str): Captured stdout/stderr
    • variables (dict): Workspace variables
    • figures (list): Generated figure files
    • execution_time (float): Duration in seconds
    • status (str): "completed" or "promoted_to_async"
    • For async: job_id (str): ID for later retrieval

Raises:

  • BlockedFunctionError: If code violates security policy
  • TimeoutError: If execution exceeds timeout

Example:

executor = JobExecutor(pool, tracker, config, ...)
result = await executor.execute("x = randn(1000, 1000); s = svd(x);", "session-1")
if result["status"] == "completed":
    print(f"Done in {result['execution_time']}s")
    print(f"Variables: {result['variables']}")
elif result["status"] == "promoted_to_async":
    print(f"Job {result['job_id']} running in background")

JobTracker

Class: Thread-safe registry of all jobs (active and completed).

Constructor:

JobTracker(
    event_collector: Optional[MetricsCollector] = None
)

Methods:

def create_job(code: str, session_id: str) → Job

Create and register a new job.

Parameters:

  • code (str): MATLAB code to execute
  • session_id (str): Session owner

Returns:

  • Job: New job with auto-generated ID

Example:

job = tracker.create_job("disp('hello')", "session-1")
print(f"Job ID: {job.job_id}")

def get_job(job_id: str) → Optional[Job]

Retrieve a job by ID.

Parameters:

  • job_id (str): Job identifier

Returns:

  • Optional[Job]: Job if found, None otherwise

def list_jobs(session_id: str) → list[Job]

List all jobs for a session.

Parameters:

  • session_id (str): Session identifier

Returns:

  • list[Job]: Jobs in creation order

def mark_completed(job_id: str, result: Any) → None

Mark a job as complete.

Parameters:

  • job_id (str): Job identifier
  • result (Any): Result data

Job

Class: Dataclass representing a single MATLAB execution job.

Fields:

  • job_id (str): Unique identifier
  • session_id (str): Session owner
  • code (str): MATLAB code
  • status (JobStatus): Current status (PENDING, RUNNING, COMPLETED, FAILED, CANCELLED)
  • created_at (datetime): Creation timestamp
  • started_at (Optional[datetime]): Start timestamp
  • completed_at (Optional[datetime]): Completion timestamp
  • result (Any): Execution result
  • error (Optional[str]): Error message if failed
  • background_future (Optional[Future]): For async execution

Properties:

elapsed_seconds: float

Time elapsed (or time taken if complete).

Example:

job = tracker.get_job("job-123")
print(f"Status: {job.status}")
print(f"Elapsed: {job.elapsed_seconds:.2f}s")
if job.status == JobStatus.COMPLETED:
    print(f"Result: {job.result}")

Session Module (session/)

SessionManager

Class: Manages user sessions with workspace isolation.

Constructor:

SessionManager(
    config: AppConfig,
    event_collector: Optional[MetricsCollector] = None
)

Methods:

def create_session() → Session

Create a new session with temp directory.

Returns:

  • Session: New session with unique ID

Example:

session = session_manager.create_session()
print(f"Session: {session.session_id}")
print(f"Temp dir: {session.temp_dir}")

def get_session(session_id: str) → Optional[Session]

Retrieve a session by ID.

Parameters:

  • session_id (str): Session identifier

Returns:

  • Optional[Session]: Session if found

async destroy_session(session_id: str) → None

Destroy a session and clean up temp directory.

Parameters:

  • session_id (str): Session identifier

def cleanup_expired() → None

Remove idle sessions exceeding timeout.

Session

Class: Dataclass representing a user session.

Fields:

  • session_id (str): Unique identifier
  • temp_dir (str): Temporary workspace directory
  • created_at (datetime): Creation timestamp
  • last_activity_at (datetime): Last activity timestamp

Methods:

def touch() → None

Update last activity timestamp to now.

def idle_seconds() → float

Get idle duration in seconds.

Example:

session = session_manager.get_session("session-1")
session.touch()
print(f"Idle: {session.idle_seconds():.1f}s")

Security Module (security/)

SecurityValidator

Class: Validates MATLAB code and sanitizes filenames.

Constructor:

SecurityValidator(
    config: SecurityConfig,
    event_collector: Optional[MetricsCollector] = None
)

Methods:

def check_code(code: str) → None

Validate MATLAB code against security policy.

Parameters:

  • code (str): MATLAB code to validate

Raises:

  • BlockedFunctionError: If code violates blocklist

Example:

validator = SecurityValidator(config.security)
try:
    validator.check_code("result = system('ls');")
except BlockedFunctionError as e:
    print(f"Blocked: {e}")

def sanitize_filename(filename: str) → str

Sanitize filename to prevent path traversal and control characters.

Parameters:

  • filename (str): Filename to sanitize

Returns:

  • str: Sanitized filename (alphanumeric, dots, dashes, underscores only)

Example:

safe = validator.sanitize_filename("../../etc/passwd.txt")
# Returns: "etcpasswd.txt"

Tools Module (tools/)

Core Execution Tools

execute_code_impl(code: str, ctx: Context, state: MatlabMCPServer, config: AppConfig) → dict

Execute MATLAB code with optional HITL approval gating.

Parameters:

  • code (str): MATLAB code to execute
  • ctx (Context): MCP request context
  • state (MatlabMCPServer): Server state
  • config (AppConfig): Application config

Returns:

  • dict: Execution result with output, variables, figures, timing

MCP Tool Definition:

{
  "name": "execute_code",
  "description": "Execute MATLAB code and return results",
  "inputSchema": {
    "type": "object",
    "properties": {
      "code": {
        "type": "string",
        "description": "MATLAB code to execute"
      }
    },
    "required": ["code"]
  }
}

Example Usage (Agent):

{
  "method": "tools/call",
  "params": {
    "name": "execute_code",
    "arguments": {"code": "x = 1:10; plot(x, x.^2);"}
  }
}

check_code_impl(code: str, state: MatlabMCPServer) → dict

Lint MATLAB code using checkcode().

Parameters:

  • code (str): MATLAB code to lint
  • state (MatlabMCPServer): Server state

Returns:

  • dict: Linting results with issues (line, column, message, id)

MCP Tool Definition:

{
  "name": "check_code",
  "description": "Lint MATLAB code for issues",
  "inputSchema": {
    "type": "object",
    "properties": {
      "code": {"type": "string"}
    },
    "required": ["code"]
  }
}

get_workspace_impl(state: MatlabMCPServer, session_id: str) → dict

List workspace variables.

Parameters:

  • state (MatlabMCPServer): Server state
  • session_id (str): Session ID

Returns:

  • dict: Variables with types and sizes

MCP Tool Definition:

{
  "name": "get_workspace",
  "description": "Get all workspace variables",
  "inputSchema": {"type": "object"}
}

File Operations

upload_data_impl(data: str, filename: str, ctx: Context, state: MatlabMCPServer, config: AppConfig, session_id: str) → dict

Upload a file (base64-encoded) to session temp directory.

Parameters:

  • data (str): Base64-encoded file contents
  • filename (str): Desired filename
  • ctx (Context): MCP context for HITL approval
  • state (MatlabMCPServer): Server state
  • config (AppConfig): Application config
  • session_id (str): Session ID

Returns:

  • dict: {"file_path": str, "size_bytes": int}

Raises:

  • ValueError: If file too large
  • BlockedFunctionError: If filename contains traversal attempts

delete_file_impl(filename: str, ctx: Context, state: MatlabMCPServer, config: AppConfig, session_id: str) → dict

Delete a file from session temp directory.

Parameters:

  • filename (str): Filename to delete
  • ctx (Context): MCP context for HITL approval
  • state (MatlabMCPServer): Server state
  • config (AppConfig): Application config
  • session_id (str): Session ID

Returns:

  • dict: {"deleted": True, "path": str}

list_files_impl(state: MatlabMCPServer, session_id: str) → dict

List files in session temp directory.

Parameters:

  • state (MatlabMCPServer): Server state
  • session_id (str): Session ID

Returns:

  • dict: {"files": [{"name": str, "size_bytes": int}, ...]}

read_script_impl(filename: str, state: MatlabMCPServer, session_id: str) → dict

Read a .m file from session.

Parameters:

  • filename (str): Script filename
  • state (MatlabMCPServer): Server state
  • session_id (str): Session ID

Returns:

  • dict: {"content": str, "lines": int}

read_data_impl(filename: str, state: MatlabMCPServer, session_id: str) → dict

Read data file (.mat, .csv, .json) from session.

Parameters:

  • filename (str): Data filename
  • state (MatlabMCPServer): Server state
  • session_id (str): Session ID

Returns:

  • dict: Parsed data contents

read_image_impl(filename: str, state: MatlabMCPServer, session_id: str) → Image

Read image file (.png, .jpg) from session.

Parameters:

  • filename (str): Image filename
  • state (MatlabMCPServer): Server state
  • session_id (str): Session ID

Returns:

  • Image: MCP Image object for display

Job Management

get_job_status_impl(job_id: str, state: MatlabMCPServer) → dict

Get status of a job.

Parameters:

  • job_id (str): Job identifier
  • state (MatlabMCPServer): Server state

Returns:

  • dict: Job status with progress if available

get_job_result_impl(job_id: str, state: MatlabMCPServer) → dict

Retrieve result of a completed job.

Parameters:

  • job_id (str): Job identifier
  • state (MatlabMCPServer): Server state

Returns:

  • dict: Execution result with output, variables, figures

cancel_job_impl(job_id: str, state: MatlabMCPServer) → dict

Cancel a pending or running job.

Parameters:

  • job_id (str): Job identifier
  • state (MatlabMCPServer): Server state

Returns:

  • dict: {"cancelled": True, "job_id": str}

list_jobs_impl(state: MatlabMCPServer, session_id: str) → dict

List jobs in a session.

Parameters:

  • state (MatlabMCPServer): Server state
  • session_id (str): Session ID

Returns:

  • dict: {"jobs": [{"job_id": str, "status": str, "created_at": str}, ...]}

Discovery

list_toolboxes_impl(state: MatlabMCPServer) → dict

List installed MATLAB toolboxes.

Parameters:

  • state (MatlabMCPServer): Server state

Returns:

  • dict: {"toolboxes": [{"name": str, "version": str}, ...]}

list_functions_impl(toolbox: str, state: MatlabMCPServer) → dict

List functions in a toolbox.

Parameters:

  • toolbox (str): Toolbox name
  • state (MatlabMCPServer): Server state

Returns:

  • dict: {"functions": [str, ...]}

get_help_impl(function: str, state: MatlabMCPServer) → dict

Get help text for a function.

Parameters:

  • function (str): Function name
  • state (MatlabMCPServer): Server state

Returns:

  • dict: {"help": str, "function": str}

Monitoring

get_server_metrics_impl(state: MatlabMCPServer) → dict

Get current server metrics snapshot.

Parameters:

  • state (MatlabMCPServer): Server state

Returns:

  • dict: Current metrics (pool utilization, job count, error rate, etc.)

get_server_health_impl(state: MatlabMCPServer) → dict

Get server health status.

Parameters:

  • state (MatlabMCPServer): Server state

Returns:

  • dict: Health status (healthy/degraded/unhealthy) with details

get_error_log_impl(limit: int, state: MatlabMCPServer) → dict

Get recent error events.

Parameters:

  • limit (int): Maximum errors to return
  • state (MatlabMCPServer): Server state

Returns:

  • dict: {"errors": [{"timestamp": str, "type": str, "message": str}, ...]}

Admin

get_pool_status_impl(state: MatlabMCPServer) → dict

Get engine pool status.

Parameters:

  • state (MatlabMCPServer): Server state

Returns:

  • dict: {"total": int, "available": int, "busy": int, "max": int}

Output Module (output/)

ResultFormatter

Class: Structures MATLAB execution results for MCP responses.

Constructor:

ResultFormatter(
    config: OutputConfig,
    event_collector: Optional[MetricsCollector] = None
)

Methods:

def format_text(text: str) → str

Format and optionally truncate captured text output.

Parameters:

  • text (str): Raw stdout/stderr

Returns:

  • str: Formatted text (truncated if exceeds limit)

def format_variables(variables: dict) → dict

Format workspace variables for display.

Parameters:

  • variables (dict): Raw workspace variables

Returns:

  • dict: Formatted variables with type/size info, large objects masked

def build_success_response(output: str, variables: dict, figures: list, timing: float) → dict

Build complete success response.

Parameters:

  • output (str): Captured text
  • variables (dict): Workspace variables
  • figures (list): Generated figure file paths
  • timing (float): Execution duration in seconds

Returns:

  • dict: Structured response with all components

Example:

formatter = ResultFormatter(

Clone this wiki locally