Method call_tool

src/services/mcp/client.py:471–527

(
        self,
        tool_name: str,
        arguments: dict[str, Any] | None = None,
        meta: dict[str, Any] | None = None,
    )

        return tools

    async def call_tool(
        self,
        tool_name: str,
        arguments: dict[str, Any] | None = None,
        meta: dict[str, Any] | None = None,
    ) -> McpToolResult:
        params: dict[str, Any] = {
            "name": tool_name,
            "arguments": arguments or {},
        }
        if meta:
            params["_meta"] = meta

        # Phase 6a WI-6.1 (gap #8): on Streamable-HTTP session expiry,
        # the chapter §"Session Expiry Detection" specifies clear-cache +
        # retry-once. Mirrors typescript/src/services/mcp/client.ts: the
        # cache is cleared on detection so the next request reconnects
        # against a fresh session rather than reusing the expired one.
        try:
            result = await self._send_request("tools/call", params)
        except McpToolCallError as err:
            if not is_mcp_session_expired_error(err):
                # Regular tool error (invalid params, server-rejected, etc.) —
                # propagate untouched. No reconnect, no retry.
                raise
            await self._recover_from_session_expiry(err, tool_name=tool_name)
            # Retry once after the recovery routine returned (it either
            # reconnected, or another concurrent caller did, or recovery
            # failed and re-raised). A second session-expired here means
            # the server is unstable / the retry hit a fresh session that
            # already expired — propagate so we don't loop indefinitely.
            result = await self._send_request("tools/call", params)
        if not result or not isinstance(result, dict):
            return McpToolResult()

        is_error = result.get("isError", False)
        content = result.get("content", [])
        result_meta = result.get("_meta")
        structured = result.get("structuredContent")

        if is_error:
            error_text = ""
            for item in content:
                if isinstance(item, dict) and item.get("type") == "text":
                    error_text += item.get("text", "")
            raise McpToolCallError(
                error_text or "MCP tool returned an error",
                "MCP tool error",
                {"_meta": result_meta} if result_meta else None,
            )

        return McpToolResult(
            content=content,
            is_error=False,
            meta=result_meta,
            structured_content=structured,
        )