vllm_mlx.mcp.executor

Tool executor for handling tool calls from model responses.

View the complete module source at #L1-L500.

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.executor

Tool executor for handling tool calls from model responses.

vllm_mlx.mcp.executor.logger module-attribute

logger = logging.getLogger(__name__)

vllm_mlx.mcp.executor.ToolArgumentValidationError

Bases: Exception

Raised when tool arguments fail validation against schema.

vllm_mlx.mcp.executor.ToolExecutor

ToolExecutor(manager: MCPClientManager, max_parallel: int = 5, default_timeout: Optional[float] = None, validate_arguments: bool = True, sandbox: Optional[ToolSandbox] = None)

Handles execution of tool calls from model responses.

Provides utilities for: - Extracting tool calls from responses - Executing multiple tool calls (parallel or sequential) - Formatting results for conversation - Validating tool arguments against schemas

Initialize tool executor.

Parameters:

• manager (MCPClientManager) –

  MCP client manager

• max_parallel (int, default: 5 ) –

  Maximum parallel tool executions

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

  Default timeout for tool calls

• validate_arguments (bool, default: True ) –

  If True, validate arguments against tool schemas

• sandbox (Optional[ToolSandbox], default: None ) –

  Optional tool sandbox for security controls. Uses global if None.

Source code in vllm_mlx/mcp/executor.py

def __init__(
    self,
    manager: MCPClientManager,
    max_parallel: int = 5,
    default_timeout: Optional[float] = None,
    validate_arguments: bool = True,
    sandbox: Optional[ToolSandbox] = None,
):
    """
    Initialize tool executor.

    Args:
        manager: MCP client manager
        max_parallel: Maximum parallel tool executions
        default_timeout: Default timeout for tool calls
        validate_arguments: If True, validate arguments against tool schemas
        sandbox: Optional tool sandbox for security controls. Uses global if None.
    """
    self.manager = manager
    self.max_parallel = max_parallel
    self.default_timeout = default_timeout or manager.config.default_timeout
    self.validate_arguments = validate_arguments
    self.sandbox = sandbox or get_sandbox()

vllm_mlx.mcp.executor.ToolExecutor.manager instance-attribute

manager = manager

vllm_mlx.mcp.executor.ToolExecutor.max_parallel instance-attribute

max_parallel = max_parallel

vllm_mlx.mcp.executor.ToolExecutor.default_timeout instance-attribute

default_timeout = default_timeout or manager.config.default_timeout

vllm_mlx.mcp.executor.ToolExecutor.validate_arguments instance-attribute

validate_arguments = validate_arguments

vllm_mlx.mcp.executor.ToolExecutor.sandbox instance-attribute

sandbox = sandbox or get_sandbox()

vllm_mlx.mcp.executor.ToolExecutor.execute_tool_calls async

execute_tool_calls(tool_calls: List[Dict[str, Any]], parallel: bool = True) -> List[Tuple[MCPToolResult, str]]

Execute multiple tool calls.

Parameters:

• tool_calls (List[Dict[str, Any]]) –

  List of OpenAI tool call objects

• parallel (bool, default: True ) –

  Execute in parallel (True) or sequential (False)

Returns:

• List[Tuple[MCPToolResult, str]] –

  List of (MCPToolResult, tool_call_id) tuples

Source code in vllm_mlx/mcp/executor.py

async def execute_tool_calls(
    self,
    tool_calls: List[Dict[str, Any]],
    parallel: bool = True,
) -> List[Tuple[MCPToolResult, str]]:
    """
    Execute multiple tool calls.

    Args:
        tool_calls: List of OpenAI tool call objects
        parallel: Execute in parallel (True) or sequential (False)

    Returns:
        List of (MCPToolResult, tool_call_id) tuples
    """
    if not tool_calls:
        return []

    if parallel:
        return await self._execute_parallel(tool_calls)
    else:
        return await self._execute_sequential(tool_calls)

vllm_mlx.mcp.executor.ToolExecutor._get_tool_by_name

_get_tool_by_name(full_name: str) -> Optional[MCPTool]

Get a tool by its full name (server__tool or just tool).

Source code in vllm_mlx/mcp/executor.py

