Method execute_tool

src/mcp_cli/tools/manager.py:709–833

Execute tool and return ToolCallResult. When middleware is enabled (default), CTP handles: - Retry with exponential backoff for transient errors - Circuit breaker pattern for failing servers - Rate limiting (if configured) OAuth handling: - If a tool

(
        self,
        tool_name: str,
        arguments: dict[str, Any],
        namespace: str | None = None,
        timeout: float | None = None,
        _oauth_retry: bool = False,
    )

                return False

    async def execute_tool(
        self,
        tool_name: str,
        arguments: dict[str, Any],
        namespace: str | None = None,
        timeout: float | None = None,
        _oauth_retry: bool = False,
    ) -> ToolCallResult:
        """Execute tool and return ToolCallResult.

        When middleware is enabled (default), CTP handles:
        - Retry with exponential backoff for transient errors
        - Circuit breaker pattern for failing servers
        - Rate limiting (if configured)

        OAuth handling:
        - If a tool fails with OAuth authorization error, automatically
          triggers the OAuth flow and retries the tool call once.
        """
        # Check if this is a dynamic tool
        if self.dynamic_tool_provider.is_dynamic_tool(tool_name):
            logger.info(f"Executing dynamic tool: {tool_name}")
            try:
                result = await self.dynamic_tool_provider.execute_dynamic_tool(
                    tool_name, arguments
                )
                return ToolCallResult(tool_name=tool_name, success=True, result=result)
            except Exception as e:
                error_msg = str(e)
                logger.error(f"Dynamic tool execution failed: {error_msg}")
                return ToolCallResult(
                    tool_name=tool_name, success=False, error=error_msg
                )

        # Regular MCP tool execution (middleware handles retries if enabled)
        if not self.stream_manager:
            return ToolCallResult(
                tool_name=tool_name, success=False, error="ToolManager not initialized"
            )

        try:
            result = await self.stream_manager.call_tool(
                tool_name=tool_name,
                arguments=arguments,
                server_name=namespace,
                timeout=timeout or self._get_server_timeout(namespace),
            )

            # Check if result contains an OAuth error (some servers return errors in content)
            # Only check results that are flagged as errors — scanning successful
            # payloads causes false positives (e.g. the number "401" in data).
            result_is_error = (
                getattr(result, "isError", False)
                or (isinstance(result, dict) and result.get("isError", False))
                or (isinstance(result, dict) and "error" in result)
            )
            result_str = str(result) if result_is_error else ""
            if result_str and _is_oauth_error(result_str) and not _oauth_retry: