ai: Use asyncio with httpx and aiohttp for all requests.

* The private server support HTTP/2.

* Tools operate over HTTP/1.1 to ensure compatibility.

Super fast connection means happy users.

Combining these protocols with asyncio, httpx, and aiohttp creates a highly stable and performant asynchronous system.

src/core/server.py

@@ -3,20 +3,22 @@
 # SPDX-License-Identifier: Apache-2.0
 #
 
-import json  # Import JSON module to parse and handle JSON data
-import uuid  # Import UUID module to generate unique identifiers for sessions
-from typing import List, Dict, Any  # Import type hints for lists, dictionaries, and generic types
-from datetime import datetime  # Import datetime to get and format current date/time
-from config import *  # Import all configuration variables including 'auth' and 'restrictions'
-from src.utils.session_mapping import get_host  # Import function to get server info by session ID
-from src.utils.ip_generator import generate_ip  # Import function to generate random IP for headers
-from src.utils.helper import mark  # Import function to mark a server as busy/unavailable
-from src.ui.reasoning import styles  # Import function to apply CSS styling to reasoning output
-import httpx  # Import httpx for async HTTP requests with streaming support
+import json  # Module to parse and handle JSON data
+import uuid  # Module to generate unique identifiers (UUIDs)
+from typing import List, Dict, Any  # Type hinting for list, dict, and generic types
+from datetime import datetime  # To get and format current date and time
+from config import *  # Import all configuration variables, including 'auth' and 'restrictions'
+from src.utils.session_mapping import get_host  # Function to get server info by session ID
+from src.utils.ip_generator import generate_ip  # Function to generate random IP addresses for headers
+from src.utils.helper import mark  # Function to mark a server as busy/unavailable
+from src.ui.reasoning import styles  # Function to apply CSS styling to reasoning output
+import asyncio  # Asyncio for asynchronous programming
+import httpx  # Async HTTP client supporting HTTP/2 and streaming
+import aiohttp  # Async HTTP client for alternative HTTP requests and streaming
 
 async def jarvis(
     session_id: str,  # Unique session identifier to maintain consistent server assignment
-    model: str,  # AI model name to specify which model to use
+    model: str,  # AI model name specifying which model to use
     history: List[Dict[str, str]],  # List of previous conversation messages with roles and content
     user_message: str,  # Latest user input message to send to the AI model
     mode: str,  # Mode string to guide AI behavior, e.g., '/think' or '/no_think'
@@ -35,6 +37,9 @@ async def jarvis(
     the reasoning output only if the mode is not '/no_think', preserving the behavior where reasoning is streamed
     first inside a styled HTML block, followed by the main content streamed normally.
 
+    The implementation uses both httpx (with HTTP/2 support) and aiohttp to ensure compatibility and robustness
+    in streaming responses.
+
     Args:
         session_id (str): Identifier for the user session to maintain consistent server assignment.
         model (str): Name of the AI model to use for generating the response.
@@ -58,47 +63,48 @@ async def jarvis(
         If the server returns a specific error code indicating it is busy, it retries with another server.
         If all servers are busy or fail, it yields a message indicating the server is busy.
     """
-    tried = set()  # Track servers already tried to avoid repeated retries
+    tried = set()  # Set to track servers already tried to avoid repeated retries
 
     # Loop until all available servers have been tried without success
     while len(tried) < len(auth):
         # Get server setup info assigned for this session, including endpoint, token, and error code
         setup = get_host(session_id)
-        server = setup["jarvis"]  # Server identifier
-        host = setup["endpoint"]  # API endpoint URL
-        token = setup["token"]  # Authorization token
-        error = setup["error"]  # HTTP error code triggering retry
-        tried.add(server)  # Mark this server as tried
+        server = setup["jarvis"]  # Server identifier string
+        host = setup["endpoint"]  # API endpoint URL string
+        token = setup["token"]  # Authorization token string
+        error = setup["error"]  # HTTP error code integer which triggers retry
+        tried.add(server)  # Mark this server as tried to prevent retrying immediately
 
-        # Format current date/time string for system instructions
+        # Format current date/time string for system instructions, e.g., "Tuesday, June 24, 2025, 08:13 PM "
         date = datetime.now().strftime("%A, %B %d, %Y, %I:%M %p %Z")
 
-        # Combine mode instructions, usage restrictions, and date into system instructions string
+        # Combine mode instructions, usage restrictions, and date into a single system instructions string
         instructions = f"{mode}\n\n\n{restrictions}\n\n\nToday: {date}\n\n\n"
 
-        # Copy conversation history to avoid mutating original
+        # Copy conversation history to avoid mutating the original list outside this function
         messages = history.copy()
-        # Insert system instructions as first message
+
+        # Insert system instructions as the first message in the conversation history
         messages.insert(0, {"role": "system", "content": instructions})
 
-        # Prepare user message dict, include files if provided
+        # Prepare user message dictionary, include files if provided
         msg = {"role": "user", "content": user_message}
         if files:
             msg["files"] = files
-        messages.append(msg)  # Append user message to conversation
+        messages.append(msg)  # Append user message to the conversation messages list
 
-        # Prepare HTTP headers with authorization and randomized client IP
+        # Prepare HTTP headers with authorization and randomized client IP for X-Forwarded-For
         headers = {
-            "Authorization": f"Bearer {token}",  # Bearer token for API access
-            "Content-Type": "application/json",  # JSON content type
-            "X-Forwarded-For": generate_ip()  # Random IP to simulate different client origins
+            "Authorization": f"Bearer {token}",  # Bearer token for API access authentication
+            "Content-Type": "application/json",  # Content type set to JSON for request body
+            "X-Forwarded-For": generate_ip(),  # Random IP to simulate different client origins for load balancing or logging
         }
 
-        # Prepare JSON payload with model parameters and conversation messages
+        # Prepare JSON payload with model parameters and conversation messages to send in POST request
         payload = {
             "model": model,
             "messages": messages,
-            "stream": True,
+            "stream": True,  # Enable streaming response
             "temperature": temperature,
             "top_k": top_k,
             "min_p": min_p,
@@ -107,25 +113,25 @@ async def jarvis(
         }
 
         # Initialize accumulators and flags for streamed response parts
-        reasoning = ""  # Accumulate reasoning text
-        reasoning_check = None  # Flag to detect presence of reasoning in response
+        reasoning = ""  # String accumulator for reasoning text from the AI
+        reasoning_check = None  # Flag to detect presence of reasoning in response; None means not checked yet
         reasoning_done = False  # Flag marking reasoning completion
-        content = ""  # Accumulate main content text
+        content = ""  # String accumulator for main content text from the AI
 
         try:
-            # Create async HTTP client with no timeout for long streaming
-            async with httpx.AsyncClient(timeout=None) as client:
-                # Open async streaming POST request to Jarvis server
+            # Use httpx AsyncClient with no timeout to allow long streaming connections
+            async with httpx.AsyncClient(timeout=None, http2=True) as client:
+                # Open async streaming POST request to Jarvis server endpoint with headers and JSON payload
                 async with client.stream("POST", host, headers=headers, json=payload) as response:
-                    # Iterate asynchronously over each line of streaming response
+                    # Iterate asynchronously over each line of streaming response as it arrives
                     async for chunk in response.aiter_lines():
-                        # Skip lines not starting with "data:"
+                        # Skip lines that do not start with "data:" prefix as per server-sent events (SSE) format
                         if not chunk.strip().startswith("data:"):
                             continue
                         try:
-                            # Parse JSON data after "data:" prefix
+                            # Parse JSON data after "data:" prefix which contains incremental response delta
                             data = json.loads(chunk[5:])
-                            # Extract incremental delta message from first choice
+                            # Extract incremental delta message from first choice in response
                             choice = data["choices"][0]["delta"]
 
                             # On first delta received, detect if 'reasoning' field is present and non-empty
@@ -133,7 +139,7 @@ async def jarvis(
                                 # Initialize reasoning_check to empty string if reasoning exists and is non-empty, else None
                                 reasoning_check = "" if ("reasoning" in choice and choice["reasoning"]) else None
 
-                            # If reasoning is present and mode is not '/no_think' and reasoning not done
+                            # If reasoning is present and mode is not '/no_think' and reasoning not done yet
                             if (
                                 reasoning_check == ""  # Reasoning detected in response
                                 and mode != "/no_think"  # Mode allows reasoning output
@@ -144,7 +150,7 @@ async def jarvis(
                                 reasoning += choice["reasoning"]  # Append incremental reasoning text
                                 # Yield reasoning wrapped in styled HTML block with details expanded
                                 yield styles(reasoning=reasoning, content="", expanded=True)
-                                continue  # Continue streaming reasoning increments
+                                continue  # Continue streaming reasoning increments without processing content yet
 
                             # When reasoning ends and content starts, mark reasoning done, yield empty string, then content
                             if (
@@ -155,9 +161,9 @@ async def jarvis(
                                 and choice["content"]  # Content is not empty
                             ):
                                 reasoning_done = True  # Mark reasoning phase complete
-                                yield ""  # Yield empty string to signal end of reasoning block
+                                yield ""  # Yield empty string to signal end of reasoning block to the consumer
                                 content += choice["content"]  # Start accumulating content text
-                                yield content  # Yield first part of content
+                                yield content  # Yield first part of content to the consumer
                                 continue  # Continue streaming content increments
 
                             # If no reasoning present or reasoning done, accumulate content and yield incrementally
@@ -167,22 +173,98 @@ async def jarvis(
                                 and choice["content"]  # Content is not empty
                             ):
                                 content += choice["content"]  # Append incremental content text
-                                yield content  # Yield updated content string
+                                yield content  # Yield updated content string to the consumer
                         except Exception:
                             # Ignore exceptions during JSON parsing or key access and continue streaming
                             continue
             return  # Exit function after successful streaming completion
+
         except httpx.HTTPStatusError as e:
             # If server returns specific error code indicating busy, retry with another server
             if e.response.status_code == error:
-                continue  # Try next available server
+                # Continue to next iteration to try a different server
+                continue
             else:
-                # For other HTTP errors, mark this server as busy
+                # For other HTTP errors, mark this server as busy/unavailable
                 mark(server)
         except Exception:
-            # For other exceptions (network errors, timeouts), mark server as busy
+            # For other exceptions (network errors, timeouts), mark server as busy/unavailable
+            mark(server)
+
+        # If httpx fails or server is busy, fallback to aiohttp for robustness and compatibility
+        try:
+            # Create aiohttp client session with no timeout for streaming
+            async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=None)) as session:
+                # Open async streaming POST request to Jarvis server endpoint with headers and JSON payload
+                async with session.post(host, headers=headers, json=payload) as resp:
+                    # Raise for status to catch HTTP errors
+                    resp.raise_for_status()
+                    # Iterate asynchronously over each line of streaming response as it arrives
+                    async for line_bytes in resp.content:
+                        # Decode bytes to string and strip whitespace
+                        line = line_bytes.decode("utf-8").strip()
+                        # Skip lines that do not start with "data:" prefix as per SSE format
+                        if not line.startswith("data:"):
+                            continue
+                        try:
+                            # Parse JSON data after "data:" prefix which contains incremental response delta
+                            data = json.loads(line[5:])
+                            # Extract incremental delta message from first choice in response
+                            choice = data["choices"][0]["delta"]
+
+                            # On first delta received, detect if 'reasoning' field is present and non-empty
+                            if reasoning_check is None:
+                                reasoning_check = "" if ("reasoning" in choice and choice["reasoning"]) else None
+
+                            # If reasoning is present and mode is not '/no_think' and reasoning not done yet
+                            if (
+                                reasoning_check == ""
+                                and mode != "/no_think"
+                                and not reasoning_done
+                                and "reasoning" in choice
+                                and choice["reasoning"]
+                            ):
+                                reasoning += choice["reasoning"]
+                                yield styles(reasoning=reasoning, content="", expanded=True)
+                                continue
+
+                            # When reasoning ends and content starts, mark reasoning done, yield empty string, then content
+                            if (
+                                reasoning_check == ""
+                                and mode != "/no_think"
+                                and not reasoning_done
+                                and "content" in choice
+                                and choice["content"]
+                            ):
+                                reasoning_done = True
+                                yield ""
+                                content += choice["content"]
+                                yield content
+                                continue
+
+                            # If no reasoning present or reasoning done, accumulate content and yield incrementally
+                            if (
+                                (reasoning_check is None or reasoning_done or mode == "/no_think")
+                                and "content" in choice
+                                and choice["content"]
+                            ):
+                                content += choice["content"]
+                                yield content
+                        except Exception:
+                            # Ignore exceptions during JSON parsing or key access and continue streaming
+                            continue
+            return  # Exit function after successful streaming completion with aiohttp
+
+        except aiohttp.ClientResponseError as e:
+            # If server returns specific error code indicating busy, retry with another server
+            if e.status == error:
+                continue  # Try next available server
+            else:
+                mark(server)  # Mark server as busy/unavailable for other HTTP errors
+        except Exception:
+            # For other exceptions (network errors, timeouts), mark server as busy/unavailable
             mark(server)
 
-    # If all servers tried and none succeeded, yield busy message
+    # If all servers tried and none succeeded, yield a message indicating server busy status
     yield "The server is currently busy. Please wait a moment or try again later."
     return  # End of function

src/tools/audio.py

@@ -3,66 +3,126 @@
 # SPDX-License-Identifier: Apache-2.0
 #
 
-import asyncio  # Import asyncio to enable asynchronous waiting and retries
-import httpx  # Import the httpx library to perform asynchronous HTTP requests efficiently
-from urllib.parse import quote  # Import the quote function to safely encode strings for use in URLs
+import asyncio  # Import asyncio to enable asynchronous programming, including async/await syntax and event loop management
+import httpx  # Import httpx library to perform asynchronous HTTP requests with advanced features like connection pooling and timeout control
+import aiohttp  # Import aiohttp library to provide an alternative asynchronous HTTP client for flexible request handling
+from urllib.parse import quote  # Import quote function to safely encode strings for URL usage, escaping special characters
 from src.utils.ip_generator import generate_ip  # Import a custom utility function to generate random IP addresses
-from config import auth  # Import authentication configuration or credentials from the config module
-from src.utils.tools import initialize_tools  # Import a utility function to initialize and retrieve tool endpoints or resources
+from config import auth  # Import authentication credentials or configuration from the config module (not used directly here but imported for completeness)
+from src.utils.tools import initialize_tools  # Import utility function to initialize and retrieve service endpoints or tool URLs
 
 # Define a class named AudioGeneration to encapsulate functionalities related to generating audio content
 class AudioGeneration:
     # This class provides methods to create audio files based on text instructions and voice parameters
 
-    @staticmethod  # Decorator indicating that the following method does not depend on instance state and can be called on the class itself
-    # Define an asynchronous method to create audio from a text instruction, optionally specifying a voice style
+    """
+    This class provides methods to generate audio files from text instructions asynchronously.
+    It supports retrying requests until successful audio generation is confirmed.
+    """
+
+    @staticmethod  # This method does not depend on class instance state and can be called directly on the class
     async def create_audio(generate_audio_instruction: str, voice: str = "echo") -> str:
         """
-        Generate an audio file URL by sending a request to an audio generation service.
-        This method will keep retrying until a successful response with status code 200 and audio content is received.
+        Asynchronously generate an audio file URL by sending a request to an audio generation service.
+        The method continuously retries until it receives a successful HTTP 200 response with audio content.
 
         Args:
-            generate_audio_instruction (str): The textual instruction or content to convert into audio.
+            generate_audio_instruction (str): The text instruction or content to convert into audio.
             voice (str, optional): The voice style or effect to apply on the generated audio. Defaults to "echo".
 
         Returns:
-            str: The URL to the generated audio file if successful.
+            str: The URL of the generated audio file upon successful retrieval.
 
         Raises:
-            Exception: If the audio generation continuously fails after retries (optional, currently infinite retry).
+            Exception: Currently, the method retries indefinitely and does not raise exceptions on failure.
+        """
+        # Encode the text instruction to make it safe for URL path inclusion by escaping special characters
+        generate_audio_instruct = quote(generate_audio_instruction)
+
+        # Initialize tools and extract the audio generation service endpoint (third element in the returned tuple)
+        _, _, audio_tool = initialize_tools()
+
+        # Construct the full URL by appending the encoded instruction to the base audio tool URL
+        url = f"{audio_tool}/{generate_audio_instruct}"
+
+        # Define query parameters specifying the audio generation model and voice effect
+        params = {
+            "model": "openai-audio",  # Specify the model used by the audio generation service
+            "voice": voice            # Specify the voice style or effect for the generated audio
+        }
+
+        # Create an aiohttp asynchronous HTTP client session with no timeout to allow long-running requests
+        async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=None)) as session:
+            # Enter an infinite loop to retry the request until success criteria are met
+            while True:
+                # Generate a random IP address to spoof the client's origin in the request headers
+                headers = {
+                    "X-Forwarded-For": generate_ip()  # Set the X-Forwarded-For header to a random IP address
+                }
+
+                try:
+                    # Perform an asynchronous GET request to the audio generation service with URL, parameters, and headers
+                    async with session.get(url, params=params, headers=headers) as resp:
+                        # Check if the response status code is 200 (OK) and content type indicates MPEG audio stream
+                        content_type = resp.headers.get('Content-Type', '')
+                        if resp.status == 200 and 'audio/mpeg' in content_type:
+                            # Return the final URL of the generated audio resource as a string
+                            return str(resp.url)
+                        else:
+                            # If the response is not successful or content type is unexpected, wait before retrying
+                            await asyncio.sleep(15)  # Pause for 15 seconds to avoid overwhelming the server
+                except aiohttp.ClientError:
+                    # Catch network-related errors such as connection issues and wait before retrying
+                    await asyncio.sleep(15)  # Pause for 15 seconds before retrying after an exception
+
+    @staticmethod  # Provide an alternative implementation using httpx for flexibility or fallback
+    async def create_audio_httpx(generate_audio_instruction: str, voice: str = "echo") -> str:
         """
-        # Encode the text instruction to make it safe for inclusion in a URL path segment
+        Alternative asynchronous method to generate audio using httpx client.
+        This method also retries indefinitely until a successful response with audio content is received.
+
+        Args:
+            generate_audio_instruction (str): The text instruction to convert into audio.
+            voice (str, optional): Voice style or effect. Defaults to "echo".
+
+        Returns:
+            str: URL of the generated audio file.
+        """
+        # Encode instruction for safe URL usage
         generate_audio_instruct = quote(generate_audio_instruction)
 
-        # Initialize tools and retrieve the audio generation service endpoint from the returned tuple
+        # Initialize tools and get audio generation endpoint
         _, _, audio_tool = initialize_tools()
 
-        # Construct the full URL by appending the encoded instruction to the audio tool's base URL
+        # Construct request URL
         url = f"{audio_tool}/{generate_audio_instruct}"
 
-        # Define query parameters for the HTTP request specifying the model and voice to use for audio generation
+        # Define query parameters for the request
         params = {
-            "model": "openai-audio",  # Specify the audio generation model to be used by the service
-            "voice": voice            # Specify the desired voice style or effect
+            "model": "openai-audio",
+            "voice": voice
         }
 
-        # Create an asynchronous HTTP client with no timeout limit to perform the request
+        # Create an asynchronous HTTP client with no timeout limit
         async with httpx.AsyncClient(timeout=None) as client:
-            # Enter an infinite loop to keep retrying the request until success criteria are met
+            # Retry loop until success
             while True:
-                # Define HTTP headers for the request, including random IP address to simulate different client origins
+              # Define HTTP headers for the request, including random IP address to simulate different client origins
                 headers = {
                     "X-Forwarded-For": generate_ip()  # Generate and set a random IP address for the request header
                 }
 
-                # Send a GET request to the audio generation service with specified URL, parameters, and headers
-                resp = await client.get(url, params=params, headers=headers)
-
-                # Check if the response status code indicates success and the content type is an audio MPEG stream
-                if resp.status_code == 200 and 'audio/mpeg' in resp.headers.get('Content-Type', ''):
-                    # Return the final URL of the generated audio resource as a string
-                    return str(resp.url)
-                else:
-                    # If the response is not successful, wait for a short delay before retrying to avoid hammering the server
-                    await asyncio.sleep(15)  # Pause for 15 second before retrying the request
-                    # The loop will continue and try again without closing the connection prematurely
+                try:
+                    # Send GET request asynchronously
+                    resp = await client.get(url, params=params, headers=headers)
+
+                    # Check for successful response with audio content type
+                    if resp.status_code == 200 and 'audio/mpeg' in resp.headers.get('Content-Type', ''):
+                        # Return the URL of generated audio
+                        return str(resp.url)
+                    else:
+                        # Wait before retrying on failure
+                        await asyncio.sleep(15)
+                except httpx.RequestError:
+                    # Handle network errors and wait before retrying
+                    await asyncio.sleep(15)

src/tools/deep_search.py

@@ -3,8 +3,9 @@
 # SPDX-License-Identifier: Apache-2.0
 #
 
-import aiohttp  # Import the aiohttp library to perform asynchronous HTTP requests
-import asyncio  # Import asyncio library to handle asynchronous operations and implement delay mechanisms
+import asyncio  # Import asyncio for managing asynchronous operations and concurrency
+import aiohttp  # Import aiohttp to perform asynchronous HTTP requests efficiently
+import httpx  # Import httpx library for asynchronous HTTP requests as an alternative client
 from src.utils.ip_generator import generate_ip  # Import function to generate random IP addresses for request headers
 
 # Define the main SearchTools class that provides web searching and URL reading capabilities
@@ -12,124 +13,132 @@ class SearchTools:
     # This class provides methods to connect to the web
 
     """
-    A comprehensive class providing tools to perform web searches and read content from URLs using various search engines
-    and a reader API service. This class implements full asynchronous operations with robust retry mechanisms to ensure
-    connections remain active even when encountering errors.
+    A robust asynchronous class that provides tools to perform web searches and read content from URLs.
+    This implementation uses both aiohttp and httpx libraries to demonstrate usage of multiple async HTTP clients.
+    It includes infinite retry mechanisms to ensure reliability when network errors or timeouts occur.
 
     Attributes:
-        searxng_url (str): Base URL for the SearXNG search proxy service that handles Google and other search engines.
-        baidu_url (str): Base URL for Baidu search engine for Chinese language searches.
+        searxng_url (str): Base URL for the SearXNG search proxy service for Google and Bing searches.
+        baidu_url (str): Base URL for Baidu search engine for Chinese language queries.
         timeout (int): Timeout duration in seconds for HTTP requests to prevent indefinite hanging.
         reader_api (str): Base URL for the reader API service used to extract clean content from URLs.
 
     Methods:
-        read_url(url): Asynchronously reads and returns the textual content of the specified URL using the reader API.
+        read_url(url): Asynchronously reads and returns textual content of the specified URL using the reader API.
         search(query, engine): Asynchronously performs a web search with the given query on the specified search engine,
                                returning the raw HTML response text.
     """
 
-    # Constructor method to initialize the SearchTools instance with all necessary configuration values
     def __init__(self):
         """
-        Initialize the SearchTools instance with predefined URLs and timeout settings.
-        This method sets up all the base URLs and configuration parameters needed for web searching and content reading.
+        Initialize the SearchTools instance with all necessary URLs and timeout settings.
+        This sets up the base URLs for search engines and reader API, along with a default timeout.
         """
-        # Set the base URL for SearXNG search proxy service which provides access to multiple search engines
-        self.searxng_url = "https://paulgo.io/search"
-        # Set the base URL for Baidu search engine for handling Chinese language queries
-        self.baidu_url = "https://www.baidu.com/s"
-        # Set timeout duration to 30 seconds to balance between allowing slow responses and preventing infinite waits
-        self.timeout = 30
-        # Set the reader API endpoint that converts web pages into clean, readable text format
-        self.reader_api = "https://r.jina.ai/"
-
-    # Private helper method that implements the core retry logic for all HTTP requests
-    async def _fetch_with_retry(self, session, method, url, **kwargs):
+        self.searxng_url = "https://paulgo.io/search"  # SearXNG proxy URL for Google/Bing searches
+        self.baidu_url = "https://www.baidu.com/s"  # Baidu search engine base URL
+        self.timeout = 30  # Timeout for HTTP requests in seconds
+        self.reader_api = "https://r.jina.ai/"  # Reader API endpoint to extract clean text from web pages
+
+    async def _fetch_with_retry_aiohttp(self, session: aiohttp.ClientSession, method: str, url: str, **kwargs) -> str:
         """
-        Helper method to perform HTTP requests with infinite retry until a valid response is obtained.
-        This method ensures that connections never fail permanently and will keep trying until success.
+        Internal helper method to perform HTTP requests using aiohttp with infinite retry until success.
 
         Args:
-            session (aiohttp.ClientSession): The aiohttp session object to use for making HTTP requests.
-            method (str): HTTP method to use for the request (e.g., 'get', 'post', 'put', 'delete').
-            url (str): The complete URL to send the request to.
-            **kwargs: Additional keyword arguments to pass to the aiohttp request method (headers, data, etc.).
+            session (aiohttp.ClientSession): aiohttp session object for making requests.
+            method (str): HTTP method to use ('get', 'post', etc.).
+            url (str): The full URL to send the request to.
+            **kwargs: Additional parameters passed to session.request (headers, data, etc.).
 
         Returns:
-            str: The response text content when a successful request is finally achieved.
+            str: The response text content upon successful request.
         """
-        # Create an infinite loop that will only break when a successful response is received
-        while True:
-            # Use a try-except block to catch any type of exception that might occur during the request
+        while True:  # Loop indefinitely until a successful response is received
             try:
-                # Make the actual HTTP request using the provided session, method, URL and additional arguments
+                # Perform the HTTP request asynchronously using aiohttp session
                 async with session.request(method, url, **kwargs) as response:
-                    # Check if the response status indicates success, raise exception if it's an error status
+                    # Raise exception if HTTP status indicates an error (4xx or 5xx)
                     response.raise_for_status()
-                    # Return the text content of the successful response
+                    # Return the response body as text
                     return await response.text()
-            # Catch any exception that occurs during the request process
-            except Exception:
-                # Retry on any exception without stopping the loop or raising the error
-                # Wait for 5 second before attempting the next retry to avoid overwhelming the server
+            except Exception as e:
+                # On any exception (network error, timeout, HTTP error), wait 5 seconds before retrying
+                await asyncio.sleep(5)
+
+    async def _fetch_with_retry_httpx(self, client: httpx.AsyncClient, method: str, url: str, **kwargs) -> str:
+        """
+        Internal helper method to perform HTTP requests using httpx with infinite retry until success.
+
+        Args:
+            client (httpx.AsyncClient): httpx asynchronous client instance.
+            method (str): HTTP method to use ('get', 'post', etc.).
+            url (str): The full URL to send the request to.
+            **kwargs: Additional parameters passed to client.request (headers, data, etc.).
+
+        Returns:
+            str: The response text content upon successful request.
+        """
+        while True:  # Loop indefinitely until a successful response is received
+            try:
+                # Perform the HTTP request asynchronously using httpx client
+                response = await client.request(method, url, **kwargs)
+                # Raise exception if HTTP status indicates an error (4xx or 5xx)
+                response.raise_for_status()
+                # Return the response body as text
+                return response.text
+            except Exception as e:
+                # On any exception (network error, timeout, HTTP error), wait 5 seconds before retrying
                 await asyncio.sleep(5)
 
-    # Public method to read and extract content from any given URL
     async def read_url(self, url: str) -> str:
         """
-        Asynchronously read and retrieve the textual content of a given URL using the reader API with infinite retry.
-        This method will keep trying until it successfully retrieves the content from the specified URL.
+        Asynchronously read and extract textual content from the specified URL using the reader API.
+        This method uses aiohttp client with infinite retry to ensure reliable content retrieval.
 
         Args:
-            url (str): The complete URL of the webpage to read content from.
+            url (str): The full URL of the webpage to extract content from.
 
         Returns:
-            str: The clean textual content extracted from the URL by the reader API service.
+            str: The clean textual content extracted by the reader API.
         """
-        # Prepare the POST data payload containing the target URL for the reader API
+        # Prepare POST data payload containing the target URL for the reader API
         data = {"url": url}
-        # Create an aiohttp client session with the configured timeout settings
+        # Create aiohttp client session with a total timeout configured
         async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=self.timeout)) as session:
-            # Use the retry helper method to POST the URL to the reader API and get the content
-            return await self._fetch_with_retry(session, 'post', self.reader_api, data=data)
+            # Use the internal retry helper to POST the URL to the reader API and return the extracted content
+            return await self._fetch_with_retry_aiohttp(session, 'post', self.reader_api, data=data)
 
-    # Public method to perform web searches using different search engines
     async def search(self, query: str, engine: str = "google") -> str:
         """
-        Asynchronously perform a web search for the given query using the specified search engine with infinite retry.
-        This method will keep trying until it successfully retrieves search results from the chosen search engine.
+        Asynchronously perform a web search for the given query using the specified search engine.
+        This method uses httpx client with infinite retry to reliably fetch search results.
 
         Args:
-            query (str): The search query string containing the terms to search for.
-            engine (str, optional): The search engine to use for the search. Supported values are "google" and "baidu".
-                                    Defaults to "google" if not specified.
+            query (str): The search query string.
+            engine (str, optional): The search engine to use ("google" or "baidu"). Defaults to "google".
 
         Returns:
-            str: The raw HTML content of the search results page from the specified search engine.
+            str: The raw HTML content of the search results page.
         """
-        # Check if the user wants to use Baidu search engine for the query
         if engine == "baidu":
-            # Construct the full URL by combining reader API, Baidu URL and the search query parameter
+            # Construct the full URL for Baidu search with the query parameter 'wd'
             url = f"{self.reader_api}{self.baidu_url}?wd={query}"
-            # Set HTTP headers specific to Baidu search results extraction
+            # Set HTTP headers to target Baidu's main content container and spoof IP address
             headers = {
-                # Target the main content container where Baidu displays search results
-                "X-Target-Selector": "#content_left",
+                "X-Target-Selector": "#content_left",  # CSS selector for Baidu search results container
                 "X-Forwarded-For": generate_ip()  # Random IP address to simulate different client origins
             }
-        # Handle all other search engines (Google, Bing, etc.) through SearXNG proxy
         else:
-            # Determine the search prefix based on the requested engine (Google or Bing)
+            # Determine prefix for SearXNG proxy based on engine: '!go' for Google, '!bi' for Bing
             prefix = "!go" if engine == "google" else "!bi"
-            # Construct the full URL by combining reader API, SearXNG URL, prefix and query
+            # Construct the full URL for SearXNG search proxy with query and prefix
             url = f"{self.reader_api}{self.searxng_url}?q={prefix} {query}"
-            # Set HTTP headers specific to SearXNG search results extraction
+            # Set HTTP headers to target SearXNG search results container and spoof IP address
             headers = {
-                # Target the URLs container where SearXNG displays search result links
-                "X-Target-Selector": "#urls",
+                "X-Target-Selector": "#urls",  # CSS selector for SearXNG search results container
                 "X-Forwarded-For": generate_ip()  # Random IP address to simulate different client origins
             }
-        # Create an aiohttp client session with the configured timeout settings
-        async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=self.timeout)) as session:
-            # Use the retry helper method to GET the search results and return the HTML content
-            return await self._fetch_with_retry(session, 'get', url, headers=headers)
+
+        # Create httpx asynchronous client with timeout configured
+        async with httpx.AsyncClient(timeout=self.timeout) as client:
+            # Use the internal retry helper to GET the search results and return the HTML content
+            return await self._fetch_with_retry_httpx(client, 'get', url, headers=headers)

src/tools/image.py

@@ -3,114 +3,127 @@
 # SPDX-License-Identifier: Apache-2.0
 #
 
-import httpx  # Import httpx library for performing asynchronous HTTP requests efficiently
-from urllib.parse import quote  # Import quote function to safely encode strings for use in URLs
-from typing import Optional  # Import Optional type hint for parameters that can be None
+import asyncio  # Import asyncio for asynchronous programming and managing event loops
+import httpx  # Import httpx for async HTTP requests with HTTP/1.1 and HTTP/2 support
+import aiohttp  # Import aiohttp for alternative async HTTP client capabilities
+from urllib.parse import quote  # Import quote to safely encode URL path components
+from typing import Optional  # Import Optional for type hinting parameters that may be None
 from src.utils.ip_generator import generate_ip  # Import custom utility to generate random IP addresses for request headers
-from src.utils.tools import initialize_tools  # Import utility function to initialize and retrieve tool endpoints
+from src.utils.tools import initialize_tools  # Import utility to initialize and get tool endpoints
 
 # Define a class named ImageGeneration to encapsulate functionalities related to generating image content
 class ImageGeneration:
     # This class provides methods to create image files based on text instructions
 
     """
-    A class to handle image generation requests to an external image generation service.
+    Class to handle asynchronous image generation requests to an external service.
 
     Attributes:
-        FORMATS (dict): A dictionary mapping image format names to their (width, height) dimensions.
+        FORMATS (dict): Maps image format names to (width, height) tuples.
 
     Methods:
-        create_image: Asynchronously generates an image based on a textual instruction and parameters,
-                      returning the URL of the generated image.
+        create_image: Async method to generate an image URL from a text prompt,
+                      retrying until successful, using httpx and aiohttp.
     """
 
-    # Image formats
+    # Supported image formats with their dimensions (width, height)
     FORMATS = {
-        "default": (1024, 1024),  # Default square image size (width x height)
-        "square": (1024, 1024),  # Square image format with equal width and height
-        "landscape": (1024, 768),  # Landscape format with wider width than height
-        "landscape_large": (1440, 1024),  # Larger landscape format with increased resolution
-        "portrait": (768, 1024),  # Portrait format with taller height than width
-        "portrait_large": (1024, 1440),  # Larger portrait format with increased resolution
+        "default": (1024, 1024),
+        "square": (1024, 1024),
+        "landscape": (1024, 768),
+        "landscape_large": (1440, 1024),
+        "portrait": (768, 1024),
+        "portrait_large": (1024, 1440),
     }
 
-    @staticmethod  # Decorator indicating that the following method does not depend on instance state and can be called on the class itself
-    # Define an asynchronous method to create image from a text instruction
+    @staticmethod
     async def create_image(
-        generate_image_instruction: str,  # Text instruction describing the image to generate
-        image_format: str = "default",  # Desired image format key from FORMATS dictionary
-        model: Optional[str] = "flux-realism",  # Optional model name for image generation; defaults to 'flux-realism'
-        seed: Optional[int] = None,  # Optional seed value for randomization control in image generation
-        nologo: bool = True,  # Whether to generate image without logo watermark; defaults to True
-        private: bool = True,  # Whether the generated image should be private; defaults to True
-        enhance: bool = True,  # Whether to apply enhancement filters to the generated image; defaults to True
+        generate_image_instruction: str,  # Text description for the image to generate
+        image_format: str = "default",  # Format key from FORMATS dict
+        model: Optional[str] = "flux-realism",  # Model name for generation, default 'flux-realism'
+        seed: Optional[int] = None,  # Optional seed for reproducible randomness
+        nologo: bool = True,  # Whether to exclude logo watermark
+        private: bool = True,  # Whether the image should be private
+        enhance: bool = True,  # Whether to apply enhancement filters
     ) -> str:
         """
-        Asynchronously generate an image URL by sending a request to the image generation service.
-        This method will keep retrying until a successful response with status code 200 is received.
+        Asynchronously generate an image URL by sending requests to the image generation service.
+        Uses httpx for initial requests and aiohttp as fallback, retrying indefinitely until success.
 
         Args:
-            generate_image_instruction (str): The textual instruction or description for the desired image.
-            image_format (str, optional): The format key specifying image dimensions. Defaults to "default".
-            model (Optional[str], optional): The image generation model to use. Defaults to "flux-realism".
-            seed (Optional[int], optional): Seed for randomization to reproduce images. Defaults to None.
-            nologo (bool, optional): Flag to exclude logo watermark. Defaults to True.
-            private (bool, optional): Flag to mark image as private. Defaults to True.
-            enhance (bool, optional): Flag to apply image enhancement. Defaults to True.
+            generate_image_instruction (str): Text prompt describing the desired image.
+            image_format (str): Key for image dimensions.
+            model (Optional[str]): Model to use for generation.
+            seed (Optional[int]): Seed for randomization control.
+            nologo (bool): Flag to exclude logo watermark.
+            private (bool): Flag to mark image as private.
+            enhance (bool): Flag to apply image enhancement.
 
         Returns:
-            str: The URL of the generated image if the request is successful.
+            str: URL of the generated image on success.
 
         Raises:
-            ValueError: If the specified image_format is not supported.
-            Exception: If the image generation continuously fails (currently infinite retry).
+            ValueError: If image_format is invalid.
         """
-        # Validate that the requested image format exists in the FORMATS dictionary
+        # Validate image format key
         if image_format not in ImageGeneration.FORMATS:
             raise ValueError("Invalid image format.")
 
-        # Retrieve width and height based on the requested image format
+        # Extract width and height for the requested format
         width, height = ImageGeneration.FORMATS[image_format]
 
-        # Initialize tools and retrieve the image generation service endpoint
+        # Initialize tools and get image generation service endpoint URL
         _, image_tool, _ = initialize_tools()
 
-        # Encode the image instruction to safely include it in the URL path
+        # Encode instruction safely for URL path usage
         generate_image_instruct = quote(generate_image_instruction)
 
-        # Construct the full URL for the image generation request by appending the encoded instruction
-        url = f"{image_tool}{generate_image_instruct}"  # Full endpoint URL for image generation
+        # Construct the full URL endpoint for image generation
+        url = f"{image_tool}{generate_image_instruct}"
 
-        # Prepare query parameters including image dimensions, model, and flags converted to string "true"/"false"
+        # Prepare query parameters with image size, model, flags as strings
         params = {
-            "width": width,  # Image width parameter
-            "height": height,  # Image height parameter
-            "model": model,  # Model name for image generation
-            "nologo": "true" if nologo else "false",  # Flag to exclude logo watermark as string
-            "private": "true" if private else "false",  # Flag to mark image as private as string
-            "enhance": "true" if enhance else "false"  # Flag to apply enhancement as string
+            "width": width,
+            "height": height,
+            "model": model,
+            "nologo": "true" if nologo else "false",
+            "private": "true" if private else "false",
+            "enhance": "true" if enhance else "false",
         }
 
-        # Include seed parameter if provided to control randomness in image generation
+        # Add seed parameter if provided
         if seed is not None:
-            params["seed"] = seed  # Add seed to parameters to reproduce images
+            params["seed"] = seed
 
-        # Prepare HTTP headers with a generated random IP to simulate different client origins
+        # Prepare headers
         headers = {
             "X-Forwarded-For": generate_ip()  # Random IP address for request header to simulate client origin
         }
 
-        # Create an asynchronous HTTP client with no timeout limit to perform the request
+        # Use httpx.AsyncClient with no timeout for initial requests
         async with httpx.AsyncClient(timeout=None) as client:
-            # Keep retrying the request until a successful response with status 200 is received
             while True:
-                # Send a GET request to the image generation service with URL, parameters, and headers
-                resp = await client.get(url, params=params, headers=headers)
-
-                # Check if the response status code indicates success
-                if resp.status_code == 200:
-                    # Return the URL of the generated image as a string
-                    return str(resp.url)
-                else:
-                    # Wait briefly before retrying to avoid overwhelming the server
-                    await asyncio.sleep(15)  # Pause 15 second before retrying
+                try:
+                    # Send GET request to the image generation endpoint
+                    resp = await client.get(url, params=params, headers=headers)
+
+                    # If response is successful, return the final URL
+                    if resp.status_code == 200:
+                        return str(resp.url)
+                except httpx.HTTPError:
+                    # On httpx errors, fallback to aiohttp for robustness
+                    pass
+
+                # Fallback retry with aiohttp client
+                async with aiohttp.ClientSession() as session:
+                    try:
+                        async with session.get(url, params=params, headers=headers) as resp:
+                            if resp.status == 200:
+                                # Return the final URL (aiohttp does not provide direct URL property)
+                                return str(resp.url)
+                    except aiohttp.ClientError:
+                        # Ignore aiohttp errors and retry
+                        pass
+
+                # Wait 15 seconds before retrying to avoid overwhelming the server
+                await asyncio.sleep(15)