Upload folder using huggingface_hub

gemini_web2api/__init__.py

@@ -0,0 +1,2 @@
+"""gemini-web2api: Gemini Web to OpenAI API proxy."""
+__version__ = "1.1.0"

gemini_web2api/__main__.py

@@ -0,0 +1,50 @@
+"""Entry point: python -m gemini_web2api"""
+import argparse
+import os
+
+from .config import CONFIG, load_config, find_config
+from .models import MODELS
+from .gemini import HAS_HTTPX
+from .server import GeminiHandler, ThreadedServer
+from . import __version__
+
+
+def main():
+    parser = argparse.ArgumentParser(description="Gemini Web to OpenAI API")
+    parser.add_argument("--port", type=int, default=None)
+    parser.add_argument("--config", type=str, default=None)
+    parser.add_argument("--cookie-file", type=str, default=None)
+    parser.add_argument("--proxy", type=str, default=None, help="HTTP proxy, e.g. http://127.0.0.1:7890")
+    parser.add_argument("--version", action="version", version=f"gemini-web2api {__version__}")
+    args = parser.parse_args()
+
+    config_path = args.config or os.environ.get("GEMINI_WEB2API_CONFIG") or find_config()
+    if config_path:
+        load_config(config_path)
+
+    if args.port:
+        CONFIG["port"] = args.port
+    if args.cookie_file:
+        CONFIG["cookie_file"] = args.cookie_file
+    if args.proxy:
+        CONFIG["proxy"] = args.proxy
+
+    port = CONFIG["port"]
+    server = ThreadedServer((CONFIG["host"], port), GeminiHandler)
+    print(f"gemini-web2api v{__version__}")
+    print(f"  Listening: http://0.0.0.0:{port}")
+    print(f"  Base URL:  http://localhost:{port}/v1")
+    print(f"  Models:    {', '.join(MODELS.keys())}")
+    print(f"  Cookie:    {'yes' if CONFIG.get('cookie_file') else 'none (anonymous)'}")
+    print(f"  Proxy:     {CONFIG.get('proxy') or 'system env'}")
+    print(f"  Streaming: {'httpx (true streaming)' if HAS_HTTPX else 'urllib (buffered)'}")
+    print()
+    try:
+        server.serve_forever()
+    except KeyboardInterrupt:
+        print("\nStopped.")
+        server.shutdown()
+
+
+if __name__ == "__main__":
+    main()

gemini_web2api/config.py

@@ -0,0 +1,37 @@
+"""Configuration management."""
+import json
+import os
+
+DEFAULT_CONFIG = {
+    "port": 8081,
+    "host": "0.0.0.0",
+    "retry_attempts": 3,
+    "retry_delay_sec": 2,
+    "request_timeout_sec": 180,
+    "gemini_bl": "boq_assistant-bard-web-server_20260525.09_p0",
+    "auth_user": None,
+    "xsrf_token": None,
+    "default_model": "gemini-3.5-flash",
+    "log_requests": True,
+    "cookie_file": None,
+    "proxy": None,
+    "api_keys": [],
+}
+
+CONFIG = dict(DEFAULT_CONFIG)
+
+
+def load_config(path: str = None):
+    """Load config from JSON file."""
+    if path and os.path.exists(path):
+        with open(path) as f:
+            CONFIG.update(json.load(f))
+    return CONFIG
+
+
+def find_config():
+    """Search for config file in standard locations."""
+    for p in ["./config.json", os.path.expanduser("~/.config/gemini-web2api/config.json")]:
+        if os.path.exists(p):
+            return p
+    return None

gemini_web2api/gemini.py

