vllm_mlx.mcp.security

MCP security module for command validation and sandboxing.

View the complete module source at #L1-L852.

API details

Each callable below includes its exact signature, type annotations, inputs, defaults, return contract, documented exceptions, implementation source, and parsed docstring sections when the source provides them.

vllm_mlx.mcp.security

MCP security module for command validation and sandboxing.

This module provides security controls to prevent command injection and other attacks via MCP server configurations.

vllm_mlx.mcp.security.logger module-attribute

logger = logging.getLogger(__name__)

vllm_mlx.mcp.security.ALLOW_UNSAFE_ENV_VAR module-attribute

ALLOW_UNSAFE_ENV_VAR = 'VLLM_MCP_ALLOW_UNSAFE'

vllm_mlx.mcp.security.ALLOWED_COMMANDS module-attribute

ALLOWED_COMMANDS: Set[str] = {'npx', 'npm', 'node', 'uvx', 'uv', 'python', 'python3', 'pip', 'pipx', 'mcp-server-filesystem', 'mcp-server-sqlite', 'mcp-server-postgres', 'mcp-server-github', 'mcp-server-slack', 'mcp-server-memory', 'mcp-server-puppeteer', 'mcp-server-brave-search', 'mcp-server-google-maps', 'mcp-server-fetch', 'docker'}

vllm_mlx.mcp.security.DANGEROUS_PATTERNS module-attribute

DANGEROUS_PATTERNS: List[Pattern] = [re.compile(';\\s*'), re.compile('\\|\\s*'), re.compile('&&\\s*'), re.compile('\\|\\|\\s*'), re.compile('`'), re.compile('\\$\\('), re.compile('>\\s*'), re.compile('<\\s*'), re.compile('\\.\\./'), re.compile('~')]

vllm_mlx.mcp.security.DANGEROUS_ARG_PATTERNS module-attribute

DANGEROUS_ARG_PATTERNS: List[Pattern] = [re.compile(';\\s*'), re.compile('\\|\\s*'), re.compile('&&\\s*'), re.compile('\\|\\|\\s*'), re.compile('`'), re.compile('\\$\\('), re.compile('\\$\\{'), re.compile('>\\s*/'), re.compile('<\\s*/')]

vllm_mlx.mcp.security.BLOCKED_COMMAND_ARG_RULES module-attribute

BLOCKED_COMMAND_ARG_RULES: Dict[str, Dict[str, str]] = {'python': {'-c': 'inline Python execution'}, 'python3': {'-c': 'inline Python execution'}, 'node': {'-e': 'inline JavaScript evaluation', '--eval': 'inline JavaScript evaluation', '-p': 'JavaScript evaluation/print', '--print': 'JavaScript evaluation/print'}, 'npx': {'-c': 'shell command execution', '--call': 'shell command execution'}}

vllm_mlx.mcp.security.CONTROL_CHARS module-attribute

CONTROL_CHARS = ('\n', '\r')

vllm_mlx.mcp.security._validator module-attribute

_validator: Optional[MCPCommandValidator] = None

vllm_mlx.mcp.security.DANGEROUS_TOOL_ARG_PATTERNS module-attribute

DANGEROUS_TOOL_ARG_PATTERNS: List[Pattern] = [re.compile('\\.\\./'), re.compile('/etc/'), re.compile('/proc/'), re.compile('/sys/'), re.compile('~root'), re.compile('/root/')]

vllm_mlx.mcp.security.HIGH_RISK_TOOL_PATTERNS module-attribute

HIGH_RISK_TOOL_PATTERNS: List[str] = ['execute', 'run_command', 'shell', 'eval', 'exec', 'system', 'subprocess']

vllm_mlx.mcp.security._sandbox module-attribute

_sandbox: Optional[ToolSandbox] = None

vllm_mlx.mcp.security.MCPSecurityError

Bases: Exception

Raised when MCP security validation fails.

vllm_mlx.mcp.security.MCPCommandValidator

MCPCommandValidator(allowed_commands: Optional[Set[str]] = None, allow_unsafe: bool = False, custom_whitelist: Optional[Set[str]] = None, check_path_exists: bool = True)

Validates MCP server commands for security.

This class provides methods to validate commands and arguments before they are executed, preventing command injection attacks.

Initialize the command validator.

Parameters:

• allowed_commands (Optional[Set[str]], default: None ) –

  Set of allowed command names. If None, uses default whitelist.

• allow_unsafe (bool, default: False ) –

  If True, allows any command (for development only). WARNING: This disables security checks!

• custom_whitelist (Optional[Set[str]], default: None ) –

  Additional commands to allow beyond the default whitelist.

• check_path_exists (bool, default: True ) –

  If True, verify command exists in PATH. Set to False for testing.

Source code in vllm_mlx/mcp/security.py

def __init__(
    self,
    allowed_commands: Optional[Set[str]] = None,
    allow_unsafe: bool = False,
    custom_whitelist: Optional[Set[str]] = None,
    check_path_exists: bool = True,
):
    """
    Initialize the command validator.

    Args:
        allowed_commands: Set of allowed command names. If None, uses default whitelist.
        allow_unsafe: If True, allows any command (for development only).
                     WARNING: This disables security checks!
        custom_whitelist: Additional commands to allow beyond the default whitelist.
        check_path_exists: If True, verify command exists in PATH. Set to False for testing.
    """
    self.allow_unsafe = allow_unsafe
    self.allowed_commands = allowed_commands or ALLOWED_COMMANDS.copy()
    self.check_path_exists = check_path_exists

    if custom_whitelist:
        self.allowed_commands.update(custom_whitelist)

    if allow_unsafe:
        logger.warning(
            "MCP SECURITY WARNING: Unsafe mode enabled. "
            "All commands will be allowed without validation. "
            "This should NEVER be used in production!"
        )

vllm_mlx.mcp.security.MCPCommandValidator.allow_unsafe instance-attribute

allow_unsafe = allow_unsafe

vllm_mlx.mcp.security.MCPCommandValidator.allowed_commands instance-attribute

allowed_commands = allowed_commands or ALLOWED_COMMANDS.copy()

vllm_mlx.mcp.security.MCPCommandValidator.check_path_exists instance-attribute

check_path_exists = check_path_exists

vllm_mlx.mcp.security.MCPCommandValidator._check_control_chars

_check_control_chars(value: str, context: str, server_name: str) -> None

Block command separators carried via literal newlines.

Source code in vllm_mlx/mcp/security.py

def _check_control_chars(self, value: str, context: str, server_name: str) -> None:
    """Block command separators carried via literal newlines."""
    if any(ch in value for ch in CONTROL_CHARS):
        raise MCPSecurityError(
            f"MCP server '{server_name}': {context} contains newline characters. "
            "Potential command injection blocked."
        )

vllm_mlx.mcp.security.MCPCommandValidator._check_path_traversal

_check_path_traversal(value: str, context: str, server_name: str) -> None

Block parent-directory traversal, including URL-encoded forms.

This normalizes likely path-like inputs rather than relying only on the simple ../ regex, which can be bypassed by percent-encoding.

Source code in vllm_mlx/mcp/security.py

def _check_path_traversal(self, value: str, context: str, server_name: str) -> None:
    """
    Block parent-directory traversal, including URL-encoded forms.

    This normalizes likely path-like inputs rather than relying only on
    the simple ``../`` regex, which can be bypassed by percent-encoding.
    """
    candidates = [value]
    decoded = unquote(value)
    if decoded != value:
        candidates.append(decoded)

    for candidate in candidates:
        if (
            "/" not in candidate
            and "\\" not in candidate
            and "%2e" not in value.lower()
        ):
            continue

        normalized = posixpath.normpath(candidate.replace("\\", "/"))
        if normalized == ".." or normalized.startswith("../"):
            raise MCPSecurityError(
                f"MCP server '{server_name}': {context} contains path traversal: "
                f"'{value}'."
            )

        # Also reject any explicit parent segments before or after normalization.
        path_parts = [
            part for part in candidate.replace("\\", "/").split("/") if part
        ]
        if any(part == ".." for part in path_parts):
            raise MCPSecurityError(
                f"MCP server '{server_name}': {context} contains path traversal: "
                f"'{value}'."
            )

vllm_mlx.mcp.security.MCPCommandValidator.validate_command

validate_command(command: str, server_name: str) -> None

Validate that a command is safe to execute.

Parameters:

• command (str) –

  The command to validate