def _get_tool_by_name(self, full_name: str) -> Optional[MCPTool]:
    """Get a tool by its full name (server__tool or just tool)."""
    for tool in self.manager.get_all_tools():
        if tool.full_name == full_name:
            return tool
    # Try without server prefix
    if "__" not in full_name:
        for tool in self.manager.get_all_tools():
            if tool.name == full_name:
                return tool
    return None

vllm_mlx.mcp.executor.ToolExecutor._validate_tool_call

_validate_tool_call(tool_call: Dict[str, Any]) -> Optional[str]

Validate a tool call's arguments against the tool's schema.

Returns:

• Optional[str] –

  Error message if validation fails, None if valid

Source code in vllm_mlx/mcp/executor.py

def _validate_tool_call(self, tool_call: Dict[str, Any]) -> Optional[str]:
    """
    Validate a tool call's arguments against the tool's schema.

    Returns:
        Error message if validation fails, None if valid
    """
    if not self.validate_arguments:
        return None

    func = tool_call.get("function", {})
    name = func.get("name", "")
    arguments = func.get("arguments", {})

    # Parse arguments if string
    if isinstance(arguments, str):
        import json

        try:
            arguments = json.loads(arguments)
        except json.JSONDecodeError:
            return f"Invalid JSON in arguments for tool '{name}'"

    tool = self._get_tool_by_name(name)
    if not tool:
        return None  # Let execution handle missing tool

    try:
        validate_tool_arguments(tool, arguments, strict=True)
        return None
    except ToolArgumentValidationError as e:
        return str(e)

vllm_mlx.mcp.executor.ToolExecutor._validate_sandbox

_validate_sandbox(tool_name: str, server_name: str, arguments: Dict[str, Any]) -> Optional[str]

Validate tool execution against sandbox policy.

Returns:

• Optional[str] –

  Error message if blocked, None if allowed

Source code in vllm_mlx/mcp/executor.py

def _validate_sandbox(
    self,
    tool_name: str,
    server_name: str,
    arguments: Dict[str, Any],
) -> Optional[str]:
    """
    Validate tool execution against sandbox policy.

    Returns:
        Error message if blocked, None if allowed
    """
    try:
        self.sandbox.validate_tool_execution(tool_name, server_name, arguments)
        return None
    except MCPSecurityError as e:
        return str(e)

vllm_mlx.mcp.executor.ToolExecutor._get_server_for_tool

_get_server_for_tool(full_name: str) -> str

Extract server name from full tool name or find it.

Source code in vllm_mlx/mcp/executor.py

def _get_server_for_tool(self, full_name: str) -> str:
    """Extract server name from full tool name or find it."""
    if "__" in full_name:
        return full_name.split("__")[0]
    # Find which server has this tool
    for tool in self.manager.get_all_tools():
        if tool.name == full_name:
            return tool.server_name
    return "unknown"

vllm_mlx.mcp.executor.ToolExecutor._execute_parallel async

_execute_parallel(tool_calls: List[Dict[str, Any]]) -> List[Tuple[MCPToolResult, str]]

Execute tool calls in parallel with concurrency limit.

Source code in vllm_mlx/mcp/executor.py

async def _execute_parallel(
    self,
    tool_calls: List[Dict[str, Any]],
) -> List[Tuple[MCPToolResult, str]]:
    """Execute tool calls in parallel with concurrency limit."""
    semaphore = asyncio.Semaphore(self.max_parallel)

    async def execute_with_semaphore(tool_call: Dict[str, Any]):
        async with semaphore:
            func = tool_call.get("function", {})
            name = func.get("name", "")
            arguments = func.get("arguments", {})
            call_id = tool_call.get("id", "")

            # Parse arguments if string
            if isinstance(arguments, str):
                import json

                try:
                    arguments = json.loads(arguments)
                except json.JSONDecodeError:
                    arguments = {}

            server_name = self._get_server_for_tool(name)
            tool_name = name.split("__")[-1] if "__" in name else name

            # Validate arguments before execution
            validation_error = self._validate_tool_call(tool_call)
            if validation_error:
                self.sandbox.record_execution(
                    tool_name,
                    server_name,
                    arguments,
                    success=False,
                    error_message=validation_error,
                )
                return (
                    MCPToolResult(
                        tool_name=name,
                        content=None,
                        is_error=True,
                        error_message=validation_error,
                    ),
                    call_id,
                )

            # Validate sandbox policy
            sandbox_error = self._validate_sandbox(
                tool_name, server_name, arguments
            )
            if sandbox_error:
                self.sandbox.record_execution(
                    tool_name,
                    server_name,
                    arguments,
                    success=False,
                    error_message=sandbox_error,
                )
                return (
                    MCPToolResult(
                        tool_name=name,
                        content=None,
                        is_error=True,
                        error_message=sandbox_error,
                    ),
                    call_id,
                )

            # Execute with timing for audit
            start_time = time.time()
            result = await self.manager.execute_tool_call(
                tool_call,
                timeout=self.default_timeout,
            )
            execution_time_ms = (time.time() - start_time) * 1000

            # Record execution in audit log
            self.sandbox.record_execution(
                tool_name,
                server_name,
                arguments,
                success=not result.is_error,
                error_message=result.error_message,
                execution_time_ms=execution_time_ms,
            )

            return (result, call_id)

    tasks = [execute_with_semaphore(tc) for tc in tool_calls]
    results = await asyncio.gather(*tasks, return_exceptions=True)

    # Handle exceptions
    processed = []
    for i, result in enumerate(results):
        call_id = tool_calls[i].get("id", "")
        if isinstance(result, Exception):
            processed.append(
                (
                    MCPToolResult(
                        tool_name=tool_calls[i].get("function", {}).get("name", ""),
                        content=None,
                        is_error=True,
                        error_message=str(result),
                    ),
                    call_id,
                )
            )
        else:
            processed.append(result)

    return processed

vllm_mlx.mcp.executor.ToolExecutor._execute_sequential async

_execute_sequential(tool_calls: List[Dict[str, Any]]) -> List[Tuple[MCPToolResult, str]]

Execute tool calls sequentially.

Source code in vllm_mlx/mcp/executor.py

async def _execute_sequential(
    self,
    tool_calls: List[Dict[str, Any]],
) -> List[Tuple[MCPToolResult, str]]:
    """Execute tool calls sequentially."""
    results = []
    for tool_call in tool_calls:
        func = tool_call.get("function", {})
        name = func.get("name", "")
        arguments = func.get("arguments", {})
        call_id = tool_call.get("id", "")

        # Parse arguments if string
        if isinstance(arguments, str):
            import json

            try:
                arguments = json.loads(arguments)
            except json.JSONDecodeError:
                arguments = {}

        server_name = self._get_server_for_tool(name)
        tool_name = name.split("__")[-1] if "__" in name else name

        # Validate arguments before execution
        validation_error = self._validate_tool_call(tool_call)
        if validation_error:
            self.sandbox.record_execution(
                tool_name,
                server_name,
                arguments,
                success=False,
                error_message=validation_error,
            )
            results.append(
                (
                    MCPToolResult(
                        tool_name=name,
                        content=None,
                        is_error=True,
                        error_message=validation_error,
                    ),
                    call_id,
                )
            )
            continue

        # Validate sandbox policy
        sandbox_error = self._validate_sandbox(tool_name, server_name, arguments)
        if sandbox_error:
            self.sandbox.record_execution(
                tool_name,
                server_name,
                arguments,
                success=False,
                error_message=sandbox_error,
            )
            results.append(
                (
                    MCPToolResult(
                        tool_name=name,
                        content=None,
                        is_error=True,
                        error_message=sandbox_error,
                    ),
                    call_id,
                )
            )
            continue

        try:
            # Execute with timing for audit
            start_time = time.time()
            result = await self.manager.execute_tool_call(
                tool_call,
                timeout=self.default_timeout,
            )
            execution_time_ms = (time.time() - start_time) * 1000

            # Record execution in audit log
            self.sandbox.record_execution(
                tool_name,
                server_name,
                arguments,
                success=not result.is_error,
                error_message=result.error_message,
                execution_time_ms=execution_time_ms,
            )
            results.append((result, call_id))
        except Exception as e:
            self.sandbox.record_execution(
                tool_name,
                server_name,
                arguments,
                success=False,
                error_message=str(e),
            )
            results.append(
                (
                    MCPToolResult(
                        tool_name=name,
                        content=None,
                        is_error=True,
                        error_message=str(e),
                    ),
                    call_id,
                )
            )
    return results