@@ -0,0 +1,257 @@
+"""Gemini StreamGenerate protocol implementation with httpx streaming."""
+import json
+import time
+import uuid
+import re
+import urllib.request
+import urllib.parse
+import ssl
+import os
+import hashlib
+
+try:
+    import httpx
+    HAS_HTTPX = True
+except ImportError:
+    HAS_HTTPX = False
+
+from .config import CONFIG
+
+_ssl_ctx = None
+_cookie_cache = {"str": "", "sapisid": None, "mtime": 0}
+_httpx_client = None
+
+
+def log(msg: str):
+    if CONFIG["log_requests"]:
+        import sys
+        sys.stderr.write(f"[{time.strftime('%H:%M:%S')}] {msg}\n")
+        sys.stderr.flush()
+
+
+def _get_ssl_ctx():
+    global _ssl_ctx
+    if _ssl_ctx is None:
+        _ssl_ctx = ssl.create_default_context()
+    return _ssl_ctx
+
+
+def _get_httpx_client():
+    global _httpx_client
+    if _httpx_client is None and HAS_HTTPX:
+        proxy = CONFIG.get("proxy")
+        transport = httpx.HTTPTransport(proxy=proxy) if proxy else None
+        _httpx_client = httpx.Client(transport=transport, timeout=CONFIG["request_timeout_sec"], verify=True)
+    return _httpx_client
+
+
+def load_cookie() -> tuple:
+    """Load cookie from file with mtime-based caching."""
+    cookie_file = CONFIG.get("cookie_file")
+    if not cookie_file or not os.path.exists(cookie_file):
+        return "", None
+    try:
+        mtime = os.path.getmtime(cookie_file)
+        if mtime == _cookie_cache["mtime"] and _cookie_cache["str"]:
+            return _cookie_cache["str"], _cookie_cache["sapisid"]
+        with open(cookie_file, "r") as f:
+            content = f.read().strip()
+        if content.startswith("{"):
+            data = json.loads(content)
+            cookie_str = data.get("cookie", "")
+            sapisid = data.get("sapisid", "")
+        else:
+            cookie_str = content
+            pairs = dict(p.split("=", 1) for p in cookie_str.split("; ") if "=" in p)
+            sapisid = pairs.get("SAPISID", "")
+        _cookie_cache.update({"str": cookie_str, "sapisid": sapisid or None, "mtime": mtime})
+        return cookie_str, sapisid if sapisid else None
+    except Exception as e:
+        log(f"Cookie load error: {e}")
+        return _cookie_cache["str"], _cookie_cache["sapisid"]
+
+
+def make_sapisidhash(sapisid: str) -> str:
+    ts = int(time.time())
+    h = hashlib.sha1(f"{ts} {sapisid} https://gemini.google.com".encode()).hexdigest()
+    return f"SAPISIDHASH {ts}_{h}"
+
+
+def _account_prefix() -> str:
+    """Return the Gemini account path prefix for non-default Google accounts."""
+    auth_user = CONFIG.get("auth_user")
+    if auth_user is None or auth_user == "":
+        return ""
+    return f"/u/{auth_user}"
+
+
+def _build_headers() -> dict:
+    account_prefix = _account_prefix()
+    headers = {
+        "Content-Type": "application/x-www-form-urlencoded",
+        "Origin": "https://gemini.google.com",
+        "Referer": f"https://gemini.google.com{account_prefix}/app",
+        "X-Same-Domain": "1",
+        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
+    }
+    if account_prefix:
+        headers["X-Goog-AuthUser"] = str(CONFIG["auth_user"])
+    cookie_str, sapisid = load_cookie()
+    if cookie_str:
+        headers["Cookie"] = cookie_str
+    if sapisid:
+        headers["Authorization"] = make_sapisidhash(sapisid)
+    return headers
+
+
+def _build_payload(prompt: str, model_id: int, think_mode: int, file_refs: list = None, extra_fields: dict = None) -> str:
+    inner = [None] * 102
+    if file_refs:
+        refs = [[None, None, ref] for ref in file_refs]
+        inner[0] = [prompt, 0, None, refs, None, None, 0]
+    else:
+        inner[0] = [prompt, 0, None, None, None, None, 0]
+    inner[1] = ["en"]
+    inner[2] = ["", "", "", None, None, None, None, None, None, ""]
+    inner[6] = [0]
+    inner[7] = 1
+    inner[10] = 1
+    inner[11] = 0
+    inner[17] = [[think_mode]]
+    inner[18] = 0
+    inner[27] = 1
+    inner[30] = [4]
+    inner[41] = [2]
+    inner[53] = 0
+    inner[59] = str(uuid.uuid4())
+    inner[61] = []
+    inner[68] = 1
+    inner[79] = model_id
+    if extra_fields:
+        for k, v in extra_fields.items():
+            inner[k] = v
+    outer = [None, json.dumps(inner)]
+    params = {"f.req": json.dumps(outer)}
+    if CONFIG.get("xsrf_token"):
+        params["at"] = CONFIG["xsrf_token"]
+    return urllib.parse.urlencode(params)
+
+
+def _get_url() -> str:
+    reqid = int(time.time()) % 1000000
+    account_prefix = _account_prefix()
+    return (
+        f"https://gemini.google.com{account_prefix}/_/BardChatUi/data/"
+        "assistant.lamda.BardFrontendService/StreamGenerate"
+        f"?bl={CONFIG['gemini_bl']}&hl=en&_reqid={reqid}&rt=c"
+    )
+
+
+def clean_text(text: str) -> str:
+    text = re.sub(
+        r'```(?:python|javascript|text)\?code_(?:reference|stdout)&code_event_index=\d+\n.*?```\n?',
+        '', text, flags=re.DOTALL
+    )
+    text = re.sub(r'http://googleusercontent\.com/card_content/\d+\n?', '', text)
+    return text.strip()
+
+
+def _extract_texts_from_line(line: str) -> list:
+    """Parse a single wrb.fr line and return list of text strings found."""
+    if '"wrb.fr"' not in line or len(line) < 200:
+        return []
+    try:
+        arr = json.loads(line)
+        inner_str = arr[0][2]
+        if not inner_str or len(inner_str) < 50:
+            return []
+        inner = json.loads(inner_str)
+        if not (isinstance(inner, list) and len(inner) > 4 and inner[4]):
+            return []
+        texts = []
+        for part in inner[4]:
+            if isinstance(part, list) and len(part) > 1 and part[1] and isinstance(part[1], list):
+                for t in part[1]:
+                    if isinstance(t, str) and t:
+                        texts.append(t)
+        return texts
+    except (json.JSONDecodeError, IndexError, TypeError):
+        return []
+
+
+def extract_response_text(raw: str) -> str:
+    """Parse full response to get final text."""
+    last_text = ""
+    for line in raw.split("\n"):
+        for t in _extract_texts_from_line(line):
+            if len(t) > len(last_text):
+                last_text = t
+    return clean_text(last_text)
+
+
+def generate(prompt: str, model_id: int, think_mode: int, file_refs: list = None, extra_fields: dict = None) -> str:
+    """Non-streaming generation with retry."""
+    body = _build_payload(prompt, model_id, think_mode, file_refs, extra_fields).encode()
+    url = _get_url()
+    headers = _build_headers()
+    ctx = _get_ssl_ctx()
+    proxy = CONFIG.get("proxy")
+
+    last_err = None
+    for attempt in range(CONFIG["retry_attempts"]):
+        try:
+            req = urllib.request.Request(url, data=body, headers=headers, method="POST")
+            if proxy:
+                opener = urllib.request.build_opener(
+                    urllib.request.ProxyHandler({"http": proxy, "https": proxy}),
+                    urllib.request.HTTPSHandler(context=ctx)
+                )
+                resp = opener.open(req, timeout=CONFIG["request_timeout_sec"])
+            else:
+                resp = urllib.request.urlopen(req, context=ctx, timeout=CONFIG["request_timeout_sec"])
+            raw = resp.read().decode("utf-8", errors="replace")
+            return extract_response_text(raw)
+        except Exception as e:
+            last_err = e
+            if attempt < CONFIG["retry_attempts"] - 1:
+                log(f"Retry {attempt+1}/{CONFIG['retry_attempts']}: {e}")
+                time.sleep(CONFIG["retry_delay_sec"])
+    raise last_err
+
+
+def generate_stream(prompt: str, model_id: int, think_mode: int, file_refs: list = None, extra_fields: dict = None):
+    """Streaming generation via httpx with retry on connection failure."""
+    if not HAS_HTTPX:
+        text = generate(prompt, model_id, think_mode, file_refs, extra_fields)
+        if text:
+            yield text
+        return
+
+    body = _build_payload(prompt, model_id, think_mode, file_refs, extra_fields)
+    url = _get_url()
+    headers = _build_headers()
+    client = _get_httpx_client()
+
+    last_err = None
+    for attempt in range(CONFIG["retry_attempts"]):
+        try:
+            prev_text = ""
+            with client.stream("POST", url, content=body, headers=headers) as resp:
+                buf = ""
+                for chunk in resp.iter_text():
+                    buf += chunk
+                    while "\n" in buf:
+                        line, buf = buf.split("\n", 1)
+                        for t in _extract_texts_from_line(line):
+                            if len(t) > len(prev_text):
+                                delta = clean_text(t[len(prev_text):])
+                                if delta:
+                                    yield delta
+                                prev_text = t
+            return
+        except Exception as e:
+            last_err = e
+            if attempt < CONFIG["retry_attempts"] - 1:
+                log(f"Stream retry {attempt+1}/{CONFIG['retry_attempts']}: {e}")
+                time.sleep(CONFIG["retry_delay_sec"])
+    raise last_err

