Upload 49 files

gemini_webapi/__init__.py

@@ -0,0 +1,6 @@
+# flake8: noqa
+
+from .client import GeminiClient, ChatSession
+from .exceptions import *
+from .types import *
+from .utils import set_log_level, logger

gemini_webapi/client.py

@@ -0,0 +1,1871 @@
+import asyncio
+import codecs
+import io
+import random
+import re
+import time
+from asyncio import Task
+from pathlib import Path
+from typing import Any, AsyncGenerator, Optional
+
+import orjson as json
+from curl_cffi.requests import AsyncSession, Cookies, Response
+from curl_cffi.requests.exceptions import ReadTimeout
+
+from .components import GemMixin
+from .constants import (
+    Endpoint,
+    ErrorCode,
+    GRPC,
+    Model,
+    TEMPORARY_CHAT_FLAG_INDEX,
+    STREAMING_FLAG_INDEX,
+    GEM_FLAG_INDEX,
+)
+from .exceptions import (
+    APIError,
+    AuthError,
+    GeminiError,
+    ModelInvalid,
+    TemporarilyBlocked,
+    TimeoutError,
+    UsageLimitExceeded,
+)
+from .types import (
+    Candidate,
+    Gem,
+    GeneratedImage,
+    ModelOutput,
+    RPCData,
+    WebImage,
+    AvailableModel,
+    ChatInfo,
+    ChatTurn,
+    ChatHistory,
+    GeneratedVideo,
+)
+from .utils import (
+    extract_json_from_response,
+    get_access_token,
+    get_delta_by_fp_len,
+    get_nested_value,
+    logger,
+    parse_file_name,
+    parse_response_by_frame,
+    rotate_1psidts,
+    running,
+    upload_file,
+)
+
+_CARD_CONTENT_RE = re.compile(r"^http://googleusercontent\.com/card_content/\d+")
+_ARTIFACTS_RE = re.compile(r"http://googleusercontent\.com/\w+/\d+\n*")
+_DEFAULT_METADATA: list[Any] = ["", "", "", None, None, None, None, None, None, ""]
+
+
+class GeminiClient(GemMixin):
+    """
+    Async requests client interface for gemini.google.com.
+
+    `secure_1psid` must be provided unless the optional dependency `browser-cookie3` is installed, and
+    you have logged in to google.com in your local browser.
+
+    Parameters
+    ----------
+    secure_1psid: `str`, optional
+        __Secure-1PSID cookie value.
+    secure_1psidts: `str`, optional
+        __Secure-1PSIDTS cookie value, some Google accounts don't require this value, provide only if it's in the cookie list.
+    proxy: `str`, optional
+        Proxy URL.
+    kwargs: `dict`, optional
+        Additional arguments which will be passed to the http client.
+        Refer to `curl_cffi.requests.AsyncSession` for more information.
+
+    Raises
+    ------
+    `ValueError`
+        If `browser-cookie3` is installed but cookies for google.com are not found in your local browser storage.
+    """
+
+    __slots__ = [
+        "_cookies",
+        "proxy",
+        "_running",
+        "client",
+        "access_token",
+        "build_label",
+        "session_id",
+        "timeout",
+        "auto_close",
+        "close_delay",
+        "close_task",
+        "auto_refresh",
+        "refresh_interval",
+        "refresh_task",
+        "verbose",
+        "watchdog_timeout",
+        "_lock",
+        "_reqid",
+        "_gems",  # From GemMixin
+        "_available_models",
+        "_recent_chats",
+        "kwargs",
+    ]
+
+    def __init__(
+        self,
+        secure_1psid: str | None = None,
+        secure_1psidts: str | None = None,
+        proxy: str | None = None,
+        **kwargs,
+    ):
+        super().__init__()
+        self._cookies = Cookies()
+        self.proxy = proxy
+        self._running: bool = False
+        self.client: AsyncSession | None = None
+        self.access_token: str | None = None
+        self.build_label: str | None = None
+        self.session_id: str | None = None
+        self.timeout: float = 600
+        self.auto_close: bool = False
+        self.close_delay: float = 600
+        self.close_task: Task | None = None
+        self.auto_refresh: bool = True
+        self.refresh_interval: float = 600
+        self.refresh_task: Task | None = None
+        self.verbose: bool = True
+        self.watchdog_timeout: float = 90
+        self._lock = asyncio.Lock()
+        self._reqid: int = random.randint(10000, 99999)
+
+        self._available_models: list[AvailableModel] | None = None
+        self._recent_chats: list[ChatInfo] | None = None
+        self.kwargs = kwargs
+
+        if secure_1psid:
+            self._cookies.set("__Secure-1PSID", secure_1psid, domain=".google.com")
+            if secure_1psidts:
+                self._cookies.set(
+                    "__Secure-1PSIDTS", secure_1psidts, domain=".google.com"
+                )
+
+    @property
+    def cookies(self) -> Cookies:
+        """
+        Returns the cookies used for the current session.
+        """
+        return self.client.cookies if self.client else self._cookies
+
+    @cookies.setter
+    def cookies(self, value: Cookies | dict):
+        if isinstance(value, Cookies):
+            self._cookies.update(value)
+        elif isinstance(value, dict):
+            for k, v in value.items():
+                self._cookies.set(k, v, domain=".google.com")
+
+        if self.client:
+            self.client.cookies.update(self._cookies)
+
+    async def init(
+        self,
+        timeout: float = 600,
+        auto_close: bool = False,
+        close_delay: float = 600,
+        auto_refresh: bool = True,
+        refresh_interval: float = 600,
+        verbose: bool = True,
+        watchdog_timeout: float = 90,
+    ) -> None:
+        """
+        Get SNlM0e value as access token. Without this token posting will fail with 400 bad request.
+
+        Parameters
+        ----------
+        timeout: `float`, optional
+            Request timeout of the client in seconds. Used to limit the max waiting time when sending a request.
+        auto_close: `bool`, optional
+            If `True`, the client will close connections and clear resource usage after a certain period
+            of inactivity. Useful for always-on services.
+        close_delay: `float`, optional
+            Time to wait before auto-closing the client in seconds. Effective only if `auto_close` is `True`.
+        auto_refresh: `bool`, optional
+            If `True`, will schedule a task to automatically refresh cookies and access token in the background.
+        refresh_interval: `float`, optional
+            Time interval for background cookie and access token refresh in seconds.
+            Effective only if `auto_refresh` is `True`.
+        verbose: `bool`, optional
+            If `True`, will print more infomation in logs.
+        watchdog_timeout: `float`, optional
+            Timeout in seconds for shadow retry watchdog. If no data receives from stream but connection is active,
+            client will retry automatically after this duration.
+        """
+
+        async with self._lock:
+            if self._running:
+                return
+
+            try:
+                self.verbose = verbose
+                self.watchdog_timeout = watchdog_timeout
+                access_token, build_label, session_id, session = await get_access_token(
+                    base_cookies=self.cookies,
+                    proxy=self.proxy,
+                    verbose=self.verbose,
+                    verify=self.kwargs.get("verify", True),
+                )
+
+                session.timeout = timeout
+                self.client = session
+                self._cookies.update(self.client.cookies)
+                self.access_token = access_token
+                self.build_label = build_label
+                self.session_id = session_id
+                self._running = True
+                self._reqid = random.randint(10000, 99999)
+
+                self.timeout = timeout
+                self.auto_close = auto_close
+                self.close_delay = close_delay
+                if self.auto_close:
+                    await self.reset_close_task()
+
+                self.auto_refresh = auto_refresh
+                self.refresh_interval = refresh_interval
+
+                if self.refresh_task:
+                    self.refresh_task.cancel()
+                    self.refresh_task = None
+
+                if self.auto_refresh:
+                    self.refresh_task = asyncio.create_task(self.start_auto_refresh())
+
+                await self._init_rpc()
+
+                if self.verbose:
+                    logger.success("Gemini client initialized successfully.")
+            except Exception:
+                await self.close()
+                raise
+
+    async def close(self, delay: float = 0) -> None:
+        """
+        Close the client after a certain period of inactivity, or call manually to close immediately.
+
+        Parameters
+        ----------
+        delay: `float`, optional
+            Time to wait before closing the client in seconds.
+        """
+
+        if delay:
+            await asyncio.sleep(delay)
+
+        self._running = False
+
+        if self.close_task:
+            self.close_task.cancel()
+            self.close_task = None
+
+        if self.refresh_task:
+            self.refresh_task.cancel()
+            self.refresh_task = None
+
+        if self.client:
+            self._cookies.update(self.client.cookies)
+            await self.client.close()
+            self.client = None
+
+    async def reset_close_task(self) -> None:
+        """
+        Reset the timer for closing the client when a new request is made.
+        """
+
+        if self.close_task:
+            self.close_task.cancel()
+            self.close_task = None
+
+        self.close_task = asyncio.create_task(self.close(self.close_delay))
+
+    async def start_auto_refresh(self) -> None:
+        """
+        Start the background task to automatically refresh cookies.
+        """
+        if self.refresh_interval < 60:
+            self.refresh_interval = 60
+
+        while self._running:
+            await asyncio.sleep(self.refresh_interval)
+
+            if not self._running:
+                break
+
+            try:
+                async with self._lock:
+                    # Refresh all cookies in the background to keep the session alive.
+                    new_1psidts = await rotate_1psidts(self.client, self.verbose)
+
+                    if new_1psidts:
+                        logger.debug("Cookies refreshed (network update).")
+                    else:
+                        logger.warning(
+                            "Rotation response did not contain a new __Secure-1PSIDTS. "
+                            "Session might expire soon if this persists."
+                        )
+            except asyncio.CancelledError:
+                raise
+            except AuthError:
+                logger.warning(
+                    "AuthError: Failed to refresh cookies. Retrying in next interval."
+                )
+            except Exception:
+                logger.warning(
+                    "Unexpected error while refreshing cookies. Retrying in next interval."
+                )
+
+    async def _init_rpc(self) -> None:
+        """
+        Send initial RPC calls to set up the session.
+        """
+        await self._fetch_models()
+        await self._send_bard_settings()
+        await self._send_bard_activity()
+        await self._fetch_recent_chats()
+
+    async def _fetch_models(self) -> None:
+        """
+        Fetch and parse available models.
+        """
+        response = await self._batch_execute(
+            [
+                RPCData(
+                    rpcid=GRPC.LIST_MODELS,
+                    payload="[]",
+                )
+            ]
+        )
+
+        response_json = extract_json_from_response(response.text)
+
+        available_models = []
+        for part in response_json:
+            part_body_str = get_nested_value(part, [2])
+            if not part_body_str:
+                continue
+
+            part_body = json.loads(part_body_str)
+
+            models_list = get_nested_value(part_body, [15])
+            if isinstance(models_list, list):
+                for model_data in models_list:
+                    if isinstance(model_data, list) and len(model_data) > 2:
+                        model_id = get_nested_value(model_data, [0], "")
+                        name = get_nested_value(model_data, [10]) or get_nested_value(
+                            model_data, [1], ""
+                        )
+                        description = get_nested_value(
+                            model_data, [12]
+                        ) or get_nested_value(model_data, [2], "")
+                        core_model = Model.UNSPECIFIED
+                        code_name = "unspecified"
+                        for enum_model in Model:
+                            val = enum_model.model_header.get(
+                                "x-goog-ext-525001261-jspb", ""
+                            )
+                            if val and (model_id in val):
+                                core_model = enum_model
+                                code_name = enum_model.model_name
+                                break
+
+                        if model_id and name:
+                            available_models.append(
+                                AvailableModel(
+                                    id=code_name,
+                                    name=name,
+                                    model=core_model,
+                                    description=description,
+                                )
+                            )
+                break
+
+        self._available_models = available_models
+
+    async def _fetch_recent_chats(self, recent: int = 13) -> None:
+        """
+        Fetch and parse recent chats.
+        """
+        response_chats1 = await self._batch_execute(
+            [
+                RPCData(
+                    rpcid=GRPC.LIST_CHATS,
+                    payload=json.dumps([recent, None, [1, None, 1]]).decode("utf-8"),
+                ),
+            ]
+        )
+        response_chats2 = await self._batch_execute(
+            [
+                RPCData(
+                    rpcid=GRPC.LIST_CHATS,
+                    payload=json.dumps([recent, None, [0, None, 1]]).decode("utf-8"),
+                ),
+            ]
+        )
+
+        recent_chats: list[ChatInfo] = []
+        for response_chats in (response_chats1, response_chats2):
+            chats_json = extract_json_from_response(response_chats.text)
+            for part in chats_json:
+                part_body_str = get_nested_value(part, [2])
+                if not part_body_str:
+                    continue
+
+                try:
+                    part_body = json.loads(part_body_str)
+                except json.JSONDecodeError:
+                    continue
+
+                chat_list = get_nested_value(part_body, [2])
+                if isinstance(chat_list, list):
+                    for chat_data in chat_list:
+                        if isinstance(chat_data, list) and len(chat_data) > 1:
+                            cid = get_nested_value(chat_data, [0], "")
+                            title = get_nested_value(chat_data, [1], "")
+                            is_pinned = bool(get_nested_value(chat_data, [2]))
+
+                            if cid and title:
+                                if not any(c.cid == cid for c in recent_chats):
+                                    recent_chats.append(
+                                        ChatInfo(
+                                            cid=cid, title=title, is_pinned=is_pinned
+                                        )
+                                    )
+                    break
+
+        self._recent_chats = recent_chats
+
+    async def _send_bard_settings(self) -> None:
+        """
+        Send required setup activity to Gemini.
+        """
+        await self._batch_execute(
+            [
+                RPCData(
+                    rpcid=GRPC.BARD_SETTINGS,
+                    payload='[[["adaptive_device_responses_enabled","advanced_mode_theme_override_triggered","advanced_zs_upsell_dismissal_count","advanced_zs_upsell_last_dismissed","ai_transparency_notice_dismissed","audio_overview_discovery_dismissal_count","audio_overview_discovery_last_dismissed","bard_in_chrome_link_sharing_enabled","bard_sticky_mode_disabled_count","canvas_create_discovery_tooltip_seen_count","combined_files_button_tag_seen_count","indigo_banner_explicit_dismissal_count","indigo_banner_impression_count","indigo_banner_last_seen_sec","current_popup_id","deep_research_has_seen_file_upload_tooltip","deep_research_model_update_disclaimer_display_count","default_bot_id","disabled_discovery_card_feature_ids","disabled_model_discovery_tooltip_feature_ids","disabled_mode_disclaimers","disabled_new_model_badge_mode_ids","disabled_settings_discovery_tooltip_feature_ids","disablement_disclaimer_last_dismissed_sec","disable_advanced_beta_dialog","disable_advanced_beta_non_en_banner","disable_advanced_resubscribe_ui","disable_at_mentions_discovery_tooltip","disable_autorun_fact_check_u18","disable_bot_create_tips_card","disable_bot_docs_in_gems_disclaimer","disable_bot_onboarding_dialog","disable_bot_save_reminder_tips_card","disable_bot_send_prompt_tips_card","disable_bot_shared_in_drive_disclaimer","disable_bot_try_create_tips_card","disable_colab_tooltip","disable_collapsed_tool_menu_tooltip","disable_continue_discovery_tooltip","disable_debug_info_moved_tooltip_v2","disable_enterprise_mode_dialog","disable_export_python_tooltip","disable_extensions_discovery_dialog","disable_extension_one_time_badge","disable_fact_check_tooltip_v2","disable_free_file_upload_tips_card","disable_generated_image_download_dialog","disable_get_app_banner","disable_get_app_desktop_dialog","disable_googler_in_enterprise_mode","disable_human_review_disclosure","disable_ice_open_vega_editor_tooltip","disable_image_upload_tooltip","disable_legal_concern_tooltip","disable_llm_history_import_disclaimer","disable_location_popup","disable_memory_discovery","disable_memory_extraction_discovery","disable_new_conversation_dialog","disable_onboarding_experience","disable_personal_context_tooltip","disable_photos_upload_disclaimer","disable_power_up_intro_tooltip","disable_scheduled_actions_mobile_notification_snackbar","disable_storybook_listen_button_tooltip","disable_streaming_settings_tooltip","disable_take_control_disclaimer","disable_teens_only_english_language_dialog","disable_tier1_rebranding_tooltip","disable_try_advanced_mode_dialog","enable_advanced_beta_mode","enable_advanced_mode","enable_googler_in_enterprise_mode","enable_memory","enable_memory_extraction","enable_personal_context","enable_personal_context_gemini","enable_personal_context_gemini_using_photos","enable_personal_context_gemini_using_workspace","enable_personal_context_search","enable_personal_context_youtube","enable_token_streaming","enforce_default_to_fast_version","mayo_discovery_banner_dismissal_count","mayo_discovery_banner_last_dismissed_sec","gempix_discovery_banner_dismissal_count","gempix_discovery_banner_last_dismissed","get_app_banner_ack_count","get_app_banner_seen_count","get_app_mobile_dialog_ack_count","guided_learning_banner_dismissal_count","guided_learning_banner_last_dismissed","has_accepted_agent_mode_fre_disclaimer","has_received_streaming_response","has_seen_agent_mode_tooltip","has_seen_bespoke_tooltip","has_seen_deepthink_mustard_tooltip","has_seen_deepthink_v2_tooltip","has_seen_deep_think_tooltip","has_seen_first_youtube_video_disclaimer","has_seen_ggo_tooltip","has_seen_image_grams_discovery_banner","has_seen_image_preview_in_input_area_tooltip","has_seen_kallo_discovery_banner","has_seen_kallo_tooltip","has_seen_model_picker_in_input_area_tooltip","has_seen_model_tooltip_in_input_area_for_gempix","has_seen_redo_with_gempix2_tooltip","has_seen_veograms_discovery_banner","has_seen_video_generation_discovery_banner","is_imported_chats_panel_open_by_default","jumpstart_onboarding_dismissal_count","last_dismissed_deep_research_implicit_invite","last_dismissed_discovery_feature_implicit_invites","last_dismissed_immersives_canvas_implicit_invite","last_dismissed_immersive_share_disclaimer_sec","last_dismissed_strike_timestamp_sec","last_dismissed_zs_student_aip_banner_sec","last_get_app_banner_ack_timestamp_sec","last_get_app_mobile_dialog_ack_timestamp_sec","last_human_review_disclosure_ack","last_selected_mode_id_in_embedded","last_selected_mode_id_on_web","last_two_up_activation_timestamp_sec","last_winter_olympics_interaction_timestamp_sec","memory_extracted_greeting_name","mini_gemini_tos_closed","mode_switcher_soft_badge_disabled_ids","mode_switcher_soft_badge_seen_count","personalization_first_party_onboarding_cross_surface_clicked","personalization_first_party_onboarding_cross_surface_seen_count","personalization_one_p_discovery_card_seen_count","personalization_one_p_discovery_last_consented","personalization_zero_state_card_last_interacted","personalization_zero_state_card_seen_count","popup_zs_visits_cooldown","require_reconsent_setting_for_personalization_banner_seen_count","show_debug_info","side_nav_open_by_default","student_verification_dismissal_count","student_verification_last_dismissed","task_viewer_cc_banner_dismissed_count","task_viewer_cc_banner_dismissed_time_sec","tool_menu_new_badge_disabled_ids","tool_menu_new_badge_impression_counts","tool_menu_soft_badge_disabled_ids","tool_menu_soft_badge_impression_counts","upload_disclaimer_last_consent_time_sec","viewed_student_aip_upsell_campaign_ids","voice_language","voice_name","web_and_app_activity_enabled","wellbeing_nudge_notice_last_dismissed_sec","zs_student_aip_banner_dismissal_count"]]]',
+                )
+            ]
+        )
+
+    async def _send_bard_activity(self) -> None:
+        """
+        Send warmup RPC calls before querying.
+        """
+        await self._batch_execute(
+            [
+                RPCData(
+                    rpcid=GRPC.BARD_SETTINGS,
+                    payload='[[["bard_activity_enabled"]]]',
+                )
+            ]
+        )
+
+    def list_models(self) -> list[AvailableModel] | None:
+        """
+        List all available models for the current account.
+
+        Returns
+        -------
+        `list[gemini_webapi.types.AvailableModel]`
+            List of models with their name and description. Returns `None` if the client holds no session cache.
+        """
+        return self._available_models
+
+    async def generate_content(
+        self,
+        prompt: str,
+        files: list[str | Path | bytes | io.BytesIO] | None = None,
+        model: Model | str | dict = Model.UNSPECIFIED,
+        gem: Gem | str | None = None,
+        chat: Optional["ChatSession"] = None,
+        temporary: bool = False,
+        **kwargs,
+    ) -> ModelOutput:
+        """
+        Generates contents with prompt.
+
+        Parameters
+        ----------
+        prompt: `str`
+            Text prompt provided by user.
+        files: `list[str | Path | bytes | io.BytesIO]`, optional
+            List of file paths or byte streams to be attached.
+        model: `Model | str | dict`, optional
+            Specify the model to use for generation.
+            Pass either a `gemini_webapi.constants.Model` enum or a model name string to use predefined models.
+            Pass a dictionary to use custom model header strings ("model_name" and "model_header" keys must be provided).
+        gem: `Gem | str`, optional
+            Specify a gem to use as system prompt for the chat session.
+            Pass either a `gemini_webapi.types.Gem` object or a gem id string.
+        chat: `ChatSession`, optional
+            Chat data to retrieve conversation history.
+            If None, will automatically generate a new chat id when sending post request.
+        temporary: `bool`, optional
+            If set to `True`, the ongoing conversation will not show up in Gemini history.
+        kwargs: `dict`, optional
+            Additional arguments which will be passed to the post request.
+            Refer to `curl_cffi.requests.AsyncSession.request` for more information.
+
+        Returns
+        -------
+        :class:`ModelOutput`
+            Output data from gemini.google.com.
+
+        Raises
+        ------
+        `AssertionError`
+            If prompt is empty.
+        `gemini_webapi.TimeoutError`
+            If request timed out.
+        `gemini_webapi.GeminiError`
+            If no reply candidate found in response.
+        `gemini_webapi.APIError`
+            - If request failed with status code other than 200.
+            - If response structure is invalid and failed to parse.
+        """
+
+        if self.auto_close:
+            await self.reset_close_task()
+
+        file_data = None
+        if files:
+            await self._send_bard_activity()
+
+            uploaded_urls = await asyncio.gather(
+                *(
+                    upload_file(file, client=self.client, verbose=self.verbose)
+                    for file in files
+                )
+            )
+            file_data = [
+                [[url], parse_file_name(file)]
+                for url, file in zip(uploaded_urls, files)
+            ]
+
+        try:
+            await self._send_bard_activity()
+
+            session_state = {
+                "last_texts": {},
+                "last_thoughts": {},
+                "last_progress_time": time.time(),
+                "is_thinking": False,
+                "is_queueing": False,
+                "title": None,
+            }
+            output = None
+            async for output in self._generate(
+                prompt=prompt,
+                req_file_data=file_data,
+                model=model,
+                gem=gem,
+                chat=chat,
+                temporary=temporary,
+                session_state=session_state,
+                **kwargs,
+            ):
+                pass
+
+            if output is None:
+                raise GeminiError(
+                    "Failed to generate contents. No output data found in response."
+                )
+
+            if isinstance(chat, ChatSession):
+                output.metadata = chat.metadata
+                chat.last_output = output
+
+            return output
+
+        finally:
+            if files:
+                for file in files:
+                    if isinstance(file, io.BytesIO):
+                        file.close()
+
+    async def generate_content_stream(
+        self,
+        prompt: str,
+        files: list[str | Path | bytes | io.BytesIO] | None = None,
+        model: Model | str | dict = Model.UNSPECIFIED,
+        gem: Gem | str | None = None,
+        chat: Optional["ChatSession"] = None,
+        temporary: bool = False,
+        **kwargs,
+    ) -> AsyncGenerator[ModelOutput, None]:
+        """
+        Generates contents with prompt in streaming mode.
+
+        This method sends a request to Gemini and yields partial responses as they arrive.
+        It automatically calculates the text delta (new characters) to provide a smooth
+        streaming experience. It also continuously updates chat metadata and candidate IDs.
+
+        Parameters
+        ----------
+        prompt: `str`
+            Text prompt provided by user.
+        files: `list[str | Path | bytes | io.BytesIO]`, optional
+            List of file paths or byte streams to be attached.
+        model: `Model | str | dict`, optional
+            Specify the model to use for generation.
+        gem: `Gem | str`, optional
+            Specify a gem to use as system prompt for the chat session.
+        chat: `ChatSession`, optional
+            Chat data to retrieve conversation history.
+        temporary: `bool`, optional
+            If set to `True`, the ongoing conversation will not show up in Gemini history.
+        kwargs: `dict`, optional
+            Additional arguments passed to `curl_cffi.requests.AsyncSession.stream`.
+
+        Yields
+        ------
+        :class:`ModelOutput`
+            Partial output data. The `text_delta` attribute contains only the NEW characters
+            received since the last yield.
+
+        Raises
+        ------
+        `gemini_webapi.APIError`
+            If the request fails or response structure is invalid.
+        `gemini_webapi.TimeoutError`
+            If the stream request times out.
+        """
+
+        if self.auto_close:
+            await self.reset_close_task()
+
+        file_data = None
+        if files:
+            await self._send_bard_activity()
+
+            uploaded_urls = await asyncio.gather(
+                *(
+                    upload_file(file, client=self.client, verbose=self.verbose)
+                    for file in files
+                )
+            )
+            file_data = [
+                [[url], parse_file_name(file)]
+                for url, file in zip(uploaded_urls, files)
+            ]
+
+        try:
+            await self._send_bard_activity()
+
+            session_state = {
+                "last_texts": {},
+                "last_thoughts": {},
+                "last_progress_time": time.time(),
+                "is_thinking": False,
+                "is_queueing": False,
+                "title": None,
+            }
+            output = None
+            async for output in self._generate(
+                prompt=prompt,
+                req_file_data=file_data,
+                model=model,
+                gem=gem,
+                chat=chat,
+                temporary=temporary,
+                session_state=session_state,
+                **kwargs,
+            ):
+                yield output
+
+            if output and isinstance(chat, ChatSession):
+                output.metadata = chat.metadata
+                chat.last_output = output
+
+        finally:
+            if files:
+                for file in files:
+                    if isinstance(file, io.BytesIO):
+                        file.close()
+
+    @running(retry=5)
+    async def _generate(
+        self,
+        prompt: str,
+        req_file_data: list[Any] | None = None,
+        model: Model | str | dict = Model.UNSPECIFIED,
+        gem: Gem | str | None = None,
+        chat: Optional["ChatSession"] = None,
+        temporary: bool = False,
+        session_state: dict[str, Any] | None = None,
+        **kwargs,
+    ) -> AsyncGenerator[ModelOutput, None]:
+        """
+        Internal method which actually sends content generation requests.
+        """
+
+        assert prompt, "Prompt cannot be empty."
+
+        if isinstance(model, str):
+            model = Model.from_name(model)
+        elif isinstance(model, dict):
+            model = Model.from_dict(model)
+        elif not isinstance(model, Model):
+            raise TypeError(
+                f"'model' must be a `gemini_webapi.constants.Model` instance, "
+                f"string, or dictionary; got `{type(model).__name__}`"
+            )
+
+        _reqid = self._reqid
+        self._reqid += 100000
+
+        gem_id = gem.id if isinstance(gem, Gem) else gem
+
+        chat_backup: dict[str, Any] | None = None
+        if chat:
+            chat_backup = {
+                "metadata": (
+                    list(chat.metadata)
+                    if getattr(chat, "metadata", None)
+                    else list(_DEFAULT_METADATA)
+                ),
+                "cid": getattr(chat, "cid", ""),
+                "rid": getattr(chat, "rid", ""),
+                "rcid": getattr(chat, "rcid", ""),
+            }
+
+        if session_state is None:
+            session_state = {
+                "last_texts": {},
+                "last_thoughts": {},
+                "last_progress_time": time.time(),
+                "is_thinking": False,
+                "is_queueing": False,
+                "title": None,
+            }
+        else:
+            # Reset connection-specific states during a retry attempt
+            session_state["last_progress_time"] = time.time()
+            session_state["is_thinking"] = False
+            session_state["is_queueing"] = False
+
+        has_generated_text = False
+        sleep_time = 10
+
+        message_content = [
+            prompt,
+            0,
+            None,
+            req_file_data,
+            None,
+            None,
+            0,
+        ]
+
+        params: dict[str, Any] = {"_reqid": _reqid, "rt": "c"}
+        if self.build_label:
+            params["bl"] = self.build_label
+        if self.session_id:
+            params["f.sid"] = self.session_id
+
+        while True:
+            try:
+                inner_req_list: list[Any] = [None] * 69
+                inner_req_list[0] = message_content
+                inner_req_list[2] = chat.metadata if chat else list(_DEFAULT_METADATA)
+                inner_req_list[STREAMING_FLAG_INDEX] = 1
+                if gem_id:
+                    inner_req_list[GEM_FLAG_INDEX] = gem_id
+                if temporary:
+                    inner_req_list[TEMPORARY_CHAT_FLAG_INDEX] = 1
+
+                request_data = {
+                    "at": self.access_token,
+                    "f.req": json.dumps(
+                        [
+                            None,
+                            json.dumps(inner_req_list).decode("utf-8"),
+                        ]
+                    ).decode("utf-8"),
+                }
+
+                async with self.client.stream(
+                    "POST",
+                    Endpoint.GENERATE,
+                    params=params,
+                    headers=model.model_header,
+                    data=request_data,
+                    **kwargs,
+                ) as response:
+                    if self.verbose:
+                        logger.debug(
+                            f"HTTP Request: POST {Endpoint.GENERATE} [{response.status_code}]"
+                        )
+                    if response.status_code != 200:
+                        await self.close()
+                        raise APIError(
+                            f"Failed to generate contents. Status: {response.status_code}"
+                        )
+
+                    buffer = ""
+                    decoder = codecs.getincrementaldecoder("utf-8")(errors="replace")
+
+                    last_texts: dict[str, str] = session_state["last_texts"]
+                    last_thoughts: dict[str, str] = session_state["last_thoughts"]
+                    last_progress_time = session_state["last_progress_time"]
+
+                    is_thinking = session_state["is_thinking"]
+                    is_queueing = session_state["is_queueing"]
+                    has_candidates = False
+                    is_completed = False  # Check if this conversation turn has been fully answered.
+                    is_final_chunk = False  # Check if this turn is saved to history and marked complete or still pending (e.g., video generation).
+                    cid = chat.cid if chat else ""
+                    rid = chat.rid if chat else ""
+
+                    async def _process_parts(
+                        parts: list[Any],
+                    ) -> AsyncGenerator[ModelOutput, None]:
+                        nonlocal is_thinking, is_queueing, has_candidates, is_completed, is_final_chunk, cid, rid
+                        for part in parts:
+                            # Check for fatal error codes
+                            error_code = get_nested_value(part, [5, 2, 0, 1, 0])
+                            if error_code:
+                                await self.close()
+                                match error_code:
+                                    case ErrorCode.USAGE_LIMIT_EXCEEDED:
+                                        raise UsageLimitExceeded(
+                                            f"Usage limit exceeded for model '{model.model_name}'. Please wait a few minutes, "
+                                            "switch to a different model (e.g., Gemini Flash), or check your account limits on gemini.google.com."
+                                        )
+                                    case ErrorCode.MODEL_INCONSISTENT:
+                                        raise ModelInvalid(
+                                            "The specified model is inconsistent with the conversation history. "
+                                            "Please ensure you are using the same 'model' parameter throughout the entire ChatSession."
+                                        )
+                                    case ErrorCode.MODEL_HEADER_INVALID:
+                                        raise ModelInvalid(
+                                            f"The model '{model.model_name}' is currently unavailable or the request structure is outdated. "
+                                            "Please update 'gemini_webapi' to the latest version or report this on GitHub if the problem persists."
+                                        )
+                                    case ErrorCode.IP_TEMPORARILY_BLOCKED:
+                                        raise TemporarilyBlocked(
+                                            "Your IP address has been temporarily flagged or blocked by Google. "
+                                            "Please try using a proxy, a different network, or wait for a while before retrying."
+                                        )
+                                    case ErrorCode.TEMPORARY_ERROR_1013:
+                                        raise APIError(
+                                            "Gemini encountered a temporary error (1013). Retrying..."
+                                        )
+                                    case _:
+                                        raise APIError(
+                                            f"Failed to generate contents (stream). Unknown API error code: {error_code}. "
+                                            "This might be a temporary Google service issue."
+                                        )
+
+                            # Check for queueing status
+                            status = get_nested_value(part, [5])
+                            if isinstance(status, list) and status:
+                                if not is_thinking:
+                                    is_queueing = True
+                                    session_state["is_queueing"] = True
+                                    if not has_candidates:
+                                        logger.debug(
+                                            "Model is in a waiting state (queueing)..."
+                                        )
+
+                            inner_json_str = get_nested_value(part, [2])
+                            if inner_json_str:
+                                try:
+                                    part_json = json.loads(inner_json_str)
+                                    m_data = get_nested_value(part_json, [1])
+                                    cid = get_nested_value(m_data, [0], "")
+                                    rid = get_nested_value(m_data, [1], "")
+                                    if m_data and isinstance(chat, ChatSession):
+                                        chat.metadata = m_data
+
+                                    # Check for busy analyzing data
+                                    tool_name = get_nested_value(part_json, [6, 1, 0])
+                                    if tool_name == "data_analysis_tool":
+                                        is_thinking = True
+                                        session_state["is_thinking"] = True
+                                        is_queueing = False
+                                        session_state["is_queueing"] = False
+                                        if not has_candidates:
+                                            logger.debug(
+                                                f"Model is active (thinking/analyzing)... Raw: {str(part_json)[:500]}"
+                                            )
+
+                                    context_str = get_nested_value(part_json, [25])
+                                    if isinstance(context_str, str):
+                                        is_final_chunk = True
+                                        is_thinking = False
+                                        session_state["is_thinking"] = False
+                                        is_queueing = False
+                                        session_state["is_queueing"] = False
+                                        if isinstance(chat, ChatSession):
+                                            chat.metadata = [None] * 9 + [context_str]
+
+                                    title = get_nested_value(part_json, [10, 0])
+                                    if title:
+                                        session_state["title"] = title
+
+                                    candidates_list = get_nested_value(
+                                        part_json, [4], []
+                                    )
+                                    if candidates_list:
+                                        output_candidates = []
+                                        for i, candidate_data in enumerate(
+                                            candidates_list
+                                        ):
+                                            rcid = get_nested_value(candidate_data, [0])
+                                            if not rcid:
+                                                continue
+                                            if isinstance(chat, ChatSession):
+                                                chat.rcid = rcid
+
+                                            (
+                                                text,
+                                                thoughts,
+                                                web_images,
+                                                generated_images,
+                                                generated_videos,
+                                            ) = self._parse_candidate(
+                                                candidate_data, cid, rid, rcid
+                                            )
+
+                                            # Check if this frame represents the complete state of the message
+                                            is_completed = (
+                                                get_nested_value(
+                                                    candidate_data, [8, 0], 1
+                                                )
+                                                == 2
+                                            )
+
+                                            # Save this conversation turn to list_chats whenever it is stored in history.
+                                            if is_final_chunk:
+                                                cid = get_nested_value(
+                                                    part_json, [1, 0]
+                                                )
+                                                if cid and isinstance(
+                                                    self._recent_chats, list
+                                                ):
+                                                    chat_title = session_state.get(
+                                                        "title"
+                                                    )
+                                                    if not chat_title:
+                                                        for c in self._recent_chats:
+                                                            if c.cid == cid:
+                                                                chat_title = c.title
+                                                                break
+
+                                                    if chat_title:
+                                                        is_pinned = False
+                                                        for c in self._recent_chats:
+                                                            if c.cid == cid:
+                                                                is_pinned = c.is_pinned
+                                                                break
+
+                                                        expected_idx = (
+                                                            0
+                                                            if is_pinned
+                                                            else sum(
+                                                                1
+                                                                for c in self._recent_chats
+                                                                if c.cid != cid
+                                                                and c.is_pinned
+                                                            )
+                                                        )
+
+                                                        if not (
+                                                            len(self._recent_chats)
+                                                            > expected_idx
+                                                            and self._recent_chats[
+                                                                expected_idx
+                                                            ].cid
+                                                            == cid
+                                                            and self._recent_chats[
+                                                                expected_idx
+                                                            ].title
+                                                            == chat_title
+                                                        ):
+                                                            self._recent_chats = [
+                                                                c
+                                                                for c in self._recent_chats
+                                                                if c.cid != cid
+                                                            ]
+                                                            self._recent_chats.insert(
+                                                                expected_idx,
+                                                                ChatInfo(
+                                                                    cid=cid,
+                                                                    title=chat_title,
+                                                                    is_pinned=is_pinned,
+                                                                ),
+                                                            )
+
+                                            last_sent_text = last_texts.get(
+                                                rcid
+                                            ) or last_texts.get(f"idx_{i}", "")
+                                            text_delta, new_full_text = (
+                                                get_delta_by_fp_len(
+                                                    text,
+                                                    last_sent_text,
+                                                    is_final=is_completed,
+                                                )
+                                            )
+                                            last_sent_thought = last_thoughts.get(
+                                                rcid
+                                            ) or last_thoughts.get(f"idx_{i}", "")
+                                            if thoughts:
+                                                thoughts_delta, new_full_thought = (
+                                                    get_delta_by_fp_len(
+                                                        thoughts,
+                                                        last_sent_thought,
+                                                        is_final=is_completed,
+                                                    )
+                                                )
+                                            else:
+                                                thoughts_delta = ""
+                                                new_full_thought = ""
+
+                                            if (
+                                                text_delta
+                                                or thoughts_delta
+                                                or web_images
+                                                or generated_images
+                                            ):
+                                                has_candidates = True
+                                                if thoughts_delta:
+                                                    logger.debug(f"[Thinking]: {thoughts_delta.strip()}")
+                                                if text_delta:
+                                                    logger.debug(f"[Generating]: {text_delta.strip()}")
+
+                                            # Update state with the provider's cleaned state to handle drift
+                                            last_texts[rcid] = last_texts[
+                                                f"idx_{i}"
+                                            ] = new_full_text
+
+                                            last_thoughts[rcid] = last_thoughts[
+                                                f"idx_{i}"
+                                            ] = new_full_thought
+
+                                            output_candidates.append(
+                                                Candidate(
+                                                    rcid=rcid,
+                                                    text=text,
+                                                    text_delta=text_delta,
+                                                    thoughts=thoughts or None,
+                                                    thoughts_delta=thoughts_delta,
+                                                    web_images=web_images,
+                                                    generated_images=generated_images,
+                                                    generated_videos=generated_videos,
+                                                )
+                                            )
+
+                                        if output_candidates:
+                                            is_thinking = False
+                                            session_state["is_thinking"] = False
+                                            is_queueing = False
+                                            session_state["is_queueing"] = False
+                                            yield ModelOutput(
+                                                metadata=get_nested_value(
+                                                    part_json, [1], []
+                                                ),
+                                                candidates=output_candidates,
+                                            )
+                                except json.JSONDecodeError:
+                                    continue
+
+                    chunk_iterator = response.aiter_content().__aiter__()
+                    while True:
+                        try:
+                            stall_threshold = (
+                                self.timeout
+                                if (is_thinking or is_queueing)
+                                else min(self.timeout, self.watchdog_timeout)
+                            )
+                            chunk = await asyncio.wait_for(
+                                chunk_iterator.__anext__(), timeout=stall_threshold + 5
+                            )
+                        except StopAsyncIteration:
+                            break
+                        except asyncio.TimeoutError:
+                            logger.debug(
+                                f"[Watchdog] Socket idle for {stall_threshold + 5}s. Refreshing connection..."
+                            )
+                            break
+
+                        buffer += decoder.decode(chunk, final=False)
+                        if buffer.startswith(")]}'"):
+                            buffer = buffer[4:].lstrip()
+                        parsed_parts, buffer = parse_response_by_frame(buffer)
+
+                        got_update = False
+                        async for out in _process_parts(parsed_parts):
+                            has_generated_text = True
+                            yield out
+                            got_update = True
+
+                        if got_update:
+                            last_progress_time = time.time()
+                            session_state["last_progress_time"] = last_progress_time
+                        else:
+                            stall_threshold = (
+                                self.timeout
+                                if (is_thinking or is_queueing)
+                                else min(self.timeout, self.watchdog_timeout)
+                            )
+                            if (time.time() - last_progress_time) > stall_threshold:
+                                if is_thinking:
+                                    logger.debug(
+                                        f"[Watchdog] Model is taking its time thinking ({int(time.time() - last_progress_time)}s). Reconnecting to poll..."
+                                    )
+                                    break
+                                else:
+                                    logger.debug(
+                                        f"[Watchdog] Connection idle for {stall_threshold}s (queueing={is_queueing}). "
+                                        "Attempting recovery..."
+                                    )
+                                    await self.close()
+                                    break
+
+                    # Final flush
+                    buffer += decoder.decode(b"", final=True)
+                    if buffer:
+                        parsed_parts, _ = parse_response_by_frame(buffer)
+                        async for out in _process_parts(parsed_parts):
+                            has_generated_text = True
+                            yield out
+
+                    if not is_completed or is_thinking or is_queueing:
+                        stall_threshold = (
+                            self.timeout
+                            if (is_thinking or is_queueing)
+                            else min(self.timeout, self.watchdog_timeout)
+                        )
+                        if (time.time() - last_progress_time) > stall_threshold:
+                            if not is_thinking:
+                                logger.debug(
+                                    f"[Watchdog] Stream ended after {stall_threshold}s without completing. Triggering recovery..."
+                                )
+                            else:
+                                logger.debug(
+                                    "[Watchdog] Stream finished but model is still thinking. Polling again..."
+                                )
+
+                        if cid:
+                            logger.debug(
+                                f"Stream incomplete. Checking conversation history for {cid}..."
+                            )
+
+                            poll_start_time = time.time()
+
+                            while True:
+                                if (time.time() - poll_start_time) > self.timeout:
+                                    logger.warning(
+                                        f"[Recovery] Polling for {cid} timed out after {self.timeout}s."
+                                    )
+                                    if has_generated_text:
+                                        raise GeminiError(
+                                            "The connection to Gemini was lost while generating the response, and recovery timed out. "
+                                            "Please try sending your prompt again."
+                                        )
+                                    else:
+                                        raise APIError(
+                                            "read_chat polling timed out waiting for the model to finish. "
+                                            "The original request may have been silently aborted by Google."
+                                        )
+                                await self._send_bard_activity()
+                                recovered_history = await self.read_chat(cid)
+                                if (
+                                    recovered_history
+                                    and recovered_history.turns
+                                    and recovered_history.turns[-1].role == "model"
+                                ):
+                                    recovered = recovered_history.turns[-1].info
+                                    if (
+                                        recovered
+                                        and recovered.candidates
+                                        and (
+                                            recovered.candidates[0].text.strip()
+                                            or recovered.candidates[0].generated_images
+                                            or recovered.candidates[0].web_images
+                                        )
+                                    ):
+                                        rec_rcid = recovered.candidates[0].rcid
+                                        prev_rcid = (
+                                            chat_backup["rcid"] if chat_backup else ""
+                                        )
+                                        current_expected_rcid = (
+                                            getattr(chat, "rcid", "") if chat else ""
+                                        )
+
+                                        is_new_turn = (
+                                            rec_rcid == current_expected_rcid
+                                            if current_expected_rcid
+                                            else rec_rcid != prev_rcid
+                                        )
+
+                                        if is_new_turn:
+                                            logger.debug(
+                                                f"[Recovery] Successfully recovered response for CID: {cid} (RCID: {rec_rcid})"
+                                            )
+                                            if chat:
+                                                recovered.metadata = chat.metadata
+                                                chat.rcid = rec_rcid
+                                            yield recovered
+                                            break
+                                        else:
+                                            logger.debug(
+                                                f"[Recovery] Recovered turn is not the target turn (target: {current_expected_rcid or 'NEW'}, got {rec_rcid}). Waiting..."
+                                            )
+
+                                logger.debug(
+                                    f"[Recovery] Response not ready, waiting {sleep_time}s..."
+                                )
+                                await asyncio.sleep(sleep_time)
+                            break
+                        else:
+                            logger.debug(
+                                f"Stream suspended (completed={is_completed}, final_chunk={is_final_chunk}, thinking={is_thinking}, queueing={is_queueing}). "
+                                f"No CID found to recover. (Request ID: {_reqid})"
+                            )
+                            raise APIError(
+                                "The original request may have been silently aborted by Google."
+                            )
+
+                break
+
+            except ReadTimeout:
+                raise TimeoutError(
+                    "The request timed out while waiting for Gemini to respond. This often happens with very long prompts "
+                    "or complex file analysis. Try increasing the 'timeout' value when initializing GeminiClient."
+                )
+            except (GeminiError, APIError):
+                if not has_generated_text and chat and chat_backup:
+                    chat.metadata = list(chat_backup["metadata"])  # type: ignore
+                    chat.cid = chat_backup["cid"]
+                    chat.rid = chat_backup["rid"]
+                    chat.rcid = chat_backup["rcid"]
+                raise
+            except Exception:
+                if not has_generated_text and chat and chat_backup:
+                    chat.metadata = list(chat_backup["metadata"])  # type: ignore
+                    chat.cid = chat_backup["cid"]
+                    chat.rid = chat_backup["rid"]
+                    chat.rcid = chat_backup["rcid"]
+                logger.debug(
+                    "Stream parsing interrupted. Attempting to recover conversation context..."
+                )
+                raise APIError(
+                    "Failed to parse response body from Google. This might be a temporary API change or invalid data."
+                )
+
+    def start_chat(self, **kwargs) -> "ChatSession":
+        """
+        Returns a `ChatSession` object attached to this client.
+
+        Parameters
+        ----------
+        kwargs: `dict`, optional
+            Additional arguments which will be passed to the chat session.
+            Refer to `gemini_webapi.ChatSession` for more information.
+
+        Returns
+        -------
+        :class:`ChatSession`
+            Empty chat session object for retrieving conversation history.
+        """
+
+        return ChatSession(geminiclient=self, **kwargs)
+
+    async def delete_chat(self, cid: str) -> None:
+        """
+        Delete a specific conversation by chat id.
+
+        Parameters
+        ----------
+        cid: `str`
+            The ID of the chat requiring deletion (e.g. "c_...").
+        """
+
+        await self._batch_execute(
+            [
+                RPCData(
+                    rpcid=GRPC.DELETE_CHAT,
+                    payload=json.dumps([cid]).decode("utf-8"),
+                ),
+            ]
+        )
+        await self._batch_execute(
+            [
+                RPCData(
+                    rpcid=GRPC.DELETE_CHAT_SECOND,
+                    payload=json.dumps([cid, [1, None, 0, 1]]).decode("utf-8"),
+                ),
+            ]
+        )
+
+    def list_chats(self) -> list[ChatInfo] | None:
+        """
+        List all conversations.
+
+        Returns
+        -------
+        `list[gemini_webapi.types.ChatInfo] | None`
+            The list of conversations. Returns `None` if the client holds no session cache.
+        """
+        return self._recent_chats
+
+    async def read_chat(self, cid: str, limit: int = 10) -> ChatHistory | None:
+        """
+        Fetch the full conversation history by chat id.
+
+        Parameters
+        ----------
+        cid: `str`
+            The ID of the conversation to read (e.g. "c_...").
+        limit: `int`, optional
+            The maximum number of turns to fetch, by default 10.
+
+        Returns
+        -------
+        :class:`ChatHistory` | None
+            The conversation history, or None if reading failed.
+        """
+        try:
+            response = await self._batch_execute(
+                [
+                    RPCData(
+                        rpcid=GRPC.READ_CHAT,
+                        payload=json.dumps(
+                            [cid, limit, None, 1, [1], [4], None, 1]
+                        ).decode("utf-8"),
+                    ),
+                ]
+            )
+
+            response_json = extract_json_from_response(response.text)
+
+            for part in response_json:
+                part_body_str = get_nested_value(part, [2])
+                if not part_body_str:
+                    continue
+
+                part_body = json.loads(part_body_str)
+                turns_data = get_nested_value(part_body, [0])
+                if not turns_data:
+                    continue
+
+                chat_turns = []
+                for conv_turn in turns_data:
+                    # User turn
+                    user_text = get_nested_value(conv_turn, [2, 0, 0], "")
+                    if user_text:
+                        chat_turns.append(ChatTurn(role="user", text=user_text))
+
+                    # Model turn
+                    candidates_list = get_nested_value(conv_turn, [3, 0])
+                    if candidates_list:
+                        output_candidates = []
+                        rid = get_nested_value(conv_turn, [1], "")
+                        for candidate_data in candidates_list:
+                            rcid = get_nested_value(candidate_data, [0], "")
+                            (
+                                text,
+                                thoughts,
+                                web_images,
+                                generated_images,
+                                generated_videos,
+                            ) = self._parse_candidate(candidate_data, cid, rid, rcid)
+                            output_candidates.append(
+                                Candidate(
+                                    rcid=rcid,
+                                    text=text,
+                                    thoughts=thoughts,
+                                    web_images=web_images,
+                                    generated_images=generated_images,
+                                    generated_videos=generated_videos,
+                                )
+                            )
+
+                        if output_candidates:
+                            model_output = ModelOutput(
+                                metadata=[cid, rid, output_candidates[0].rcid],
+                                candidates=output_candidates,
+                            )
+                            chat_turns.append(
+                                ChatTurn(
+                                    role="model",
+                                    text=output_candidates[0].text,
+                                    info=model_output,
+                                )
+                            )
+
+                return ChatHistory(cid=cid, metadata=[cid], turns=chat_turns)
+
+            return None
+        except Exception:
+            logger.debug(
+                f"[read_chat] Response data for {cid!r} is still incomplete (model is still processing)..."
+            )
+            return None
+
+    def _parse_candidate(
+        self, candidate_data: list[Any], cid: str, rid: str, rcid: str
+    ) -> tuple[str, str, list[WebImage], list[GeneratedImage], list[GeneratedVideo]]:
+        """
+        Parses individual candidate data from the Gemini response.
+
+        Args:
+            candidate_data (list[Any]): The raw candidate list from the API response.
+            cid (str): Conversation ID.
+            rid (str): Response ID.
+            rcid (str): Response Candidate ID.
+
+        Returns:
+            tuple: A tuple containing:
+                - text (str): The main response text.
+                - thoughts (str): The model's reasoning or internal thoughts.
+                - web_images (list[WebImage]): List of images found on the web.
+                - generated_images (list[GeneratedImage]): List of images generated by the model.
+                - generated_videos (list[GeneratedVideo]): List of videos generated by the model.
+        """
+        text = get_nested_value(candidate_data, [1, 0], "")
+        if _CARD_CONTENT_RE.match(text):
+            text = get_nested_value(candidate_data, [22, 0]) or text
+
+        # Cleanup googleusercontent artifacts
+        text = _ARTIFACTS_RE.sub("", text)
+
+        thoughts = get_nested_value(candidate_data, [37, 0, 0]) or ""
+
+        # Image handling
+        web_images = []
+        for img_idx, web_img_data in enumerate(
+            get_nested_value(candidate_data, [12, 1], [])
+        ):
+            url = get_nested_value(web_img_data, [0, 0, 0])
+            if url:
+                web_images.append(
+                    WebImage(
+                        url=url,
+                        title=f"[Image {img_idx + 1}]",
+                        alt=get_nested_value(web_img_data, [0, 4], ""),
+                        proxy=self.proxy,
+                        client=self.client,
+                    )
+                )
+
+        generated_images = []
+        for img_idx, gen_img_data in enumerate(
+            get_nested_value(candidate_data, [12, 7, 0], [])
+        ):
+            url = get_nested_value(gen_img_data, [0, 3, 3])
+            if url:
+                image_id = get_nested_value(gen_img_data, [1, 0])
+                if not image_id:
+                    image_id = f"http://googleusercontent.com/image_generation_content/{img_idx}"
+
+                generated_images.append(
+                    GeneratedImage(
+                        url=url,
+                        title=f"[Generated Image {img_idx}]",
+                        alt=get_nested_value(gen_img_data, [0, 3, 2], ""),
+                        proxy=self.proxy,
+                        client=self.client,
+                        client_ref=self,
+                        cid=cid,
+                        rid=rid,
+                        rcid=rcid,
+                        image_id=image_id,
+                    )
+                )
+
+        # Video handling
+        generated_videos = []
+        for video_root in get_nested_value(candidate_data, [12, 59, 0], []):
+            video_info = get_nested_value(video_root, [0])
+            if video_info:
+                urls = get_nested_value(video_info, [0, 7], [])
+                if len(urls) >= 2:
+                    generated_videos.append(
+                        GeneratedVideo(
+                            url=urls[1],
+                            thumbnail=urls[0],
+                            cid=cid,
+                            rid=rid,
+                            rcid=rcid,
+                            client_ref=self,
+                            proxy=self.proxy,
+                        )
+                    )
+
+        return text, thoughts, web_images, generated_images, generated_videos
+
+    async def _get_image_full_size(
+        self, cid: str, rid: str, rcid: str, image_id: str
+    ) -> str | None:
+        """
+        Get the full size URL of an image.
+        """
+        try:
+            payload = [
+                [
+                    [None, None, None, [None, None, None, None, None, ""]],
+                    [image_id, 0],
+                    None,
+                    [19, ""],
+                    None,
+                    None,
+                    None,
+                    None,
+                    None,
+                    "",
+                ],
+                [rid, rcid, cid, None, ""],
+                1,
+                0,
+                1,
+            ]
+
+            response = await self._batch_execute(
+                [
+                    RPCData(
+                        rpcid=GRPC.IMAGE_FULL_SIZE,
+                        payload=json.dumps(payload).decode("utf-8"),
+                    ),
+                ]
+            )
+
+            response_data = extract_json_from_response(response.text)
+            return get_nested_value(
+                json.loads(get_nested_value(response_data, [0, 2], "[]")), [0]
+            )
+        except Exception:
+            logger.debug(
+                "[_get_image_full_size] Could not retrieve full size URL via RPC."
+            )
+            return None
+
+    @running(retry=2)
+    async def _batch_execute(self, payloads: list[RPCData], **kwargs) -> Response:
+        """
+        Execute a batch of requests to Gemini API.
+
+        Parameters
+        ----------
+        payloads: `list[RPCData]`
+            List of `gemini_webapi.types.RPCData` objects to be executed.
+        kwargs: `dict`, optional
+            Additional arguments which will be passed to the post request.
+            Refer to `curl_cffi.requests.AsyncSession.request` for more information.
+
+        Returns
+        -------
+        :class:`curl_cffi.requests.Response`
+            Response object containing the result of the batch execution.
+        """
+
+        _reqid = self._reqid
+        self._reqid += 100000
+
+        try:
+            params: dict[str, Any] = {
+                "rpcids": ",".join([p.rpcid for p in payloads]),
+                "_reqid": _reqid,
+                "rt": "c",
+                "source-path": "/app",
+            }
+            if self.build_label:
+                params["bl"] = self.build_label
+            if self.session_id:
+                params["f.sid"] = self.session_id
+
+            response = await self.client.post(
+                Endpoint.BATCH_EXEC,
+                params=params,
+                data={
+                    "at": self.access_token,
+                    "f.req": json.dumps(
+                        [[payload.serialize() for payload in payloads]]
+                    ).decode("utf-8"),
+                },
+                **kwargs,
+            )
+            if self.verbose:
+                logger.debug(
+                    f"HTTP Request: POST {Endpoint.BATCH_EXEC} [{response.status_code}]"
+                )
+        except ReadTimeout:
+            raise TimeoutError(
+                "The request timed out while waiting for Gemini to respond. This often happens with very long prompts "
+                "or complex file analysis. Try increasing the 'timeout' value when initializing GeminiClient."
+            )
+
+        if response.status_code != 200:
+            await self.close()
+            raise APIError(
+                f"Batch execution failed with status code {response.status_code}"
+            )
+
+        return response
+
+
+class ChatSession:
+    """
+    Chat data to retrieve conversation history. Only if all 3 ids are provided will the conversation history be retrieved.
+
+    Parameters
+    ----------
+    geminiclient: `GeminiClient`
+        Async requests client interface for gemini.google.com.
+    metadata: `list[str]`, optional
+        List of chat metadata `[cid, rid, rcid]`, can be shorter than 3 elements, like `[cid, rid]` or `[cid]` only.
+    cid: `str`, optional
+        Chat id, if provided together with metadata, will override the first value in it.
+    rid: `str`, optional
+        Reply id, if provided together with metadata, will override the second value in it.
+    rcid: `str`, optional
+        Reply candidate id, if provided together with metadata, will override the third value in it.
+    model: `Model | str | dict`, optional
+        Specify the model to use for generation.
+        Pass either a `gemini_webapi.constants.Model` enum or a model name string to use predefined models.
+        Pass a dictionary to use custom model header strings ("model_name" and "model_header" keys must be provided).
+    gem: `Gem | str`, optional
+        Specify a gem to use as system prompt for the chat session.
+        Pass either a `gemini_webapi.types.Gem` object or a gem id string.
+    """
+
+    __slots__ = [
+        "__metadata",
+        "geminiclient",
+        "last_output",
+        "model",
+        "gem",
+    ]
+
+    def __init__(
+        self,
+        geminiclient: GeminiClient,
+        metadata: list[str | None] | None = None,
+        cid: str = "",  # chat id
+        rid: str = "",  # reply id
+        rcid: str = "",  # reply candidate id
+        model: Model | str | dict = Model.UNSPECIFIED,
+        gem: Gem | str | None = None,
+    ):
+        self.__metadata: list[Any] = list(_DEFAULT_METADATA)
+        self.geminiclient: GeminiClient = geminiclient
+        self.last_output: ModelOutput | None = None
+        self.model: Model | str | dict = model
+        self.gem: Gem | str | None = gem
+
+        if metadata:
+            self.metadata = metadata
+        if cid:
+            self.cid = cid
+        if rid:
+            self.rid = rid
+        if rcid:
+            self.rcid = rcid
+
+    def __str__(self):
+        return f"ChatSession(cid='{self.cid}', rid='{self.rid}', rcid='{self.rcid}')"
+
+    __repr__ = __str__
+
+    def __setattr__(self, name: str, value: Any) -> None:
+        super().__setattr__(name, value)
+        # update conversation history when last output is updated
+        if name == "last_output" and isinstance(value, ModelOutput):
+            self.metadata = value.metadata
+            self.rcid = value.rcid
+
+    async def send_message(
+        self,
+        prompt: str,
+        files: list[str | Path | bytes | io.BytesIO] | None = None,
+        temporary: bool = False,
+        **kwargs,
+    ) -> ModelOutput:
+        """
+        Generates contents with prompt.
+        Use as a shortcut for `GeminiClient.generate_content(prompt, files, self)`.
+
+        Parameters
+        ----------
+        prompt: `str`
+            Text prompt provided by user.
+        files: `list[str | Path | bytes | io.BytesIO]`, optional
+            List of file paths or byte streams to be attached.
+        temporary: `bool`, optional
+            If set to `True`, the ongoing conversation will not show up in Gemini history.
+            Switching temporary mode within a chat session will clear the previous context
+            and create a new chat session under the hood.
+        kwargs: `dict`, optional
+            Additional arguments which will be passed to the post request.
+            Refer to `curl_cffi.requests.AsyncSession.request` for more information.
+
+        Returns
+        -------
+        :class:`ModelOutput`
+            Output data from gemini.google.com.
+
+        Raises
+        ------
+        `AssertionError`
+            If prompt is empty.
+        `gemini_webapi.TimeoutError`
+            If request timed out.
+        `gemini_webapi.GeminiError`
+            If no reply candidate found in response.
+        `gemini_webapi.APIError`
+            - If request failed with status code other than 200.
+            - If response structure is invalid and failed to parse.
+        """
+
+        return await self.geminiclient.generate_content(
+            prompt=prompt,
+            files=files,
+            model=self.model,
+            gem=self.gem,
+            chat=self,
+            temporary=temporary,
+            **kwargs,
+        )
+
+    async def send_message_stream(
+        self,
+        prompt: str,
+        files: list[str | Path | bytes | io.BytesIO] | None = None,
+        temporary: bool = False,
+        **kwargs,
+    ) -> AsyncGenerator[ModelOutput, None]:
+        """
+        Generates contents with prompt in streaming mode within this chat session.
+
+        This is a shortcut for `GeminiClient.generate_content_stream(prompt, files, self)`.
+        The session's metadata and conversation history are automatically managed.
+
+        Parameters
+        ----------
+        prompt: `str`
+            Text prompt provided by user.
+        files: `list[str | Path | bytes | io.BytesIO]`, optional
+            List of file paths or byte streams to be attached.
+        temporary: `bool`, optional
+            If set to `True`, the ongoing conversation will not show up in Gemini history.
+            Switching temporary mode within a chat session will clear the previous context
+            and create a new chat session under the hood.
+        kwargs: `dict`, optional
+            Additional arguments passed to the streaming request.
+
+        Yields
+        ------
+        :class:`ModelOutput`
+            Partial output data containing text deltas.
+        """
+
+        async for output in self.geminiclient.generate_content_stream(
+            prompt=prompt,
+            files=files,
+            model=self.model,
+            gem=self.gem,
+            chat=self,
+            temporary=temporary,
+            **kwargs,
+        ):
+            yield output
+
+    def choose_candidate(self, index: int) -> ModelOutput:
+        """
+        Choose a candidate from the last `ModelOutput` to control the ongoing conversation flow.
+
+        Parameters
+        ----------
+        index: `int`
+            Index of the candidate to choose, starting from 0.
+
+        Returns
+        -------
+        :class:`ModelOutput`
+            Output data of the chosen candidate.
+
+        Raises
+        ------
+        `ValueError`
+            If no previous output data found in this chat session, or if index exceeds the number of candidates in last model output.
+        """
+
+        if not self.last_output:
+            raise ValueError("No previous output data found in this chat session.")
+
+        if index >= len(self.last_output.candidates):
+            raise ValueError(
+                f"Index {index} exceeds the number of candidates in last model output."
+            )
+
+        self.last_output.chosen = index
+        self.rcid = self.last_output.rcid
+        return self.last_output
+
+    async def read_history(self, limit: int = 10) -> ChatHistory | None:
+        """
+        Fetch the conversation history for this session.
+
+        Parameters
+        ----------
+        limit: `int`, optional
+            The maximum number of turns to fetch, by default 10.
+
+        Returns
+        -------
+        :class:`ChatHistory` | None
+            The conversation history, or None if reading failed or cid is missing.
+        """
+        if not self.cid:
+            return None
+        return await self.geminiclient.read_chat(self.cid, limit=limit)
+
+    @property
+    def metadata(self):
+        return self.__metadata
+
+    @metadata.setter
+    def metadata(self, value: list[str]):
+        if not isinstance(value, list):
+            return
+
+        # Update only non-None elements to preserve existing CID/RID/RCID/Context
+        for i, val in enumerate(value):
+            if i < 10 and val is not None:
+                self.__metadata[i] = val
+
+    @property
+    def cid(self):
+        return self.__metadata[0]
+
+    @cid.setter
+    def cid(self, value: str):
+        self.__metadata[0] = value
+
+    @property
+    def rcid(self):
+        return self.__metadata[2]
+
+    @rcid.setter
+    def rcid(self, value: str):
+        self.__metadata[2] = value
+
+    @property
+    def rid(self):
+        return self.__metadata[1]
+
+    @rid.setter
+    def rid(self, value: str):
+        self.__metadata[1] = value