vllm_mlx.mcp.executor.ToolExecutor.execute_and_format async

execute_and_format(tool_calls: List[Dict[str, Any]], parallel: bool = True) -> List[Dict[str, Any]]

Execute tool calls and format results as messages.

Parameters:

• tool_calls (List[Dict[str, Any]]) –

  List of OpenAI tool call objects

• parallel (bool, default: True ) –

  Execute in parallel

Returns:

• List[Dict[str, Any]] –

  List of tool result messages ready for conversation

Source code in vllm_mlx/mcp/executor.py

async def execute_and_format(
    self,
    tool_calls: List[Dict[str, Any]],
    parallel: bool = True,
) -> List[Dict[str, Any]]:
    """
    Execute tool calls and format results as messages.

    Args:
        tool_calls: List of OpenAI tool call objects
        parallel: Execute in parallel

    Returns:
        List of tool result messages ready for conversation
    """
    results = await self.execute_tool_calls(tool_calls, parallel)
    return [format_tool_result(result, call_id) for result, call_id in results]

vllm_mlx.mcp.executor.ToolExecutor.extract_and_validate

extract_and_validate(response: Dict[str, Any]) -> Tuple[List[Dict[str, Any]], bool]

Extract tool calls from response and validate them.

Parameters:

• response (Dict[str, Any]) –

  Model response in OpenAI format

Returns:

• Tuple[List[Dict[str, Any]], bool] –

  Tuple of (tool_calls, all_valid)

Source code in vllm_mlx/mcp/executor.py

def extract_and_validate(
    self,
    response: Dict[str, Any],
) -> Tuple[List[Dict[str, Any]], bool]:
    """
    Extract tool calls from response and validate them.

    Args:
        response: Model response in OpenAI format

    Returns:
        Tuple of (tool_calls, all_valid)
    """
    tool_calls = extract_tool_calls(response)

    if not tool_calls:
        return [], True

    # Validate each tool call
    all_valid = True
    for tc in tool_calls:
        func = tc.get("function", {})
        name = func.get("name", "")

        # Check if tool exists
        if not self._tool_exists(name):
            logger.warning(f"Tool '{name}' not found in any MCP server")
            all_valid = False

    return tool_calls, all_valid

vllm_mlx.mcp.executor.ToolExecutor._tool_exists

_tool_exists(full_name: str) -> bool

Check if a tool exists in any connected server.

Source code in vllm_mlx/mcp/executor.py

def _tool_exists(self, full_name: str) -> bool:
    """Check if a tool exists in any connected server."""
    # Check by full name
    for tool in self.manager.get_all_tools():
        if tool.full_name == full_name:
            return True

    # Check by just tool name (without server prefix)
    if "__" not in full_name:
        for tool in self.manager.get_all_tools():
            if tool.name == full_name:
                return True

    return False

vllm_mlx.mcp.executor.validate_tool_arguments

validate_tool_arguments(tool: MCPTool, arguments: Dict[str, Any], strict: bool = True) -> None

Validate tool arguments against the tool's input schema.

Parameters:

• tool (MCPTool) –

  The MCP tool with input_schema

• arguments (Dict[str, Any]) –

  Arguments to validate

• strict (bool, default: True ) –

  If True, raise exception on validation failure

Raises:

• ToolArgumentValidationError –

  If validation fails and strict=True

Source code in vllm_mlx/mcp/executor.py

def validate_tool_arguments(
    tool: MCPTool,
    arguments: Dict[str, Any],
    strict: bool = True,
) -> None:
    """
    Validate tool arguments against the tool's input schema.

    Args:
        tool: The MCP tool with input_schema
        arguments: Arguments to validate
        strict: If True, raise exception on validation failure

    Raises:
        ToolArgumentValidationError: If validation fails and strict=True
    """
    schema = tool.input_schema
    if not schema:
        logger.debug(
            f"Tool '{tool.full_name}' has no input schema, skipping validation"
        )
        return

    try:
        jsonschema.validate(instance=arguments, schema=schema)
        logger.debug(f"Tool '{tool.full_name}' arguments validated successfully")
    except ValidationError as e:
        error_msg = (
            f"Tool '{tool.full_name}' argument validation failed: {e.message}. "
            f"Path: {'.'.join(str(p) for p in e.path) or 'root'}"
        )
        logger.warning(error_msg)
        if strict:
            raise ToolArgumentValidationError(error_msg) from e

