feat(auth): introduce iFlow provider integration via OAuth

Implement full integration for iFlow, supporting automatic credential discovery and management using the OAuth 2.0 authorization code flow.

- Added `IFlowAuthBase` to handle the interactive OAuth flow and token refresh using a local `aiohttp` callback server.
- Enabled automatic discovery of iFlow credentials from the standard `~/.iflow/` directory.
- Ensures the dedicated API key (derived from user info after OAuth) is always used for API calls and refreshed proactively.
- Updated `requirements.txt` to include `aiohttp` for the local server functionality.
- Refactored Qwen providers to clean tool schemas and standardize streaming assembly logic for better robustness.

README.md

@@ -28,7 +28,7 @@ This project provides a powerful solution for developers building complex applic
 -   **Resilient Performance**: A global timeout on all requests prevents your application from hanging on unresponsive provider APIs.
 -   **Efficient Concurrency**: Maximizes throughput by allowing a single API key to handle multiple concurrent requests to different models.
 -   **Intelligent Key Management**: Optimizes request distribution across your pool of keys by selecting the best available one for each call.
--   **Automated OAuth Discovery**: Automatically discovers, validates, and manages OAuth credentials from standard provider directories (e.g., `~/.gemini/`, `~/.qwen/`). No manual `.env` configuration is required for supported providers.
+-   **Automated OAuth Discovery**: Automatically discovers, validates, and manages OAuth credentials from standard provider directories (e.g., `~/.gemini/`, `~/.qwen/`, `~/.iflow/`). No manual `.env` configuration is required for supported providers.
 -   **Duplicate Credential Detection**: Intelligently detects if multiple local credential files belong to the same user account and logs a warning, preventing redundancy in your key pool.
 -   **Escalating Per-Model Cooldowns**: If a key fails for a specific model, it's placed on a temporary, escalating cooldown for that model, allowing it to be used with others.
 -   **Automatic Daily Resets**: Cooldowns and usage statistics are automatically reset daily, making the system self-maintaining.
@@ -115,7 +115,7 @@ The proxy supports two types of credentials:
 
 For many providers, **no configuration is necessary**. The proxy automatically discovers and manages credentials from their default locations:
 -   **API Keys**: Scans your environment variables for keys matching the format `PROVIDER_API_KEY_1` (e.g., `GEMINI_API_KEY_1`).
--   **OAuth Credentials**: Scans default system directories (e.g., `~/.gemini/`, `~/.qwen/`) for all `*.json` credential files.
+-   **OAuth Credentials**: Scans default system directories (e.g., `~/.gemini/`, `~/.qwen/`, `~/.iflow/`) for all `*.json` credential files.
 
 You only need to create a `.env` file to set your `PROXY_API_KEY` and to override or add credentials if the automatic discovery doesn't suit your needs.
 

requirements.txt

@@ -14,6 +14,7 @@ litellm
 filelock
 httpx
 aiofiles
+aiohttp
 
 colorlog
 

src/rotator_library/credential_manager.py

@@ -13,6 +13,7 @@ OAUTH_BASE_DIR.mkdir(exist_ok=True)
 DEFAULT_OAUTH_DIRS = {
     "gemini_cli": Path.home() / ".gemini",
     "qwen_code": Path.home() / ".qwen",
+    "iflow": Path.home() / ".iflow",
     # Add other providers like 'claude' here if they have a standard CLI path
 }
 

src/rotator_library/credential_tool.py

@@ -80,7 +80,7 @@ async def setup_api_key():
     }
 
     # Discover custom providers and add them to the list