gemini_webapi/components/__init__.py

@@ -0,0 +1,3 @@
+# flake8: noqa
+
+from .gem_mixin import GemMixin

gemini_webapi/components/gem_mixin.py

@@ -0,0 +1,311 @@
+import itertools
+
+import orjson as json
+
+from ..constants import GRPC
+from ..exceptions import APIError
+from ..types import Gem, GemJar, RPCData
+from ..utils import extract_json_from_response, get_nested_value, logger
+
+
+class GemMixin:
+    """
+    Mixin class providing gem-related functionality for GeminiClient.
+    """
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self._gems: GemJar | None = None
+
+    @property
+    def gems(self) -> GemJar:
+        """
+        Returns a `GemJar` object containing cached gems.
+        Only available after calling `GeminiClient.fetch_gems()`.
+
+        Returns
+        -------
+        :class:`GemJar`
+            Refer to `gemini_webapi.types.GemJar`.
+
+        Raises
+        ------
+        `RuntimeError`
+            If `GeminiClient.fetch_gems()` has not been called before accessing this property.
+        """
+
+        if self._gems is None:
+            raise RuntimeError(
+                "Gems not fetched yet. Call `GeminiClient.fetch_gems()` method to fetch gems from gemini.google.com."
+            )
+
+        return self._gems
+
+    async def fetch_gems(
+        self, include_hidden: bool = False, language: str = "en", **kwargs
+    ) -> GemJar:
+        """
+        Get a list of available gems from gemini, including system predefined gems and user-created custom gems.
+
+        Note that network request will be sent every time this method is called.
+        Once the gems are fetched, they will be cached and accessible via `GeminiClient.gems` property.
+
+        Parameters
+        ----------
+        include_hidden: `bool`, optional
+            There are some predefined gems that by default are not shown to users (and therefore may not work properly).
+            Set this parameter to `True` to include them in the fetched gem list.
+        language: `str`, optional
+            Language code for the gems to fetch. Default is 'en'.
+
+        Returns
+        -------
+        :class:`GemJar`
+            Refer to `gemini_webapi.types.GemJar`.
+        """
+
+        response = await self._batch_execute(
+            [
+                RPCData(
+                    rpcid=GRPC.LIST_GEMS,
+                    payload=(
+                        f"[4,['{language}'],0]"
+                        if include_hidden
+                        else f"[3,['{language}'],0]"
+                    ),
+                    identifier="system",
+                ),
+                RPCData(
+                    rpcid=GRPC.LIST_GEMS,
+                    payload=f"[2,['{language}'],0]",
+                    identifier="custom",
+                ),
+            ],
+            **kwargs,
+        )
+
+        try:
+            response_json = extract_json_from_response(response.text)
+
+            predefined_gems, custom_gems = [], []
+
+            for part in response_json:
+                try:
+                    identifier = get_nested_value(part, [-1])
+                    part_body_str = get_nested_value(part, [2])
+                    if not part_body_str:
+                        continue
+
+                    part_body = json.loads(part_body_str)
+                    if identifier == "system":
+                        predefined_gems = get_nested_value(part_body, [2], [])
+                    elif identifier == "custom":
+                        custom_gems = get_nested_value(part_body, [2], [])
+                except json.JSONDecodeError:
+                    continue
+
+            if not predefined_gems and not custom_gems:
+                raise Exception
+        except Exception:
+            await self.close()
+            logger.debug(f"Unexpected response data structure: {response.text}")
+            raise APIError(
+                "Failed to fetch gems. Unexpected response data structure. Client will try to re-initialize on next request."
+            )
+
+        self._gems = GemJar(
+            itertools.chain(
+                (
+                    (
+                        gem[0],
+                        Gem(
+                            id=gem[0],
+                            name=gem[1][0],
+                            description=gem[1][1],
+                            prompt=gem[2] and gem[2][0] or None,
+                            predefined=True,
+                        ),
+                    )
+                    for gem in predefined_gems
+                ),
+                (
+                    (
+                        gem[0],
+                        Gem(
+                            id=gem[0],
+                            name=gem[1][0],
+                            description=gem[1][1],
+                            prompt=gem[2] and gem[2][0] or None,
+                            predefined=False,
+                        ),
+                    )
+                    for gem in custom_gems
+                ),
+            )
+        )
+
+        return self._gems
+
+    async def create_gem(self, name: str, prompt: str, description: str = "") -> Gem:
+        """
+        Create a new custom gem.
+
+        Parameters
+        ----------
+        name: `str`
+            Name of the custom gem.
+        prompt: `str`
+            System instructions for the custom gem.
+        description: `str`, optional
+            Description of the custom gem (has no effect on the model's behavior).
+
+        Returns
+        -------
+        :class:`Gem`
+            The created gem.
+        """
+
+        response = await self._batch_execute(
+            [
+                RPCData(
+                    rpcid=GRPC.CREATE_GEM,
+                    payload=json.dumps(
+                        [
+                            [
+                                name,
+                                description,
+                                prompt,
+                                None,
+                                None,
+                                None,
+                                None,
+                                None,
+                                0,
+                                None,
+                                1,
+                                None,
+                                None,
+                                None,
+                                [],
+                            ]
+                        ]
+                    ).decode("utf-8"),
+                )
+            ]
+        )
+
+        try:
+            response_json = extract_json_from_response(response.text)
+            part_body_str = get_nested_value(response_json, [0, 2], verbose=True)
+            if not part_body_str:
+                raise Exception
+
+            part_body = json.loads(part_body_str)
+            gem_id = get_nested_value(part_body, [0], verbose=True)
+            if not gem_id:
+                raise Exception
+        except Exception:
+            await self.close()
+            logger.debug(f"Unexpected response data structure: {response.text}")
+            raise APIError(
+                "Failed to create gem. Unexpected response data structure. Client will try to re-initialize on next request."
+            )
+
+        return Gem(
+            id=gem_id,
+            name=name,
+            description=description,
+            prompt=prompt,
+            predefined=False,
+        )
+
+    async def update_gem(
+        self, gem: Gem | str, name: str, prompt: str, description: str = ""
+    ) -> Gem:
+        """
+        Update an existing custom gem.
+
+        Parameters
+        ----------
+        gem: `Gem | str`
+            Gem to update, can be either a `gemini_webapi.types.Gem` object or a gem id string.
+        name: `str`
+            New name for the custom gem.
+        prompt: `str`
+            New system instructions for the custom gem.
+        description: `str`, optional
+            New description of the custom gem (has no effect on the model's behavior).
+
+        Returns
+        -------
+        :class:`Gem`
+            The updated gem.
+        """
+
+        if isinstance(gem, Gem):
+            gem_id = gem.id
+        else:
+            gem_id = gem
+
+        await self._batch_execute(
+            [
+                RPCData(
+                    rpcid=GRPC.UPDATE_GEM,
+                    payload=json.dumps(
+                        [
+                            gem_id,
+                            [
+                                name,
+                                description,
+                                prompt,
+                                None,
+                                None,
+                                None,
+                                None,
+                                None,
+                                0,
+                                None,
+                                1,
+                                None,
+                                None,
+                                None,
+                                [],
+                                0,
+                            ],
+                        ]
+                    ).decode("utf-8"),
+                )
+            ]
+        )
+
+        return Gem(
+            id=gem_id,
+            name=name,
+            description=description,
+            prompt=prompt,
+            predefined=False,
+        )
+
+    async def delete_gem(self, gem: Gem | str, **kwargs) -> None:
+        """
+        Delete a custom gem.
+
+        Parameters
+        ----------
+        gem: `Gem | str`
+            Gem to delete, can be either a `gemini_webapi.types.Gem` object or a gem id string.
+        """
+
+        if isinstance(gem, Gem):
+            gem_id = gem.id
+        else:
+            gem_id = gem
+
+        await self._batch_execute(
+            [
+                RPCData(
+                    rpcid=GRPC.DELETE_GEM, payload=json.dumps([gem_id]).decode("utf-8")
+                )
+            ],
+            **kwargs,
+        )