vllm_mlx.mcp.executor.execute_single_tool async

execute_single_tool(manager: MCPClientManager, tool_name: str, arguments: Dict[str, Any], timeout: Optional[float] = None) -> MCPToolResult

Convenience function to execute a single tool.

Parameters:

• manager (MCPClientManager) –

  MCP client manager

• tool_name (str) –

  Full tool name (server__tool)

• arguments (Dict[str, Any]) –

  Tool arguments

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

  Optional timeout

Returns:

• MCPToolResult –

  MCPToolResult

Source code in vllm_mlx/mcp/executor.py

async def execute_single_tool(
    manager: MCPClientManager,
    tool_name: str,
    arguments: Dict[str, Any],
    timeout: Optional[float] = None,
) -> MCPToolResult:
    """
    Convenience function to execute a single tool.

    Args:
        manager: MCP client manager
        tool_name: Full tool name (server__tool)
        arguments: Tool arguments
        timeout: Optional timeout

    Returns:
        MCPToolResult
    """
    return await manager.execute_tool(tool_name, arguments, timeout)

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.executor.ToolArgumentValidationError · class

vllm_mlx.mcp.executor.ToolArgumentValidationError()

Raised when tool arguments fail validation against schema.

Parameters

This callable has no explicit inputs.

Returns

• Constructs: vllm_mlx.mcp.executor.ToolArgumentValidationError

Exceptions and behavior

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

View source #L22-L25.

vllm_mlx.mcp.executor.validate_tool_arguments · function

vllm_mlx.mcp.executor.validate_tool_arguments(tool: MCPTool, arguments: Dict[str, Any], strict: bool = True) -> None

Validate tool arguments against the tool's input schema.

Parameters

Name	Type	Required	Default	Description
tool	MCPTool	yes	none	The MCP tool with input_schema
arguments	Dict[str, Any]	yes	none	Arguments to validate
strict	bool	no	True	If True, raise exception on validation failure

Returns

• Type: None
• Direct return expressions: None

Exceptions and behavior

Function validate_tool_arguments calls logger.debug, jsonschema.validate, '.'.join, str; can raise ToolArgumentValidationError; returns None. Directly raised exceptions: ToolArgumentValidationError.

View source #L28-L61.

vllm_mlx.mcp.executor.ToolExecutor · class

vllm_mlx.mcp.executor.ToolExecutor(manager: MCPClientManager, max_parallel: int = 5, default_timeout: Optional[float] = None, validate_arguments: bool = True, sandbox: Optional[ToolSandbox] = None)

Handles execution of tool calls from model responses.

Parameters

Name	Type	Required	Default	Description
manager	MCPClientManager	yes	none	MCP client manager
max_parallel	int	no	5	Maximum parallel tool executions
default_timeout	Optional[float]	no	None	Default timeout for tool calls
validate_arguments	bool	no	True	If True, validate arguments against tool schemas
sandbox	Optional[ToolSandbox]	no	None	Optional tool sandbox for security controls. Uses global if None.

Returns

• Constructs: vllm_mlx.mcp.executor.ToolExecutor

Exceptions and behavior

Class ToolExecutor declares 11 direct member(s). No direct raise statement appears in this definition.

View source #L64-L479.

vllm_mlx.mcp.executor.ToolExecutor.__init__ · method

vllm_mlx.mcp.executor.ToolExecutor.__init__(manager: MCPClientManager, max_parallel: int = 5, default_timeout: Optional[float] = None, validate_arguments: bool = True, sandbox: Optional[ToolSandbox] = None) -> not annotated

Initialize tool executor.

Parameters

Name	Type	Required	Default	Description
manager	MCPClientManager	yes	none	MCP client manager
max_parallel	int	no	5	Maximum parallel tool executions
default_timeout	Optional[float]	no	None	Default timeout for tool calls
validate_arguments	bool	no	True	If True, validate arguments against tool schemas
sandbox	Optional[ToolSandbox]	no	None	Optional tool sandbox for security controls. Uses global if None.

Returns

• Type: not annotated

