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