Class ToolManager

src/mcp_cli/tools/manager.py:79–1314

Facade over chuk-tool-processor with mcp-cli specific features. Responsibilities: 1. Parse MCP config files and detect server types (HTTP/SSE/STDIO) 2. Integrate with mcp-cli's OAuth TokenManager 3. Filter and validate tools for LLM compatibility 4. Convert between chuk and

class ToolManager:
    """
    Facade over chuk-tool-processor with mcp-cli specific features.

    Responsibilities:
    1. Parse MCP config files and detect server types (HTTP/SSE/STDIO)
    2. Integrate with mcp-cli's OAuth TokenManager
    3. Filter and validate tools for LLM compatibility
    4. Convert between chuk and mcp-cli data models

    For direct tool operations, use the exposed stream_manager property.
    """

    def __init__(
        self,
        config_file: str,
        servers: list[str],
        server_names: dict[int, str | None] | None = None,
        tool_timeout: float | None = None,
        max_concurrency: int = 4,
        initialization_timeout: float = 120.0,
        runtime_config: RuntimeConfig | None = None,
        middleware_config: MiddlewareConfig | None = None,
        middleware_enabled: bool = DEFAULT_MIDDLEWARE_ENABLED,
    ):
        self.config_file = config_file
        self.servers = servers
        self.server_names = server_names or {}

        # Use runtime config for timeout management (type-safe!)
        self.runtime_config = runtime_config or load_runtime_config()

        # Tool timeout with priority: param > runtime_config
        if tool_timeout is not None:
            self.tool_timeout = tool_timeout
        else:
            self.tool_timeout = self.runtime_config.get_timeout(
                TimeoutType.TOOL_EXECUTION
            )

        # Initialization timeout with priority: param > runtime_config
        if initialization_timeout != 120.0:  # User provided non-default
            self.initialization_timeout = initialization_timeout
        else:
            self.initialization_timeout = self.runtime_config.get_timeout(
                TimeoutType.SERVER_INIT
            )

        self.max_concurrency = max_concurrency

        # Middleware configuration (retry, circuit breaker, rate limiting)
        self._middleware_enabled = middleware_enabled
        self._middleware_config = middleware_config

        logger.debug(
            f"ToolManager initialized with timeouts: "
            f"tool={self.tool_timeout}s, init={self.initialization_timeout}s, "
            f"middleware_enabled={middleware_enabled}"