Exceptions and behavior

Method ToolExecutor.__init__ updates self.manager, self.max_parallel, self.default_timeout, self.validate_arguments; calls get_sandbox. No direct raise statement appears in this definition.

View source #L75-L97.

vllm_mlx.mcp.executor.ToolExecutor.execute_tool_calls · method

async vllm_mlx.mcp.executor.ToolExecutor.execute_tool_calls(tool_calls: List[Dict[str, Any]], parallel: bool = True) -> List[Tuple[MCPToolResult, str]]

Execute multiple tool calls.

Parameters

Name	Type	Required	Default	Description
tool_calls	List[Dict[str, Any]]	yes	none	List of OpenAI tool call objects
parallel	bool	no	True	Execute in parallel (True) or sequential (False)

Returns

• Type: List[Tuple[MCPToolResult, str]]
• Direct return expressions: []; await self._execute_parallel(tool_calls); await self._execute_sequential(tool_calls)

Exceptions and behavior

Method ToolExecutor.execute_tool_calls calls self._execute_parallel, self._execute_sequential; awaits asynchronous work; has 3 explicit return paths. No direct raise statement appears in this definition.

View source #L99-L120.

vllm_mlx.mcp.executor.ToolExecutor._get_tool_by_name · method

vllm_mlx.mcp.executor.ToolExecutor._get_tool_by_name(full_name: str) -> Optional[MCPTool]

Get a tool by its full name (server__tool or just tool).

Parameters

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

Returns

• Type: Optional[MCPTool]
• Direct return expressions: tool; None

Exceptions and behavior

Method ToolExecutor._get_tool_by_name calls self.manager.get_all_tools; has 2 explicit return paths. No direct raise statement appears in this definition.

View source #L122-L132.

vllm_mlx.mcp.executor.ToolExecutor._validate_tool_call · method

vllm_mlx.mcp.executor.ToolExecutor._validate_tool_call(tool_call: Dict[str, Any]) -> Optional[str]

Validate a tool call's arguments against the tool's schema.

Parameters

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

Returns

• Type: Optional[str]
• Direct return expressions: None; f"Invalid JSON in arguments for tool '{name}'"; str(e)

Exceptions and behavior

Method ToolExecutor._validate_tool_call calls tool_call.get, func.get, isinstance, json.loads; has 3 explicit return paths. No direct raise statement appears in this definition.

View source #L134-L165.

vllm_mlx.mcp.executor.ToolExecutor._validate_sandbox · method

vllm_mlx.mcp.executor.ToolExecutor._validate_sandbox(tool_name: str, server_name: str, arguments: Dict[str, Any]) -> Optional[str]

Validate tool execution against sandbox policy.

Parameters

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

Returns

• Type: Optional[str]
• Direct return expressions: None; str(e)

Exceptions and behavior

Method ToolExecutor._validate_sandbox calls self.sandbox.validate_tool_execution, str; has 2 explicit return paths. No direct raise statement appears in this definition.

View source #L167-L183.

vllm_mlx.mcp.executor.ToolExecutor._get_server_for_tool · method

vllm_mlx.mcp.executor.ToolExecutor._get_server_for_tool(full_name: str) -> str

Extract server name from full tool name or find it.

Parameters

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

Returns

• Type: str
• Direct return expressions: full_name.split('__')[0]; tool.server_name; 'unknown'

Exceptions and behavior

Method ToolExecutor._get_server_for_tool calls full_name.split, self.manager.get_all_tools; has 3 explicit return paths. No direct raise statement appears in this definition.

View source #L185-L193.

vllm_mlx.mcp.executor.ToolExecutor._execute_parallel · method

async vllm_mlx.mcp.executor.ToolExecutor._execute_parallel(tool_calls: List[Dict[str, Any]]) -> List[Tuple[MCPToolResult, str]]

Execute tool calls in parallel with concurrency limit.

Parameters

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

Returns

• Type: List[Tuple[MCPToolResult, str]]
• Direct return expressions: processed

Exceptions and behavior

Method ToolExecutor._execute_parallel calls asyncio.Semaphore, execute_with_semaphore, asyncio.gather, enumerate; awaits asynchronous work; returns processed. No direct raise statement appears in this definition.

View source #L195-L305.