• server_name (str) –

  Name of the MCP server (for logging)

Raises:

• MCPSecurityError –

  If the command is not allowed

Source code in vllm_mlx/mcp/security.py

def validate_command(self, command: str, server_name: str) -> None:
    """
    Validate that a command is safe to execute.

    Args:
        command: The command to validate
        server_name: Name of the MCP server (for logging)

    Raises:
        MCPSecurityError: If the command is not allowed
    """
    if self.allow_unsafe:
        logger.warning(
            f"MCP security bypassed for server '{server_name}': "
            f"allowing command '{command}' (unsafe mode)"
        )
        return

    self._check_control_chars(command, "Command", server_name)
    self._check_path_traversal(command, "Command", server_name)

    # Check for dangerous patterns in command
    for pattern in DANGEROUS_PATTERNS:
        if pattern.search(command):
            raise MCPSecurityError(
                f"MCP server '{server_name}': Command contains dangerous pattern: "
                f"'{command}'. Command injection attempt blocked."
            )

    # Extract base command name (without path)
    base_command = Path(command).name

    # Check if command is in whitelist
    if base_command not in self.allowed_commands:
        # Check if it's an absolute path to an allowed command
        if os.path.isabs(command):
            resolved_name = Path(command).name
            if resolved_name in self.allowed_commands:
                # Verify the path actually exists and is executable
                if os.path.isfile(command) and os.access(command, os.X_OK):
                    logger.info(
                        f"MCP server '{server_name}': Allowing absolute path "
                        f"to whitelisted command: {command}"
                    )
                    return

        raise MCPSecurityError(
            f"MCP server '{server_name}': Command '{base_command}' is not in the "
            f"allowed commands whitelist. Allowed commands: {sorted(self.allowed_commands)}"
        )

    # Verify command exists in PATH (for non-absolute paths)
    if self.check_path_exists and not os.path.isabs(command):
        resolved_path = shutil.which(command)
        if resolved_path is None:
            raise MCPSecurityError(
                f"MCP server '{server_name}': Command '{command}' not found in PATH. "
                f"Ensure the command is installed and accessible."
            )

    logger.debug(
        f"MCP server '{server_name}': Command '{command}' validated successfully"
    )

vllm_mlx.mcp.security.MCPCommandValidator.validate_args

validate_args(args: List[str], server_name: str) -> None

Validate command arguments for dangerous patterns.

Parameters:

• args (List[str]) –

  List of command arguments

• server_name (str) –

  Name of the MCP server (for logging)

Raises:

• MCPSecurityError –

  If any argument contains dangerous patterns

Source code in vllm_mlx/mcp/security.py

def validate_args(self, args: List[str], server_name: str) -> None:
    """
    Validate command arguments for dangerous patterns.

    Args:
        args: List of command arguments
        server_name: Name of the MCP server (for logging)

    Raises:
        MCPSecurityError: If any argument contains dangerous patterns
    """
    if self.allow_unsafe:
        return

    for i, arg in enumerate(args):
        self._check_control_chars(arg, f"Argument {i}", server_name)
        self._check_path_traversal(arg, f"Argument {i}", server_name)
        for pattern in DANGEROUS_ARG_PATTERNS:
            if pattern.search(arg):
                raise MCPSecurityError(
                    f"MCP server '{server_name}': Argument {i} contains dangerous "
                    f"pattern: '{arg}'. Potential command injection blocked."
                )

    logger.debug(
        f"MCP server '{server_name}': {len(args)} arguments validated successfully"
    )

vllm_mlx.mcp.security.MCPCommandValidator.validate_command_args

validate_command_args(command: str, args: List[str], server_name: str) -> None

Validate command-specific argument combinations.

Some whitelisted runtimes (python, node, npx) remain acceptable for launching packaged MCP servers, but inline evaluator flags such as python -c and node -e must be rejected.

Source code in vllm_mlx/mcp/security.py

def validate_command_args(
    self,
    command: str,
    args: List[str],
    server_name: str,
) -> None:
    """
    Validate command-specific argument combinations.

    Some whitelisted runtimes (python, node, npx) remain acceptable for
    launching packaged MCP servers, but inline evaluator flags such as
    ``python -c`` and ``node -e`` must be rejected.
    """
    if self.allow_unsafe or not args:
        return

    base_command = Path(command).name
    blocked_rules = BLOCKED_COMMAND_ARG_RULES.get(base_command)
    if not blocked_rules:
        return

    for i, arg in enumerate(args):
        if arg in blocked_rules:
            raise MCPSecurityError(
                f"MCP server '{server_name}': Argument {i} '{arg}' enables "
                f"{blocked_rules[arg]} for '{base_command}', which is not allowed."
            )

        if base_command == "node" and arg.startswith("--eval="):
            raise MCPSecurityError(
                f"MCP server '{server_name}': Argument {i} '{arg}' enables "
                "inline JavaScript evaluation for 'node', which is not allowed."
            )

        if base_command == "npx" and arg.startswith("--call="):
            raise MCPSecurityError(
                f"MCP server '{server_name}': Argument {i} '{arg}' enables "
                "shell command execution for 'npx', which is not allowed."
            )

    logger.debug(
        f"MCP server '{server_name}': command-specific arguments validated"
    )

vllm_mlx.mcp.security.MCPCommandValidator.validate_env

validate_env(env: Optional[Dict[str, str]], server_name: str) -> None

Validate environment variables for dangerous values.

Parameters:

• env (Optional[Dict[str, str]]) –

  Dictionary of environment variables

• server_name (str) –

  Name of the MCP server (for logging)

Raises:

• MCPSecurityError –

  If any env var contains dangerous patterns

Source code in vllm_mlx/mcp/security.py

def validate_env(self, env: Optional[Dict[str, str]], server_name: str) -> None:
    """
    Validate environment variables for dangerous values.

    Args:
        env: Dictionary of environment variables
        server_name: Name of the MCP server (for logging)

    Raises:
        MCPSecurityError: If any env var contains dangerous patterns
    """
    if self.allow_unsafe or not env:
        return

    # Dangerous environment variables that could affect execution
    dangerous_env_vars = {
        "LD_PRELOAD",
        "LD_LIBRARY_PATH",
        "DYLD_INSERT_LIBRARIES",
        "DYLD_LIBRARY_PATH",
        "PATH",  # Modifying PATH could redirect commands
        "PYTHONPATH",
        "NODE_PATH",
    }

    for key, value in env.items():
        self._check_control_chars(
            value, f"Environment variable '{key}'", server_name
        )
        self._check_path_traversal(
            value,
            f"Environment variable '{key}'",
            server_name,
        )
        # Check for dangerous env var names
        if key.upper() in dangerous_env_vars:
            raise MCPSecurityError(
                f"MCP server '{server_name}': Setting '{key}' environment variable "
                f"is not allowed for security reasons."
            )

        # Check for dangerous patterns in values
        for pattern in DANGEROUS_ARG_PATTERNS:
            if pattern.search(value):
                raise MCPSecurityError(
                    f"MCP server '{server_name}': Environment variable '{key}' "
                    f"contains dangerous pattern. Potential injection blocked."
                )

    logger.debug(
        f"MCP server '{server_name}': {len(env)} environment variables validated"
    )

vllm_mlx.mcp.security.MCPCommandValidator.validate_url

validate_url(url: str, server_name: str) -> None

Validate SSE URL for security.

Parameters:

• url (str) –

  The SSE URL to validate

• server_name (str) –

  Name of the MCP server (for logging)

Raises:

• MCPSecurityError –

  If the URL is not safe

Source code in vllm_mlx/mcp/security.py

def validate_url(self, url: str, server_name: str) -> None:
    """
    Validate SSE URL for security.

    Args:
        url: The SSE URL to validate
        server_name: Name of the MCP server (for logging)

    Raises:
        MCPSecurityError: If the URL is not safe
    """
    if self.allow_unsafe:
        return

    self._check_control_chars(url, "URL", server_name)

    # Must be http or https
    if not url.startswith(("http://", "https://")):
        raise MCPSecurityError(
            f"MCP server '{server_name}': URL must use http:// or https:// scheme. "
            f"Got: {url}"
        )

    # Warn about non-HTTPS URLs
    if url.startswith("http://") and not url.startswith("http://localhost"):
        logger.warning(
            f"MCP server '{server_name}': Using insecure HTTP connection to {url}. "
            f"Consider using HTTPS for production environments."
        )

    parsed = urlparse(url)
    self._check_path_traversal(parsed.path, "URL", server_name)
    if parsed.query:
        self._check_control_chars(parsed.query, "URL query", server_name)

    # Check for dangerous patterns
    for pattern in DANGEROUS_PATTERNS:
        if pattern.search(url):
            raise MCPSecurityError(
                f"MCP server '{server_name}': URL contains dangerous pattern: {url}"
            )

    logger.debug(f"MCP server '{server_name}': URL '{url}' validated successfully")