gemini_webapi/constants.py

@@ -0,0 +1,142 @@
+from enum import Enum, IntEnum, StrEnum
+
+STREAMING_FLAG_INDEX = 7
+GEM_FLAG_INDEX = 19
+TEMPORARY_CHAT_FLAG_INDEX = 45
+
+
+class Endpoint(StrEnum):
+    GOOGLE = "https://www.google.com"
+    INIT = "https://gemini.google.com/app"
+    GENERATE = "https://gemini.google.com/_/BardChatUi/data/assistant.lamda.BardFrontendService/StreamGenerate"
+    ROTATE_COOKIES = "https://accounts.google.com/RotateCookies"
+    UPLOAD = "https://content-push.googleapis.com/upload"
+    BATCH_EXEC = "https://gemini.google.com/_/BardChatUi/data/batchexecute"
+
+
+class GRPC(StrEnum):
+    """
+    Google RPC ids used in Gemini API.
+    """
+
+    # Chat methods
+    LIST_CHATS = "MaZiqc"
+    READ_CHAT = "hNvQHb"
+    DELETE_CHAT = "GzXR5e"
+    DELETE_CHAT_SECOND = "qWymEb"
+
+    # Gem methods
+    LIST_GEMS = "CNgdBe"
+    CREATE_GEM = "oMH3Zd"
+    UPDATE_GEM = "kHv0Vd"
+    DELETE_GEM = "UXcSJb"
+
+    BARD_SETTINGS = "ESY5D"
+
+    LIST_MODELS = "otAQ7b"
+
+    IMAGE_FULL_SIZE = "c8o8Fe"
+
+
+class Headers(Enum):
+    GEMINI = {
+        "Content-Type": "application/x-www-form-urlencoded;charset=utf-8",
+        "Host": "gemini.google.com",
+        "Origin": "https://gemini.google.com",
+        "Referer": "https://gemini.google.com/",
+        "X-Same-Domain": "1",
+    }
+    ROTATE_COOKIES = {
+        "Content-Type": "application/json",
+    }
+    UPLOAD = {"Push-ID": "feeds/mcudyrk2a4khkz"}
+
+
+class Model(Enum):
+    UNSPECIFIED = ("unspecified", {}, False)
+    G_3_PRO_AI_FREE = (
+        "gemini-3-pro-ai-free",
+        {
+            "x-goog-ext-525001261-jspb": '[1,null,null,null,"9d8ca3786ebdfbea",null,null,0,[4],null,null,1]'
+        },
+        False,
+    )
+    G_3_FLASH_AI_FREE = (
+        "gemini-3-flash-ai-free",
+        {
+            "x-goog-ext-525001261-jspb": '[1,null,null,null,"fbb127bbb056c959",null,null,0,[4],null,null,1]'
+        },
+        False,
+    )
+    G_3_FLASH_THINKING_AI_FREE = (
+        "gemini-3-flash-thinking-ai-free",
+        {
+            "x-goog-ext-525001261-jspb": '[1,null,null,null,"5bf011840784117a",null,null,0,[4],null,null,1]'
+        },
+        False,
+    )
+    G_3_PRO_AI_PRO = (
+        "gemini-3-pro-ai-pro",
+        {
+            "x-goog-ext-525001261-jspb": '[1,null,null,null,"e6fa609c3fa255c0",null,null,0,[4],null,null,2]'
+        },
+        True,
+    )
+    G_3_FLASH_AI_PRO = (
+        "gemini-3-flash-ai-pro",
+        {
+            "x-goog-ext-525001261-jspb": '[1,null,null,null,"56fdd199312815e2",null,null,0,[4],null,null,2]'
+        },
+        True,
+    )
+    G_3_FLASH_THINKING_AI_PRO = (
+        "gemini-3-flash-thinking-ai-pro",
+        {
+            "x-goog-ext-525001261-jspb": '[1,null,null,null,"e051ce1aa80aa576",null,null,0,[4],null,null,2]'
+        },
+        True,
+    )
+
+    def __init__(self, name, header, advanced_only):
+        self.model_name = name
+        self.model_header = header
+        self.advanced_only = advanced_only
+
+    @classmethod
+    def from_name(cls, name: str):
+        for model in cls:
+            if model.model_name == name:
+                return model
+
+        raise ValueError(
+            f"Unknown model name: {name}. Available models: {', '.join([model.model_name for model in cls])}"
+        )
+
+    @classmethod
+    def from_dict(cls, model_dict: dict):
+        if "model_name" not in model_dict or "model_header" not in model_dict:
+            raise ValueError(
+                "When passing a custom model as a dictionary, 'model_name' and 'model_header' keys must be provided."
+            )
+
+        if not isinstance(model_dict["model_header"], dict):
+            raise ValueError(
+                "When passing a custom model as a dictionary, 'model_header' must be a dictionary containing valid header strings."
+            )
+
+        custom_model = cls.UNSPECIFIED
+        custom_model.model_name = model_dict["model_name"]
+        custom_model.model_header = model_dict["model_header"]
+        return custom_model
+
+
+class ErrorCode(IntEnum):
+    """
+    Known error codes returned from server.
+    """
+
+    TEMPORARY_ERROR_1013 = 1013  # Randomly raised when generating with certain models, but disappears soon after
+    USAGE_LIMIT_EXCEEDED = 1037
+    MODEL_INCONSISTENT = 1050
+    MODEL_HEADER_INVALID = 1052
+    IP_TEMPORARILY_BLOCKED = 1060