-    oauth_providers = {'gemini_cli', 'qwen_code'}
+    oauth_providers = {'gemini_cli', 'qwen_code', 'iflow'}
     discovered_providers = {
         p.replace('_', ' ').title(): p.upper() + "_API_KEY"
         for p in PROVIDER_PLUGINS.keys()
@@ -222,7 +222,8 @@ async def main():
             available_providers = get_available_providers()
             oauth_friendly_names = {
                 "gemini_cli": "Gemini CLI (OAuth)",
-                "qwen_code": "Qwen Code (OAuth)"
+                "qwen_code": "Qwen Code (OAuth)",
+                "iflow": "iFlow (OAuth)"
             }
             
             provider_text = Text()

src/rotator_library/provider_factory.py

@@ -2,10 +2,12 @@
 
 from .providers.gemini_auth_base import GeminiAuthBase
 from .providers.qwen_auth_base import QwenAuthBase
+from .providers.iflow_auth_base import IFlowAuthBase
 
 PROVIDER_MAP = {
     "gemini_cli": GeminiAuthBase,
     "qwen_code": QwenAuthBase,
+    "iflow": IFlowAuthBase,
 }
 
 def get_provider_auth_class(provider_name: str):

src/rotator_library/providers/iflow_auth_base.py

@@ -0,0 +1,543 @@
+# src/rotator_library/providers/iflow_auth_base.py
+
+import secrets
+import base64
+import json
+import time
+import asyncio
+import logging
+import webbrowser
+import socket
+from pathlib import Path
+from typing import Dict, Any, Tuple, Union, Optional
+from urllib.parse import urlencode, parse_qs, urlparse
+
+import httpx
+from aiohttp import web
+from rich.console import Console
+from rich.panel import Panel
+from rich.prompt import Prompt
+from rich.text import Text
+
+lib_logger = logging.getLogger('rotator_library')
+
+# OAuth endpoints and credentials from Go example
+IFLOW_OAUTH_AUTHORIZE_ENDPOINT = "https://iflow.cn/oauth"
+IFLOW_OAUTH_TOKEN_ENDPOINT = "https://iflow.cn/oauth/token"
+IFLOW_USER_INFO_ENDPOINT = "https://iflow.cn/api/oauth/getUserInfo"
+IFLOW_SUCCESS_REDIRECT_URL = "https://iflow.cn/oauth/success"
+IFLOW_ERROR_REDIRECT_URL = "https://iflow.cn/oauth/error"
+
+# Client credentials provided by iFlow
+IFLOW_CLIENT_ID = "10009311001"
+IFLOW_CLIENT_SECRET = "4Z3YjXycVsQvyGF1etiNlIBB4RsqSDtW"
+
+# Local callback server port
+CALLBACK_PORT = 11451
+
+# Refresh tokens 24 hours before expiry (from Go example)
+REFRESH_EXPIRY_BUFFER_SECONDS = 24 * 60 * 60
+
+console = Console()
+
+
+class OAuthCallbackServer:
+    """
+    Minimal HTTP server for handling iFlow OAuth callbacks.
+    Based on the Go example's oauth_server.go implementation.
+    """
+
+    def __init__(self, port: int = CALLBACK_PORT):
+        self.port = port
+        self.app = web.Application()
+        self.runner: Optional[web.AppRunner] = None
+        self.site: Optional[web.TCPSite] = None
+        self.result_future: Optional[asyncio.Future] = None
+        self.expected_state: Optional[str] = None
+
+    def _is_port_available(self) -> bool:
+        """Checks if the callback port is available."""
+        try:
+            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+            sock.bind(('', self.port))
+            sock.close()
+            return True
+        except OSError:
+            return False
+
+    async def start(self, expected_state: str):
+        """Starts the OAuth callback server."""
+        if not self._is_port_available():
+            raise RuntimeError(f"Port {self.port} is already in use")
+
+        self.expected_state = expected_state
+        self.result_future = asyncio.Future()
+
+        # Setup route
+        self.app.router.add_get('/oauth2callback', self._handle_callback)
+
+        # Start server
+        self.runner = web.AppRunner(self.app)
+        await self.runner.setup()
+        self.site = web.TCPSite(self.runner, 'localhost', self.port)
+        await self.site.start()
+
+        lib_logger.debug(f"iFlow OAuth callback server started on port {self.port}")
+
+    async def stop(self):
+        """Stops the OAuth callback server."""
+        if self.site:
+            await self.site.stop()
+        if self.runner:
+            await self.runner.cleanup()
+        lib_logger.debug("iFlow OAuth callback server stopped")
+
+    async def _handle_callback(self, request: web.Request) -> web.Response:
+        """Handles the OAuth callback request."""
+        query = request.query
+
+        # Check for error parameter
+        if 'error' in query:
+            error = query.get('error', 'unknown_error')
+            lib_logger.error(f"iFlow OAuth callback received error: {error}")
+            if not self.result_future.done():
+                self.result_future.set_exception(ValueError(f"OAuth error: {error}"))
+            return web.Response(status=302, headers={'Location': IFLOW_ERROR_REDIRECT_URL})
+
+        # Check for authorization code
+        code = query.get('code')
+        if not code:
+            lib_logger.error("iFlow OAuth callback missing authorization code")
+            if not self.result_future.done():
+                self.result_future.set_exception(ValueError("Missing authorization code"))
+            return web.Response(status=302, headers={'Location': IFLOW_ERROR_REDIRECT_URL})
+
+        # Validate state parameter
+        state = query.get('state', '')
+        if state != self.expected_state:
+            lib_logger.error(f"iFlow OAuth state mismatch. Expected: {self.expected_state}, Got: {state}")
+            if not self.result_future.done():
+                self.result_future.set_exception(ValueError("State parameter mismatch"))
+            return web.Response(status=302, headers={'Location': IFLOW_ERROR_REDIRECT_URL})
+
+        # Success - set result and redirect to success page
+        if not self.result_future.done():
+            self.result_future.set_result(code)
+
+        return web.Response(status=302, headers={'Location': IFLOW_SUCCESS_REDIRECT_URL})
+
+    async def wait_for_callback(self, timeout: float = 300.0) -> str:
+        """Waits for the OAuth callback and returns the authorization code."""
+        try:
+            code = await asyncio.wait_for(self.result_future, timeout=timeout)
+            return code
+        except asyncio.TimeoutError:
+            raise TimeoutError("Timeout waiting for OAuth callback")
+
+
+class IFlowAuthBase:
+    """
+    iFlow OAuth authentication base class.
+    Implements authorization code flow with local callback server.
+    Based on the Go example implementation.
+    """
+
+    def __init__(self):
+        self._credentials_cache: Dict[str, Dict[str, Any]] = {}
+        self._refresh_locks: Dict[str, asyncio.Lock] = {}
+
+    async def _read_creds_from_file(self, path: str) -> Dict[str, Any]:
+        """Reads credentials from file and populates the cache. No locking."""
+        try:
+            lib_logger.debug(f"Reading iFlow credentials from file: {path}")
+            with open(path, 'r') as f:
+                creds = json.load(f)
+            self._credentials_cache[path] = creds
+            return creds
+        except FileNotFoundError:
+            raise IOError(f"iFlow OAuth credential file not found at '{path}'")
+        except Exception as e:
+            raise IOError(f"Failed to load iFlow OAuth credentials from '{path}': {e}")
+
+    async def _load_credentials(self, path: str) -> Dict[str, Any]:
+        """Loads credentials from cache or file."""
+        if path in self._credentials_cache:
+            return self._credentials_cache[path]
+
+        async with self._get_lock(path):
+            # Re-check cache after acquiring lock
+            if path in self._credentials_cache:
+                return self._credentials_cache[path]
+            return await self._read_creds_from_file(path)
+
+    async def _save_credentials(self, path: str, creds: Dict[str, Any]):
+        """Saves credentials to cache and file."""
+        self._credentials_cache[path] = creds
+        try:
+            with open(path, 'w') as f:
+                json.dump(creds, f, indent=2)
+            lib_logger.debug(f"Saved updated iFlow OAuth credentials to '{path}'.")
+        except Exception as e:
+            lib_logger.error(f"Failed to save updated iFlow OAuth credentials to '{path}': {e}")
+
+    def _is_token_expired(self, creds: Dict[str, Any]) -> bool:
+        """Checks if the token is expired (with buffer for proactive refresh)."""
+        # Try to parse expiry_date as ISO 8601 string (from Go example)
+        expiry_str = creds.get("expiry_date")
+        if not expiry_str:
+            return True
+
+        try:
+            # Parse ISO 8601 format (e.g., "2025-01-17T12:00:00Z")
+            from datetime import datetime
+            expiry_dt = datetime.fromisoformat(expiry_str.replace('Z', '+00:00'))
+            expiry_timestamp = expiry_dt.timestamp()
+        except (ValueError, AttributeError):
+            # Fallback: treat as numeric timestamp
+            try:
+                expiry_timestamp = float(expiry_str)
+            except (ValueError, TypeError):
+                lib_logger.warning(f"Could not parse expiry_date: {expiry_str}")
+                return True
+
+        return expiry_timestamp < time.time() + REFRESH_EXPIRY_BUFFER_SECONDS
+
+    async def _fetch_user_info(self, access_token: str) -> Dict[str, Any]:
+        """
+        Fetches user info (including API key) from iFlow API.
+        This is critical: iFlow uses a separate API key for actual API calls.
+        """
+        if not access_token or not access_token.strip():
+            raise ValueError("Access token is empty")
+
+        url = f"{IFLOW_USER_INFO_ENDPOINT}?accessToken={access_token}"
+        headers = {"Accept": "application/json"}
+
+        async with httpx.AsyncClient(timeout=30.0) as client:
+            response = await client.get(url, headers=headers)
+            response.raise_for_status()
+            result = response.json()
+
+        if not result.get("success"):
+            raise ValueError("iFlow user info request not successful")
+
+        data = result.get("data", {})
+        api_key = data.get("apiKey", "").strip()
+        if not api_key:
+            raise ValueError("Missing API key in user info response")
+
+        email = data.get("email", "").strip()
+        if not email:
+            email = data.get("phone", "").strip()
+        if not email:
+            raise ValueError("Missing email/phone in user info response")
+
+        return {"api_key": api_key, "email": email}
+
+    async def _exchange_code_for_tokens(self, code: str, redirect_uri: str) -> Dict[str, Any]:
+        """
+        Exchanges authorization code for access and refresh tokens.
+        Uses Basic Auth with client credentials (from Go example).
+        """
+        # Create Basic Auth header
+        auth_string = f"{IFLOW_CLIENT_ID}:{IFLOW_CLIENT_SECRET}"
+        basic_auth = base64.b64encode(auth_string.encode()).decode()
+
+        headers = {
+            "Content-Type": "application/x-www-form-urlencoded",
+            "Accept": "application/json",
+            "Authorization": f"Basic {basic_auth}"
+        }
+
+        data = {
+            "grant_type": "authorization_code",
+            "code": code,
+            "redirect_uri": redirect_uri,
+            "client_id": IFLOW_CLIENT_ID,
+            "client_secret": IFLOW_CLIENT_SECRET
+        }
+
+        async with httpx.AsyncClient(timeout=30.0) as client:
+            response = await client.post(IFLOW_OAUTH_TOKEN_ENDPOINT, headers=headers, data=data)
+
+            if response.status_code != 200:
+                error_text = response.text
+                lib_logger.error(f"iFlow token exchange failed: {response.status_code} {error_text}")
+                raise ValueError(f"Token exchange failed: {response.status_code} {error_text}")
+
+            token_data = response.json()
+
+        access_token = token_data.get("access_token")
+        if not access_token:
+            raise ValueError("Missing access_token in token response")
+
+        refresh_token = token_data.get("refresh_token", "")
+        expires_in = token_data.get("expires_in", 3600)
+        token_type = token_data.get("token_type", "Bearer")
+        scope = token_data.get("scope", "")
+
+        # Fetch user info to get API key
+        user_info = await self._fetch_user_info(access_token)
+
+        # Calculate expiry date
+        from datetime import datetime, timedelta
+        expiry_date = (datetime.utcnow() + timedelta(seconds=expires_in)).isoformat() + 'Z'
+
+        return {
+            "access_token": access_token,
+            "refresh_token": refresh_token,
+            "api_key": user_info["api_key"],
+            "email": user_info["email"],
+            "expiry_date": expiry_date,
+            "token_type": token_type,
+            "scope": scope
+        }
+
+    async def _refresh_token(self, path: str, force: bool = False) -> Dict[str, Any]:
+        """
+        Refreshes the OAuth tokens and re-fetches the API key.
+        CRITICAL: Must re-fetch user info to get potentially updated API key.
+        """
+        async with self._get_lock(path):
+            cached_creds = self._credentials_cache.get(path)
+            if not force and cached_creds and not self._is_token_expired(cached_creds):
+                return cached_creds
+
+            # If cache is empty, read from file
+            if path not in self._credentials_cache:
+                await self._read_creds_from_file(path)
+
+            creds_from_file = self._credentials_cache[path]
+
+            lib_logger.info(f"Refreshing iFlow OAuth token for '{Path(path).name}'...")
+            refresh_token = creds_from_file.get("refresh_token")
+            if not refresh_token:
+                raise ValueError("No refresh_token found in iFlow credentials file.")
+
+            # Create Basic Auth header
+            auth_string = f"{IFLOW_CLIENT_ID}:{IFLOW_CLIENT_SECRET}"
+            basic_auth = base64.b64encode(auth_string.encode()).decode()
+
+            headers = {
+                "Content-Type": "application/x-www-form-urlencoded",
+                "Accept": "application/json",
+                "Authorization": f"Basic {basic_auth}"
+            }
+
+            data = {
+                "grant_type": "refresh_token",
+                "refresh_token": refresh_token,
+                "client_id": IFLOW_CLIENT_ID,
+                "client_secret": IFLOW_CLIENT_SECRET
+            }
+
+            async with httpx.AsyncClient(timeout=30.0) as client:
+                response = await client.post(IFLOW_OAUTH_TOKEN_ENDPOINT, headers=headers, data=data)
+                response.raise_for_status()
+                new_token_data = response.json()
+
+            # Update tokens
+            access_token = new_token_data.get("access_token")
+            if not access_token:
+                raise ValueError("Missing access_token in refresh response")
+
+            creds_from_file["access_token"] = access_token
+            creds_from_file["refresh_token"] = new_token_data.get("refresh_token", creds_from_file["refresh_token"])
+
+            expires_in = new_token_data.get("expires_in", 3600)
+            from datetime import datetime, timedelta
+            creds_from_file["expiry_date"] = (datetime.utcnow() + timedelta(seconds=expires_in)).isoformat() + 'Z'
+
+            creds_from_file["token_type"] = new_token_data.get("token_type", creds_from_file.get("token_type", "Bearer"))
+            creds_from_file["scope"] = new_token_data.get("scope", creds_from_file.get("scope", ""))
+
+            # CRITICAL: Re-fetch user info to get potentially updated API key
+            try:
+                user_info = await self._fetch_user_info(access_token)
+                if user_info.get("api_key"):
+                    creds_from_file["api_key"] = user_info["api_key"]
+                if user_info.get("email"):
+                    creds_from_file["email"] = user_info["email"]
+            except Exception as e:
+                lib_logger.warning(f"Failed to update API key during token refresh: {e}")
+
+            # Update timestamp in metadata if it exists
+            if creds_from_file.get("_proxy_metadata"):
+                creds_from_file["_proxy_metadata"]["last_check_timestamp"] = time.time()
+
+            await self._save_credentials(path, creds_from_file)
+            lib_logger.info(f"Successfully refreshed iFlow OAuth token for '{Path(path).name}'.")
+            return creds_from_file
+
+    async def get_api_details(self, credential_path: str) -> Tuple[str, str]:
+        """
+        Returns the API base URL and API key (NOT access_token).
+        CRITICAL: iFlow uses the api_key for API requests, not the OAuth access_token.
+        """
+        creds = await self._load_credentials(credential_path)
+
+        # Check if token needs refresh
+        if self._is_token_expired(creds):
+            creds = await self._refresh_token(credential_path)
+
+        api_key = creds.get("api_key")
+        if not api_key:
+            raise ValueError("Missing api_key in iFlow credentials")
+
+        base_url = "https://apis.iflow.cn/v1"
+        return base_url, api_key
+
+    async def proactively_refresh(self, credential_path: str):
+        """Proactively refreshes tokens if they're close to expiry."""
+        creds = await self._load_credentials(credential_path)
+        if self._is_token_expired(creds):
+            await self._refresh_token(credential_path)
+
+    def _get_lock(self, path: str) -> asyncio.Lock:
+        """Gets or creates a lock for the given credential path."""
+        if path not in self._refresh_locks:
+            self._refresh_locks[path] = asyncio.Lock()
+        return self._refresh_locks[path]
+
+    async def initialize_token(self, creds_or_path: Union[Dict[str, Any], str]) -> Dict[str, Any]:
+        """
+        Initiates OAuth authorization code flow if tokens are missing or invalid.
+        Uses local callback server to receive authorization code.
+        """
+        path = creds_or_path if isinstance(creds_or_path, str) else None
+        file_name = Path(path).name if path else "in-memory object"
+        lib_logger.debug(f"Initializing iFlow token for '{file_name}'...")
+
+        try:
+            creds = await self._load_credentials(creds_or_path) if path else creds_or_path
+
+            reason = ""
+            if not creds.get("refresh_token"):
+                reason = "refresh token is missing"
+            elif self._is_token_expired(creds):
+                reason = "token is expired"
+
+            if reason:
+                # Try automatic refresh first if we have a refresh token
+                if reason == "token is expired" and creds.get("refresh_token"):
+                    try:
+                        return await self._refresh_token(path)
+                    except Exception as e:
+                        lib_logger.warning(f"Automatic token refresh for '{file_name}' failed: {e}. Proceeding to interactive login.")
+
+                # Interactive OAuth flow
+                lib_logger.warning(f"iFlow OAuth token for '{file_name}' needs setup: {reason}.")
+
+                # Generate random state for CSRF protection
+                state = secrets.token_urlsafe(32)
+
+                # Build authorization URL
+                redirect_uri = f"http://localhost:{CALLBACK_PORT}/oauth2callback"
+                auth_params = {
+                    "loginMethod": "phone",
+                    "type": "phone",
+                    "redirect": redirect_uri,
+                    "state": state,
+                    "client_id": IFLOW_CLIENT_ID
+                }
+                auth_url = f"{IFLOW_OAUTH_AUTHORIZE_ENDPOINT}?{urlencode(auth_params)}"
+
+                # Start OAuth callback server
+                callback_server = OAuthCallbackServer(port=CALLBACK_PORT)
+                try:
+                    await callback_server.start(expected_state=state)
+
+                    # Display instructions to user
+                    auth_panel_text = Text.from_markup(
+                        "1. Visit the URL below to sign in with your phone number.\n"
+                        "2. [bold]Authorize the application[/bold] to access your account.\n"
+                        "3. You will be automatically redirected after authorization."
+                    )
+                    console.print(Panel(auth_panel_text, title=f"iFlow OAuth Setup for [bold yellow]{file_name}[/bold yellow]", style="bold blue"))
+                    console.print(f"[bold]URL:[/bold] [link={auth_url}]{auth_url}[/link]\n")
+
+                    # Open browser
+                    webbrowser.open(auth_url)
+
+                    # Wait for callback
+                    with console.status("[bold green]Waiting for authorization in the browser...[/bold green]", spinner="dots"):
+                        code = await callback_server.wait_for_callback(timeout=300.0)
+
+                    lib_logger.info("Received authorization code, exchanging for tokens...")
+
+                    # Exchange code for tokens and API key
+                    token_data = await self._exchange_code_for_tokens(code, redirect_uri)
+
+                    # Update credentials
+                    creds.update({
+                        "access_token": token_data["access_token"],
+                        "refresh_token": token_data["refresh_token"],
+                        "api_key": token_data["api_key"],
+                        "email": token_data["email"],
+                        "expiry_date": token_data["expiry_date"],
+                        "token_type": token_data["token_type"],
+                        "scope": token_data["scope"]
+                    })
+
+                    # Create metadata object
+                    if not creds.get("_proxy_metadata"):
+                        creds["_proxy_metadata"] = {
+                            "email": token_data["email"],
+                            "last_check_timestamp": time.time()
+                        }
+
+                    if path:
+                        await self._save_credentials(path, creds)
+
+                    lib_logger.info(f"iFlow OAuth initialized successfully for '{file_name}'.")
+                    return creds
+
+                finally:
+                    await callback_server.stop()
+
+            lib_logger.info(f"iFlow OAuth token at '{file_name}' is valid.")
+            return creds
+
+        except Exception as e:
+            raise ValueError(f"Failed to initialize iFlow OAuth for '{path}': {e}")
+
+    async def get_auth_header(self, credential_path: str) -> Dict[str, str]:
+        """
+        Returns auth header with API key (NOT OAuth access_token).
+        CRITICAL: iFlow API requests use the api_key, not the OAuth tokens.
+        """
+        creds = await self._load_credentials(credential_path)
+        if self._is_token_expired(creds):
+            creds = await self._refresh_token(credential_path)
+
+        api_key = creds.get("api_key")
+        if not api_key:
+            raise ValueError("Missing api_key in iFlow credentials")
+
+        return {"Authorization": f"Bearer {api_key}"}
+
+    async def get_user_info(self, creds_or_path: Union[Dict[str, Any], str]) -> Dict[str, Any]:
+        """Retrieves user info from the _proxy_metadata in the credential file."""
+        try:
+            path = creds_or_path if isinstance(creds_or_path, str) else None
+            creds = await self._load_credentials(creds_or_path) if path else creds_or_path
+
+            # Ensure the token is valid
+            if path:
+                await self.initialize_token(path)
+                creds = await self._load_credentials(path)
+
+            email = creds.get("email") or creds.get("_proxy_metadata", {}).get("email")
+
+            if not email:
+                lib_logger.warning(f"No email found in iFlow credentials for '{path or 'in-memory object'}'.")
+
+            # Update timestamp on check
+            if path and "_proxy_metadata" in creds:
+                creds["_proxy_metadata"]["last_check_timestamp"] = time.time()
+                await self._save_credentials(path, creds)
+
+            return {"email": email}
+        except Exception as e:
+            lib_logger.error(f"Failed to get iFlow user info from credentials: {e}")
+            return {"email": None}

src/rotator_library/providers/iflow_provider.py

@@ -0,0 +1,348 @@
+# src/rotator_library/providers/iflow_provider.py
+
+import json
+import time
+import httpx
+import logging
+from typing import Union, AsyncGenerator, List, Dict, Any
+from .provider_interface import ProviderInterface
+from .iflow_auth_base import IFlowAuthBase
+import litellm
+from litellm.exceptions import RateLimitError, AuthenticationError
+
+lib_logger = logging.getLogger('rotator_library')
+
+# Model list can be expanded as iFlow supports more models
+HARDCODED_MODELS = [
+    "deepseek-v3",
+    "deepseek-chat",
+    "deepseek-coder"
+]
+
+# OpenAI-compatible parameters supported by iFlow API
+SUPPORTED_PARAMS = {
+    'model', 'messages', 'temperature', 'top_p', 'max_tokens',
+    'stream', 'tools', 'tool_choice', 'presence_penalty',
+    'frequency_penalty', 'n', 'stop', 'seed', 'response_format'
+}
+
+
+class IFlowProvider(IFlowAuthBase, ProviderInterface):
+    """
+    iFlow provider using OAuth authentication with local callback server.
+    API requests use the derived API key (NOT OAuth access_token).
+    Based on the Go example implementation.
+    """
+    skip_cost_calculation = True
+
+    def __init__(self):
+        super().__init__()
+
+    def has_custom_logic(self) -> bool:
+        return True
+
+    async def get_models(self, credential: str, client: httpx.AsyncClient) -> List[str]:
+        """Returns a hardcoded list of known compatible iFlow models."""
+        return [f"iflow/{model_id}" for model_id in HARDCODED_MODELS]
+
+    def _clean_tool_schemas(self, tools: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
+        """
+        Removes unsupported properties from tool schemas to prevent API errors.
+        Similar to Qwen Code implementation.
+        """
+        import copy
+        cleaned_tools = []
+
+        for tool in tools:
+            cleaned_tool = copy.deepcopy(tool)
+
+            if "function" in cleaned_tool:
+                func = cleaned_tool["function"]
+
+                # Remove strict mode (may not be supported)
+                func.pop("strict", None)
+
+                # Clean parameter schema if present
+                if "parameters" in func and isinstance(func["parameters"], dict):
+                    params = func["parameters"]
+
+                    # Remove additionalProperties if present
+                    params.pop("additionalProperties", None)
+
+                    # Recursively clean nested properties
+                    if "properties" in params:
+                        self._clean_schema_properties(params["properties"])
+
+            cleaned_tools.append(cleaned_tool)
+
+        return cleaned_tools
+
+    def _clean_schema_properties(self, properties: Dict[str, Any]) -> None:
+        """Recursively cleans schema properties."""
+        for prop_name, prop_schema in properties.items():
+            if isinstance(prop_schema, dict):
+                # Remove unsupported fields
+                prop_schema.pop("strict", None)
+                prop_schema.pop("additionalProperties", None)
+
+                # Recurse into nested properties
+                if "properties" in prop_schema:
+                    self._clean_schema_properties(prop_schema["properties"])
+
+                # Recurse into array items
+                if "items" in prop_schema and isinstance(prop_schema["items"], dict):
+                    self._clean_schema_properties({"item": prop_schema["items"]})
+
+    def _build_request_payload(self, **kwargs) -> Dict[str, Any]:
+        """
+        Builds a clean request payload with only supported parameters.
+        This prevents 400 Bad Request errors from litellm-internal parameters.
+        """
+        # Extract only supported OpenAI parameters
+        payload = {k: v for k, v in kwargs.items() if k in SUPPORTED_PARAMS}
+
+        # Always force streaming for internal processing
+        payload['stream'] = True
+
+        # Always include usage data in stream
+        payload['stream_options'] = {"include_usage": True}
+
+        # Handle tool schema cleaning
+        if "tools" in payload and payload["tools"]:
+            payload["tools"] = self._clean_tool_schemas(payload["tools"])
+            lib_logger.debug(f"Cleaned {len(payload['tools'])} tool schemas")
+
+        return payload
+
+    def _convert_chunk_to_openai(self, chunk: Dict[str, Any], model_id: str):
+        """
+        Converts a raw iFlow SSE chunk to an OpenAI-compatible chunk.
+        Since iFlow is OpenAI-compatible, minimal conversion is needed.
+        """
+        if not isinstance(chunk, dict):
+            return
+
+        # Handle usage data
+        if usage_data := chunk.get("usage"):
+            yield {
+                "choices": [], "model": model_id, "object": "chat.completion.chunk",
+                "id": chunk.get("id", f"chatcmpl-iflow-{time.time()}"),
+                "created": chunk.get("created", int(time.time())),
+                "usage": {
+                    "prompt_tokens": usage_data.get("prompt_tokens", 0),
+                    "completion_tokens": usage_data.get("completion_tokens", 0),
+                    "total_tokens": usage_data.get("total_tokens", 0),
+                }
+            }
+            return
+
+        # Handle content data
+        choices = chunk.get("choices", [])
+        if not choices:
+            return
+
+        # iFlow returns OpenAI-compatible format, so we can mostly pass through
+        yield {
+            "choices": choices,
+            "model": model_id,
+            "object": "chat.completion.chunk",
+            "id": chunk.get("id", f"chatcmpl-iflow-{time.time()}"),
+            "created": chunk.get("created", int(time.time()))
+        }
+
+    def _stream_to_completion_response(self, chunks: List[litellm.ModelResponse]) -> litellm.ModelResponse:
+        """
+        Manually reassembles streaming chunks into a complete response.
+        """
+        if not chunks:
+            raise ValueError("No chunks provided for reassembly")
+
+        # Initialize the final response structure
+        final_message = {"role": "assistant"}
+        aggregated_tool_calls = {}
+        usage_data = None
+        finish_reason = None
+
+        # Get the first chunk for basic response metadata
+        first_chunk = chunks[0]
+
+        # Process each chunk to aggregate content
+        for chunk in chunks:
+            if not hasattr(chunk, 'choices') or not chunk.choices:
+                continue
+
+            choice = chunk.choices[0]
+            delta = choice.get("delta", {})
+
+            # Aggregate content
+            if "content" in delta and delta["content"] is not None:
+                if "content" not in final_message:
+                    final_message["content"] = ""
+                final_message["content"] += delta["content"]
+
+            # Aggregate reasoning content (if supported by iFlow)
+            if "reasoning_content" in delta and delta["reasoning_content"] is not None:
+                if "reasoning_content" not in final_message:
+                    final_message["reasoning_content"] = ""
+                final_message["reasoning_content"] += delta["reasoning_content"]
+
+            # Aggregate tool calls
+            if "tool_calls" in delta and delta["tool_calls"]:
+                for tc_chunk in delta["tool_calls"]:
+                    index = tc_chunk["index"]
+                    if index not in aggregated_tool_calls:
+                        aggregated_tool_calls[index] = {"function": {"name": "", "arguments": ""}}
+                    if "id" in tc_chunk:
+                        aggregated_tool_calls[index]["id"] = tc_chunk["id"]
+                    if "type" in tc_chunk:
+                        aggregated_tool_calls[index]["type"] = tc_chunk["type"]
+                    if "function" in tc_chunk:
+                        if "name" in tc_chunk["function"] and tc_chunk["function"]["name"] is not None:
+                            aggregated_tool_calls[index]["function"]["name"] += tc_chunk["function"]["name"]
+                        if "arguments" in tc_chunk["function"] and tc_chunk["function"]["arguments"] is not None:
+                            aggregated_tool_calls[index]["function"]["arguments"] += tc_chunk["function"]["arguments"]
+
+            # Aggregate function calls (legacy format)
+            if "function_call" in delta and delta["function_call"] is not None:
+                if "function_call" not in final_message:
+                    final_message["function_call"] = {"name": "", "arguments": ""}
+                if "name" in delta["function_call"] and delta["function_call"]["name"] is not None:
+                    final_message["function_call"]["name"] += delta["function_call"]["name"]
+                if "arguments" in delta["function_call"] and delta["function_call"]["arguments"] is not None:
+                    final_message["function_call"]["arguments"] += delta["function_call"]["arguments"]
+
+            # Get finish reason from the last chunk that has it
+            if choice.get("finish_reason"):
+                finish_reason = choice["finish_reason"]
+
+        # Handle usage data from the last chunk that has it
+        for chunk in reversed(chunks):
+            if hasattr(chunk, 'usage') and chunk.usage:
+                usage_data = chunk.usage
+                break
+
+        # Add tool calls to final message if any
+        if aggregated_tool_calls:
+            final_message["tool_calls"] = list(aggregated_tool_calls.values())
+
+        # Ensure standard fields are present for consistent logging
+        for field in ["content", "tool_calls", "function_call"]:
+            if field not in final_message:
+                final_message[field] = None
+
+        # Construct the final response
+        final_choice = {
+            "index": 0,
+            "message": final_message,
+            "finish_reason": finish_reason
+        }
+
+        # Create the final ModelResponse
+        final_response_data = {
+            "id": first_chunk.id,
+            "object": "chat.completion",
+            "created": first_chunk.created,
+            "model": first_chunk.model,
+            "choices": [final_choice],
+            "usage": usage_data
+        }
+
+        return litellm.ModelResponse(**final_response_data)
+
+    async def acompletion(self, client: httpx.AsyncClient, **kwargs) -> Union[litellm.ModelResponse, AsyncGenerator[litellm.ModelResponse, None]]:
+        credential_path = kwargs.pop("credential_identifier")
+        enable_request_logging = kwargs.pop("enable_request_logging", False)
+        model = kwargs["model"]
+
+        async def make_request():
+            """Prepares and makes the actual API call."""
+            # CRITICAL: get_api_details returns api_key, NOT access_token
+            api_base, api_key = await self.get_api_details(credential_path)
+
+            # Build clean payload with only supported parameters
+            payload = self._build_request_payload(**kwargs)
+
+            headers = {
+                "Authorization": f"Bearer {api_key}",  # Uses api_key from user info
+                "Content-Type": "application/json",
+                "Accept": "text/event-stream",
+                "User-Agent": "iFlow-Cli"
+            }
+
+            url = f"{api_base.rstrip('/')}/chat/completions"
+
+            if enable_request_logging:
+                lib_logger.info(f"iFlow Request URL: {url}")
+                lib_logger.info(f"iFlow Request Payload: {json.dumps(payload, indent=2)}")
+            else:
+                lib_logger.debug(f"iFlow Request URL: {url}")
+
+            return client.stream("POST", url, headers=headers, json=payload, timeout=600)
+
+        async def stream_handler(response_stream, attempt=1):
+            """Handles the streaming response and converts chunks."""
+            try:
+                async with response_stream as response:
+                    # Check for HTTP errors before processing stream
+                    if response.status_code >= 400:
+                        error_text = await response.aread()
+                        error_text = error_text.decode('utf-8') if isinstance(error_text, bytes) else error_text
+
+                        # Handle 401: Force token refresh and retry once
+                        if response.status_code == 401 and attempt == 1:
+                            lib_logger.warning("iFlow returned 401. Forcing token refresh and retrying once.")
+                            await self._refresh_token(credential_path, force=True)
+                            retry_stream = await make_request()
+                            async for chunk in stream_handler(retry_stream, attempt=2):
+                                yield chunk
+                            return
+
+                        # Handle 429: Rate limit
+                        elif response.status_code == 429 or "slow_down" in error_text.lower():
+                            raise RateLimitError(
+                                f"iFlow rate limit exceeded: {error_text}",
+                                llm_provider="iflow",
+                                model=model,
+                                response=response
+                            )
+
+                        # Handle other errors
+                        else:
+                            if enable_request_logging:
+                                lib_logger.error(f"iFlow HTTP {response.status_code} error: {error_text}")
+                            raise httpx.HTTPStatusError(
+                                f"HTTP {response.status_code}: {error_text}",
+                                request=response.request,
+                                response=response
+                            )
+
+                    # Process successful streaming response
+                    async for line in response.aiter_lines():
+                        if line.startswith('data: '):
+                            data_str = line[6:]
+                            if data_str == "[DONE]":
+                                break
+                            try:
+                                chunk = json.loads(data_str)
+                                for openai_chunk in self._convert_chunk_to_openai(chunk, model):
+                                    yield litellm.ModelResponse(**openai_chunk)
+                            except json.JSONDecodeError:
+                                lib_logger.warning(f"Could not decode JSON from iFlow: {line}")
+
+            except httpx.HTTPStatusError:
+                raise  # Re-raise HTTP errors we already handled
+            except Exception as e:
+                if enable_request_logging:
+                    lib_logger.error(f"Error during iFlow stream processing: {e}", exc_info=True)
+                raise
+
+        http_response_stream = await make_request()
+        response_generator = stream_handler(http_response_stream)
+
+        if kwargs.get("stream"):
+            return response_generator
+        else:
+            async def non_stream_wrapper():
+                chunks = [chunk async for chunk in response_generator]
+                return self._stream_to_completion_response(chunks)
+            return await non_stream_wrapper()

src/rotator_library/providers/qwen_auth_base.py

@@ -110,8 +110,8 @@ class QwenAuthBase:
             lib_logger.info(f"Successfully refreshed Qwen OAuth token for '{Path(path).name}'.")
             return creds_from_file
 
-    def get_api_details(self, credential_path: str) -> Tuple[str, str]:
-        creds = self._credentials_cache[credential_path]
+    async def get_api_details(self, credential_path: str) -> Tuple[str, str]:
+        creds = await self._load_credentials(credential_path)
         base_url = creds.get("resource_url", "https://portal.qwen.ai/v1")
         if not base_url.startswith("http"):
             base_url = f"https://{base_url}"

src/rotator_library/providers/qwen_code_provider.py

@@ -17,8 +17,16 @@ HARDCODED_MODELS = [
     "qwen3-coder-flash"
 ]
 
+# OpenAI-compatible parameters supported by Qwen Code API
+SUPPORTED_PARAMS = {
+    'model', 'messages', 'temperature', 'top_p', 'max_tokens',
+    'stream', 'tools', 'tool_choice', 'presence_penalty',
+    'frequency_penalty', 'n', 'stop', 'seed', 'response_format'
+}
+
 class QwenCodeProvider(QwenAuthBase, ProviderInterface):
     skip_cost_calculation = True
+    REASONING_START_MARKER = 'THINK||'
 
     def __init__(self):
         super().__init__()
@@ -30,6 +38,87 @@ class QwenCodeProvider(QwenAuthBase, ProviderInterface):
         """Returns a hardcoded list of known compatible Qwen models."""
         return [f"qwen_code/{model_id}" for model_id in HARDCODED_MODELS]
 
+    def _clean_tool_schemas(self, tools: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
+        """
+        Removes unsupported properties from tool schemas to prevent API errors.
+        Based on Gemini CLI's approach but adapted for Qwen's API requirements.
+        """
+        import copy
+        cleaned_tools = []
+
+        for tool in tools:
+            cleaned_tool = copy.deepcopy(tool)
+
+            if "function" in cleaned_tool:
+                func = cleaned_tool["function"]
+
+                # Remove strict mode (not supported by Qwen)
+                func.pop("strict", None)
+
+                # Clean parameter schema if present
+                if "parameters" in func and isinstance(func["parameters"], dict):
+                    params = func["parameters"]
+
+                    # Remove additionalProperties if present
+                    params.pop("additionalProperties", None)
+
+                    # Recursively clean nested properties
+                    if "properties" in params:
+                        self._clean_schema_properties(params["properties"])
+
+            cleaned_tools.append(cleaned_tool)
+
+        return cleaned_tools
+
+    def _clean_schema_properties(self, properties: Dict[str, Any]) -> None:
+        """Recursively cleans schema properties."""
+        for prop_name, prop_schema in properties.items():
+            if isinstance(prop_schema, dict):
+                # Remove unsupported fields
+                prop_schema.pop("strict", None)
+                prop_schema.pop("additionalProperties", None)
+
+                # Recurse into nested properties
+                if "properties" in prop_schema:
+                    self._clean_schema_properties(prop_schema["properties"])
+
+                # Recurse into array items
+                if "items" in prop_schema and isinstance(prop_schema["items"], dict):
+                    self._clean_schema_properties({"item": prop_schema["items"]})
+
+    def _build_request_payload(self, **kwargs) -> Dict[str, Any]:
+        """
+        Builds a clean request payload with only supported parameters.
+        This prevents 400 Bad Request errors from litellm-internal parameters.
+        """
+        # Extract only supported OpenAI parameters
+        payload = {k: v for k, v in kwargs.items() if k in SUPPORTED_PARAMS}
+
+        # Always force streaming for internal processing
+        payload['stream'] = True
+
+        # Always include usage data in stream
+        payload['stream_options'] = {"include_usage": True}
+
+        # Handle tool schema cleaning
+        if "tools" in payload and payload["tools"]:
+            payload["tools"] = self._clean_tool_schemas(payload["tools"])
+            lib_logger.debug(f"Cleaned {len(payload['tools'])} tool schemas")
+        elif not payload.get("tools"):
+            # Per Qwen Code API bug (see: https://github.com/qianwen-team/flash-dance/issues/2),
+            # injecting a dummy tool prevents stream corruption when no tools are provided
+            payload["tools"] = [{
+                "type": "function",
+                "function": {
+                    "name": "do_not_call_me",
+                    "description": "Do not call this tool.",
+                    "parameters": {"type": "object", "properties": {}}
+                }
+            }]
+            lib_logger.debug("Injected dummy tool to prevent Qwen API stream corruption")
+
+        return payload
+
     def _convert_chunk_to_openai(self, chunk: Dict[str, Any], model_id: str):
         """Converts a raw Qwen SSE chunk to an OpenAI-compatible chunk."""
         if not isinstance(chunk, dict):
@@ -60,14 +149,14 @@ class QwenCodeProvider(QwenAuthBase, ProviderInterface):
         # Handle <think> tags for reasoning content
         content = delta.get("content")
         if content and ("<think>" in content or "</think>" in content):
-            parts = content.replace("<think>", "||THINK||").replace("</think>", "||/THINK||").split("||")
+            parts = content.replace("<think>", f"||{self.REASONING_START_MARKER}").replace("</think>", f"||/{self.REASONING_START_MARKER}").split("||")
             for part in parts:
                 if not part: continue
                 
                 new_delta = {}
-                if part.startswith("THINK||"):
-                    new_delta['reasoning_content'] = part.replace("THINK||", "")
-                elif part.startswith("/THINK||"):
+                if part.startswith(self.REASONING_START_MARKER):
+                    new_delta['reasoning_content'] = part.replace(self.REASONING_START_MARKER, "")
+                elif part.startswith(f"/{self.REASONING_START_MARKER}"):
                     continue
                 else:
                     new_delta['content'] = part
@@ -85,71 +174,199 @@ class QwenCodeProvider(QwenAuthBase, ProviderInterface):
                 "id": f"chatcmpl-qwen-{time.time()}", "created": int(time.time())
             }
 
+    def _stream_to_completion_response(self, chunks: List[litellm.ModelResponse]) -> litellm.ModelResponse:
+        """
+        Manually reassembles streaming chunks into a complete response.
+        This replaces the non-existent litellm.utils.stream_to_completion_response function.
+        """
+        if not chunks:
+            raise ValueError("No chunks provided for reassembly")
+
+        # Initialize the final response structure
+        final_message = {"role": "assistant"}
+        aggregated_tool_calls = {}
+        usage_data = None
+        finish_reason = None
+
+        # Get the first chunk for basic response metadata
+        first_chunk = chunks[0]
+
+        # Process each chunk to aggregate content
+        for chunk in chunks:
+            if not hasattr(chunk, 'choices') or not chunk.choices:
+                continue
+
+            choice = chunk.choices[0]
+            delta = choice.get("delta", {})
+
+            # Aggregate content
+            if "content" in delta and delta["content"] is not None:
+                if "content" not in final_message:
+                    final_message["content"] = ""
+                final_message["content"] += delta["content"]
+
+            # Aggregate reasoning content
+            if "reasoning_content" in delta and delta["reasoning_content"] is not None:
+                if "reasoning_content" not in final_message:
+                    final_message["reasoning_content"] = ""
+                final_message["reasoning_content"] += delta["reasoning_content"]
+
+            # Aggregate tool calls
+            if "tool_calls" in delta and delta["tool_calls"]:
+                for tc_chunk in delta["tool_calls"]:
+                    index = tc_chunk["index"]
+                    if index not in aggregated_tool_calls:
+                        aggregated_tool_calls[index] = {"function": {"name": "", "arguments": ""}}
+                    if "id" in tc_chunk:
+                        aggregated_tool_calls[index]["id"] = tc_chunk["id"]
+                    if "function" in tc_chunk:
+                        if "name" in tc_chunk["function"] and tc_chunk["function"]["name"] is not None:
+                            aggregated_tool_calls[index]["function"]["name"] += tc_chunk["function"]["name"]
+                        if "arguments" in tc_chunk["function"] and tc_chunk["function"]["arguments"] is not None:
+                            aggregated_tool_calls[index]["function"]["arguments"] += tc_chunk["function"]["arguments"]
+
+            # Aggregate function calls (legacy format)
+            if "function_call" in delta and delta["function_call"] is not None:
+                if "function_call" not in final_message:
+                    final_message["function_call"] = {"name": "", "arguments": ""}
+                if "name" in delta["function_call"] and delta["function_call"]["name"] is not None:
+                    final_message["function_call"]["name"] += delta["function_call"]["name"]
+                if "arguments" in delta["function_call"] and delta["function_call"]["arguments"] is not None:
+                    final_message["function_call"]["arguments"] += delta["function_call"]["arguments"]
+
+            # Get finish reason from the last chunk that has it
+            if choice.get("finish_reason"):
+                finish_reason = choice["finish_reason"]
+
+        # Handle usage data from the last chunk that has it
+        for chunk in reversed(chunks):
+            if hasattr(chunk, 'usage') and chunk.usage:
+                usage_data = chunk.usage
+                break
+
+        # Add tool calls to final message if any
+        if aggregated_tool_calls:
+            final_message["tool_calls"] = list(aggregated_tool_calls.values())
+
+        # Ensure standard fields are present for consistent logging
+        for field in ["content", "tool_calls", "function_call"]:
+            if field not in final_message:
+                final_message[field] = None
+
+        # Construct the final response
+        final_choice = {
+            "index": 0,
+            "message": final_message,
+            "finish_reason": finish_reason
+        }
+
+        # Create the final ModelResponse
+        final_response_data = {
+            "id": first_chunk.id,
+            "object": "chat.completion",
+            "created": first_chunk.created,
+            "model": first_chunk.model,
+            "choices": [final_choice],
+            "usage": usage_data
+        }
+
+        return litellm.ModelResponse(**final_response_data)
+
     async def acompletion(self, client: httpx.AsyncClient, **kwargs) -> Union[litellm.ModelResponse, AsyncGenerator[litellm.ModelResponse, None]]:
         credential_path = kwargs.pop("credential_identifier")
+        enable_request_logging = kwargs.pop("enable_request_logging", False)
         model = kwargs["model"]
 
-        async def do_call():
+        async def make_request():
+            """Prepares and makes the actual API call."""
             api_base, access_token = await self.get_api_details(credential_path)
-            
-            # Prepare payload
-            payload = kwargs.copy()
-            payload.pop("litellm_params", None) # Clean up internal params
-            
-            # Per Go example, inject dummy tool to prevent stream corruption
-            if not payload.get("tools"):
-                payload["tools"] = [{"type": "function", "function": {"name": "do_not_call_me", "description": "Do not call this tool under any circumstances.", "parameters": {"type": "object", "properties": {}}}}]
-
-            # Ensure usage is included in stream
-            payload["stream_options"] = {"include_usage": True}
+
+            # Build clean payload with only supported parameters
+            payload = self._build_request_payload(**kwargs)
 
             headers = {
                 "Authorization": f"Bearer {access_token}",
                 "Content-Type": "application/json",
-                "Accept": "text/event-stream" if kwargs.get("stream") else "application/json",
+                "Accept": "text/event-stream",
                 "User-Agent": "google-api-nodejs-client/9.15.1",
                 "X-Goog-Api-Client": "gl-node/22.17.0",
                 "Client-Metadata": "ideType=IDE_UNSPECIFIED,platform=PLATFORM_UNSPECIFIED,pluginType=GEMINI",
             }
-            
-            url = f"{api_base.rstrip('/')}/chat/completions"
-            lib_logger.debug(f"Qwen Code Request URL: {url}")
-            lib_logger.debug(f"Qwen Code Request Payload: {json.dumps(payload, indent=2)}")
-
-            async def stream_handler():
-                async with client.stream("POST", url, headers=headers, json=payload, timeout=600) as response:
-                    response.raise_for_status()
+
+            url = f"{api_base.rstrip('/')}/v1/chat/completions"
+
+            if enable_request_logging:
+                lib_logger.info(f"Qwen Code Request URL: {url}")
+                lib_logger.info(f"Qwen Code Request Payload: {json.dumps(payload, indent=2)}")
+            else:
+                lib_logger.debug(f"Qwen Code Request URL: {url}")
+
+            return client.stream("POST", url, headers=headers, json=payload, timeout=600)
+
+        async def stream_handler(response_stream, attempt=1):
+            """Handles the streaming response and converts chunks."""
+            try:
+                async with response_stream as response:
+                    # Check for HTTP errors before processing stream
+                    if response.status_code >= 400:
+                        error_text = await response.aread()
+                        error_text = error_text.decode('utf-8') if isinstance(error_text, bytes) else error_text
+
+                        # Handle 401: Force token refresh and retry once
+                        if response.status_code == 401 and attempt == 1:
+                            lib_logger.warning("Qwen Code returned 401. Forcing token refresh and retrying once.")
+                            await self._refresh_token(credential_path, force=True)
+                            retry_stream = await make_request()
+                            async for chunk in stream_handler(retry_stream, attempt=2):
+                                yield chunk
+                            return
+
+                        # Handle 429: Rate limit
+                        elif response.status_code == 429 or "slow_down" in error_text.lower():
+                            raise RateLimitError(
+                                f"Qwen Code rate limit exceeded: {error_text}",
+                                llm_provider="qwen_code",
+                                model=model,
+                                response=response
+                            )
+
+                        # Handle other errors
+                        else:
+                            if enable_request_logging:
+                                lib_logger.error(f"Qwen Code HTTP {response.status_code} error: {error_text}")
+                            raise httpx.HTTPStatusError(
+                                f"HTTP {response.status_code}: {error_text}",
+                                request=response.request,
+                                response=response
+                            )
+
+                    # Process successful streaming response
                     async for line in response.aiter_lines():
                         if line.startswith('data: '):
                             data_str = line[6:]
-                            if data_str == "[DONE]": break
+                            if data_str == "[DONE]":
+                                break
                             try:
                                 chunk = json.loads(data_str)
                                 for openai_chunk in self._convert_chunk_to_openai(chunk, model):
                                     yield litellm.ModelResponse(**openai_chunk)
                             except json.JSONDecodeError:
                                 lib_logger.warning(f"Could not decode JSON from Qwen Code: {line}")
-            
-            return stream_handler()
-
-        try:
-            response_gen = await do_call()
-        except httpx.HTTPStatusError as e:
-            if e.response.status_code == 401:
-                lib_logger.warning("Qwen Code returned 401. Forcing token refresh and retrying once.")
-                await self._refresh_token(credential_path, force=True)
-                response_gen = await do_call()
-            elif e.response.status_code == 429 or "slow_down" in e.response.text.lower():
-                raise RateLimitError(
-                    message=f"Qwen Code rate limit exceeded: {e.response.text}",
-                    llm_provider="qwen_code",
-                    response=e.response
-                )
-            else:
-                raise e
+
+            except httpx.HTTPStatusError:
+                raise  # Re-raise HTTP errors we already handled
+            except Exception as e:
+                if enable_request_logging:
+                    lib_logger.error(f"Error during Qwen Code stream processing: {e}", exc_info=True)
+                raise
+
+        http_response_stream = await make_request()
+        response_generator = stream_handler(http_response_stream)
 
         if kwargs.get("stream"):
-            return response_gen
+            return response_generator
         else:
-            chunks = [chunk async for chunk in response_gen]
-            return litellm.utils.stream_to_completion_response(chunks)
+            async def non_stream_wrapper():
+                chunks = [chunk async for chunk in response_generator]
+                return self._stream_to_completion_response(chunks)
+            return await non_stream_wrapper()