vllm_mlx.mcp.security.ToolExecutionAudit dataclass

ToolExecutionAudit(timestamp: float, tool_name: str, server_name: str, arguments: Dict[str, Any], success: bool, error_message: Optional[str] = None, execution_time_ms: Optional[float] = None)

Record of a tool execution for audit purposes.

vllm_mlx.mcp.security.ToolExecutionAudit.timestamp instance-attribute

timestamp: float

vllm_mlx.mcp.security.ToolExecutionAudit.tool_name instance-attribute

tool_name: str

vllm_mlx.mcp.security.ToolExecutionAudit.server_name instance-attribute

server_name: str

vllm_mlx.mcp.security.ToolExecutionAudit.arguments instance-attribute

arguments: Dict[str, Any]

vllm_mlx.mcp.security.ToolExecutionAudit.success instance-attribute

success: bool

vllm_mlx.mcp.security.ToolExecutionAudit.error_message class-attribute instance-attribute

error_message: Optional[str] = None

vllm_mlx.mcp.security.ToolExecutionAudit.execution_time_ms class-attribute instance-attribute

execution_time_ms: Optional[float] = None

vllm_mlx.mcp.security.ToolSandbox

ToolSandbox(allowed_tools: Optional[Set[str]] = None, blocked_tools: Optional[Set[str]] = None, allowed_high_risk_tools: Optional[Set[str]] = None, blocked_arg_patterns: Optional[List[Pattern]] = None, max_calls_per_minute: int = 60, audit_callback: Optional[Callable[[ToolExecutionAudit], None]] = None, enabled: bool = True)

Sandboxing controls for MCP tool execution.

Provides: - Tool allowlisting/blocklisting - Argument sanitization - Audit logging - Rate limiting

Initialize tool sandbox.

Parameters:

• allowed_tools (Optional[Set[str]], default: None ) –

  If set, only these tools can be executed (whitelist mode).

• blocked_tools (Optional[Set[str]], default: None ) –

  Tools that are always blocked (blacklist mode).

• allowed_high_risk_tools (Optional[Set[str]], default: None ) –

  High-risk tools that are explicitly allowed.

• blocked_arg_patterns (Optional[List[Pattern]], default: None ) –

  Patterns to block in tool arguments.

• max_calls_per_minute (int, default: 60 ) –

  Rate limit for tool calls (0 = unlimited).

• audit_callback (Optional[Callable[[ToolExecutionAudit], None]], default: None ) –

  Optional callback for audit events.

• enabled (bool, default: True ) –

  If False, sandbox checks are bypassed (dev mode only).

Source code in vllm_mlx/mcp/security.py

def __init__(
    self,
    allowed_tools: Optional[Set[str]] = None,
    blocked_tools: Optional[Set[str]] = None,
    allowed_high_risk_tools: Optional[Set[str]] = None,
    blocked_arg_patterns: Optional[List[re.Pattern]] = None,
    max_calls_per_minute: int = 60,
    audit_callback: Optional[Callable[[ToolExecutionAudit], None]] = None,
    enabled: bool = True,
):
    """
    Initialize tool sandbox.

    Args:
        allowed_tools: If set, only these tools can be executed (whitelist mode).
        blocked_tools: Tools that are always blocked (blacklist mode).
        allowed_high_risk_tools: High-risk tools that are explicitly allowed.
        blocked_arg_patterns: Patterns to block in tool arguments.
        max_calls_per_minute: Rate limit for tool calls (0 = unlimited).
        audit_callback: Optional callback for audit events.
        enabled: If False, sandbox checks are bypassed (dev mode only).
    """
    self.allowed_tools = allowed_tools
    self.blocked_tools = blocked_tools or set()
    self.allowed_high_risk_tools = {
        tool.lower() for tool in (allowed_high_risk_tools or set())
    }
    self.blocked_arg_patterns = (
        blocked_arg_patterns or DANGEROUS_TOOL_ARG_PATTERNS.copy()
    )
    self.max_calls_per_minute = max_calls_per_minute
    self.audit_callback = audit_callback
    self.enabled = enabled

    # Rate limiting state
    self._call_times: Dict[str, List[float]] = defaultdict(list)
    self._rate_limit_lock = Lock()

    # Audit log (in-memory, bounded)
    self._audit_log: List[ToolExecutionAudit] = []
    self._audit_log_max_size = 1000
    self._audit_lock = Lock()

    if not enabled:
        logger.warning(
            "SECURITY WARNING: Tool sandbox is DISABLED. "
            "All tool executions will be allowed without checks."
        )

vllm_mlx.mcp.security.ToolSandbox.allowed_tools instance-attribute

allowed_tools = allowed_tools

vllm_mlx.mcp.security.ToolSandbox.blocked_tools instance-attribute

blocked_tools = blocked_tools or set()

vllm_mlx.mcp.security.ToolSandbox.allowed_high_risk_tools instance-attribute

allowed_high_risk_tools = {tool.lower() for tool in allowed_high_risk_tools or set()}

vllm_mlx.mcp.security.ToolSandbox.blocked_arg_patterns instance-attribute

blocked_arg_patterns = blocked_arg_patterns or DANGEROUS_TOOL_ARG_PATTERNS.copy()

vllm_mlx.mcp.security.ToolSandbox.max_calls_per_minute instance-attribute

max_calls_per_minute = max_calls_per_minute

vllm_mlx.mcp.security.ToolSandbox.audit_callback instance-attribute

audit_callback = audit_callback

vllm_mlx.mcp.security.ToolSandbox.enabled instance-attribute

enabled = enabled

vllm_mlx.mcp.security.ToolSandbox._call_times instance-attribute

_call_times: Dict[str, List[float]] = defaultdict(list)

vllm_mlx.mcp.security.ToolSandbox._rate_limit_lock instance-attribute

_rate_limit_lock = Lock()

vllm_mlx.mcp.security.ToolSandbox._audit_log instance-attribute

_audit_log: List[ToolExecutionAudit] = []

vllm_mlx.mcp.security.ToolSandbox._audit_log_max_size instance-attribute

_audit_log_max_size = 1000

vllm_mlx.mcp.security.ToolSandbox._audit_lock instance-attribute

_audit_lock = Lock()

vllm_mlx.mcp.security.ToolSandbox.validate_tool_execution

validate_tool_execution(tool_name: str, server_name: str, arguments: Dict[str, Any]) -> None

Validate that a tool execution is allowed.

Parameters:

• tool_name (str) –

  Name of the tool to execute

• server_name (str) –

  MCP server providing the tool

• arguments (Dict[str, Any]) –

  Tool arguments

Raises:

• MCPSecurityError –

  If execution is not allowed

Source code in vllm_mlx/mcp/security.py

def validate_tool_execution(
    self,
    tool_name: str,
    server_name: str,
    arguments: Dict[str, Any],
) -> None:
    """
    Validate that a tool execution is allowed.

    Args:
        tool_name: Name of the tool to execute
        server_name: MCP server providing the tool
        arguments: Tool arguments

    Raises:
        MCPSecurityError: If execution is not allowed
    """
    if not self.enabled:
        logger.debug(f"Sandbox disabled, allowing tool '{tool_name}'")
        return

    full_name = f"{server_name}__{tool_name}"

    # Check blocklist first
    if self._is_blocked(tool_name, full_name):
        raise MCPSecurityError(f"Tool '{tool_name}' is blocked by security policy")

    # Check allowlist if configured
    if self.allowed_tools is not None:
        if (
            tool_name not in self.allowed_tools
            and full_name not in self.allowed_tools
        ):
            raise MCPSecurityError(
                f"Tool '{tool_name}' is not in the allowed tools list"
            )

    # Check for high-risk tool patterns
    self._check_high_risk_tool(tool_name, full_name)

    # Validate arguments
    self._validate_arguments(tool_name, arguments)

    # Check rate limit
    self._check_rate_limit(full_name)

    logger.debug(f"Tool execution validated: {full_name}")