gemini_webapi/exceptions.py

@@ -0,0 +1,62 @@
+class AuthError(Exception):
+    """
+    Exception for authentication errors caused by invalid credentials/cookies.
+    """
+
+    pass
+
+
+class APIError(Exception):
+    """
+    Exception for package-level errors which need to be fixed in the future development (e.g. validation errors).
+    """
+
+    pass
+
+
+class ImageGenerationError(APIError):
+    """
+    Exception for generated image parsing errors.
+    """
+
+    pass
+
+
+class GeminiError(Exception):
+    """
+    Exception for errors returned from Gemini server which are not handled by the package.
+    """
+
+    pass
+
+
+class TimeoutError(GeminiError):
+    """
+    Exception for request timeouts.
+    """
+
+    pass
+
+
+class UsageLimitExceeded(GeminiError):
+    """
+    Exception for model usage limit exceeded errors.
+    """
+
+    pass
+
+
+class ModelInvalid(GeminiError):
+    """
+    Exception for invalid model header string errors.
+    """
+
+    pass
+
+
+class TemporarilyBlocked(GeminiError):
+    """
+    Exception for 429 Too Many Requests when IP is temporarily blocked.
+    """
+
+    pass

gemini_webapi/types/__init__.py

@@ -0,0 +1,11 @@
+# flake8: noqa
+
+from .candidate import Candidate
+from .gem import Gem, GemJar
+from .grpc import RPCData
+from .image import Image, WebImage, GeneratedImage
+from .video import Video, GeneratedVideo
+from .modeloutput import ModelOutput
+from .availablemodel import AvailableModel
+from .chatinfo import ChatInfo
+from .chathistory import ChatTurn, ChatHistory

