"""Gemini provider — httpx (async) implementation of the P2 provider Protocol. Port of v1's ``gemini_chat_complete`` transport (main.py ≈3618-3739) and ``_chat_summarize_title`` (≈7990-8020), restructured per the plan: - streaming via ``streamGenerateContent?alt=sse`` (v1 was non-streaming); - the function-calling LOOP is NOT here — this module does ONE model call per ``chat_stream`` and reports functionCall parts as ``tool_call`` events; services/ai/chat.py owns dispatch + iteration; - tool declarations come in as provider-neutral ToolSpec and map 1:1 onto Gemini ``function_declarations`` (same dict keys, v1 parity). Config: ``GEMINI_API_KEY`` from app settings (.env, shape-verified only); ``GEMINI_CHAT_MODEL`` from the environment (default gemini-2.5-flash), matching the v1 env-override contract (same pattern as transcribe.py). Security deviation from v1: the API key travels in the ``x-goog-api-key`` header instead of the URL query string, so it can never leak through an error message or log line that includes the URL. """ from __future__ import annotations import json import os from typing import AsyncIterator import httpx from ....config import get_settings from ....errors import AppError from ..transcribe import AI_BAD_RESPONSE, AI_NOT_CONFIGURED, AI_UPSTREAM_ERROR from .base import Message, Part, StreamEvent, ToolSpec DEFAULT_CHAT_MODEL = "gemini-2.5-flash" _BASE_URL = "https://generativelanguage.googleapis.com/v1beta/models" # v1 gemini_chat_complete used temperature 0.4 for chat turns. _CHAT_TEMPERATURE = 0.4 _TIMEOUT = httpx.Timeout(120.0, connect=15.0) def chat_model() -> str: return os.environ.get("GEMINI_CHAT_MODEL", DEFAULT_CHAT_MODEL) # Provider-neutral roles -> Gemini wire roles (v1 used model/function). _ROLE_MAP = {"user": "user", "assistant": "model", "tool": "function"} def _gemini_part(part: Part) -> dict: if part.text is not None: return {"text": part.text} if part.inline_data is not None: return { "inline_data": { "mime_type": part.inline_data.mime_type, "data": part.inline_data.data, } } if part.function_call is not None: return { "functionCall": { "name": part.function_call.name, "args": part.function_call.args or {}, } } if part.function_response is not None: return { "functionResponse": { "name": part.function_response.name, "response": part.function_response.response or {}, } } return {"text": ""} def _gemini_contents(messages: list[Message]) -> list[dict]: return [ {"role": _ROLE_MAP[m.role], "parts": [_gemini_part(p) for p in m.parts]} for m in messages if m.parts ] class GeminiProvider: """Implements the AIProvider Protocol (providers/base.py).""" async def chat_stream( self, *, model: str, system: str, messages: list[Message], tools: list[ToolSpec], ) -> AsyncIterator[StreamEvent]: settings = get_settings() if not settings.gemini_api_key: yield StreamEvent( type="error", data={ "code": AI_NOT_CONFIGURED, "message": "GEMINI_API_KEY is not configured", }, ) return payload: dict = { "contents": _gemini_contents(messages), "systemInstruction": {"parts": [{"text": system}]}, "generationConfig": {"temperature": _CHAT_TEMPERATURE}, } if tools: payload["tools"] = [ { "function_declarations": [ { "name": t.name, "description": t.description, "parameters": t.parameters, } for t in tools ] } ] url = f"{_BASE_URL}/{model}:streamGenerateContent?alt=sse" text_chunks: list[str] = [] tool_calls: list[dict] = [] try: async with httpx.AsyncClient(timeout=_TIMEOUT) as client: async with client.stream( "POST", url, json=payload, headers={"x-goog-api-key": settings.gemini_api_key}, ) as resp: if resp.status_code != 200: body = (await resp.aread()).decode("utf-8", "replace") yield StreamEvent( type="error", data={ "code": AI_UPSTREAM_ERROR, "message": ( f"Gemini API error {resp.status_code}: " f"{body[:500]}" ), }, ) return async for line in resp.aiter_lines(): if not line.startswith("data:"): continue raw = line[len("data:"):].strip() if not raw or raw == "[DONE]": continue try: chunk = json.loads(raw) except ValueError: continue # keepalive / partial frame candidates = chunk.get("candidates") or [] if not candidates: continue parts = (candidates[0].get("content") or {}).get( "parts" ) or [] for p in parts: if p.get("text"): text_chunks.append(p["text"]) yield StreamEvent( type="text_delta", data={"text": p["text"]} ) if "functionCall" in p: fc = { "name": p["functionCall"].get("name", ""), "args": p["functionCall"].get("args", {}) or {}, } tool_calls.append(fc) yield StreamEvent(type="tool_call", data=fc) except httpx.HTTPError as e: yield StreamEvent( type="error", data={ "code": AI_UPSTREAM_ERROR, "message": f"Gemini API unreachable: {e}", }, ) return yield StreamEvent( type="message_done", data={"text": "".join(text_chunks), "tool_calls": tool_calls}, ) async def complete( self, *, model: str, prompt: str, temperature: float = 0.0 ) -> str: """Non-streaming one-shot text call (session titling).""" settings = get_settings() if not settings.gemini_api_key: raise AppError(AI_NOT_CONFIGURED, "GEMINI_API_KEY is not configured", 503) payload = { "contents": [{"parts": [{"text": prompt}]}], "generationConfig": { "temperature": temperature, "responseMimeType": "text/plain", }, } url = f"{_BASE_URL}/{model}:generateContent" try: async with httpx.AsyncClient(timeout=_TIMEOUT) as client: resp = await client.post( url, json=payload, headers={"x-goog-api-key": settings.gemini_api_key}, ) except httpx.HTTPError as e: raise AppError(AI_UPSTREAM_ERROR, f"Gemini API unreachable: {e}", 502) if resp.status_code != 200: raise AppError( AI_UPSTREAM_ERROR, f"Gemini API error {resp.status_code}: {resp.text[:500]}", 502, ) try: data = resp.json() return data["candidates"][0]["content"]["parts"][0]["text"].strip() except (KeyError, IndexError, TypeError, ValueError): raise AppError( AI_BAD_RESPONSE, f"Unexpected Gemini response shape: {resp.text[:300]}", 502, )