vllm_mlx.mcp.security.ToolSandbox._is_blocked

_is_blocked(tool_name: str, full_name: str) -> bool

Check if tool is in blocklist.

Source code in vllm_mlx/mcp/security.py

def _is_blocked(self, tool_name: str, full_name: str) -> bool:
    """Check if tool is in blocklist."""
    return (
        tool_name in self.blocked_tools
        or full_name in self.blocked_tools
        or tool_name.lower() in self.blocked_tools
    )

vllm_mlx.mcp.security.ToolSandbox._check_high_risk_tool

_check_high_risk_tool(tool_name: str, full_name: str) -> None

Check if tool matches high-risk patterns.

Source code in vllm_mlx/mcp/security.py

def _check_high_risk_tool(self, tool_name: str, full_name: str) -> None:
    """Check if tool matches high-risk patterns."""
    tool_lower = tool_name.lower()
    full_lower = full_name.lower()
    for pattern in HIGH_RISK_TOOL_PATTERNS:
        if pattern in tool_lower:
            if (
                tool_lower in self.allowed_high_risk_tools
                or full_lower in self.allowed_high_risk_tools
            ):
                logger.warning(
                    "Allowing high-risk tool '%s' due to explicit allowlist entry",
                    full_name,
                )
                return
            raise MCPSecurityError(
                f"High-risk tool '{tool_name}' is blocked by security policy. "
                f"Add '{full_name}' or '{tool_name}' to allowed_high_risk_tools "
                f"to allow it explicitly."
            )

vllm_mlx.mcp.security.ToolSandbox._validate_arguments

_validate_arguments(tool_name: str, arguments: Dict[str, Any]) -> None

Validate tool arguments for dangerous patterns.

Source code in vllm_mlx/mcp/security.py

def _validate_arguments(self, tool_name: str, arguments: Dict[str, Any]) -> None:
    """Validate tool arguments for dangerous patterns."""

    def check_value(key: str, value: Any, path: str = "") -> None:
        current_path = f"{path}.{key}" if path else key

        if isinstance(value, str):
            for pattern in self.blocked_arg_patterns:
                if pattern.search(value):
                    raise MCPSecurityError(
                        f"Tool '{tool_name}' argument '{current_path}' contains "
                        f"blocked pattern: {pattern.pattern}"
                    )
        elif isinstance(value, dict):
            for k, v in value.items():
                check_value(k, v, current_path)
        elif isinstance(value, list):
            for i, item in enumerate(value):
                check_value(f"[{i}]", item, current_path)

    for key, value in arguments.items():
        check_value(key, value)

vllm_mlx.mcp.security.ToolSandbox._check_rate_limit

_check_rate_limit(full_name: str) -> None

Check and enforce rate limit for tool calls.

Source code in vllm_mlx/mcp/security.py

def _check_rate_limit(self, full_name: str) -> None:
    """Check and enforce rate limit for tool calls."""
    if self.max_calls_per_minute <= 0:
        return

    now = time.time()
    window_start = now - 60  # 1 minute window

    with self._rate_limit_lock:
        # Clean old entries
        self._call_times[full_name] = [
            t for t in self._call_times[full_name] if t > window_start
        ]

        # Check limit
        if len(self._call_times[full_name]) >= self.max_calls_per_minute:
            raise MCPSecurityError(
                f"Rate limit exceeded for tool '{full_name}': "
                f"max {self.max_calls_per_minute} calls per minute"
            )

        # Record this call
        self._call_times[full_name].append(now)

vllm_mlx.mcp.security.ToolSandbox.record_execution

record_execution(tool_name: str, server_name: str, arguments: Dict[str, Any], success: bool, error_message: Optional[str] = None, execution_time_ms: Optional[float] = None) -> ToolExecutionAudit

Record a tool execution for audit purposes.

Parameters:

• tool_name (str) –

  Name of the executed tool

• server_name (str) –

  MCP server that executed the tool

• arguments (Dict[str, Any]) –

  Arguments passed to the tool

• success (bool) –

  Whether execution succeeded

• error_message (Optional[str], default: None ) –

  Error message if failed

• execution_time_ms (Optional[float], default: None ) –

  Execution time in milliseconds

Returns:

• ToolExecutionAudit –

  The audit record

Source code in vllm_mlx/mcp/security.py

def record_execution(
    self,
    tool_name: str,
    server_name: str,
    arguments: Dict[str, Any],
    success: bool,
    error_message: Optional[str] = None,
    execution_time_ms: Optional[float] = None,
) -> ToolExecutionAudit:
    """
    Record a tool execution for audit purposes.

    Args:
        tool_name: Name of the executed tool
        server_name: MCP server that executed the tool
        arguments: Arguments passed to the tool
        success: Whether execution succeeded
        error_message: Error message if failed
        execution_time_ms: Execution time in milliseconds

    Returns:
        The audit record
    """
    audit = ToolExecutionAudit(
        timestamp=time.time(),
        tool_name=tool_name,
        server_name=server_name,
        arguments=self._sanitize_arguments_for_log(arguments),
        success=success,
        error_message=error_message,
        execution_time_ms=execution_time_ms,
    )

    # Store in audit log
    with self._audit_lock:
        self._audit_log.append(audit)
        # Trim if over max size
        if len(self._audit_log) > self._audit_log_max_size:
            self._audit_log = self._audit_log[-self._audit_log_max_size :]

    # Log the execution
    if success:
        logger.info(
            f"AUDIT: Tool executed - {server_name}__{tool_name} "
            f"(took {execution_time_ms:.1f}ms)"
            if execution_time_ms is not None
            else f"AUDIT: Tool executed - {server_name}__{tool_name}"
        )
    else:
        logger.warning(
            f"AUDIT: Tool failed - {server_name}__{tool_name}: {error_message}"
        )

    # Call callback if configured
    if self.audit_callback:
        try:
            self.audit_callback(audit)
        except Exception as e:
            logger.error(f"Audit callback failed: {e}")

    return audit

vllm_mlx.mcp.security.ToolSandbox._sanitize_arguments_for_log

_sanitize_arguments_for_log(arguments: Dict[str, Any]) -> Dict[str, Any]

Sanitize arguments for logging (redact sensitive data).

Source code in vllm_mlx/mcp/security.py