gemini_webapi/types/availablemodel.py

@@ -0,0 +1,24 @@
+from pydantic import BaseModel
+from ..constants import Model
+
+
+class AvailableModel(BaseModel):
+    """
+    Available model configuration for the current account.
+
+    Parameters
+    ----------
+    id: `str`
+        The explicit internal code name of the model.
+    name: `str`
+        The display name of the model on the web UI.
+    model: `gemini_webapi.constants.Model`
+        The core model variation enum.
+    description: `str`
+        A brief description of the model's capabilities.
+    """
+
+    id: str
+    name: str
+    model: Model
+    description: str

gemini_webapi/types/candidate.py

@@ -0,0 +1,58 @@
+import html
+import reprlib
+
+from pydantic import BaseModel, field_validator
+
+from .image import Image, WebImage, GeneratedImage
+from .video import GeneratedVideo
+
+
+class Candidate(BaseModel):
+    """
+    A single reply candidate object in the model output. A full response from Gemini usually contains multiple reply candidates.
+
+    Parameters
+    ----------
+    rcid: `str`
+        Reply candidate ID to build the metadata
+    text: `str`
+        Text output
+    thoughts: `str`, optional
+        Model's thought process, can be empty. Only populated with `-thinking` models
+    web_images: `list[WebImage]`, optional
+        List of web images in reply, can be empty.
+    generated_images: `list[GeneratedImage]`, optional
+        List of generated images in reply, can be empty
+    generated_videos: `list[GeneratedVideo]`, optional
+        List of generated videos in reply, can be empty
+    """
+
+    rcid: str
+    text: str
+    text_delta: str | None = None
+    thoughts: str | None = None
+    thoughts_delta: str | None = None
+    web_images: list[WebImage] = []
+    generated_images: list[GeneratedImage] = []
+    generated_videos: list[GeneratedVideo] = []
+
+    def __str__(self):
+        return self.text
+
+    def __repr__(self):
+        return f"Candidate(rcid='{self.rcid}', text='{reprlib.repr(self.text)}', images={self.images}, videos={self.generated_videos})"
+
+    @field_validator("text", "thoughts")
+    @classmethod
+    def decode_html(cls, value: str) -> str:
+        """
+        Auto unescape HTML entities in text/thoughts if any.
+        """
+
+        if value:
+            value = html.unescape(value)
+        return value
+
+    @property
+    def images(self) -> list[Image]:
+        return list(self.web_images) + list(self.generated_images)  # type: ignore

gemini_webapi/types/chathistory.py

@@ -0,0 +1,53 @@
+import reprlib
+from typing import List, Optional
+
+from pydantic import BaseModel
+
+from .modeloutput import ModelOutput
+
+
+class ChatTurn(BaseModel):
+    """
+    Represents a single turn (message) in a chat conversation.
+
+    Parameters
+    ----------
+    role: `str`
+        The role of the message sender, either "user" or "model".
+    text: `str`
+        The text content of the message.
+    info: `ModelOutput`, optional
+        The full model output if the role is "model". This contains candidates, images, and metadata.
+    """
+
+    role: str
+    text: str
+    info: Optional[ModelOutput] = None
+
+    def __str__(self):
+        return f"{self.role.upper()}: {self.text}"
+
+    def __repr__(self):
+        return f"ChatTurn(role='{self.role}', text='{reprlib.repr(self.text)}')"
+
+
+class ChatHistory(BaseModel):
+    """
+    Represents the full history of a chat conversation.
+
+    Parameters
+    ----------
+    cid: `str`
+        The chat ID.
+    metadata: `list[str]`
+        The chat metadata.
+    turns: `list[ChatTurn]`
+        The list of messages in the conversation.
+    """
+
+    cid: str
+    metadata: List[str]
+    turns: List[ChatTurn]
+
+    def __repr__(self):
+        return f"ChatHistory(cid='{self.cid}', turns={len(self.turns)})"