gemini_web2api/models.py

@@ -0,0 +1,60 @@
+"""Model definitions and mapping from Gemini frontend JS source."""
+
+# MODE_CATEGORY enum from 028-6eb337387583.js:
+#   1=FAST, 2=THINKING, 3=PRO, 4=AUTO, 5=FAST_DYNAMIC_THINKING, 6=FLASH_LITE
+
+MODELS = {
+    "gemini-3.5-flash": {
+        "mode": 1, "think": 4,
+        "desc": "Fast general-purpose model",
+    },
+    "gemini-3.5-flash-thinking": {
+        "mode": 2, "think": 0,
+        "desc": "Deep thinking mode, longest output (~20k chars)",
+    },
+    "gemini-3.1-pro": {
+        "mode": 3, "think": 4,
+        "desc": "Pro model (requires cookie for real routing)",
+    },
+    "gemini-3.1-pro-enhanced": {
+        "mode": 3, "think": 4, "extra": {31: 2, 80: 3},
+        "desc": "Pro with enhanced output (experimental)",
+    },
+    "gemini-auto": {
+        "mode": 4, "think": 4,
+        "desc": "Auto model selection",
+    },
+    "gemini-3.5-flash-thinking-lite": {
+        "mode": 5, "think": 0,
+        "desc": "Dynamic thinking with adaptive depth",
+    },
+    "gemini-flash-lite": {
+        "mode": 6, "think": 4,
+        "desc": "Lightweight fast model",
+    },
+}
+
+
+def resolve_model(model_name: str, default: str = "gemini-3.5-flash"):
+    """Resolve model name to (name, mode_id, think_mode, error, extra_fields).
+
+    Unknown model names fall back to default rather than erroring,
+    since upstream clients may request arbitrary model identifiers.
+    """
+    think_override = None
+    if "@think=" in model_name:
+        model_name, think_str = model_name.rsplit("@think=", 1)
+        try:
+            think_override = int(think_str)
+        except ValueError:
+            return None, None, None, f"Invalid think level: {think_str}", None
+    cfg = MODELS.get(model_name)
+    if not cfg:
+        from .gemini import log
+        log(f"Unknown model '{model_name}', falling back to '{default}'")
+        model_name = default
+        cfg = MODELS[default]
+    mode_id = cfg["mode"]
+    think_mode = think_override if think_override is not None else cfg["think"]
+    extra = cfg.get("extra")
+    return model_name, mode_id, think_mode, None, extra

gemini_web2api/multimodal.py

@@ -0,0 +1,127 @@
+"""Multimodal: Scotty resumable upload for Gemini image input."""
+import json
+import base64
+import urllib.request
+import urllib.parse
+import time
+import ssl
+import re
+
+from .config import CONFIG
+from .gemini import load_cookie, make_sapisidhash, _get_ssl_ctx, log
+
+
+def _get_page_tokens() -> dict:
+    """Fetch WIZ_global_data tokens from Gemini page (Push-ID, X-Client-Pctx)."""
+    headers = {
+        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
+    }
+    cookie_str, sapisid = load_cookie()
+    if cookie_str:
+        headers["Cookie"] = cookie_str
+    try:
+        req = urllib.request.Request("https://gemini.google.com/app", headers=headers)
+        resp = urllib.request.urlopen(req, context=_get_ssl_ctx(), timeout=30)
+        html = resp.read().decode()
+        tokens = {}
+        for key, pattern in [
+            ("push_id", r'"qKIAYe":"([^"]+)"'),
+            ("pctx", r'"Ylro7b":"([^"]+)"'),
+            ("at", r'"thykhd":"([^"]+)"'),
+        ]:
+            m = re.search(pattern, html)
+            if m:
+                tokens[key] = m.group(1)
+        return tokens
+    except Exception as e:
+        log(f"Page token fetch failed: {e}")
+        return {}
+
+
+_page_tokens_cache = {"tokens": {}, "ts": 0}
+
+
+def _cached_page_tokens() -> dict:
+    now = time.time()
+    if now - _page_tokens_cache["ts"] > 600:
+        _page_tokens_cache["tokens"] = _get_page_tokens()
+        _page_tokens_cache["ts"] = now
+    return _page_tokens_cache["tokens"]
+
+
+def upload_image(image_bytes: bytes, filename: str = "image.png", mime_type: str = "image/png") -> str:
+    """Upload image via Scotty resumable upload. Returns file reference path."""
+    tokens = _cached_page_tokens()
+    push_id = tokens.get("push_id", "feeds/mcudyrk2a4khkz")
+    pctx = tokens.get("pctx", "CgcSBWjK7pYx")
+
+    cookie_str, sapisid = load_cookie()
+    ctx = _get_ssl_ctx()
+    proxy = CONFIG.get("proxy")
+
+    # Step 1: Initiate resumable upload
+    start_headers = {
+        "Push-ID": push_id,
+        "X-Tenant-Id": "bard-storage",
+        "X-Client-Pctx": pctx,
+        "X-Goog-Upload-Header-Content-Length": str(len(image_bytes)),
+        "X-Goog-Upload-Header-Content-Type": mime_type,
+        "X-Goog-Upload-Protocol": "resumable",
+        "X-Goog-Upload-Command": "start",
+        "Content-Type": "application/x-www-form-urlencoded;charset=utf-8",
+        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
+    }
+    if cookie_str:
+        start_headers["Cookie"] = cookie_str
+    if sapisid:
+        start_headers["Authorization"] = make_sapisidhash(sapisid)
+
+    start_url = "https://content-push.googleapis.com/upload/"
+    req = urllib.request.Request(start_url, data=b"", headers=start_headers, method="POST")
+
+    if proxy:
+        opener = urllib.request.build_opener(
+            urllib.request.ProxyHandler({"http": proxy, "https": proxy}),
+            urllib.request.HTTPSHandler(context=ctx)
+        )
+        resp = opener.open(req, timeout=30)
+    else:
+        resp = urllib.request.urlopen(req, context=ctx, timeout=30)
+
+    upload_url = resp.headers.get("X-Goog-Upload-URL") or resp.headers.get("x-goog-upload-url")
+    if not upload_url:
+        raise RuntimeError(f"No upload URL in response headers: {dict(resp.headers)}")
+
+    log(f"Upload session started: {upload_url[:80]}...")
+
+    # Step 2: Upload file data + finalize
+    upload_headers = {
+        "X-Goog-Upload-Command": "upload, finalize",
+        "X-Goog-Upload-Offset": "0",
+        "Content-Type": "application/octet-stream",
+        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
+    }
+
+    req2 = urllib.request.Request(upload_url, data=image_bytes, headers=upload_headers, method="POST")
+    if proxy:
+        resp2 = opener.open(req2, timeout=60)
+    else:
+        resp2 = urllib.request.urlopen(req2, context=ctx, timeout=60)
+
+    file_ref = resp2.read().decode().strip()
+    if not file_ref or not file_ref.startswith("/"):
+        raise RuntimeError(f"Invalid file reference: {file_ref[:100]}")
+
+    log(f"Image uploaded: {filename} -> {file_ref[:50]}...")
+    return file_ref
+
+
+def fetch_image_bytes(url: str) -> bytes:
+    """Fetch image from URL."""
+    try:
+        req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"})
+        resp = urllib.request.urlopen(req, timeout=30)
+        return resp.read()
+    except Exception as e:
+        log(f"Image fetch failed: {e}")
+        return b""