vllm_mlx.mcp.executor.ToolExecutor._execute_parallel.execute_with_semaphore · nested function

async vllm_mlx.mcp.executor.ToolExecutor._execute_parallel.execute_with_semaphore(tool_call: Dict[str, Any]) -> not annotated

Nested Function ToolExecutor._execute_parallel.execute_with_semaphore calls tool_call.get, func.get, isinstance, json.loads; awaits asynchronous work; has 3 explicit return paths.

Parameters

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

Returns

• Type: not annotated
• Direct return expressions: (MCPToolResult(tool_name=name, content=None, is_error=True, error_message=validation_error), call_id); (MCPToolResult(tool_name=name, content=None, is_error=True, error_message=sandbox_error), call_id); (result, call_id)

Exceptions and behavior

Nested Function ToolExecutor._execute_parallel.execute_with_semaphore calls tool_call.get, func.get, isinstance, json.loads; awaits asynchronous work; has 3 explicit return paths. No direct raise statement appears in this definition.

View source #L202-L281.

vllm_mlx.mcp.executor.ToolExecutor._execute_sequential · method

async vllm_mlx.mcp.executor.ToolExecutor._execute_sequential(tool_calls: List[Dict[str, Any]]) -> List[Tuple[MCPToolResult, str]]

Execute tool calls sequentially.

Parameters

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

Returns

• Type: List[Tuple[MCPToolResult, str]]
• Direct return expressions: results

Exceptions and behavior

Method ToolExecutor._execute_sequential calls tool_call.get, func.get, isinstance, json.loads; awaits asynchronous work; returns results. No direct raise statement appears in this definition.

View source #L307-L415.

vllm_mlx.mcp.executor.ToolExecutor.execute_and_format · method

async vllm_mlx.mcp.executor.ToolExecutor.execute_and_format(tool_calls: List[Dict[str, Any]], parallel: bool = True) -> List[Dict[str, Any]]

Execute tool calls and format results as messages.

Parameters

Name	Type	Required	Default	Description
tool_calls	List[Dict[str, Any]]	yes	none	List of OpenAI tool call objects
parallel	bool	no	True	Execute in parallel

Returns

• Type: List[Dict[str, Any]]
• Direct return expressions: [format_tool_result(result, call_id) for result, call_id in results]

Exceptions and behavior

Method ToolExecutor.execute_and_format calls self.execute_tool_calls, format_tool_result; awaits asynchronous work; returns [format_tool_result(result, call_id) for result, call_id in results]. No direct raise statement appears in this definition.

View source #L417-L433.

vllm_mlx.mcp.executor.ToolExecutor.extract_and_validate · method

vllm_mlx.mcp.executor.ToolExecutor.extract_and_validate(response: Dict[str, Any]) -> Tuple[List[Dict[str, Any]], bool]

Extract tool calls from response and validate them.

Parameters

Name	Type	Required	Default	Description
response	Dict[str, Any]	yes	none	Model response in OpenAI format

Returns

• Type: Tuple[List[Dict[str, Any]], bool]
• Direct return expressions: ([], True); (tool_calls, all_valid)

Exceptions and behavior

Method ToolExecutor.extract_and_validate calls extract_tool_calls, tc.get, func.get, self._tool_exists; has 2 explicit return paths. No direct raise statement appears in this definition.

View source #L435-L464.

vllm_mlx.mcp.executor.ToolExecutor._tool_exists · method

vllm_mlx.mcp.executor.ToolExecutor._tool_exists(full_name: str) -> bool

Check if a tool exists in any connected server.

Parameters

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

Returns

• Type: bool
• Direct return expressions: True; False

Exceptions and behavior

Method ToolExecutor._tool_exists calls self.manager.get_all_tools; has 2 explicit return paths. No direct raise statement appears in this definition.

View source #L466-L479.

vllm_mlx.mcp.executor.execute_single_tool · function

async vllm_mlx.mcp.executor.execute_single_tool(manager: MCPClientManager, tool_name: str, arguments: Dict[str, Any], timeout: Optional[float] = None) -> MCPToolResult

Convenience function to execute a single tool.

Parameters