gemini_webapi/types/chatinfo.py

@@ -0,0 +1,18 @@
+from pydantic import BaseModel
+
+
+class ChatInfo(BaseModel):
+    """
+    Chat information from the user's account.
+
+    Parameters
+    ----------
+    cid: `str`
+        The ID of the chat conversation (cid).
+    title: `str`
+        The display title of the chat conversation.
+    """
+
+    cid: str
+    title: str
+    is_pinned: bool = False

gemini_webapi/types/gem.py

@@ -0,0 +1,132 @@
+from pydantic import BaseModel
+
+
+class Gem(BaseModel):
+    """
+    Reusable Gemini Gem object working as a system prompt, providing additional context to the model.
+    Gemini provides a set of predefined gems, and users can create custom gems as well.
+
+    Parameters
+    ----------
+    id: `str`
+        Unique identifier for the gem.
+    name: `str`
+        User-friendly name of the gem.
+    description: `str`, optional
+        Brief description of the gem's purpose or content.
+    prompt: `str`, optional
+        The system prompt text that the gem provides to the model.
+    predefined: `bool`
+        Indicates whether the gem is predefined by Gemini or created by the user.
+    """
+
+    id: str
+    name: str
+    description: str | None = None
+    prompt: str | None = None
+    predefined: bool
+
+    def __str__(self) -> str:
+        return (
+            f"Gem(id='{self.id}', name='{self.name}', description='{self.description}', "
+            f"prompt='{self.prompt}', predefined={self.predefined})"
+        )
+
+
+class GemJar(dict[str, Gem]):
+    """
+    Helper class for handling a collection of `Gem` objects, stored by their ID.
+    This class extends `dict` to allows retrieving gems with extra filtering options.
+    """
+
+    def __iter__(self):
+        """
+        Iter over the gems in the jar.
+        """
+
+        return self.values().__iter__()
+
+    def get(
+        self, id: str | None = None, name: str | None = None, default: Gem | None = None
+    ) -> Gem | None:
+        """
+        Retrieves a gem by its id and/or name.
+        If both id and name are provided, returns the gem that matches both id and name.
+        If only id is provided, it's a direct lookup.
+        If only name is provided, it searches through the gems.
+
+        Parameters
+        ----------
+        id: `str`, optional
+            The unique identifier of the gem to retrieve.
+        name: `str`, optional
+            The user-friendly name of the gem to retrieve.
+        default: `Gem`, optional
+            The default value to return if no matching gem is found.
+
+        Returns
+        -------
+        `Gem` | None
+            The matching gem if found, otherwise return the default value.
+
+        Raises
+        ------
+        `AssertionError`
+            If neither id nor name is provided.
+        """
+
+        assert not (
+            id is None and name is None
+        ), "At least one of gem id or name must be provided."
+
+        if id is not None:
+            gem_candidate = super().get(id)
+            if gem_candidate:
+                if name is not None:
+                    if gem_candidate.name == name:
+                        return gem_candidate
+                    else:
+                        return default
+                else:
+                    return gem_candidate
+            else:
+                return default
+        elif name is not None:
+            for gem_obj in self.values():
+                if gem_obj.name == name:
+                    return gem_obj
+            return default
+
+        # Should be unreachable due to the assertion.
+        return default
+
+    def filter(
+        self, predefined: bool | None = None, name: str | None = None
+    ) -> "GemJar":
+        """
+        Returns a new `GemJar` containing gems that match the given filters.
+
+        Parameters
+        ----------
+        predefined: `bool`, optional
+            If provided, filters gems by whether they are predefined (True) or user-created (False).
+        name: `str`, optional
+            If provided, filters gems by their name (exact match).
+
+        Returns
+        -------
+        `GemJar`
+            A new `GemJar` containing the filtered gems. Can be empty if no gems match the criteria.
+        """
+
+        filtered_gems = GemJar()
+
+        for gem_id, gem in self.items():
+            if predefined is not None and gem.predefined != predefined:
+                continue
+            if name is not None and gem.name != name:
+                continue
+
+            filtered_gems[gem_id] = gem
+
+        return GemJar(filtered_gems)

gemini_webapi/types/grpc.py

@@ -0,0 +1,34 @@
+from pydantic import BaseModel
+
+from ..constants import GRPC
+
+
+class RPCData(BaseModel):
+    """
+    Helper class containing necessary data for Google RPC calls.
+
+    Parameters
+    ----------
+    rpcid : GRPC
+        Google RPC ID.
+    payload : str
+        Payload for the RPC call.
+    identifier : str, optional
+        Identifier/order for the RPC call, defaults to "generic".
+        Makes sense if there are multiple RPC calls in a batch, where this identifier
+        can be used to distinguish between responses.
+    """
+
+    rpcid: GRPC
+    payload: str
+    identifier: str = "generic"
+
+    def __repr__(self):
+        return f"GRPC(rpcid='{self.rpcid}', payload='{self.payload}', identifier='{self.identifier}')"
+
+    def serialize(self) -> list:
+        """
+        Serializes object into formatted payload ready for RPC call.
+        """
+
+        return [self.rpcid, self.payload, None, self.identifier]

gemini_webapi/types/image.py

@@ -0,0 +1,234 @@
+import hashlib
+import mimetypes
+import reprlib
+from datetime import datetime
+from pathlib import Path
+from typing import Any
+
+from curl_cffi.requests import AsyncSession
+from curl_cffi.requests.exceptions import HTTPError
+from pydantic import BaseModel
+
+from ..utils import logger
+
+
+class Image(BaseModel):
+    """A single image object returned from Gemini.
+
+    Attributes:
+        url (str): URL of the image.
+        title (str, optional): Title of the image. Defaults to "[Image]".
+        alt (str, optional): Optional description of the image.
+        proxy (str, optional): Proxy used when saving image.
+        client (Any, optional): Reference to the client object.
+    """
+
+    url: str
+    title: str = "[Image]"
+    alt: str = ""
+    proxy: str | None = None
+    client: Any = None
+
+    def __str__(self):
+        return f"Image(title='{self.title}', alt='{self.alt}', url='{reprlib.repr(self.url)}')"
+
+    async def save(
+        self,
+        path: str = "temp",
+        filename: str | None = None,
+        verbose: bool = False,
+        skip_invalid_filename: bool = False,
+        client: AsyncSession | None = None,
+    ) -> str | None:
+        """Saves the image to disk.
+
+        Args:
+            path (str, optional): Path to save the image. Defaults to "./temp".
+            filename (str | None, optional): File name to save the image. Defaults to
+                a unique generated name.
+            verbose (bool, optional): If True, will print the path of the saved file
+                or warning for invalid file name. Defaults to False.
+            skip_invalid_filename (bool, optional): If True, will only save the image
+                if the file name and extension are valid. Defaults to False.
+            client (AsyncSession | None, optional): Client used for requests.
+
+        Returns:
+            str | None: Absolute path of the saved image if successful, None if
+                filename is invalid and `skip_invalid_filename` is True.
+
+        Raises:
+            curl_cffi.requests.exceptions.HTTPError: If the network request failed.
+        """
+
+        if not filename or not Path(filename).suffix:
+            timestamp = datetime.now().strftime("%Y%m%d%H%M%S")
+            url_hash = hashlib.sha256(self.url.encode()).hexdigest()[:10]
+            base_name = Path(filename).stem if filename else "image"
+            filename = f"{timestamp}_{url_hash}_{base_name}"
+
+        close_client = False
+        req_client = client or self.client
+        if not req_client:
+            client_ref = getattr(self, "client_ref", None)
+            cookies = getattr(client_ref, "cookies", None) if client_ref else None
+            req_client = AsyncSession(
+                impersonate="chrome",
+                allow_redirects=True,
+                cookies=cookies,
+                proxy=self.proxy,
+            )
+            close_client = True
+
+        try:
+            response = await req_client.get(self.url)
+            if verbose:
+                logger.debug(f"HTTP Request: GET {self.url} [{response.status_code}]")
+
+            if response.status_code == 200:
+                path_obj_file = Path(filename)
+                if not path_obj_file.suffix:
+                    content_type = (
+                        response.headers.get("content-type", "")
+                        .split(";")[0]
+                        .strip()
+                        .lower()
+                    )
+                    ext = mimetypes.guess_extension(content_type) or ".png"
+                    filename = f"{filename}{ext}"
+
+                path_obj = Path(path)
+                path_obj.mkdir(parents=True, exist_ok=True)
+
+                dest = path_obj / filename
+                dest.write_bytes(response.content)
+
+                if verbose:
+                    logger.info(f"Image saved as {dest.resolve()}")
+
+                return str(dest.resolve())
+            else:
+                raise HTTPError(
+                    f"Error downloading image: {response.status_code} {response.reason}"
+                )
+        finally:
+            if close_client:
+                await req_client.close()
+
+
+class WebImage(Image):
+    """Image retrieved from web.
+
+    Returned when asking Gemini to "SEND an image of [something]".
+    """
+
+    pass
+
+
+class GeneratedImage(Image):
+    """Image generated by Google's AI image generator.
+
+    Returned when asking Gemini to "GENERATE an image of [something]".
+
+    Attributes:
+        client_ref (Any, optional): Reference to the GeminiClient instance.
+        cid (str, optional): Chat ID.
+        rid (str, optional): Response ID.
+        rcid (str, optional): Response candidate ID.
+        image_id (str, optional): Image ID generated.
+    """
+
+    client_ref: Any = None
+    cid: str = ""
+    rid: str = ""
+    rcid: str = ""
+    image_id: str = ""
+
+    # @override
+    async def save(
+        self,
+        path: str = "temp",
+        filename: str | None = None,
+        verbose: bool = False,
+        skip_invalid_filename: bool = False,
+        client: AsyncSession | None = None,
+        full_size: bool = True,
+    ) -> str | None:
+        """Saves the generated image to disk.
+
+        Args:
+            path (str, optional): Path to save the image. Defaults to "temp".
+            filename (str | None, optional): Filename to save the image.
+            verbose (bool, optional): Prints status. Defaults to False.
+            skip_invalid_filename (bool, optional): Skip if invalid. Defaults to False.
+            client (AsyncSession | None, optional): An existing AsyncSession client.
+            full_size (bool, optional): Modifies preview URLs to fetch full-size images if True. Defaults to True.
+
+        Returns:
+            str | None: Absolute path of the saved image if successfully saved, None otherwise.
+        """
+        if full_size:
+            if all([self.client_ref, self.cid, self.rid, self.rcid, self.image_id]):
+                try:
+                    original_url = await self.client_ref._get_image_full_size(
+                        cid=self.cid,
+                        rid=self.rid,
+                        rcid=self.rcid,
+                        image_id=self.image_id,
+                    )
+                    if original_url:
+                        req_url = f"{original_url}=d-I?alr=yes"
+
+                        req_client = client or self.client
+                        close_client = False
+
+                        if not req_client:
+                            req_client = AsyncSession(
+                                impersonate="chrome",
+                                allow_redirects=True,
+                                cookies=getattr(self.client_ref, "cookies", None),
+                                proxy=self.proxy,
+                            )
+                            close_client = True
+
+                        try:
+                            response = await req_client.get(req_url)
+                            response.raise_for_status()
+                            url_text = response.text
+
+                            response = await req_client.get(url_text)
+                            response.raise_for_status()
+                            self.url = response.text
+
+                            return await super().save(
+                                path=path,
+                                filename=filename,
+                                verbose=verbose,
+                                skip_invalid_filename=skip_invalid_filename,
+                                client=req_client,
+                            )
+                        finally:
+                            if close_client and req_client:
+                                await req_client.close()
+
+                except Exception as e:
+                    logger.debug(
+                        f"Failed to fetch full size image URL via RPC: {e}, falling back to default URL suffix."
+                    )
+
+            if "=s1024-rj" in self.url:
+                self.url = self.url.replace("=s1024-rj", "=s2048-rj")
+            elif "=s2048-rj" not in self.url:
+                self.url += "=s2048-rj"
+        else:
+            if "=s2048-rj" in self.url:
+                self.url = self.url.replace("=s2048-rj", "=s1024-rj")
+            elif "=s1024-rj" not in self.url:
+                self.url += "=s1024-rj"
+
+        return await super().save(
+            path=path,
+            filename=filename,
+            verbose=verbose,
+            skip_invalid_filename=skip_invalid_filename,
+            client=client,
+        )

gemini_webapi/types/modeloutput.py

@@ -0,0 +1,58 @@
+from pydantic import BaseModel
+
+from .image import Image
+from .candidate import Candidate
+from .video import GeneratedVideo
+
+
+class ModelOutput(BaseModel):
+    """
+    Classified output from gemini.google.com
+
+    Parameters
+    ----------
+    metadata: `list[str]`
+        List of chat metadata `[cid, rid, rcid]`, can be shorter than 3 elements, like `[cid, rid]` or `[cid]` only
+    candidates: `list[Candidate]`
+        List of all candidates returned from gemini
+    chosen: `int`, optional
+        Index of the chosen candidate, by default will choose the first one
+    """
+
+    metadata: list[str]
+    candidates: list[Candidate]
+    chosen: int = 0
+
+    def __str__(self):
+        return self.text
+
+    def __repr__(self):
+        return f"ModelOutput(metadata={self.metadata}, chosen={self.chosen}, candidates={self.candidates})"
+
+    @property
+    def text(self) -> str:
+        return self.candidates[self.chosen].text
+
+    @property
+    def text_delta(self) -> str:
+        return self.candidates[self.chosen].text_delta or ""
+
+    @property
+    def thoughts(self) -> str | None:
+        return self.candidates[self.chosen].thoughts
+
+    @property
+    def thoughts_delta(self) -> str:
+        return self.candidates[self.chosen].thoughts_delta or ""
+
+    @property
+    def images(self) -> list[Image]:
+        return self.candidates[self.chosen].images
+
+    @property
+    def videos(self) -> list[GeneratedVideo]:
+        return self.candidates[self.chosen].generated_videos
+
+    @property
+    def rcid(self) -> str:
+        return self.candidates[self.chosen].rcid

gemini_webapi/types/video.py