def _sanitize_arguments_for_log(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
    """Sanitize arguments for logging (redact sensitive data)."""
    sensitive_keys = {"password", "token", "secret", "key", "credential", "auth"}

    def sanitize(obj: Any) -> Any:
        if isinstance(obj, dict):
            return {
                k: (
                    "[REDACTED]"
                    if any(s in k.lower() for s in sensitive_keys)
                    else sanitize(v)
                )
                for k, v in obj.items()
            }
        elif isinstance(obj, list):
            return [sanitize(item) for item in obj]
        elif isinstance(obj, str) and len(obj) > 1000:
            return obj[:100] + f"... [truncated, {len(obj)} chars total]"
        return obj

    return sanitize(arguments)

vllm_mlx.mcp.security.ToolSandbox.get_audit_log

get_audit_log(limit: int = 100, tool_filter: Optional[str] = None, server_filter: Optional[str] = None, errors_only: bool = False) -> List[ToolExecutionAudit]

Get audit log entries.

Parameters:

• limit (int, default: 100 ) –

  Maximum entries to return

• tool_filter (Optional[str], default: None ) –

  Filter by tool name (substring match)

• server_filter (Optional[str], default: None ) –

  Filter by server name

• errors_only (bool, default: False ) –

  Only return failed executions

Returns:

• List[ToolExecutionAudit] –

  List of audit entries

Source code in vllm_mlx/mcp/security.py

def get_audit_log(
    self,
    limit: int = 100,
    tool_filter: Optional[str] = None,
    server_filter: Optional[str] = None,
    errors_only: bool = False,
) -> List[ToolExecutionAudit]:
    """
    Get audit log entries.

    Args:
        limit: Maximum entries to return
        tool_filter: Filter by tool name (substring match)
        server_filter: Filter by server name
        errors_only: Only return failed executions

    Returns:
        List of audit entries
    """
    with self._audit_lock:
        entries = self._audit_log.copy()

    # Apply filters
    if tool_filter:
        entries = [e for e in entries if tool_filter in e.tool_name]
    if server_filter:
        entries = [e for e in entries if server_filter in e.server_name]
    if errors_only:
        entries = [e for e in entries if not e.success]

    # Return most recent entries
    return entries[-limit:]

vllm_mlx.mcp.security.ToolSandbox.clear_audit_log

clear_audit_log() -> int

Clear audit log and return number of entries cleared.

Source code in vllm_mlx/mcp/security.py

def clear_audit_log(self) -> int:
    """Clear audit log and return number of entries cleared."""
    with self._audit_lock:
        count = len(self._audit_log)
        self._audit_log.clear()
    return count

vllm_mlx.mcp.security.get_validator

get_validator() -> MCPCommandValidator

Get the global command validator instance.

Source code in vllm_mlx/mcp/security.py

def get_validator() -> MCPCommandValidator:
    """Get the global command validator instance."""
    global _validator
    if _validator is None:
        _validator = MCPCommandValidator(
            allow_unsafe=os.environ.get(ALLOW_UNSAFE_ENV_VAR) == "1"
        )
    return _validator

vllm_mlx.mcp.security.set_validator

set_validator(validator: MCPCommandValidator) -> None

Set a custom global validator.

Source code in vllm_mlx/mcp/security.py

def set_validator(validator: MCPCommandValidator) -> None:
    """Set a custom global validator."""
    global _validator
    _validator = validator

vllm_mlx.mcp.security.validate_mcp_server_config

validate_mcp_server_config(server_name: str, command: Optional[str] = None, args: Optional[List[str]] = None, env: Optional[Dict[str, str]] = None, url: Optional[str] = None) -> None

Validate MCP server configuration for security.

This is a convenience function that uses the global validator.

Parameters:

• server_name (str) –

  Name of the MCP server

• command (Optional[str], default: None ) –

  Command to execute (for stdio transport)

• args (Optional[List[str]], default: None ) –

  Command arguments

• env (Optional[Dict[str, str]], default: None ) –

  Environment variables

• url (Optional[str], default: None ) –

  SSE URL (for sse transport)

Raises:

• MCPSecurityError –

  If validation fails

Source code in vllm_mlx/mcp/security.py

def validate_mcp_server_config(
    server_name: str,
    command: Optional[str] = None,
    args: Optional[List[str]] = None,
    env: Optional[Dict[str, str]] = None,
    url: Optional[str] = None,
) -> None:
    """
    Validate MCP server configuration for security.

    This is a convenience function that uses the global validator.

    Args:
        server_name: Name of the MCP server
        command: Command to execute (for stdio transport)
        args: Command arguments
        env: Environment variables
        url: SSE URL (for sse transport)

    Raises:
        MCPSecurityError: If validation fails
    """
    validator = get_validator()

    if command:
        validator.validate_command(command, server_name)

    if args:
        validator.validate_args(args, server_name)
        if command:
            validator.validate_command_args(command, args, server_name)

    if env:
        validator.validate_env(env, server_name)

    if url:
        validator.validate_url(url, server_name)

vllm_mlx.mcp.security.get_sandbox

get_sandbox() -> ToolSandbox

Get the global tool sandbox instance.

Source code in vllm_mlx/mcp/security.py

def get_sandbox() -> ToolSandbox:
    """Get the global tool sandbox instance."""
    global _sandbox
    if _sandbox is None:
        _sandbox = ToolSandbox()
    return _sandbox

vllm_mlx.mcp.security.set_sandbox

set_sandbox(sandbox: ToolSandbox) -> None

Set a custom global sandbox.

Source code in vllm_mlx/mcp/security.py

def set_sandbox(sandbox: ToolSandbox) -> None:
    """Set a custom global sandbox."""
    global _sandbox
    _sandbox = sandbox

Complete contract reference

Expand any definition for its exact inputs, annotations, defaults, return contract, directly raised exceptions, source-grounded behavior, and immutable line link. This section includes private and nested definitions that ordinary API generators omit.

vllm_mlx.mcp.security.MCPSecurityError · class

vllm_mlx.mcp.security.MCPSecurityError()

Raised when MCP security validation fails.

Parameters

This callable has no explicit inputs.

Returns

• Constructs: vllm_mlx.mcp.security.MCPSecurityError

Exceptions and behavior

Class MCPSecurityError derives from Exception and declares 0 direct member(s). No direct raise statement appears in this definition.

View source #L106-L109.

vllm_mlx.mcp.security.MCPCommandValidator · class

vllm_mlx.mcp.security.MCPCommandValidator(allowed_commands: Optional[Set[str]] = None, allow_unsafe: bool = False, custom_whitelist: Optional[Set[str]] = None, check_path_exists: bool = True)

Validates MCP server commands for security.

Parameters

Name	Type	Required	Default	Description
allowed_commands	Optional[Set[str]]	no	None	Set of allowed command names. If None, uses default whitelist.
allow_unsafe	bool	no	False	If True, allows any command (for development only).
custom_whitelist	Optional[Set[str]]	no	None	Additional commands to allow beyond the default whitelist.
check_path_exists	bool	no	True	If True, verify command exists in PATH. Set to False for testing.

Returns

• Constructs: vllm_mlx.mcp.security.MCPCommandValidator

Exceptions and behavior

Class MCPCommandValidator declares 8 direct member(s). No direct raise statement appears in this definition.

View source #L112-L427.

vllm_mlx.mcp.security.MCPCommandValidator.__init__ · method

vllm_mlx.mcp.security.MCPCommandValidator.__init__(allowed_commands: Optional[Set[str]] = None, allow_unsafe: bool = False, custom_whitelist: Optional[Set[str]] = None, check_path_exists: bool = True) -> not annotated

Initialize the command validator.

Parameters

Name	Type	Required	Default	Description
allowed_commands	Optional[Set[str]]	no	None	Set of allowed command names. If None, uses default whitelist.
allow_unsafe	bool	no	False	If True, allows any command (for development only).
custom_whitelist	Optional[Set[str]]	no	None	Additional commands to allow beyond the default whitelist.
check_path_exists	bool	no	True	If True, verify command exists in PATH. Set to False for testing.

Returns

• Type: not annotated

Exceptions and behavior

Method MCPCommandValidator.__init__ updates self.allow_unsafe, self.allowed_commands, self.check_path_exists; calls ALLOWED_COMMANDS.copy, self.allowed_commands.update, logger.warning. No direct raise statement appears in this definition.

View source #L120-L149.

vllm_mlx.mcp.security.MCPCommandValidator._check_control_chars · method

vllm_mlx.mcp.security.MCPCommandValidator._check_control_chars(value: str, context: str, server_name: str) -> None

Block command separators carried via literal newlines.

Parameters

Name	Type	Required	Default	Description
value	str	yes	none	Required positional or keyword input.
context	str	yes	none	Required positional or keyword input.
server_name	str	yes	none	Required positional or keyword input.

Returns

• Type: None

Exceptions and behavior

Method MCPCommandValidator._check_control_chars calls any, MCPSecurityError; can raise MCPSecurityError. Directly raised exceptions: MCPSecurityError.

View source #L151-L157.

vllm_mlx.mcp.security.MCPCommandValidator._check_path_traversal · method

vllm_mlx.mcp.security.MCPCommandValidator._check_path_traversal(value: str, context: str, server_name: str) -> None

Block parent-directory traversal, including URL-encoded forms.

Parameters

Name	Type	Required	Default	Description
value	str	yes	none	Required positional or keyword input.
context	str	yes	none	Required positional or keyword input.
server_name	str	yes	none	Required positional or keyword input.

Returns

• Type: None

Exceptions and behavior

Method MCPCommandValidator._check_path_traversal calls unquote, candidates.append, value.lower, posixpath.normpath; can raise MCPSecurityError. Directly raised exceptions: MCPSecurityError.

View source #L159-L194.

vllm_mlx.mcp.security.MCPCommandValidator.validate_command · method

vllm_mlx.mcp.security.MCPCommandValidator.validate_command(command: str, server_name: str) -> None

Validate that a command is safe to execute.

Parameters

Name	Type	Required	Default	Description
command	str	yes	none	The command to validate
server_name	str	yes	none	Name of the MCP server (for logging)

Returns

• Type: None
• Direct return expressions: None

Exceptions and behavior

Method MCPCommandValidator.validate_command calls logger.warning, self._check_control_chars, self._check_path_traversal, pattern.search; can raise MCPSecurityError; returns None. Directly raised exceptions: MCPSecurityError.

View source #L196-L258.

vllm_mlx.mcp.security.MCPCommandValidator.validate_args · method

vllm_mlx.mcp.security.MCPCommandValidator.validate_args(args: List[str], server_name: str) -> None

Validate command arguments for dangerous patterns.

Parameters

Name	Type	Required	Default	Description
args	List[str]	yes	none	List of command arguments
server_name	str	yes	none	Name of the MCP server (for logging)

Returns

• Type: None
• Direct return expressions: None

Exceptions and behavior

Method MCPCommandValidator.validate_args calls enumerate, self._check_control_chars, self._check_path_traversal, pattern.search; can raise MCPSecurityError; returns None. Directly raised exceptions: MCPSecurityError.

View source #L260-L286.

vllm_mlx.mcp.security.MCPCommandValidator.validate_command_args · method

vllm_mlx.mcp.security.MCPCommandValidator.validate_command_args(command: str, args: List[str], server_name: str) -> None

Validate command-specific argument combinations.

Parameters

Name	Type	Required	Default	Description
command	str	yes	none	Required positional or keyword input.
args	List[str]	yes	none	Required positional or keyword input.
server_name	str	yes	none	Required positional or keyword input.

Returns

• Type: None
• Direct return expressions: None

Exceptions and behavior

Method MCPCommandValidator.validate_command_args calls Path, BLOCKED_COMMAND_ARG_RULES.get, enumerate, MCPSecurityError; can raise MCPSecurityError; returns None. Directly raised exceptions: MCPSecurityError.

View source #L288-L330.

vllm_mlx.mcp.security.MCPCommandValidator.validate_env · method

vllm_mlx.mcp.security.MCPCommandValidator.validate_env(env: Optional[Dict[str, str]], server_name: str) -> None

Validate environment variables for dangerous values.

Parameters

Name	Type	Required	Default	Description
env	Optional[Dict[str, str]]	yes	none	Dictionary of environment variables
server_name	str	yes	none	Name of the MCP server (for logging)

Returns

• Type: None
• Direct return expressions: None

Exceptions and behavior

Method MCPCommandValidator.validate_env calls env.items, self._check_control_chars, self._check_path_traversal, key.upper; can raise MCPSecurityError; returns None. Directly raised exceptions: MCPSecurityError.

View source #L332-L383.

vllm_mlx.mcp.security.MCPCommandValidator.validate_url · method

vllm_mlx.mcp.security.MCPCommandValidator.validate_url(url: str, server_name: str) -> None

Validate SSE URL for security.

Parameters

Name	Type	Required	Default	Description
url	str	yes	none	The SSE URL to validate
server_name	str	yes	none	Name of the MCP server (for logging)

Returns

• Type: None
• Direct return expressions: None

Exceptions and behavior

Method MCPCommandValidator.validate_url calls self._check_control_chars, url.startswith, MCPSecurityError, logger.warning; can raise MCPSecurityError; returns None. Directly raised exceptions: MCPSecurityError.

View source #L385-L427.

vllm_mlx.mcp.security.get_validator · function

vllm_mlx.mcp.security.get_validator() -> MCPCommandValidator

Get the global command validator instance.

Parameters

This callable has no explicit inputs.

Returns

• Type: MCPCommandValidator
• Direct return expressions: _validator

Exceptions and behavior

Function get_validator calls MCPCommandValidator, os.environ.get; returns _validator. No direct raise statement appears in this definition.

View source #L434-L441.

vllm_mlx.mcp.security.set_validator · function

vllm_mlx.mcp.security.set_validator(validator: MCPCommandValidator) -> None

Set a custom global validator.

Parameters

Name	Type	Required	Default	Description
validator	MCPCommandValidator	yes	none	Required positional or keyword input.

Returns

• Type: None

Exceptions and behavior

Function set_validator contains no state mutation, call, raise, return, await, or yield. No direct raise statement appears in this definition.

View source #L444-L447.

vllm_mlx.mcp.security.validate_mcp_server_config · function

vllm_mlx.mcp.security.validate_mcp_server_config(server_name: str, command: Optional[str] = None, args: Optional[List[str]] = None, env: Optional[Dict[str, str]] = None, url: Optional[str] = None) -> None

Validate MCP server configuration for security.

Parameters

Name	Type	Required	Default	Description
server_name	str	yes	none	Name of the MCP server
command	Optional[str]	no	None	Command to execute (for stdio transport)
args	Optional[List[str]]	no	None	Command arguments
env	Optional[Dict[str, str]]	no	None	Environment variables
url	Optional[str]	no	None	SSE URL (for sse transport)

Returns

• Type: None

Exceptions and behavior

Function validate_mcp_server_config calls get_validator, validator.validate_command, validator.validate_args, validator.validate_command_args. No direct raise statement appears in this definition.

View source #L450-L486.

vllm_mlx.mcp.security.ToolExecutionAudit · class

vllm_mlx.mcp.security.ToolExecutionAudit(timestamp: float, tool_name: str, server_name: str, arguments: Dict[str, Any], success: bool, error_message: Optional[str] = None, execution_time_ms: Optional[float] = None)

Record of a tool execution for audit purposes.

Parameters

Name	Type	Required	Default	Description
timestamp	float	yes	none	Required constructor field.
tool_name	str	yes	none	Required constructor field.
server_name	str	yes	none	Required constructor field.
arguments	Dict[str, Any]	yes	none	Required constructor field.
success	bool	yes	none	Required constructor field.
error_message	Optional[str]	no	None	Optional constructor field; defaults to None.
execution_time_ms	Optional[float]	no	None	Optional constructor field; defaults to None.

Returns

• Constructs: vllm_mlx.mcp.security.ToolExecutionAudit

Exceptions and behavior

Class ToolExecutionAudit declares 0 direct member(s). No direct raise statement appears in this definition.

View source #L516-L525.

vllm_mlx.mcp.security.ToolSandbox · class

vllm_mlx.mcp.security.ToolSandbox(allowed_tools: Optional[Set[str]] = None, blocked_tools: Optional[Set[str]] = None, allowed_high_risk_tools: Optional[Set[str]] = None, blocked_arg_patterns: Optional[List[re.Pattern]] = None, max_calls_per_minute: int = 60, audit_callback: Optional[Callable[[ToolExecutionAudit], None]] = None, enabled: bool = True)

Sandboxing controls for MCP tool execution.

Parameters

Name	Type	Required	Default	Description
allowed_tools	Optional[Set[str]]	no	None	If set, only these tools can be executed (whitelist mode).
blocked_tools	Optional[Set[str]]	no	None	Tools that are always blocked (blacklist mode).
allowed_high_risk_tools	Optional[Set[str]]	no	None	High-risk tools that are explicitly allowed.
blocked_arg_patterns	Optional[List[re.Pattern]]	no	None	Patterns to block in tool arguments.
max_calls_per_minute	int	no	60	Rate limit for tool calls (0 = unlimited).
audit_callback	Optional[Callable[[ToolExecutionAudit], None]]	no	None	Optional callback for audit events.
enabled	bool	no	True	If False, sandbox checks are bypassed (dev mode only).

Returns

• Constructs: vllm_mlx.mcp.security.ToolSandbox

Exceptions and behavior

Class ToolSandbox declares 10 direct member(s). No direct raise statement appears in this definition.

View source #L528-L834.

vllm_mlx.mcp.security.ToolSandbox.__init__ · method

vllm_mlx.mcp.security.ToolSandbox.__init__(allowed_tools: Optional[Set[str]] = None, blocked_tools: Optional[Set[str]] = None, allowed_high_risk_tools: Optional[Set[str]] = None, blocked_arg_patterns: Optional[List[re.Pattern]] = None, max_calls_per_minute: int = 60, audit_callback: Optional[Callable[[ToolExecutionAudit], None]] = None, enabled: bool = True) -> not annotated

Initialize tool sandbox.

Parameters

Name	Type	Required	Default	Description
allowed_tools	Optional[Set[str]]	no	None	If set, only these tools can be executed (whitelist mode).
blocked_tools	Optional[Set[str]]	no	None	Tools that are always blocked (blacklist mode).
allowed_high_risk_tools	Optional[Set[str]]	no	None	High-risk tools that are explicitly allowed.
blocked_arg_patterns	Optional[List[re.Pattern]]	no	None	Patterns to block in tool arguments.
max_calls_per_minute	int	no	60	Rate limit for tool calls (0 = unlimited).
audit_callback	Optional[Callable[[ToolExecutionAudit], None]]	no	None	Optional callback for audit events.
enabled	bool	no	True	If False, sandbox checks are bypassed (dev mode only).

Returns

• Type: not annotated

Exceptions and behavior

Method ToolSandbox.__init__ updates self.allowed_tools, self.blocked_tools, self.allowed_high_risk_tools, self.blocked_arg_patterns; calls set, tool.lower, DANGEROUS_TOOL_ARG_PATTERNS.copy, defaultdict. No direct raise statement appears in this definition.

View source #L539-L586.

vllm_mlx.mcp.security.ToolSandbox.validate_tool_execution · method

vllm_mlx.mcp.security.ToolSandbox.validate_tool_execution(tool_name: str, server_name: str, arguments: Dict[str, Any]) -> None

Validate that a tool execution is allowed.

Parameters

Name	Type	Required	Default	Description
tool_name	str	yes	none	Name of the tool to execute
server_name	str	yes	none	MCP server providing the tool
arguments	Dict[str, Any]	yes	none	Tool arguments

Returns

• Type: None
• Direct return expressions: None

Exceptions and behavior

Method ToolSandbox.validate_tool_execution calls logger.debug, self._is_blocked, MCPSecurityError, self._check_high_risk_tool; can raise MCPSecurityError; returns None. Directly raised exceptions: MCPSecurityError.

View source #L588-L634.

vllm_mlx.mcp.security.ToolSandbox._is_blocked · method

vllm_mlx.mcp.security.ToolSandbox._is_blocked(tool_name: str, full_name: str) -> bool

Check if tool is in blocklist.

Parameters

Name	Type	Required	Default	Description
tool_name	str	yes	none	Required positional or keyword input.
full_name	str	yes	none	Required positional or keyword input.

Returns

• Type: bool
• Direct return expressions: tool_name in self.blocked_tools or full_name in self.blocked_tools or tool_name.lower() in self.blocked_tools

Exceptions and behavior

Method ToolSandbox._is_blocked calls tool_name.lower; returns tool_name in self.blocked_tools or full_name in self.blocked_tools or tool_name.lower() in self.blocked_tools. No direct raise statement appears in this definition.

View source #L636-L642.

vllm_mlx.mcp.security.ToolSandbox._check_high_risk_tool · method

vllm_mlx.mcp.security.ToolSandbox._check_high_risk_tool(tool_name: str, full_name: str) -> None

Check if tool matches high-risk patterns.

Parameters

Name	Type	Required	Default	Description
tool_name	str	yes	none	Required positional or keyword input.
full_name	str	yes	none	Required positional or keyword input.

Returns

• Type: None
• Direct return expressions: None

Exceptions and behavior

Method ToolSandbox._check_high_risk_tool calls tool_name.lower, full_name.lower, logger.warning, MCPSecurityError; can raise MCPSecurityError; returns None. Directly raised exceptions: MCPSecurityError.

View source #L644-L663.

vllm_mlx.mcp.security.ToolSandbox._validate_arguments · method

vllm_mlx.mcp.security.ToolSandbox._validate_arguments(tool_name: str, arguments: Dict[str, Any]) -> None

Validate tool arguments for dangerous patterns.

Parameters

Name	Type	Required	Default	Description
tool_name	str	yes	none	Required positional or keyword input.
arguments	Dict[str, Any]	yes	none	Required positional or keyword input.

Returns

• Type: None

Exceptions and behavior

Method ToolSandbox._validate_arguments calls arguments.items, check_value. No direct raise statement appears in this definition.

View source #L665-L686.

vllm_mlx.mcp.security.ToolSandbox._validate_arguments.check_value · nested function

vllm_mlx.mcp.security.ToolSandbox._validate_arguments.check_value(key: str, value: Any, path: str = '') -> None

Nested Function ToolSandbox._validate_arguments.check_value calls isinstance, pattern.search, MCPSecurityError, value.items; can raise MCPSecurityError.

Parameters

Name	Type	Required	Default	Description
key	str	yes	none	Required positional or keyword input.
value	Any	yes	none	Required positional or keyword input.
path	str	no	''	Optional positional or keyword input; defaults to ''.

Returns

• Type: None

Exceptions and behavior

Nested Function ToolSandbox._validate_arguments.check_value calls isinstance, pattern.search, MCPSecurityError, value.items; can raise MCPSecurityError. Directly raised exceptions: MCPSecurityError.

View source #L668-L683.

vllm_mlx.mcp.security.ToolSandbox._check_rate_limit · method

vllm_mlx.mcp.security.ToolSandbox._check_rate_limit(full_name: str) -> None

Check and enforce rate limit for tool calls.

Parameters

Name	Type	Required	Default	Description
full_name	str	yes	none	Required positional or keyword input.

Returns

• Type: None
• Direct return expressions: None

Exceptions and behavior

Method ToolSandbox._check_rate_limit calls time.time, len, MCPSecurityError, self._call_times[full_name].append; can raise MCPSecurityError; returns None. Directly raised exceptions: MCPSecurityError.

View source #L688-L710.

vllm_mlx.mcp.security.ToolSandbox.record_execution · method

vllm_mlx.mcp.security.ToolSandbox.record_execution(tool_name: str, server_name: str, arguments: Dict[str, Any], success: bool, error_message: Optional[str] = None, execution_time_ms: Optional[float] = None) -> ToolExecutionAudit

Record a tool execution for audit purposes.

Parameters

Name	Type	Required	Default	Description
tool_name	str	yes	none	Name of the executed tool
server_name	str	yes	none	MCP server that executed the tool
arguments	Dict[str, Any]	yes	none	Arguments passed to the tool
success	bool	yes	none	Whether execution succeeded
error_message	Optional[str]	no	None	Error message if failed
execution_time_ms	Optional[float]	no	None	Execution time in milliseconds

Returns

• Type: ToolExecutionAudit
• Direct return expressions: audit

Exceptions and behavior

Method ToolSandbox.record_execution updates self._audit_log; calls ToolExecutionAudit, time.time, self._sanitize_arguments_for_log, self._audit_log.append; returns audit. No direct raise statement appears in this definition.

View source #L712-L772.

vllm_mlx.mcp.security.ToolSandbox._sanitize_arguments_for_log · method

vllm_mlx.mcp.security.ToolSandbox._sanitize_arguments_for_log(arguments: Dict[str, Any]) -> Dict[str, Any]

Sanitize arguments for logging (redact sensitive data).

Parameters

Name	Type	Required	Default	Description
arguments	Dict[str, Any]	yes	none	Required positional or keyword input.

Returns

• Type: Dict[str, Any]
• Direct return expressions: sanitize(arguments)

Exceptions and behavior

Method ToolSandbox._sanitize_arguments_for_log calls sanitize; returns sanitize(arguments). No direct raise statement appears in this definition.

View source #L774-L794.

vllm_mlx.mcp.security.ToolSandbox._sanitize_arguments_for_log.sanitize · nested function

vllm_mlx.mcp.security.ToolSandbox._sanitize_arguments_for_log.sanitize(obj: Any) -> Any

Nested Function ToolSandbox._sanitize_arguments_for_log.sanitize calls isinstance, any, k.lower, sanitize; has 4 explicit return paths.

Parameters

Name	Type	Required	Default	Description
obj	Any	yes	none	Required positional or keyword input.

Returns

• Type: Any
• Direct return expressions: {k: '[REDACTED]' if any((s in k.lower() for s in sensitive_keys)) else sanitize(v) for k, v in obj.items()}; [sanitize(item) for item in obj]; obj[:100] + f'... [truncated, {len(obj)} chars total]'; obj

Exceptions and behavior

Nested Function ToolSandbox._sanitize_arguments_for_log.sanitize calls isinstance, any, k.lower, sanitize; has 4 explicit return paths. No direct raise statement appears in this definition.

View source #L778-L792.

vllm_mlx.mcp.security.ToolSandbox.get_audit_log · method

vllm_mlx.mcp.security.ToolSandbox.get_audit_log(limit: int = 100, tool_filter: Optional[str] = None, server_filter: Optional[str] = None, errors_only: bool = False) -> List[ToolExecutionAudit]

Get audit log entries.

Parameters

Name	Type	Required	Default	Description
limit	int	no	100	Maximum entries to return
tool_filter	Optional[str]	no	None	Filter by tool name (substring match)
server_filter	Optional[str]	no	None	Filter by server name
errors_only	bool	no	False	Only return failed executions

Returns

• Type: List[ToolExecutionAudit]
• Direct return expressions: entries[-limit:]

Exceptions and behavior

Method ToolSandbox.get_audit_log calls self._audit_log.copy; returns entries[-limit:]. No direct raise statement appears in this definition.

View source #L796-L827.

vllm_mlx.mcp.security.ToolSandbox.clear_audit_log · method

vllm_mlx.mcp.security.ToolSandbox.clear_audit_log() -> int

Clear audit log and return number of entries cleared.

Parameters

This callable has no explicit inputs.

Returns

• Type: int
• Direct return expressions: count

Exceptions and behavior

Method ToolSandbox.clear_audit_log calls len, self._audit_log.clear; returns count. No direct raise statement appears in this definition.

View source #L829-L834.

vllm_mlx.mcp.security.get_sandbox · function

vllm_mlx.mcp.security.get_sandbox() -> ToolSandbox

Get the global tool sandbox instance.

Parameters

This callable has no explicit inputs.

Returns

• Type: ToolSandbox
• Direct return expressions: _sandbox

Exceptions and behavior

Function get_sandbox calls ToolSandbox; returns _sandbox. No direct raise statement appears in this definition.

View source #L841-L846.

vllm_mlx.mcp.security.set_sandbox · function

vllm_mlx.mcp.security.set_sandbox(sandbox: ToolSandbox) -> None

Set a custom global sandbox.

Parameters

Name	Type	Required	Default	Description
sandbox	ToolSandbox	yes	none	Required positional or keyword input.

Returns

• Type: None

Exceptions and behavior

Function set_sandbox contains no state mutation, call, raise, return, await, or yield. No direct raise statement appears in this definition.

View source #L849-L852.

Complete symbol map

This map also includes private definitions and nested helpers. The signature column exposes every explicit input even when an internal helper has no dedicated parameter prose.

Symbol	Kind	Signature and inputs	What it does	Source
MCPSecurityError	class	MCPSecurityError()	Raised when MCP security validation fails.	#L106-L109
MCPCommandValidator	class	MCPCommandValidator(allowed_commands: Optional[Set[str]] = None, allow_unsafe: bool = False, custom_whitelist: Optional[Set[str]] = None, check_path_exists: bool = True)	Validates MCP server commands for security.	#L112-L427
MCPCommandValidator.__init__	method	MCPCommandValidator.__init__(allowed_commands: Optional[Set[str]] = None, allow_unsafe: bool = False, custom_whitelist: Optional[Set[str]] = None, check_path_exists: bool = True) -> not annotated	Initialize the command validator.	#L120-L149
MCPCommandValidator._check_control_chars	method	MCPCommandValidator._check_control_chars(value: str, context: str, server_name: str) -> None	Block command separators carried via literal newlines.	#L151-L157
MCPCommandValidator._check_path_traversal	method	MCPCommandValidator._check_path_traversal(value: str, context: str, server_name: str) -> None	Block parent-directory traversal, including URL-encoded forms.	#L159-L194
MCPCommandValidator.validate_command	method	MCPCommandValidator.validate_command(command: str, server_name: str) -> None	Validate that a command is safe to execute.	#L196-L258
MCPCommandValidator.validate_args	method	MCPCommandValidator.validate_args(args: List[str], server_name: str) -> None	Validate command arguments for dangerous patterns.	#L260-L286
MCPCommandValidator.validate_command_args	method	MCPCommandValidator.validate_command_args(command: str, args: List[str], server_name: str) -> None	Validate command-specific argument combinations.	#L288-L330
MCPCommandValidator.validate_env	method	MCPCommandValidator.validate_env(env: Optional[Dict[str, str]], server_name: str) -> None	Validate environment variables for dangerous values.	#L332-L383
MCPCommandValidator.validate_url	method	MCPCommandValidator.validate_url(url: str, server_name: str) -> None	Validate SSE URL for security.	#L385-L427
get_validator	function	get_validator() -> MCPCommandValidator	Get the global command validator instance.	#L434-L441
set_validator	function	set_validator(validator: MCPCommandValidator) -> None	Set a custom global validator.	#L444-L447
validate_mcp_server_config	function	validate_mcp_server_config(server_name: str, command: Optional[str] = None, args: Optional[List[str]] = None, env: Optional[Dict[str, str]] = None, url: Optional[str] = None) -> None	Validate MCP server configuration for security.	#L450-L486
ToolExecutionAudit	class	ToolExecutionAudit(timestamp: float, tool_name: str, server_name: str, arguments: Dict[str, Any], success: bool, error_message: Optional[str] = None, execution_time_ms: Optional[float] = None)	Record of a tool execution for audit purposes.	#L516-L525
ToolSandbox	class	ToolSandbox(allowed_tools: Optional[Set[str]] = None, blocked_tools: Optional[Set[str]] = None, allowed_high_risk_tools: Optional[Set[str]] = None, blocked_arg_patterns: Optional[List[re.Pattern]] = None, max_calls_per_minute: int = 60, audit_callback: Optional[Callable[[ToolExecutionAudit], None]] = None, enabled: bool = True)	Sandboxing controls for MCP tool execution.	#L528-L834
ToolSandbox.__init__	method	ToolSandbox.__init__(allowed_tools: Optional[Set[str]] = None, blocked_tools: Optional[Set[str]] = None, allowed_high_risk_tools: Optional[Set[str]] = None, blocked_arg_patterns: Optional[List[re.Pattern]] = None, max_calls_per_minute: int = 60, audit_callback: Optional[Callable[[ToolExecutionAudit], None]] = None, enabled: bool = True) -> not annotated	Initialize tool sandbox.	#L539-L586
ToolSandbox.validate_tool_execution	method	ToolSandbox.validate_tool_execution(tool_name: str, server_name: str, arguments: Dict[str, Any]) -> None	Validate that a tool execution is allowed.	#L588-L634
ToolSandbox._is_blocked	method	ToolSandbox._is_blocked(tool_name: str, full_name: str) -> bool	Check if tool is in blocklist.	#L636-L642
ToolSandbox._check_high_risk_tool	method	ToolSandbox._check_high_risk_tool(tool_name: str, full_name: str) -> None	Check if tool matches high-risk patterns.	#L644-L663
ToolSandbox._validate_arguments	method	ToolSandbox._validate_arguments(tool_name: str, arguments: Dict[str, Any]) -> None	Validate tool arguments for dangerous patterns.	#L665-L686
ToolSandbox._validate_arguments.check_value	nested function	ToolSandbox._validate_arguments.check_value(key: str, value: Any, path: str = '') -> None	Nested Function ToolSandbox._validate_arguments.check_value calls isinstance, pattern.search, MCPSecurityError, value.items; can raise MCPSecurityError.	#L668-L683
ToolSandbox._check_rate_limit	method	ToolSandbox._check_rate_limit(full_name: str) -> None	Check and enforce rate limit for tool calls.	#L688-L710
ToolSandbox.record_execution	method	ToolSandbox.record_execution(tool_name: str, server_name: str, arguments: Dict[str, Any], success: bool, error_message: Optional[str] = None, execution_time_ms: Optional[float] = None) -> ToolExecutionAudit	Record a tool execution for audit purposes.	#L712-L772
ToolSandbox._sanitize_arguments_for_log	method	ToolSandbox._sanitize_arguments_for_log(arguments: Dict[str, Any]) -> Dict[str, Any]	Sanitize arguments for logging (redact sensitive data).	#L774-L794
ToolSandbox._sanitize_arguments_for_log.sanitize	nested function	ToolSandbox._sanitize_arguments_for_log.sanitize(obj: Any) -> Any	Nested Function ToolSandbox._sanitize_arguments_for_log.sanitize calls isinstance, any, k.lower, sanitize; has 4 explicit return paths.	#L778-L792
ToolSandbox.get_audit_log	method	ToolSandbox.get_audit_log(limit: int = 100, tool_filter: Optional[str] = None, server_filter: Optional[str] = None, errors_only: bool = False) -> List[ToolExecutionAudit]	Get audit log entries.	#L796-L827
ToolSandbox.clear_audit_log	method	ToolSandbox.clear_audit_log() -> int	Clear audit log and return number of entries cleared.	#L829-L834
get_sandbox	function	get_sandbox() -> ToolSandbox	Get the global tool sandbox instance.	#L841-L846
set_sandbox	function	set_sandbox(sandbox: ToolSandbox) -> None	Set a custom global sandbox.	#L849-L852