gemini_web2api/server.py

@@ -0,0 +1,413 @@
+"""HTTP server: OpenAI-compatible API endpoints."""
+import json
+import time
+import uuid
+import re
+from http.server import HTTPServer, BaseHTTPRequestHandler
+from socketserver import ThreadingMixIn
+
+from .config import CONFIG
+from .models import MODELS, resolve_model
+from .gemini import generate, generate_stream, log
+from .tools import messages_to_prompt, parse_tool_calls, google_contents_to_prompt, parse_google_function_calls
+from .multimodal import upload_image, fetch_image_bytes
+from . import __version__
+
+
+def _usage(prompt: str, text: str) -> dict:
+    p = len(prompt) // 4
+    c = len(text or "") // 4
+    return {"prompt_tokens": p, "completion_tokens": c, "total_tokens": p + c}
+
+
+def _upload_images(images: list) -> list:
+    """Upload images and return list of file references. Returns None if no images."""
+    if not images:
+        return None
+    file_refs = []
+    for item in images:
+        try:
+            if isinstance(item, tuple) and len(item) == 2:
+                data, mime = item
+                if isinstance(data, str):
+                    data = fetch_image_bytes(data)
+                    mime = mime or "image/png"
+                if data:
+                    ref = upload_image(data, "image.png", mime or "image/png")
+                    file_refs.append(ref)
+        except Exception as e:
+            log(f"Image upload failed: {e}")
+    return file_refs if file_refs else None
+
+
+class GeminiHandler(BaseHTTPRequestHandler):
+    def log_message(self, fmt, *args):
+        log(fmt % args)
+
+    def send_json(self, data, status=200):
+        body = json.dumps(data, ensure_ascii=False).encode()
+        self.send_response(status)
+        self.send_header("Content-Type", "application/json")
+        self.send_header("Access-Control-Allow-Origin", "*")
+        self.send_header("Content-Length", str(len(body)))
+        self.end_headers()
+        self.wfile.write(body)
+
+    def _start_sse(self):
+        self.send_response(200)
+        self.send_header("Content-Type", "text/event-stream")
+        self.send_header("Cache-Control", "no-cache")
+        self.send_header("Access-Control-Allow-Origin", "*")
+        self.end_headers()
+
+    def _parse_body(self, body: bytes) -> dict:
+        try:
+            return json.loads(body)
+        except (json.JSONDecodeError, ValueError):
+            return None
+
+    def _authorized(self):
+        keys = CONFIG.get("api_keys") or []
+        if not keys:
+            return True
+        auth = self.headers.get("Authorization", "")
+        key = auth[7:] if auth.startswith("Bearer ") else self.headers.get("x-api-key", "")
+        return key in keys
+
+    def do_OPTIONS(self):
+        self.send_response(204)
+        self.send_header("Access-Control-Allow-Origin", "*")
+        self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS")
+        self.send_header("Access-Control-Allow-Headers", "*")
+        self.end_headers()
+
+    def do_GET(self):
+        try:
+            if self.path.startswith("/v1/") and not self._authorized():
+                self.send_json({"error": {"message": "invalid api key"}}, 401)
+                return
+            if self.path == "/v1/models":
+                self.send_json({"object": "list", "data": [
+                    {"id": n, "object": "model", "created": 1700000000,
+                     "owned_by": "google", "description": c["desc"]}
+                    for n, c in MODELS.items()
+                ]})
+            elif self.path.startswith("/v1beta/models"):
+                self.send_json({"models": [
+                    {"name": f"models/{n}", "displayName": n, "description": c["desc"],
+                     "supportedGenerationMethods": ["generateContent", "streamGenerateContent"]}
+                    for n, c in MODELS.items()
+                ]})
+            elif self.path == "/":
+                self.send_json({"status": "ok", "version": __version__, "models": list(MODELS.keys())})
+            else:
+                self.send_json({"error": "not found"}, 404)
+        except (BrokenPipeError, ConnectionResetError):
+            pass
+
+    def do_POST(self):
+        try:
+            if self.path.startswith("/v1/") and not self._authorized():
+                self.send_json({"error": {"message": "invalid api key"}}, 401)
+                return
+            length = int(self.headers.get("Content-Length", 0))
+            body = self.rfile.read(length) if length else b""
+            if self.path == "/v1/chat/completions":
+                self._handle_chat(body)
+            elif self.path == "/v1/responses":
+                self._handle_responses(body)
+            elif ":generateContent" in self.path:
+                self._handle_google_generate(body, stream=False)
+            elif ":streamGenerateContent" in self.path:
+                self._handle_google_generate(body, stream=True)
+            else:
+                self.send_json({"error": "not found"}, 404)
+        except (BrokenPipeError, ConnectionResetError):
+            pass
+        except Exception as e:
+            log(f"POST error: {e}")
+            try:
+                self.send_json({"error": {"message": str(e)}}, 500)
+            except:
+                pass
+
+    # ─── /v1/chat/completions ─────────────────────────────────────────────────
+
+    def _handle_chat(self, body: bytes):
+        req = self._parse_body(body)
+        if req is None:
+            self.send_json({"error": {"message": "invalid JSON"}}, 400)
+            return
+        model_name, model_id, think_mode, err, extra_fields = resolve_model(
+            req.get("model", CONFIG["default_model"]))
+        if err:
+            self.send_json({"error": {"message": err}}, 400)
+            return
+
+        tools = req.get("tools")
+        tool_choice = req.get("tool_choice", "auto")
+        prompt, images = messages_to_prompt(req.get("messages", []), tools, tool_choice)
+        if not prompt.strip():
+            self.send_json({"error": {"message": "empty prompt"}}, 400)
+            return
+
+        stream = req.get("stream", False)
+        cid = f"chatcmpl-{uuid.uuid4().hex[:12]}"
+
+        if stream and (not tools or tool_choice == "none"):
+            try:
+                self._start_sse()
+                for delta in generate_stream(prompt, model_id, think_mode, _upload_images(images), extra_fields):
+                    chunk = {"id": cid, "object": "chat.completion.chunk", "created": int(time.time()),
+                             "model": model_name, "choices": [{"index": 0, "delta": {"content": delta}, "finish_reason": None}]}
+                    self.wfile.write(f"data: {json.dumps(chunk, ensure_ascii=False)}\n\n".encode())
+                    self.wfile.flush()
+                end = {"id": cid, "object": "chat.completion.chunk", "created": int(time.time()),
+                       "model": model_name, "choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}]}
+                self.wfile.write(f"data: {json.dumps(end)}\n\n".encode())
+                self.wfile.write(b"data: [DONE]\n\n")
+                self.wfile.flush()
+            except (BrokenPipeError, ConnectionResetError):
+                pass
+            return
+
+        try:
+            text = generate(prompt, model_id, think_mode, _upload_images(images), extra_fields)
+        except Exception as e:
+            self.send_json({"error": {"message": f"upstream error: {e}"}}, 502)
+            return
+
+        tool_calls = None
+        if tools and text and tool_choice != "none":
+            text, tool_calls = parse_tool_calls(text)
+        msg = {"role": "assistant", "content": text or None}
+        if tool_calls:
+            msg["tool_calls"] = tool_calls
+        finish = "tool_calls" if tool_calls else "stop"
+
+        if stream:
+            self._start_sse()
+            chunk = {"id": cid, "object": "chat.completion.chunk", "created": int(time.time()),
+                     "model": model_name, "choices": [{"index": 0, "delta": msg, "finish_reason": finish}]}
+            self.wfile.write(f"data: {json.dumps(chunk, ensure_ascii=False)}\n\n".encode())
+            self.wfile.write(b"data: [DONE]\n\n")
+            self.wfile.flush()
+        else:
+            self.send_json({
+                "id": cid, "object": "chat.completion", "created": int(time.time()),
+                "model": model_name,
+                "choices": [{"index": 0, "message": msg, "finish_reason": finish}],
+                "usage": {"prompt_tokens": len(prompt)//4, "completion_tokens": len(text or "")//4,
+                          "total_tokens": (len(prompt)+len(text or ""))//4},
+            })
+
+    # ─── /v1/responses (Codex CLI) ───────────────────────────────────────────
+
+    def _handle_responses(self, body: bytes):
+        req = self._parse_body(body)
+        if req is None:
+            self.send_json({"error": {"message": "invalid JSON"}}, 400)
+            return
+        model_name, model_id, think_mode, err, extra_fields = resolve_model(
+            req.get("model", CONFIG["default_model"]))
+        if err:
+            self.send_json({"error": {"message": err}}, 400)
+            return
+
+        input_items = req.get("input", [])
+        tools = req.get("tools")
+        messages = []
+        if req.get("instructions"):
+            messages.append({"role": "system", "content": req["instructions"]})
+        if isinstance(input_items, str):
+            messages.append({"role": "user", "content": input_items})
+        elif isinstance(input_items, list):
+            for item in input_items:
+                if isinstance(item, str):
+                    messages.append({"role": "user", "content": item})
+                elif isinstance(item, dict):
+                    if item.get("type") == "function_call_output":
+                        messages.append({"role": "tool", "tool_call_id": item.get("call_id", ""),
+                                         "name": item.get("name", ""), "content": item.get("output", "")})
+                    elif item.get("role") == "assistant" or (item.get("type") == "message" and item.get("role") == "assistant"):
+                        cp = item.get("content", [])
+                        text_acc, tc_list = "", []
+                        if isinstance(cp, list):
+                            for c in cp:
+                                if isinstance(c, dict):
+                                    if c.get("type") == "output_text": text_acc += c.get("text", "")
+                                    elif c.get("type") == "function_call": tc_list.append(c)
+                        elif isinstance(cp, str):
+                            text_acc = cp
+                        m = {"role": "assistant", "content": text_acc or None}
+                        if tc_list:
+                            m["tool_calls"] = [{"id": tc.get("call_id", f"call_{i}"), "type": "function",
+                                                "function": {"name": tc.get("name",""), "arguments": tc.get("arguments","{}")}}
+                                               for i, tc in enumerate(tc_list)]
+                        messages.append(m)
+                    else:
+                        role = item.get("role", "user")
+                        content = item.get("content", "")
+                        if isinstance(content, list):
+                            content = " ".join(c.get("text", "") for c in content if c.get("type") in ("text", "input_text"))
+                        messages.append({"role": role, "content": content})
+
+        if tools:
+            tools = [{"type": "function", "function": {"name": t["name"], "description": t.get("description", ""), "parameters": t.get("parameters", {})}}
+                     if t.get("type") == "function" and "function" not in t else t for t in tools]
+
+        tool_choice = req.get("tool_choice", "auto")
+        prompt, images = messages_to_prompt(messages, tools, tool_choice)
+        if not prompt.strip():
+            self.send_json({"error": {"message": "empty input"}}, 400)
+            return
+
+        try:
+            text = generate(prompt, model_id, think_mode, _upload_images(images), extra_fields)
+        except Exception as e:
+            self.send_json({"error": {"message": f"upstream error: {e}"}}, 502)
+            return
+
+        tool_calls = None
+        if tools and text and tool_choice != "none":
+            text, tool_calls = parse_tool_calls(text)
+
+        rid = f"resp_{uuid.uuid4().hex[:16]}"
+        mid = f"msg_{uuid.uuid4().hex[:12]}"
+        output = []
+        if tool_calls:
+            for tc in tool_calls:
+                output.append({"type": "function_call", "id": tc["id"], "call_id": tc["id"],
+                               "name": tc["function"]["name"], "arguments": tc["function"]["arguments"], "status": "completed"})
+        if text or not tool_calls:
+            output.append({"type": "message", "id": mid, "role": "assistant", "status": "completed",
+                           "content": [{"type": "output_text", "text": text or "", "annotations": []}]})
+
+        if req.get("stream"):
+            self.send_response(200)
+            self.send_header("Content-Type", "text/event-stream")
+            self.send_header("Cache-Control", "no-cache")
+            self.send_header("Access-Control-Allow-Origin", "*")
+            self.end_headers()
+            ev = {"type": "response.created", "response": {"id": rid, "object": "response", "status": "in_progress", "model": model_name, "output": []}}
+            self.wfile.write(f"event: response.created\ndata: {json.dumps(ev)}\n\n".encode())
+            for item in output:
+                if item["type"] == "function_call":
+                    ev = {"type": "response.function_call_arguments.done", "item_id": item["id"], "call_id": item["call_id"], "name": item["name"], "arguments": item["arguments"]}
+                    self.wfile.write(f"event: response.function_call_arguments.done\ndata: {json.dumps(ev)}\n\n".encode())
+                elif item["type"] == "message":
+                    for ci, cp in enumerate(item["content"]):
+                        ev = {"type": "response.output_text.done", "item_id": item["id"], "content_index": ci, "text": cp["text"]}
+                        self.wfile.write(f"event: response.output_text.done\ndata: {json.dumps(ev)}\n\n".encode())
+            resp_obj = {"id": rid, "object": "response", "status": "completed", "model": model_name, "output": output,
+                        "usage": {"input_tokens": len(prompt)//4, "output_tokens": len(text or "")//4, "total_tokens": (len(prompt)+len(text or ""))//4}}
+            self.wfile.write(f"event: response.completed\ndata: {json.dumps({'type': 'response.completed', 'response': resp_obj})}\n\n".encode())
+            self.wfile.flush()
+        else:
+            self.send_json({"id": rid, "object": "response", "created_at": int(time.time()), "status": "completed",
+                            "model": model_name, "output": output,
+                            "usage": {"input_tokens": len(prompt)//4, "output_tokens": len(text or "")//4, "total_tokens": (len(prompt)+len(text or ""))//4}})
+
+    # ─── /v1beta/models (Google Gemini CLI) ──────────────────────────────────
+
+    def _handle_google_generate(self, body: bytes, stream: bool):
+        req = self._parse_body(body)
+        if req is None:
+            self.send_json({"error": {"message": "invalid JSON"}}, 400)
+            return
+        m = re.match(r'/v1beta/models/([^:?]+)', self.path)
+        model_name = m.group(1) if m else CONFIG["default_model"]
+        model_name, model_id, think_mode, err, extra_fields = resolve_model(model_name)
+        if err:
+            self.send_json({"error": {"message": err}}, 400)
+            return
+
+        tool_config = req.get("toolConfig", {})
+        fc_mode = tool_config.get("functionCallingConfig", {}).get("mode", "AUTO")
+        has_tools = bool(req.get("tools")) and fc_mode != "NONE"
+        prompt, images = google_contents_to_prompt(req)
+        if not prompt.strip():
+            self.send_json({"error": {"message": "empty content"}}, 400)
+            return
+
+        file_refs = _upload_images(images)
+        log(f"Google API: model={model_name} stream={stream} tools={has_tools} prompt_len={len(prompt)}")
+
+        if stream and not has_tools:
+            try:
+                self._start_sse()
+                full_text = ""
+                for delta in generate_stream(prompt, model_id, think_mode, file_refs, extra_fields):
+                    if not delta:
+                        continue
+                    full_text += delta
+                    chunk_obj = {
+                        "candidates": [{"content": {"parts": [{"text": delta}], "role": "model"}, "index": 0}],
+                        "modelVersion": model_name,
+                    }
+                    self.wfile.write(f"data: {json.dumps(chunk_obj, ensure_ascii=False)}\n\n".encode())
+                    self.wfile.flush()
+                final_chunk = {
+                    "candidates": [{"finishReason": "STOP", "index": 0}],
+                    "usageMetadata": {
+                        "promptTokenCount": len(prompt) // 4,
+                        "candidatesTokenCount": len(full_text) // 4,
+                        "totalTokenCount": (len(prompt) + len(full_text)) // 4,
+                    },
+                    "modelVersion": model_name,
+                }
+                self.wfile.write(f"data: {json.dumps(final_chunk, ensure_ascii=False)}\n\n".encode())
+                self.wfile.flush()
+            except (BrokenPipeError, ConnectionResetError):
+                pass
+            return
+
+        try:
+            text = generate(prompt, model_id, think_mode, file_refs, extra_fields)
+        except Exception as e:
+            self.send_json({"error": {"message": f"upstream error: {e}"}}, 502)
+            return
+
+        if not text:
+            log("Warning: empty response from Gemini")
+
+        response_parts = []
+        if has_tools and text:
+            clean_text, function_calls = parse_google_function_calls(text)
+            if function_calls:
+                if clean_text:
+                    response_parts.append({"text": clean_text})
+                for fc in function_calls:
+                    response_parts.append({"functionCall": {"name": fc["name"], "args": fc["args"]}})
+            else:
+                response_parts.append({"text": text})
+        else:
+            response_parts.append({"text": text or "I apologize, but I was unable to generate a response. Please try again."})
+
+        candidate = {
+            "content": {"parts": response_parts, "role": "model"},
+            "finishReason": "STOP",
+            "index": 0,
+        }
+        usage = {
+            "promptTokenCount": len(prompt) // 4,
+            "candidatesTokenCount": len(text or "") // 4,
+            "totalTokenCount": (len(prompt) + len(text or "")) // 4,
+        }
+        response_obj = {
+            "candidates": [candidate],
+            "usageMetadata": usage,
+            "modelVersion": model_name,
+        }
+
+        if stream:
+            self._start_sse()
+            self.wfile.write(f"data: {json.dumps(response_obj, ensure_ascii=False)}\n\n".encode())
+            self.wfile.flush()
+        else:
+            self.send_json(response_obj)
+
+
+class ThreadedServer(ThreadingMixIn, HTTPServer):
+    daemon_threads = True
+    allow_reuse_address = True

gemini_web2api/tools.py

@@ -0,0 +1,287 @@
+"""Tool calling and multimodal message parsing."""
+import json
+import re
+import uuid
+import base64
+import io
+
+MAX_IMAGE_B64_SIZE = 50000  # ~37KB raw image
+
+
+def _compress_b64_if_needed(b64: str) -> str:
+    """Compress image if base64 is too large for text embedding."""
+    if len(b64) <= MAX_IMAGE_B64_SIZE:
+        return b64
+    try:
+        from PIL import Image
+        img_data = base64.b64decode(b64)
+        img = Image.open(io.BytesIO(img_data))
+        # Resize to max 256px on longest side
+        max_dim = 256
+        ratio = min(max_dim / img.width, max_dim / img.height)
+        if ratio < 1:
+            img = img.resize((int(img.width * ratio), int(img.height * ratio)), Image.LANCZOS)
+        # Convert to JPEG with quality reduction
+        buf = io.BytesIO()
+        img.convert("RGB").save(buf, format="JPEG", quality=60)
+        compressed = base64.b64encode(buf.getvalue()).decode()
+        return compressed
+    except Exception:
+        # If PIL not available, truncate (model will get partial data)
+        return b64[:MAX_IMAGE_B64_SIZE]
+
+
+def _build_tool_choice_instruction(tool_choice, tool_defs: list) -> str:
+    """Build tool_choice constraint instruction.
+
+    tool_choice values:
+      - "none": do not call any tool
+      - "auto": decide whether to call tools (default)
+      - "required": must call at least one tool
+      - {"type": "function", "function": {"name": "xxx"}}: must call specific tool
+    """
+    if tool_choice == "none":
+        return "\n\nIMPORTANT: Do NOT call any tools. Respond with text only."
+    if tool_choice == "required":
+        return "\n\nIMPORTANT: You MUST call at least one tool. Do not respond with text only."
+    if isinstance(tool_choice, dict):
+        fn_name = tool_choice.get("function", {}).get("name", "")
+        if fn_name:
+            return f'\n\nIMPORTANT: You MUST call the tool "{fn_name}". Do not call other tools.'
+    return ""
+
+
+def messages_to_prompt(messages: list, tools: list = None, tool_choice=None) -> tuple:
+    """Convert OpenAI messages to (prompt_str, images_list).
+
+    Returns (prompt, images) where images is a list of (bytes, mime_type) tuples.
+    """
+    parts = []
+    images = []
+
+    if tools and tool_choice != "none":
+        tool_defs = []
+        for tool in tools:
+            fn = tool.get("function", tool) if tool.get("type") == "function" else tool
+            tool_defs.append({
+                "name": fn.get("name", tool.get("name", "")),
+                "description": fn.get("description", tool.get("description", "")),
+                "parameters": fn.get("parameters", tool.get("parameters", {})),
+            })
+        if tool_defs:
+            constraint = _build_tool_choice_instruction(tool_choice, tool_defs)
+            parts.append(
+                "# Tool Use\n\n"
+                "You can call the following tools. Call format:\n"
+                '```tool_call\n{"name": "func_name", "arguments": {...}}\n```\n'
+                "When calling tools, output ONLY the tool_call block(s).\n\n"
+                f"Available tools:\n{json.dumps(tool_defs, indent=2)}"
+                f"{constraint}"
+            )
+
+    for msg in messages:
+        role = msg.get("role", "user")
+        content = msg.get("content", "")
+
+        if isinstance(content, list):
+            text_parts = []
+            for c in content:
+                if c.get("type") in ("text", "input_text"):
+                    text_parts.append(c.get("text", ""))
+                elif c.get("type") == "image_url":
+                    text_parts.append("[Note: Image input not supported in this API. Please describe the image in text.]")
+                elif c.get("type") == "image":
+                    text_parts.append("[Note: Image input not supported in this API. Please describe the image in text.]")
+            content = " ".join(text_parts)
+
+        if role == "system":
+            parts.append(f"[System instruction]: {content}")
+        elif role == "assistant":
+            if msg.get("tool_calls"):
+                tc_strs = []
+                for tc in msg["tool_calls"]:
+                    fn = tc.get("function", {})
+                    tc_strs.append(
+                        f'```tool_call\n{{"name": "{fn.get("name")}", '
+                        f'"arguments": {fn.get("arguments", "{}")}}}\n```'
+                    )
+                parts.append(f"[Assistant]: {content or ''}\n" + "\n".join(tc_strs))
+            else:
+                parts.append(f"[Assistant]: {content}")
+        elif role == "tool":
+            parts.append(f"[Tool result for {msg.get('name', '')}]: {content}")
+        else:
+            parts.append(content if content else "")
+
+    prompt = "\n\n".join(p for p in parts if p)
+    return prompt, images
+
+
+def parse_tool_calls(text: str) -> tuple:
+    """Extract tool_call blocks. Returns (clean_text, tool_calls_list)."""
+    tool_calls = []
+    pattern = r'```tool_call\s*\n(.*?)\n```'
+    clean_parts = []
+    last_end = 0
+    for m in re.finditer(pattern, text, re.DOTALL):
+        clean_parts.append(text[last_end:m.start()])
+        last_end = m.end()
+        try:
+            data = json.loads(m.group(1).strip())
+            tool_calls.append({
+                "id": f"call_{uuid.uuid4().hex[:8]}",
+                "type": "function",
+                "function": {
+                    "name": data["name"],
+                    "arguments": json.dumps(data.get("arguments", {}), ensure_ascii=False),
+                },
+            })
+        except (json.JSONDecodeError, KeyError):
+            pass
+    clean_parts.append(text[last_end:])
+    clean = "".join(clean_parts).strip()
+    return clean, tool_calls
+
+
+# ─── Google Native API helpers ─────────────────────────────────────────────────
+
+
+def build_tool_prompt(tool_defs: list) -> str:
+    """Build natural tool-use prompt for Gemini Web that avoids prompt-injection detection."""
+    tool_spec = json.dumps(tool_defs, indent=2, ensure_ascii=False)
+    return (
+        "# Tool Use\n\n"
+        "You can call the following tools to help accomplish tasks. "
+        "These tools connect to the user's local environment and will execute when called.\n\n"
+        "Call format (use this exact format):\n"
+        "```function_call\n"
+        '{"name": "<tool_name>", "args": {<arguments>}}\n'
+        "```\n\n"
+        "When calling tools:\n"
+        "- Output ONLY the function_call block(s), nothing else\n"
+        "- You may call multiple tools with multiple blocks\n"
+        "- After receiving a [Tool result for ...], use that data to answer the user\n\n"
+        f"Available tools:\n{tool_spec}"
+    )
+
+
+def _google_tool_choice_instruction(req: dict) -> str:
+    """Extract tool_choice constraint from Google API toolConfig."""
+    tool_config = req.get("toolConfig", {})
+    fc_config = tool_config.get("functionCallingConfig", {})
+    mode = fc_config.get("mode", "AUTO")
+    allowed = fc_config.get("allowedFunctionNames", [])
+
+    if mode == "NONE":
+        return "\n\nIMPORTANT: Do NOT call any tools. Respond with text only."
+    if mode == "ANY":
+        if allowed:
+            names = ", ".join(f'"{n}"' for n in allowed)
+            return f"\n\nIMPORTANT: You MUST call one of these tools: {names}. Do not respond with text only."
+        return "\n\nIMPORTANT: You MUST call at least one tool. Do not respond with text only."
+    return ""
+
+
+def google_contents_to_prompt(req: dict) -> tuple:
+    """Convert Google API contents/tools/systemInstruction to (prompt_str, images_list).
+
+    Returns (prompt, images) where images is a list of (bytes, mime_type) tuples.
+    """
+    parts = []
+    images = []
+
+    tool_config = req.get("toolConfig", {})
+    fc_mode = tool_config.get("functionCallingConfig", {}).get("mode", "AUTO")
+
+    tools = req.get("tools")
+    tool_defs = []
+    if tools and fc_mode != "NONE":
+        for tool_group in tools:
+            for fn in tool_group.get("functionDeclarations", []):
+                td = {"name": fn.get("name", ""), "description": fn.get("description", "")}
+                params = fn.get("parameters") or fn.get("parametersJsonSchema")
+                if params:
+                    td["parameters"] = params
+                tool_defs.append(td)
+
+    sys_inst = req.get("systemInstruction")
+    if sys_inst:
+        sys_parts = sys_inst.get("parts", [])
+        sys_text = " ".join(p.get("text", "") for p in sys_parts if p.get("text"))
+        if sys_text:
+            if tool_defs:
+                constraint = _google_tool_choice_instruction(req)
+                parts.append(sys_text + "\n\n" + build_tool_prompt(tool_defs) + constraint)
+            else:
+                parts.append(sys_text)
+    elif tool_defs:
+        constraint = _google_tool_choice_instruction(req)
+        parts.append(build_tool_prompt(tool_defs) + constraint)
+
+    for content in req.get("contents", []):
+        role = content.get("role", "user")
+        msg_parts = []
+        for p in content.get("parts", []):
+            if p.get("text"):
+                msg_parts.append(p["text"])
+            elif p.get("inlineData"):
+                data = p["inlineData"]
+                mime = data.get("mimeType", "image/png")
+                images.append((base64.b64decode(data["data"]), mime))
+            elif p.get("functionCall"):
+                fc = p["functionCall"]
+                msg_parts.append(
+                    f'```function_call\n{json.dumps({"name": fc["name"], "args": fc.get("args", {})}, ensure_ascii=False)}\n```'
+                )
+            elif p.get("functionResponse"):
+                fr = p["functionResponse"]
+                msg_parts.append(
+                    f'[Tool result for {fr.get("name", "")}]: {json.dumps(fr.get("response", {}), ensure_ascii=False)}'
+                )
+        text = "\n".join(msg_parts)
+        if role == "model":
+            parts.append(f"[Assistant]: {text}")
+        else:
+            parts.append(text)
+
+    return "\n\n".join(p for p in parts if p), images
+
+
+def parse_google_function_calls(text: str) -> tuple:
+    """Extract function_call blocks from model output.
+
+    Handles 3 formats:
+    1. ```function_call\\n{...}\\n``` (standard)
+    2. function_call\\n{...} (without backticks)
+    3. Raw JSON with "name" + "args" keys
+
+    Returns (clean_text, [{"name": ..., "args": ...}])
+    """
+    function_calls = []
+    pattern1 = r'```function_call\s*\n(.*?)\n```'
+    pattern2 = r'(?:^|\n)function_call\s*\n(\{[^`]*?\})'
+    clean = text
+    for pattern in [pattern1, pattern2]:
+        for match in re.findall(pattern, clean, re.DOTALL):
+            try:
+                data = json.loads(match.strip())
+                if "name" in data:
+                    function_calls.append({
+                        "name": data["name"],
+                        "args": data.get("args", data.get("arguments", {})),
+                    })
+            except (json.JSONDecodeError, KeyError):
+                pass
+        clean = re.sub(pattern, '', clean, flags=re.DOTALL).strip()
+    if not function_calls and clean.strip().startswith("{"):
+        try:
+            data = json.loads(clean.strip())
+            if "name" in data and ("args" in data or "arguments" in data):
+                function_calls.append({
+                    "name": data["name"],
+                    "args": data.get("args", data.get("arguments", {})),
+                })
+                clean = ""
+        except (json.JSONDecodeError, KeyError):
+            pass
+    return clean, function_calls