@@ -0,0 +1,177 @@
+import asyncio
+import hashlib
+import mimetypes
+import reprlib
+from datetime import datetime
+from pathlib import Path
+from typing import Any
+
+from curl_cffi.requests import AsyncSession
+from curl_cffi.requests.exceptions import HTTPError
+from pydantic import BaseModel
+
+from ..utils import logger
+
+
+class Video(BaseModel):
+    """A single video object returned from Gemini.
+
+    Attributes:
+        url (str): URL of the video.
+        title (str, optional): Title of the video. Defaults to "[Video]".
+        proxy (str, optional): Proxy used when saving video.
+        client (Any, optional): Reference to the client object.
+    """
+
+    url: str
+    title: str = "[Video]"
+    proxy: str | None = None
+    client: Any = None
+
+    def __str__(self):
+        return f"Video(title='{self.title}', url='{reprlib.repr(self.url)}')"
+
+    async def save(
+        self,
+        path: str = "temp",
+        filename: str | None = None,
+        verbose: bool = False,
+        skip_invalid_filename: bool = False,
+        client: AsyncSession | None = None,
+    ) -> tuple[str | None, str | None]:
+        """Saves the video to disk.
+
+        Args:
+            path (str, optional): Path to save the video. Defaults to "./temp".
+            filename (str | None, optional): File name to save the video. Defaults to
+                a unique generated name.
+            verbose (bool, optional): If True, will print the path of the saved file.
+            skip_invalid_filename (bool, optional): If True, will only save the video
+                if the file name has a valid extension. Defaults to False.
+            client (AsyncSession | None, optional): Client used for requests.
+
+        Returns:
+            tuple[str | None, str | None]: (Absolute path of the saved video, str | None path of thumbnail)
+        """
+
+        if not filename or not Path(filename).suffix:
+            timestamp = datetime.now().strftime("%Y%m%d%H%M%S")
+            url_hash = hashlib.sha256(self.url.encode()).hexdigest()[:10]
+            base_name = Path(filename).stem if filename else "video"
+            filename = f"{timestamp}_{url_hash}_{base_name}"
+
+        close_client = False
+        req_client = client or self.client
+        if not req_client:
+            client_ref = getattr(self, "client_ref", None)
+            cookies = getattr(client_ref, "cookies", None) if client_ref else None
+            req_client = AsyncSession(
+                impersonate="chrome",
+                allow_redirects=True,
+                cookies=cookies,
+                proxy=self.proxy,
+            )
+            close_client = True
+
+        try:
+            path_obj = Path(path)
+            path_obj.mkdir(parents=True, exist_ok=True)
+            return await self._perform_save(req_client, path_obj, filename, verbose)
+        finally:
+            if close_client:
+                await req_client.close()
+
+    @staticmethod
+    async def _download_file(
+        req_client: AsyncSession,
+        url: str,
+        path_obj: Path,
+        filename: str,
+        default_ext: str = ".mp4",
+        verbose: bool = False,
+    ) -> str | None:
+        """Internal helper to download a file and determine its extension."""
+        response = await req_client.get(url)
+        if verbose:
+            logger.debug(f"HTTP Request: GET {url} [{response.status_code}]")
+
+        if response.status_code == 200:
+            path_obj_file = Path(filename)
+            if not path_obj_file.suffix:
+                content_type = (
+                    response.headers.get("content-type", "")
+                    .split(";")[0]
+                    .strip()
+                    .lower()
+                )
+                ext = mimetypes.guess_extension(content_type) or default_ext
+                filename = f"{filename}{ext}"
+
+            dest = path_obj / filename
+            dest.write_bytes(response.content)
+
+            if verbose:
+                logger.info(f"File saved as {dest.resolve()}")
+
+            return str(dest.resolve())
+        elif response.status_code == 206:
+            return "206"
+        else:
+            raise HTTPError(
+                f"Error downloading file: {response.status_code} {response.reason}"
+            )
+
+    async def _perform_save(
+        self, req_client: AsyncSession, path_obj: Path, filename: str, verbose: bool
+    ) -> tuple[str | None, str | None]:
+        """Base implementation: simple download."""
+        path = await self._download_file(
+            req_client, self.url, path_obj, filename, ".mp4", verbose
+        )
+        return path, None
+
+
+class GeneratedVideo(Video):
+    """Video generated by Google's AI Veo.
+
+    Attributes:
+        client_ref (Any, optional): Reference to the GeminiClient instance.
+        thumbnail (str, optional): URL of the video thumbnail.
+        cid (str, optional): Chat ID.
+        rid (str, optional): Response ID.
+        rcid (str, optional): Response candidate ID.
+    """
+
+    client_ref: Any = None
+    thumbnail: str = ""
+    cid: str = ""
+    rid: str = ""
+    rcid: str = ""
+
+    # @override
+    async def _perform_save(
+        self, req_client: AsyncSession, path_obj: Path, filename: str, verbose: bool
+    ) -> tuple[str | None, str | None]:
+        """Internal method for GeneratedVideo, handling thumbnails and polling."""
+        saved_thumb_path = None
+        if self.thumbnail:
+            thumb_base = Path(filename).stem
+            try:
+                saved_thumb_path = await self._download_file(
+                    req_client, self.thumbnail, path_obj, thumb_base, ".jpg", verbose
+                )
+            except Exception as e:
+                if verbose:
+                    logger.warning(f"Failed to save thumbnail: {e}")
+
+        while True:
+            path_or_marker = await self._download_file(
+                req_client, self.url, path_obj, filename, ".mp4", verbose
+            )
+
+            if path_or_marker == "206":
+                if verbose:
+                    logger.info("Video still generating (206), retrying in 10s...")
+                await asyncio.sleep(10)
+            else:
+                return path_or_marker, saved_thumb_path

gemini_webapi/utils/__init__.py

@@ -0,0 +1,14 @@
+# flake8: noqa
+
+from .decorators import running
+from .get_access_token import get_access_token
+from .load_browser_cookies import load_browser_cookies
+from .logger import logger, set_log_level
+from .parsing import (
+    extract_json_from_response,
+    get_delta_by_fp_len,
+    get_nested_value,
+    parse_response_by_frame,
+)
+from .rotate_1psidts import rotate_1psidts
+from .upload_file import upload_file, parse_file_name

gemini_webapi/utils/decorators.py

@@ -0,0 +1,99 @@
+import asyncio
+import functools
+import inspect
+from collections.abc import Callable
+
+from ..exceptions import APIError
+
+DELAY_FACTOR = 5
+
+
+def running(retry: int = 0) -> Callable:
+    """
+    Decorator to check if GeminiClient is running before making a request.
+    Supports both regular async functions and async generators.
+
+    Parameters
+    ----------
+    retry: `int`, optional
+        Max number of retries when `gemini_webapi.APIError` is raised.
+    """
+
+    def decorator(func):
+        if inspect.isasyncgenfunction(func):
+
+            @functools.wraps(func)
+            async def wrapper(client, *args, current_retry=None, **kwargs):
+                if current_retry is None:
+                    current_retry = retry
+
+                try:
+                    if not client._running:
+                        await client.init(
+                            timeout=client.timeout,
+                            auto_close=client.auto_close,
+                            close_delay=client.close_delay,
+                            auto_refresh=client.auto_refresh,
+                            refresh_interval=client.refresh_interval,
+                            verbose=client.verbose,
+                            watchdog_timeout=client.watchdog_timeout,
+                        )
+
+                    if not client._running:
+                        raise APIError(
+                            f"Invalid function call: GeminiClient.{func.__name__}. Client initialization failed."
+                        )
+
+                    async for item in func(client, *args, **kwargs):
+                        yield item
+                except APIError:
+                    if current_retry > 0:
+                        delay = (retry - current_retry + 1) * DELAY_FACTOR
+                        await asyncio.sleep(delay)
+                        async for item in wrapper(
+                            client, *args, current_retry=current_retry - 1, **kwargs
+                        ):
+                            yield item
+                    else:
+                        raise
+
+            return wrapper
+        else:
+
+            @functools.wraps(func)
+            async def wrapper(client, *args, current_retry=None, **kwargs):
+                if current_retry is None:
+                    current_retry = retry
+
+                try:
+                    if not client._running:
+                        await client.init(
+                            timeout=client.timeout,
+                            auto_close=client.auto_close,
+                            close_delay=client.close_delay,
+                            auto_refresh=client.auto_refresh,
+                            refresh_interval=client.refresh_interval,
+                            verbose=client.verbose,
+                            watchdog_timeout=client.watchdog_timeout,
+                        )
+
+                    if not client._running:
+                        raise APIError(
+                            f"Invalid function call: GeminiClient.{func.__name__}. Client initialization failed."
+                        )
+
+                    return await func(client, *args, **kwargs)
+                except APIError:
+                    if current_retry > 0:
+                        delay = (retry - current_retry + 1) * DELAY_FACTOR
+                        await asyncio.sleep(delay)
+
+                        return await wrapper(
+                            client, *args, current_retry=current_retry - 1, **kwargs
+                        )
+
+                    raise
+
+            return wrapper
+
+    return decorator

gemini_webapi/utils/get_access_token.py

@@ -0,0 +1,202 @@
+import os
+import re
+from pathlib import Path
+
+from curl_cffi.requests import AsyncSession, Cookies, Response
+
+from .load_browser_cookies import HAS_BC3, load_browser_cookies
+from .logger import logger
+from .rotate_1psidts import _extract_cookie_value
+from ..constants import Endpoint, Headers
+from ..exceptions import AuthError
+
+
+async def send_request(
+    client: AsyncSession, cookies: dict | Cookies, verbose: bool = False
+) -> Response:
+    """
+    Send http request with provided cookies using a shared session.
+    """
+    client.cookies.clear()
+    if isinstance(cookies, Cookies):
+        client.cookies.update(cookies)
+    else:
+        for k, v in cookies.items():
+            client.cookies.set(k, v, domain=".google.com")
+
+    response = await client.get(Endpoint.INIT, headers=Headers.GEMINI.value)
+    if verbose:
+        logger.debug(f"HTTP Request: GET {Endpoint.INIT} [{response.status_code}]")
+    response.raise_for_status()
+    return response
+
+
+async def get_access_token(
+    base_cookies: dict | Cookies,
+    proxy: str | None = None,
+    verbose: bool = False,
+    verify: bool = True,
+) -> tuple[str | None, str | None, str | None, AsyncSession]:
+    """
+    Send a get request to gemini.google.com for each group of available cookies and return
+    the value of "SNlM0e" as access token on the first successful request.
+
+    Returns the **live** AsyncSession that succeeded so the caller can reuse
+    the same TLS connection for subsequent requests.
+    """
+
+    client = AsyncSession(
+        impersonate="chrome", proxy=proxy, allow_redirects=True, verify=verify
+    )
+
+    try:
+        response = await client.get(Endpoint.GOOGLE)
+        if verbose:
+            logger.debug(
+                f"HTTP Request: GET {Endpoint.GOOGLE} [{response.status_code}]"
+            )
+        preflight_cookies = Cookies(client.cookies)
+    except Exception:
+        await client.close()
+        raise
+
+    extra_cookies = Cookies()
+    if response.status_code == 200:
+        extra_cookies = preflight_cookies
+
+    # Phase 1: Prepare Cache
+    cookie_jars_to_test = []
+    tried_psid_ts = set()
+
+    if isinstance(base_cookies, Cookies):
+        base_psid = _extract_cookie_value(base_cookies, "__Secure-1PSID")
+        base_psidts = _extract_cookie_value(base_cookies, "__Secure-1PSIDTS")
+    else:
+        base_psid = base_cookies.get("__Secure-1PSID")
+        base_psidts = base_cookies.get("__Secure-1PSIDTS")
+
+    gemini_cookie_path = os.getenv("GEMINI_COOKIE_PATH")
+    if gemini_cookie_path:
+        cache_dir = Path(gemini_cookie_path)
+    else:
+        cache_dir = Path(__file__).parent / "temp"
+
+    if base_psid:
+        filename = f".cached_1psidts_{base_psid}.txt"
+        cache_file = cache_dir / filename
+        if cache_file.is_file():
+            cached_1psidts = cache_file.read_text().strip()
+            if cached_1psidts:
+                jar = Cookies(extra_cookies)
+                jar.update(base_cookies)
+                jar.set("__Secure-1PSIDTS", cached_1psidts, domain=".google.com")
+                cookie_jars_to_test.append((jar, "Cache"))
+                tried_psid_ts.add((base_psid, cached_1psidts))
+            elif verbose:
+                logger.debug("Skipping loading cached cookies. Cache file is empty.")
+        elif verbose:
+            logger.debug("Skipping loading cached cookies. Cache file not found.")
+
+    if not base_psid:
+        for cache_file in cache_dir.glob(".cached_1psidts_*.txt"):
+            psid = cache_file.stem[16:]
+            cached_1psidts = cache_file.read_text().strip()
+            if cached_1psidts:
+                jar = Cookies(extra_cookies)
+                jar.set("__Secure-1PSID", psid, domain=".google.com")
+                jar.set("__Secure-1PSIDTS", cached_1psidts, domain=".google.com")
+                cookie_jars_to_test.append((jar, "Cache"))
+                tried_psid_ts.add((psid, cached_1psidts))
+
+    # Phase 2: Base Cookies
+    if base_psid and base_psidts:
+        if (base_psid, base_psidts) not in tried_psid_ts:
+            jar = Cookies(extra_cookies)
+            jar.update(base_cookies)
+            cookie_jars_to_test.append((jar, "Base Cookies"))
+            tried_psid_ts.add((base_psid, base_psidts))
+        elif verbose:
+            logger.debug("Skipping base cookies as they match cached cookies.")
+    elif verbose and not cookie_jars_to_test:
+        logger.debug(
+            "Skipping loading base cookies. Either __Secure-1PSID or __Secure-1PSIDTS is not provided."
+        )
+
+    # Phase 3: Browser Cookies
+    try:
+        browser_cookies = load_browser_cookies(
+            domain_name="google.com", verbose=verbose
+        )
+        if browser_cookies:
+            for browser, cookies in browser_cookies.items():
+                if secure_1psid := cookies.get("__Secure-1PSID"):
+                    if base_psid and base_psid != secure_1psid:
+                        if verbose:
+                            logger.debug(
+                                f"Skipping loading local browser cookies from {browser}. "
+                                "__Secure-1PSID does not match the one provided."
+                            )
+                        continue
+
+                    secure_1psidts = cookies.get("__Secure-1PSIDTS")
+                    if (secure_1psid, secure_1psidts or "") in tried_psid_ts:
+                        continue
+
+                    local_cookies = {"__Secure-1PSID": secure_1psid}
+                    if secure_1psidts:
+                        local_cookies["__Secure-1PSIDTS"] = secure_1psidts
+                    if nid := cookies.get("NID"):
+                        local_cookies["NID"] = nid
+
+                    jar = Cookies(extra_cookies)
+                    for k, v in local_cookies.items():
+                        jar.set(k, v, domain=".google.com")
+
+                    cookie_jars_to_test.append((jar, f"Browser ({browser})"))
+                    tried_psid_ts.add((secure_1psid, secure_1psidts or ""))
+                    if verbose:
+                        logger.debug(f"Prepared local browser cookies from {browser}")
+
+        if (
+            HAS_BC3
+            and not any(group.startswith("Browser") for _, group in cookie_jars_to_test)
+            and verbose
+        ):
+            logger.debug(
+                "Skipping loading local browser cookies. Login to gemini.google.com in your browser first."
+            )
+    except Exception:
+        if verbose:
+            logger.debug(
+                "Skipping loading local browser cookies (Not available or no permission)."
+            )
+
+    current_attempt = 0
+    for jar, group_name in cookie_jars_to_test:
+        current_attempt += 1
+        try:
+            res = await send_request(client, jar, verbose=verbose)
+            snlm0e = re.search(r'"SNlM0e":\s*"(.*?)"', res.text)
+            cfb2h = re.search(r'"cfb2h":\s*"(.*?)"', res.text)
+            fdrfje = re.search(r'"FdrFJe":\s*"(.*?)"', res.text)
+            if snlm0e or cfb2h or fdrfje:
+                if verbose:
+                    logger.debug(
+                        f"Init attempt ({current_attempt}) from {group_name} succeeded."
+                    )
+                return (
+                    snlm0e.group(1) if snlm0e else None,
+                    cfb2h.group(1) if cfb2h else None,
+                    fdrfje.group(1) if fdrfje else None,
+                    client,
+                )
+        except Exception:
+            if verbose:
+                logger.debug(
+                    f"Init attempt ({current_attempt}) from {group_name} failed."
+                )
+
+    await client.close()
+    raise AuthError(
+        f"Failed to initialize client after {current_attempt} attempts. SECURE_1PSIDTS could get expired frequently, please make sure cookie values are up to date."
+    )

gemini_webapi/utils/load_browser_cookies.py

@@ -0,0 +1,82 @@
+from concurrent.futures import ThreadPoolExecutor, as_completed
+from http.cookiejar import CookieJar
+
+from .logger import logger
+
+try:
+    import browser_cookie3 as bc3
+
+    HAS_BC3 = True
+except ImportError:
+    bc3 = None
+    HAS_BC3 = False
+
+
+def load_browser_cookies(domain_name: str = "", verbose: bool = False) -> dict:
+    """
+    Try to load cookies from all supported browsers and return combined cookiejar.
+    Optionally pass in a domain name to only load cookies from the specified domain.
+
+    Parameters
+    ----------
+    domain_name : str, optional
+        Domain name to filter cookies by, by default will load all cookies without filtering.
+    verbose : bool, optional
+        If `True`, will print more infomation in logs.
+
+    Returns
+    -------
+    `dict[str, dict]`
+        Dictionary with browser as keys and their cookies for the specified domain as values.
+        Only browsers that have cookies for the specified domain will be included.
+    """
+    if not HAS_BC3 or bc3 is None:
+        if verbose:
+            logger.debug(
+                "Optional dependency 'browser-cookie3' not found. Skipping browser cookie loading."
+            )
+        return {}
+
+    browser_fns = [
+        bc3.chrome,
+        bc3.chromium,
+        bc3.opera,
+        bc3.opera_gx,
+        bc3.brave,
+        bc3.edge,
+        bc3.vivaldi,
+        bc3.firefox,
+        bc3.librewolf,
+        bc3.safari,
+    ]
+
+    cookies = {}
+
+    def fetch_cookies(cookie_fn):
+        try:
+            jar: CookieJar = cookie_fn(domain_name=domain_name)
+            if jar:
+                return cookie_fn.__name__, {cookie.name: cookie.value for cookie in jar}
+        except bc3.BrowserCookieError:
+            pass
+        except PermissionError:
+            if verbose:
+                logger.warning(
+                    f"Permission denied while trying to load cookies from {cookie_fn.__name__}."
+                )
+        except Exception:
+            if verbose:
+                logger.debug(
+                    f"Failed to load cookies from {cookie_fn.__name__} (may not be installed)."
+                )
+        return None
+
+    with ThreadPoolExecutor(max_workers=len(browser_fns)) as executor:
+        futures = [executor.submit(fetch_cookies, fn) for fn in browser_fns]
+        for future in as_completed(futures):
+            result = future.result()
+            if result:
+                browser_name, cookie_dict = result
+                cookies[browser_name] = cookie_dict
+
+    return cookies