Name	Type	Required	Default	Description
manager	MCPClientManager	yes	none	MCP client manager
tool_name	str	yes	none	Full tool name (server__tool)
arguments	Dict[str, Any]	yes	none	Tool arguments
timeout	Optional[float]	no	None	Optional timeout

Returns

• Type: MCPToolResult
• Direct return expressions: await manager.execute_tool(tool_name, arguments, timeout)

Exceptions and behavior

Function execute_single_tool calls manager.execute_tool; awaits asynchronous work; returns await manager.execute_tool(tool_name, arguments, timeout). No direct raise statement appears in this definition.

View source #L482-L500.

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
ToolArgumentValidationError	class	ToolArgumentValidationError()	Raised when tool arguments fail validation against schema.	#L22-L25
validate_tool_arguments	function	validate_tool_arguments(tool: MCPTool, arguments: Dict[str, Any], strict: bool = True) -> None	Validate tool arguments against the tool's input schema.	#L28-L61
ToolExecutor	class	ToolExecutor(manager: MCPClientManager, max_parallel: int = 5, default_timeout: Optional[float] = None, validate_arguments: bool = True, sandbox: Optional[ToolSandbox] = None)	Handles execution of tool calls from model responses.	#L64-L479
ToolExecutor.__init__	method	ToolExecutor.__init__(manager: MCPClientManager, max_parallel: int = 5, default_timeout: Optional[float] = None, validate_arguments: bool = True, sandbox: Optional[ToolSandbox] = None) -> not annotated	Initialize tool executor.	#L75-L97
ToolExecutor.execute_tool_calls	method	async ToolExecutor.execute_tool_calls(tool_calls: List[Dict[str, Any]], parallel: bool = True) -> List[Tuple[MCPToolResult, str]]	Execute multiple tool calls.	#L99-L120
ToolExecutor._get_tool_by_name	method	ToolExecutor._get_tool_by_name(full_name: str) -> Optional[MCPTool]	Get a tool by its full name (server__tool or just tool).	#L122-L132
ToolExecutor._validate_tool_call	method	ToolExecutor._validate_tool_call(tool_call: Dict[str, Any]) -> Optional[str]	Validate a tool call's arguments against the tool's schema.	#L134-L165
ToolExecutor._validate_sandbox	method	ToolExecutor._validate_sandbox(tool_name: str, server_name: str, arguments: Dict[str, Any]) -> Optional[str]	Validate tool execution against sandbox policy.	#L167-L183
ToolExecutor._get_server_for_tool	method	ToolExecutor._get_server_for_tool(full_name: str) -> str	Extract server name from full tool name or find it.	#L185-L193
ToolExecutor._execute_parallel	method	async ToolExecutor._execute_parallel(tool_calls: List[Dict[str, Any]]) -> List[Tuple[MCPToolResult, str]]	Execute tool calls in parallel with concurrency limit.	#L195-L305
ToolExecutor._execute_parallel.execute_with_semaphore	nested function	async ToolExecutor._execute_parallel.execute_with_semaphore(tool_call: Dict[str, Any]) -> not annotated	Nested Function ToolExecutor._execute_parallel.execute_with_semaphore calls tool_call.get, func.get, isinstance, json.loads; awaits asynchronous work; has 3 explicit return paths.	#L202-L281
ToolExecutor._execute_sequential	method	async ToolExecutor._execute_sequential(tool_calls: List[Dict[str, Any]]) -> List[Tuple[MCPToolResult, str]]	Execute tool calls sequentially.	#L307-L415
ToolExecutor.execute_and_format	method	async ToolExecutor.execute_and_format(tool_calls: List[Dict[str, Any]], parallel: bool = True) -> List[Dict[str, Any]]	Execute tool calls and format results as messages.	#L417-L433
ToolExecutor.extract_and_validate	method	ToolExecutor.extract_and_validate(response: Dict[str, Any]) -> Tuple[List[Dict[str, Any]], bool]	Extract tool calls from response and validate them.	#L435-L464
ToolExecutor._tool_exists	method	ToolExecutor._tool_exists(full_name: str) -> bool	Check if a tool exists in any connected server.	#L466-L479
execute_single_tool	function	async execute_single_tool(manager: MCPClientManager, tool_name: str, arguments: Dict[str, Any], timeout: Optional[float] = None) -> MCPToolResult	Convenience function to execute a single tool.	#L482-L500