gemini_webapi/utils/logger.py

@@ -0,0 +1,37 @@
+import sys
+from loguru import logger as _logger
+
+_handler_id = None
+
+
+def set_log_level(level: str | int) -> None:
+    """
+    Set the log level for gemini_webapi. The default log level is "INFO".
+
+    Note: calling this function for the first time will globally remove all existing loguru
+    handlers. To avoid this, you may want to set logging behaviors directly with loguru.
+
+    Parameters
+    ----------
+    level : `str | int`
+        Log level: "TRACE", "DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"
+
+    Examples
+    --------
+    >>> from gemini_webapi import set_log_level
+    >>> set_log_level("DEBUG")  # Show debug messages
+    >>> set_log_level("ERROR")  # Only show errors
+    """
+
+    global _handler_id
+
+    _logger.remove(_handler_id)
+
+    _handler_id = _logger.add(
+        sys.stderr,
+        level=level,
+        filter=lambda record: record["extra"].get("name") == "gemini_webapi",
+    )
+
+
+logger = _logger.bind(name="gemini_webapi")

gemini_webapi/utils/parsing.py

@@ -0,0 +1,265 @@
+import difflib
+import re
+import reprlib
+from typing import Any
+
+import orjson as json
+
+from .logger import logger
+
+_LENGTH_MARKER_PATTERN = re.compile(r"(\d+)\n")
+_FLICKER_ESC_RE = re.compile(r"\\+[`*_~].*$")
+
+
+def get_clean_text(s: str) -> str:
+    """
+    Clean Gemini text by removing trailing code block artifacts and temporary escapes of Markdown markers.
+    """
+
+    if not s:
+        return ""
+
+    if s.endswith("\n```"):
+        s = s[:-4]
+
+    return _FLICKER_ESC_RE.sub("", s)
+
+
+def get_delta_by_fp_len(
+    new_raw: str, last_sent_clean: str, is_final: bool
+) -> tuple[str, str]:
+    """
+    Calculate text delta by aligning stable content and matching volatile symbols.
+    Handles temporary flicker at ends and permanent escaping drift during code block transitions.
+    Uses SequenceMatcher to robustly handle middle-string modifications.
+    """
+
+    new_c = get_clean_text(new_raw) if not is_final else new_raw
+
+    if new_c.startswith(last_sent_clean):
+        return new_c[len(last_sent_clean) :], new_c
+
+    # Find the matching suffix to handle differences gracefully
+    search_len = min(3000, max(1000, len(last_sent_clean)))
+    search_len = min(search_len, len(last_sent_clean), len(new_c))
+
+    if search_len == 0:
+        return new_c, new_c
+
+    tail_last = last_sent_clean[-search_len:]
+    tail_new = new_c[-search_len:]
+
+    sm = difflib.SequenceMatcher(None, tail_last, tail_new)
+    blocks = [b for b in sm.get_matching_blocks() if b.size > 0]
+
+    if blocks:
+        last_match = blocks[-1]
+        match_end = last_match.b + last_match.size
+        return tail_new[match_end:], new_c
+
+    # Fallback to full string if tail didn't match at all
+    sm = difflib.SequenceMatcher(None, last_sent_clean, new_c)
+    blocks = [b for b in sm.get_matching_blocks() if b.size > 0]
+
+    if blocks:
+        last_match = blocks[-1]
+        match_end = last_match.b + last_match.size
+        return new_c[match_end:], new_c
+
+    return new_c, new_c
+
+
+def _get_char_count_for_utf16_units(
+    s: str, start_idx: int, utf16_units: int
+) -> tuple[int, int]:
+    """
+    Calculate the number of Python characters (code points) and actual UTF-16
+    units found.
+    """
+
+    count = 0
+    units = 0
+    limit = len(s)
+
+    while units < utf16_units and (start_idx + count) < limit:
+        char = s[start_idx + count]
+        u = 2 if ord(char) > 0xFFFF else 1
+        if units + u > utf16_units:
+            break
+        units += u
+        count += 1
+
+    return count, units
+
+
+def get_nested_value(
+    data: Any, path: list[int | str], default: Any = None, verbose: bool = False
+) -> Any:
+    """
+    Safely navigate through a nested structure (list or dict) using a sequence of keys/indices.
+
+    Parameters
+    ----------
+    data: `Any`
+        The nested structure to traverse.
+    path: `list[int | str]`
+        A list of indices or keys representing the path.
+    default: `Any`
+        Value to return if the path is invalid.
+    verbose: `bool`
+        If True, log debug information when the path cannot be fully traversed.
+    """
+
+    current = data
+
+    for i, key in enumerate(path):
+        found = False
+        if isinstance(key, int):
+            if isinstance(current, list) and -len(current) <= key < len(current):
+                current = current[key]
+                found = True
+        elif isinstance(key, str):
+            if isinstance(current, dict) and key in current:
+                current = current[key]
+                found = True
+
+        if not found:
+            if verbose:
+                logger.debug(
+                    f"Safe navigation: path {path} ended at index {i} (key '{key}'), "
+                    f"returning default. Context: {reprlib.repr(current)}"
+                )
+            return default
+
+    return current if current is not None else default
+
+
+def parse_response_by_frame(content: str) -> tuple[list[Any], str]:
+    """
+    Core parser for Google's length-prefixed framing protocol,
+    Parse as many JSON frames as possible from an accumulated buffer received from streaming responses.
+
+    This function implements Google's length-prefixed framing protocol. Each frame starts
+    with a length marker (number of characters) followed by a newline and the JSON content.
+    If a frame is partially received, it stays in the buffer for the next call.
+
+    Each frame has the format: `[length]\n[json_payload]\n`,
+    The length value includes the newline after the number and the newline after the JSON.
+
+    Parameters
+    ----------
+    content: `str`
+        The accumulated string buffer containing raw streaming data from the API.
+
+    Returns
+    -------
+    `tuple[list[Any], str]`
+        A tuple containing:
+        - A list of parsed JSON objects (envelopes) extracted from the buffer.
+        - The remaining unparsed part of the buffer (incomplete frames).
+    """
+
+    consumed_pos = 0
+    total_len = len(content)
+    parsed_frames = []
+
+    while consumed_pos < total_len:
+        while consumed_pos < total_len and content[consumed_pos].isspace():
+            consumed_pos += 1
+
+        if consumed_pos >= total_len:
+            break
+
+        match = _LENGTH_MARKER_PATTERN.match(content, pos=consumed_pos)
+        if not match:
+            break
+
+        length_val = match.group(1)
+        length = int(length_val)
+
+        # Content starts immediately after the digits.
+        # Google uses UTF-16 code units (JavaScript `String.length`) for the length marker.
+        start_content = match.start() + len(length_val)
+        char_count, units_found = _get_char_count_for_utf16_units(
+            content, start_content, length
+        )
+
+        if units_found < length:
+            logger.debug(
+                f"Incomplete frame at position {consumed_pos}: expected {length} UTF-16 units, "
+                f"but received {units_found}. Waiting for additional data..."
+            )
+            break
+
+        end_pos = start_content + char_count
+        chunk = content[start_content:end_pos].strip()
+        consumed_pos = end_pos
+
+        if not chunk:
+            continue
+
+        try:
+            parsed = json.loads(chunk)
+            if isinstance(parsed, list):
+                parsed_frames.extend(parsed)
+            else:
+                parsed_frames.append(parsed)
+        except json.JSONDecodeError:
+            logger.debug(
+                f"Failed to parse chunk at pos {start_content} with length {length}. "
+                f"Frame content: {reprlib.repr(chunk)}"
+            )
+
+    return parsed_frames, content[consumed_pos:]
+
+
+def extract_json_from_response(text: str) -> list:
+    """
+    Extract and normalize JSON content from a Google API response.
+    """
+
+    if not isinstance(text, str):
+        raise TypeError(
+            f"Input text is expected to be a string, got {type(text).__name__} instead."
+        )
+
+    content = text
+    if content.startswith(")]}'"):
+        content = content[4:]
+
+    content = content.lstrip()
+
+    # Try extracting with framing protocol first, as it's the most structured format
+    result, _ = parse_response_by_frame(content)
+    if result:
+        return result
+
+    # Extract the entire content if parsing by frames failed
+    content_stripped = content.strip()
+    try:
+        parsed = json.loads(content_stripped)
+        return parsed if isinstance(parsed, list) else [parsed]
+    except json.JSONDecodeError:
+        pass
+
+    # Extract with NDJSON
+    collected_lines = []
+    for line in content_stripped.splitlines():
+        line = line.strip()
+        if not line:
+            continue
+
+        try:
+            parsed = json.loads(line)
+        except json.JSONDecodeError:
+            continue
+
+        if isinstance(parsed, list):
+            collected_lines.extend(parsed)
+        elif isinstance(parsed, dict):
+            collected_lines.append(parsed)
+
+    if collected_lines:
+        return collected_lines
+
+    raise ValueError("Could not find a valid JSON object or array in the response.")

gemini_webapi/utils/rotate_1psidts.py

@@ -0,0 +1,102 @@
+import os
+import time
+from pathlib import Path
+
+from curl_cffi.requests import AsyncSession, Cookies
+
+from ..constants import Endpoint, Headers
+from ..exceptions import AuthError
+from .logger import logger
+
+
+def _extract_cookie_value(cookies: Cookies, name: str) -> str | None:
+    """
+    Extract a cookie value from a curl_cffi Cookies jar, trying domain-specific
+    lookups first to avoid CookieConflict, then falling back to iteration.
+    """
+    for domain in (
+        ".google.com",
+        "google.com",
+        ".accounts.google.com",
+        "accounts.google.com",
+    ):
+        value = cookies.get(name, domain=domain)
+        if value:
+            return value
+
+    for cookie in cookies.jar:
+        if cookie.name == name:
+            return cookie.value
+
+    return None
+
+
+async def rotate_1psidts(client: AsyncSession, verbose: bool = False) -> str | None:
+    """
+    Refresh the __Secure-1PSIDTS cookie and store the refreshed cookie value in cache file.
+
+    Parameters
+    ----------
+    client : `curl_cffi.requests.AsyncSession`
+        The shared async session to use for the request.
+    verbose: `bool`, optional
+        If `True`, will print more infomation in logs.
+
+    Returns
+    -------
+    `str | None`
+        New value of the __Secure-1PSIDTS cookie if rotation was successful.
+
+    Raises
+    ------
+    `gemini_webapi.AuthError`
+        If request failed with 401 Unauthorized.
+    `curl_cffi.requests.exceptions.HTTPError`
+        If request failed with other status codes.
+    """
+
+    path = (
+        (GEMINI_COOKIE_PATH := os.getenv("GEMINI_COOKIE_PATH"))
+        and Path(GEMINI_COOKIE_PATH)
+        or (Path(__file__).parent / "temp")
+    )
+    path.mkdir(parents=True, exist_ok=True)
+
+    # Safely get __Secure-1PSID value for filename
+    secure_1psid = _extract_cookie_value(client.cookies, "__Secure-1PSID")
+
+    if not secure_1psid:
+        return None
+
+    filename = f".cached_1psidts_{secure_1psid}.txt"
+    path = path / filename
+
+    # Check if the cache file was modified in the last minute to avoid 429 Too Many Requests
+    if path.is_file() and time.time() - os.path.getmtime(path) <= 60:
+        return path.read_text()
+
+    response = await client.post(
+        url=Endpoint.ROTATE_COOKIES,
+        headers=Headers.ROTATE_COOKIES.value,
+        data='[000,"-0000000000000000000"]',
+    )
+    if verbose:
+        logger.debug(
+            f"HTTP Request: POST {Endpoint.ROTATE_COOKIES} [{response.status_code}]"
+        )
+    if response.status_code == 401:
+        raise AuthError
+    response.raise_for_status()
+
+    new_1psidts = _extract_cookie_value(client.cookies, "__Secure-1PSIDTS")
+
+    if new_1psidts:
+        path.write_text(new_1psidts)
+        logger.debug(
+            f"Rotated __Secure-1PSIDTS successfully (length={len(new_1psidts)})."
+        )
+        return new_1psidts
+
+    cookie_names = [c.name for c in client.cookies.jar]
+    logger.debug(f"Rotation response cookies: {cookie_names}")
+    return None

gemini_webapi/utils/temp/.cached_1psidts_g.a0007Qjc5GP_JJ8G6lqxKsBvwooBDG0kaQQpdrq1eVMavCuae6YHM71QR0oHtpOONkPxs87_PQACgYKAZcSARISFQHGX2MiBLscUC-RI65KuaeNsGHqgxoVAUF8yKrfh50pYTc-6ectdvp0W-we0076.txt

@@ -0,0 +1 @@
+sidts-CjEBBj1CYnSuFNTjwnWYGX5D3D4GPsxqjY4Kk2cO6xqlsGk7UJNFrK2f4SIr3Gq_jJ6rEAA

gemini_webapi/utils/upload_file.py

@@ -0,0 +1,121 @@
+import io
+import mimetypes
+import random
+from pathlib import Path
+
+import curl_cffi
+from curl_cffi.requests import AsyncSession
+from pydantic import ConfigDict, validate_call
+
+from ..constants import Endpoint, Headers
+from .logger import logger
+
+
+def _generate_random_name(extension: str = ".txt") -> str:
+    """
+    Generate a random filename using a large integer for better performance.
+    """
+
+    return f"input_{random.randint(1000000, 9999999)}{extension}"
+
+
+@validate_call(config=ConfigDict(arbitrary_types_allowed=True))
+async def upload_file(
+    file: str | Path | bytes | io.BytesIO,
+    client: AsyncSession,
+    filename: str | None = None,
+    verbose: bool = False,
+) -> str:
+    """
+    Upload a file to Google's server and return its identifier.
+
+    Parameters
+    ----------
+    file : `str` | `Path` | `bytes` | `io.BytesIO`
+        Path to the file or file content to be uploaded.
+    client: `curl_cffi.requests.AsyncSession`
+        Shared async session to use for upload.
+    filename: `str`, optional
+        Name of the file to be uploaded. Required if file is bytes or BytesIO.
+    verbose: `bool`, optional
+        If `True`, will print more infomation in logs.
+
+    Returns
+    -------
+    `str`
+        Identifier of the uploaded file.
+        E.g. "/contrib_service/ttl_1d/1709764705i7wdlyx3mdzndme3a767pluckv4flj"
+
+    Raises
+    ------
+    `curl_cffi.requests.exceptions.HTTPError`
+        If the upload request failed.
+    """
+
+    if isinstance(file, (str, Path)):
+        file_path = Path(file)
+        if not file_path.is_file():
+            raise ValueError(f"{file_path} is not a valid file.")
+        if not filename:
+            filename = file_path.name
+        file_content = file_path.read_bytes()
+    elif isinstance(file, io.BytesIO):
+        file_content = file.getvalue()
+        if not filename:
+            filename = _generate_random_name()
+    elif isinstance(file, bytes):
+        file_content = file
+        if not filename:
+            filename = _generate_random_name()
+    else:
+        raise ValueError(f"Unsupported file type: {type(file)}")
+
+    content_type = mimetypes.guess_type(filename)[0] or "application/octet-stream"
+
+    mp = curl_cffi.CurlMime()
+    mp.addpart(
+        name="file",
+        content_type=content_type,
+        filename=filename,
+        data=file_content,
+    )
+
+    try:
+        response = await client.post(
+            url=Endpoint.UPLOAD,
+            headers=Headers.UPLOAD.value,
+            multipart=mp,
+            allow_redirects=True,
+        )
+        if verbose:
+            logger.debug(
+                f"HTTP Request: POST {Endpoint.UPLOAD} [{response.status_code}]"
+            )
+        response.raise_for_status()
+        return response.text
+    finally:
+        mp.close()
+
+
+def parse_file_name(file: str | Path | bytes | io.BytesIO) -> str:
+    """
+    Parse the file name from the given path or generate a random one for in-memory data.
+
+    Parameters
+    ----------
+    file : `str` | `Path` | `bytes` | `io.BytesIO`
+        Path to the file or file content.
+
+    Returns
+    -------
+    `str`
+        File name with extension.
+    """
+
+    if isinstance(file, (str, Path)):
+        file = Path(file)
+        if not file.is_file():
+            raise ValueError(f"{file} is not a valid file.")
+        return file.name
+
+    return _generate_random_name()