Gemini и сетевой прокси: настройка клиента и диагностика соединения

Текст для тех, кто гоняет Gemini Developer API из Python через HTTP-прокси и путает обрыв CONNECT с отказом ключа. Браузерный Gemini, Vertex и CLI здесь не разбираем. Ключи Google не выдаём и IP не продаём. Ниже стенд google-genai 2.30.1 на 127.0.0.1: маршрут, 407, TLS и таймаут. Успех макета не равен доступу к аккаунту или региону.

01Какой клиент Gemini вы настраиваете#

Слово Gemini в интерфейсах Google закрывает разные продукты. Сетевой прокси у них не общий. Если смешать веб, ключ Developer API, облачный проект и CLI, вы будете чинить не тот слой.

Дальше один клиент: google-genai 2.30.1, Gemini Developer API, CPython 3.14.6, httpx 0.28.1, macOS 15.6 arm64. Документация SDK на 14 октября 2026 задаёт прокси через HTTPS_PROXY до создания клиента и необязательный SSL_CERT_FILE. SOCKS5 там же описан только как http_options.client_args={'proxy': 'socks5://...'} после httpx[socks]. Переменную окружения для SOCKS эта версия не обещает. Мы SOCKS не гоняли. Тот же текст и в README python-genai.

Официальный пример custom base URL в той же документации прямо говорит: "Currently only enterprise=True is supported". Стандартный production endpoint Developer API остаётся:

https://generativelanguage.googleapis.com/

В исходнике 2.30.1 patch_http_options после этого значения всё же подмешивает пользовательский HttpOptions.base_url. На стенде Developer-клиент с base_url=https://127.0.0.1:порт ходил на макет. Это локальный тестовый крючок этой сборки, не документированная поддержка и не рекомендация боевого API gateway.

Ключ в примерах фиктивный. Платных вызовов Google не было. Ответ макета с текстом lab-fixture-not-google не является ответом модели.

Регион и возраст живут отдельно. Available regions на 14 октября 2026 перечисляет страны для Gemini API и Google AI Studio. Russia в этом списке нет. Прокси не добавляет страну в перечень. Требование 18+ на той же странице относится к доступу Google AI Studio и проверке возраста аккаунта, не ко всем поверхностям Gemini. Troubleshooting той же даты отдельно говорит: 429 и 503 имеет смысл ретраить, 400, 402 и 403 не ретраить как "сеть моргнула". Это продуктовая рекомендация Google, не описание дефолта пакета 2.30.1. Пин конструктора ниже.

02Какие параметры прокси есть у google-genai 2.30.1#

Схема цели и схема прокси это разные поля. Цель SDK в проде это:

https://generativelanguage.googleapis.com/

Прокси до неё обычно http://хост:порт. HTTPS в URL цели не превращает прокси в "https-прокси". Для CONNECT прокси остаётся HTTP. Это видно в журнале: метод CONNECT, target 127.0.0.1:57416, а URL прокси при этом был http://127.0.0.1:57417.

Поддерживаемые на этой версии рычаги, которые мы реально трогали:

1. Явный прокси: types.HttpOptions(client_args={'proxy': 'http://USER:PASS@host:port'}). Тот же словарь можно отдать в async_client_args.
2. Переменная HTTPS_PROXY до Client(...). Документация прямо просит выставить её заранее. httpx по умолчанию с trust_env=True.
3. NO_PROXY / no_proxy. На стенде список 127.0.0.1,localhost вместе с HTTPS_PROXY дал прямой заход на origin: в журнале прокси ноль событий, ответ макета всё равно пришёл.
4. CA: SDK 2.30.1 сам собирает ssl.create_default_context из SSL_CERT_FILE или оставляет явный client_args={'verify': '/path/ca.pem'}. Это слой SDK. Отдельно httpx 0.28.1 при verify=True и trust_env=True тоже читает SSL_CERT_FILE. Комментарий в исходнике SDK, будто httpx этого не делает, для этой версии httpx устарел. Стенд измерял CA у SDK-клиента, не опровергал httpx.
5. Таймаут: HttpOptions.timeout в миллисекундах. В _api_client.get_timeout_in_seconds значение делится на 1000 и уходит в httpx. Это не секунды.
6. Ретраи: смотрите пин 2.30.1, не общую фразу troubleshooting. retry_args(None) это stop_after_attempt(1). Клиент с пустым retry_options в HttpOptions имеет _retry.stop.max_attempt_number=1. Явный HttpRetryOptions() без аргументов даёт 5 попыток всего, коды [408, 429, 500, 502, 503, 504]. attempts=1 даёт одну попытку. Транспортный таймаут тела мы гоняли с attempts=1. Дефолт конструктора эту задержку не размножает.

Неподдерживаемое в этом гайде: SOCKS из env, verify=False, правка системного хранилища сертификатов, подмена DNS через /etc/hosts, "https://" как схема самого HTTP-прокси без отдельной поддержки.

Явный client_args['proxy'] на стенде перебил враждебный HTTPS_PROXY на закрытый порт. Запрос ушёл на рабочий open-прокси и открыл CONNECT. Если чините "прокси в env есть, а SDK его не слушает", сначала смотрите, не задан ли proxy= в HttpOptions. Обратная ошибка тоже частая: в оболочке торчит чужой HTTPS_PROXY из прошлого эксперимента, а вы думаете, что клиент ходит напрямую.

Ключ API это не логин прокси. x-goog-api-key уходит уже внутри тоннеля. Proxy-Authorization смотрит прокси на CONNECT. Путать их бесполезно: 407 вы получите до TLS с origin.

03Как проверить маршрут на локальном стенде#

Не начинайте с боевого ключа и чужого биллинга. Ниже самодостаточный скрипт: ставит локальный HTTPS-origin, HTTP CONNECT-прокси, google-genai 2.30.1 и фиктивный ключ. Google не вызывается. openssl должен быть в PATH.

python3.14 -m venv .venv
.venv/bin/pip install 'google-genai==2.30.1' 'httpx==0.28.1'
.venv/bin/python public_lab.py

Скопируйте файл как public_lab.py и запустите. Это не кабинет Google и не entitlement.

"""Self-contained loopback lab for google-genai 2.30.1.

Dummy key. Local mock only. Does not call Google.
Requires openssl on PATH. Run after: pip install 'google-genai==2.30.1' 'httpx==0.28.1'
"""

from __future__ import annotations

import base64
import json
import sys
import os
import select
import socket
import ssl
import subprocess
import tempfile
import threading
import time
from contextlib import closing
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from urllib.parse import quote

from google import genai
from google.genai import types

USER = "demo"
PASSWORD = "p@ss:/#%"
DUMMY_KEY = "lab-dummy-not-a-google-key"


def _free_port() -> int:
    with closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as sock:
        sock.bind(("127.0.0.1", 0))
        sock.listen(1)
        return sock.getsockname()[1]


class _Origin(BaseHTTPRequestHandler):
    protocol_version = "HTTP/1.1"

    def log_message(self, fmt: str, *args: object) -> None:
        return

    def do_POST(self) -> None:  # noqa: N802
        length = int(self.headers.get("Content-Length") or 0)
        if length:
            self.rfile.read(length)
        payload = {
            "candidates": [
                {
                    "content": {
                        "role": "model",
                        "parts": [{"text": "lab-fixture-not-google"}],
                    },
                    "finishReason": "STOP",
                    "index": 0,
                }
            ],
            "fixture": True,
            "notGoogleUpstream": True,
        }
        body = json.dumps(payload).encode()
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.send_header("Connection", "close")
        self.end_headers()
        self.wfile.write(body)


def _certs(tmp: Path) -> tuple[Path, Path, Path]:
    ca_key, ca_pem = tmp / "ca.key", tmp / "ca.pem"
    origin_key, origin_csr, origin_pem = tmp / "origin.key", tmp / "origin.csr", tmp / "origin.pem"
    ca_cnf = tmp / "ca.cnf"
    leaf_cnf = tmp / "leaf.cnf"
    ca_cnf.write_text(
        "[req]\ndistinguished_name=req\n[req]\n[v3_ca]\n"
        "basicConstraints=critical,CA:TRUE\nsubjectKeyIdentifier=hash\n"
        "authorityKeyIdentifier=keyid:always,issuer\n"
        "keyUsage=critical,keyCertSign,cRLSign\n",
        encoding="utf-8",
    )
    leaf_cnf.write_text(
        "[v3_leaf]\nsubjectAltName=DNS:localhost,IP:127.0.0.1\n"
        "basicConstraints=CA:FALSE\nsubjectKeyIdentifier=hash\n"
        "authorityKeyIdentifier=keyid,issuer\n",
        encoding="utf-8",
    )
    subprocess.check_call(
        [
            "openssl", "req", "-x509", "-newkey", "rsa:2048", "-nodes",
            "-keyout", str(ca_key), "-out", str(ca_pem), "-days", "1",
            "-subj", "/CN=gemini-public-lab-ca", "-config", str(ca_cnf),
            "-extensions", "v3_ca",
        ],
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
    )
    subprocess.check_call(
        [
            "openssl", "req", "-newkey", "rsa:2048", "-nodes",
            "-keyout", str(origin_key), "-out", str(origin_csr),
            "-subj", "/CN=127.0.0.1",
        ],
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
    )
    subprocess.check_call(
        [
            "openssl", "x509", "-req", "-in", str(origin_csr), "-CA", str(ca_pem),
            "-CAkey", str(ca_key), "-CAcreateserial", "-out", str(origin_pem),
            "-days", "1", "-extfile", str(leaf_cnf), "-extensions", "v3_leaf",
        ],
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
    )
    return ca_pem, origin_pem, origin_key


def _proxy_loop(sock: socket.socket, stop: threading.Event, events: list[str]) -> None:
    sock.settimeout(0.4)
    while not stop.is_set():
        try:
            conn, _addr = sock.accept()
        except socket.timeout:
            continue
        except OSError:
            break
        threading.Thread(target=_handle_proxy, args=(conn, events), daemon=True).start()


def _handle_proxy(conn: socket.socket, events: list[str]) -> None:
    try:
        data = b""
        conn.settimeout(5)
        while b"\r\n\r\n" not in data:
            chunk = conn.recv(4096)
            if not chunk:
                break
            data += chunk
        line = data.split(b"\r\n", 1)[0].decode("latin1", "replace")
        parts = line.split(" ")
        method = parts[0] if parts else ""
        target = parts[1] if len(parts) > 1 else ""
        events.append(f"{method} {target}")
        if method != "CONNECT":
            conn.close()
            return
        host, _, port_s = target.partition(":")
        dest = socket.create_connection((host, int(port_s or "443")), timeout=5)
        conn.sendall(b"HTTP/1.1 200 Connection Established\r\n\r\n")
        pair = [conn, dest]
        while True:
            ready, _, _ = select.select(pair, [], [], 5)
            if not ready:
                break
            for src in ready:
                dst = dest if src is conn else conn
                chunk = src.recv(8192)
                if not chunk:
                    dest.close()
                    conn.close()
                    return
                dst.sendall(chunk)
    except OSError:
        try:
            conn.close()
        except OSError:
            pass


def main() -> int:
    inject = os.environ.get("GEMINI_PUBLIC_INJECT")
    if inject == "load_cert":
        def _boom_cert(self: ssl.SSLContext, *args: object, **kwargs: object) -> None:
            raise RuntimeError("inject load_cert_chain")

        ssl.SSLContext.load_cert_chain = _boom_cert  # type: ignore[method-assign]
    if inject == "shutdown":
        def _boom_shutdown(self: ThreadingHTTPServer, *args: object, **kwargs: object) -> None:
            raise RuntimeError("inject public shutdown")

        ThreadingHTTPServer.shutdown = _boom_shutdown  # type: ignore[method-assign]
    tmp = Path(tempfile.mkdtemp(prefix="gemini-public-lab-"))
    httpd = None
    sock = None
    serving = False
    origin_stop = threading.Event()
    proxy_stop = threading.Event()
    serve_thread: threading.Thread | None = None
    proxy_thread: threading.Thread | None = None
    events: list[str] = []
    prior_https = os.environ.get("HTTPS_PROXY")
    rc = 1
    try:
        ca_pem, origin_pem, origin_key = _certs(tmp)
        origin_port = _free_port()
        proxy_port = _free_port()
        httpd = ThreadingHTTPServer(("127.0.0.1", origin_port), _Origin)
        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
        ctx.load_cert_chain(str(origin_pem), str(origin_key))
        httpd.socket = ctx.wrap_socket(httpd.socket, server_side=True)
        serve_thread = threading.Thread(
            target=httpd.serve_forever, name="gemini-public-origin", daemon=True
        )
        serve_thread.start()
        serving = True
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        sock.bind(("127.0.0.1", proxy_port))
        sock.listen(20)
        proxy_thread = threading.Thread(
            target=_proxy_loop, args=(sock, proxy_stop, events), name="gemini-public-proxy", daemon=True
        )
        proxy_thread.start()
        time.sleep(0.2)
        proxy = (
            "http://"
            + quote(USER, safe="")
            + ":"
            + quote(PASSWORD, safe="")
            + f"@127.0.0.1:{proxy_port}"
        )
        client_args = {"proxy": proxy, "verify": str(ca_pem), "trust_env": False}
        client = genai.Client(
            api_key=DUMMY_KEY,
            http_options=types.HttpOptions(
                base_url=f"https://127.0.0.1:{origin_port}",
                timeout=8000,
                client_args=client_args,
                async_client_args=dict(client_args),
                retry_options=types.HttpRetryOptions(attempts=1),
            ),
        )
        try:
            text = client.models.generate_content(model="lab-ok", contents="lab ping").text
        finally:
            client.close()
        connect = any(item.startswith(f"CONNECT 127.0.0.1:{origin_port}") for item in events)
        print(
            json.dumps(
                {
                    "text": text,
                    "connect": connect,
                    "events": events,
                    "fixture": True,
                    "notGoogle": True,
                }
            )
        )
        rc = 0 if text == "lab-fixture-not-google" and connect else 1
    finally:
        origin_stop.set()
        proxy_stop.set()
        shutdown_error = None
        if httpd is not None and serving:
            try:
                httpd.shutdown()
            except Exception as err:
                shutdown_error = type(err).__name__
                try:
                    httpd._BaseServer__shutdown_request = True
                except Exception:
                    pass
        if serve_thread is not None:
            serve_thread.join(timeout=3.0)
        if proxy_thread is not None:
            proxy_thread.join(timeout=3.0)
        if httpd is not None:
            try:
                httpd.server_close()
            except Exception:
                pass
            try:
                httpd.socket.close()
            except Exception:
                pass
        if sock is not None:
            try:
                sock.close()
            except OSError:
                pass
        for key in tmp.glob("*.key"):
            try:
                key.unlink()
            except OSError:
                pass
        keys_left = len(list(tmp.glob("*.key")))
        if prior_https is None:
            os.environ.pop("HTTPS_PROXY", None)
        else:
            os.environ["HTTPS_PROXY"] = prior_https
        subprocess.call(["rm", "-rf", str(tmp)], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
        workers_alive = [
            thread.name
            for thread in (serve_thread, proxy_thread)
            if thread is not None and thread.is_alive()
        ]
        time.sleep(0.7)
        workers_alive_700ms = [
            thread.name
            for thread in (serve_thread, proxy_thread)
            if thread is not None and thread.is_alive()
        ]
        cleaned = not workers_alive and not workers_alive_700ms and keys_left == 0
        print(
            json.dumps(
                {
                    "cleaned": cleaned,
                    "serving": serving,
                    "keys_left": keys_left,
                    "workers_alive": workers_alive,
                    "workers_alive_700ms": workers_alive_700ms,
                    "shutdown_error": shutdown_error,
                }
            ),
            file=sys.stderr,
        )
        if workers_alive or workers_alive_700ms:
            rc = 1
        if shutdown_error is not None:
            rc = 1
    return rc


if __name__ == "__main__":
    raise SystemExit(main())

Вариант через HTTPS_PROXY до Client(...). Этот блок не запускайте с фиктивным ключом против Google: у него нет custom base_url, остаётся production endpoint. Предыдущее значение HTTPS_PROXY, включая его отсутствие, возвращается в finally.

"""Environment variant. Sets HTTPS_PROXY in-process before Client init.

Production shape keeps the default Developer endpoint. Do not run this
block with the dummy key against Google.
"""

from __future__ import annotations

import os
from urllib.parse import quote

from google import genai
from google.genai import types

USER = "demo"
PASSWORD = "p@ss:/#%"
proxy = (
    "http://"
    + quote(USER, safe="")
    + ":"
    + quote(PASSWORD, safe="")
    + "@127.0.0.1:8080"
)

_prior_https_proxy = os.environ.get("HTTPS_PROXY")
try:
    os.environ["HTTPS_PROXY"] = proxy
    client = genai.Client(
        api_key=os.environ.get("GEMINI_API_KEY", "not-a-google-key"),
        http_options=types.HttpOptions(
            timeout=8000,
            client_args={"trust_env": True},
            async_client_args={"trust_env": True},
            retry_options=types.HttpRetryOptions(attempts=1),
        ),
    )
    client.close()
finally:
    if _prior_https_proxy is None:
        os.environ.pop("HTTPS_PROXY", None)
    else:
        os.environ["HTTPS_PROXY"] = _prior_https_proxy

Порты только на 127.0.0.1, без правки системных proxy macOS. Унаследованные HTTP_PROXY, HTTPS_PROXY, ALL_PROXY, SSL_CERT_FILE, GEMINI_API_KEY, GOOGLE_APPLICATION_CREDENTIALS и CGI-подобные HTTP_* из процесса уберите. HOME укажите во временный каталог без .netrc. NETRC пусть ссылается на несуществующий файл.

Нумерованный прогон, который мы выполнили 14 октября 2026:

1. Собрать CA и сертификат origin с SAN IP:127.0.0.1. Приватные ключи после прогона удалить. На стенде privateKeysDeleted=true.
2. Слушать HTTPS-origin. Ответ generateContent пометить "fixture": true. Это не Google.
3. Слушать open-прокси без пароля и auth-прокси с паролем p@ss:/#%. В лог писать метод, target и факт заголовка, сам пароль не писать.
4. Создать:

genai.Client(api_key=..., http_options=HttpOptions(base_url=origin, retry_options=HttpRetryOptions(attempts=1)))

5. Прямой вызов без прокси. Ожидание: POST на .../models/lab-ok:generateContent, текст lab-fixture-not-google, в журнале прокси пусто.
6. Тот же вызов с client_args={'proxy': 'http://127.0.0.1:open'}. Ожидание: CONNECT на 127.0.0.1:origin.
7. Повторить через HTTPS_PROXY, затем через явный proxy при враждебном env, затем с NO_PROXY.
8. На auth-прокси: без учётки, с неверным паролем, с URL-encoded паролем.
9. Снять TLS без вашей CA, затем с SSL_CERT_FILE и с verify=.
10. Закрытый порт прокси, затем origin, который держит тело 2.5 с при timeout=400.
11. generate_content_stream и client.aio.models.generate_content через тот же CONNECT.
12. В finally закрыть клиент, остановить сокеты, удалить ключи, снять dummy GEMINI_API_KEY.

Прямой заход и CONNECT на этом прогоне разошлись как надо. Open-прокси записал CONNECT 127.0.0.1:57416. Origin при этом остался HTTPS. Если в вашем журнале вместо CONNECT виден GET http://... на HTTPS-цель, вы смотрите не тот клиент или сломали схему.

Подпись к кадру: таблица этой сессии, не кабинет Google и не страница чекера. Чекер проверяет, отвечает ли endpoint, который вы сами задали. Мой IP показывает выход браузера. Ни то ни другое не доказывает, что genai.Client открыл тот же сокет.

Пароль со слэшем и решёткой в URL прокси нужно кодировать. На стенде quote(password, safe='') дал успешный CONNECT с has_proxy_authorization=true. Сырой p@ss:/#% в URL ломает разбор. В логах стенда пароль заменён на ***, Basic на REDACTED_BASIC. Боевой -v с Proxy-Authorization в тикет не кладите.

04Как сохранить проверку TLS и не светить ключ#

Отключать проверку сертификата "чтобы прокси заработал" здесь лишнее. Без вашей CA клиент упал так: httpx.ConnectError, цепочка SSLCertVerificationError, текст unable to get local issuer certificate. Это слой TLS до разбора JSON API. verify=False в лабе не использовался.

Два рабочих пути, оба локальные для процесса:

• SSL_CERT_FILE=/path/ca.pem до Client(...). Документированный рычаг README. На стенде после этой переменной тот же CONNECT вернул текст макета.
• client_args={'verify': '/path/ca.pem'}. SDK не перетирает уже заданный verify.

Не копируйте лабораторный CA в системные ключи macOS "навсегда". Не ставьте SSL_CERT_FILE в ~/.zshrc, если не хотите, чтобы случайный curl начал доверять той же CA. Для боя корпоративный MITM-прокси требует CA, которую вам выдал админ, а не "любой pem с форума".

Ключ API не должен жить в argv и в verbose. В 2.30.1 при создании клиента ключ попадает в заголовок x-goog-api-key. На макете заголовок доходил, значение в журнале origin мы не печатали. В процессе после прогона переменные GEMINI_API_KEY и GOOGLE_API_KEY снимались. Если оба имени заданы, документация SDK отдаёт приоритет GOOGLE_API_KEY. Для отладки маршрута достаточно фиктивного значения. Боевой ключ не нужен, пока не доказан CONNECT и TLS.

.netrc и ADC сюда же. Developer-клиент с явным api_key= не должен молча уехать в облачный проект из GOOGLE_APPLICATION_CREDENTIALS. На всякий случай мы вычищали и это имя, и GOOGLE_GENAI_USE_VERTEXAI. Иначе через полчаса разбираете "почему base URL снова aiplatform".

05Как отличить сетевую ошибку от ответа API#

Смотрите слой, не настроение. Ниже факты этого прогона, не универсальная таблица всех обёрток Google.

407 в httpx и 401 в ClientError рядом не стоят. Первый вы ещё не поговорили с origin. Второй вы уже прошли прокси и TLS, а HTTP-статус пришёл от той стороны, куда открыли CONNECT. На макете 401/403/429 помечены "fixture": true. Это иллюстрация разбора клиента, не политика Google. Troubleshooting 14 октября 2026 как раз просит не ретраить 403 как сетевой сбой.

Таймаут легко перепутать с "API молчит". HttpOptions.timeout=400 это 400 миллисекунд. Мы получили ReadTimeout за 0.404 с, пока origin ещё спал. У конструктора 2.30.1 без HttpRetryOptions стоп равен 1, так что дефолт эту задержку не размножает. Если явно задать HttpRetryOptions() без attempts, получите 5 попыток на коды 408/429/5xx. Для поиска обрыва оставляйте attempts=1, затем включайте ретраи только на те статусы, которые допускает ваша политика.

Прямой отказ TCP (Connection refused) тоже не квота. Сообщение [Errno 61] на этой машине значит: на адресе прокси никто не слушает. Сначала порт и схема, потом логин.

06Что меняется для потока и async-клиента#

Синхронный generate_content на стенде ходил через CONNECT. Поток и async заявлены в том же SDK, поэтому их тоже прогнали через тот же прокси, всё ещё в макет.

generate_content_stream на 2.30.1 бьёт в {model}:streamGenerateContent?alt=sse. Макет отдал две SSE-строки data: {...}. Клиент собрал chunk-a и chunk-b. В журнале прокси снова CONNECT на loopback origin. Это конечный успешный проход двух чанков, не midstream-обрыв и не устойчивость длинного стрима на публичном API.

Async-ветка на стенде: успешный unary через client.aio, плюс отдельный таймаут lab-slow с aclose() и close() в finally. Документация отдельно предлагает google-genai[aiohttp] с trust_env=True. Этот пакет мы не ставили. Наблюдение относится к httpx async, который SDK использует по умолчанию.

Фрагмент ниже это иллюстрация стенда: ему нужны переменные GEMINI_LAB и SSE-origin из внутреннего harness, не из публичного public_lab.py. Это не отдельный reader start/run. Базовый публичный путь только скрипт выше.

"""Finite stream and async helpers. Call from an async function. Local mock only."""

from __future__ import annotations

import os

from google import genai
from google.genai import types


def make_lab_client() -> genai.Client:
    client_args = {
        "proxy": os.environ["GEMINI_LAB_PROXY"],
        "verify": os.environ["GEMINI_LAB_CA"],
        "trust_env": False,
    }
    return genai.Client(
        api_key=os.environ["GEMINI_LAB_DUMMY_KEY"],
        http_options=types.HttpOptions(
            base_url=os.environ["GEMINI_LAB_BASE_URL"],
            timeout=8000,
            client_args=client_args,
            async_client_args=dict(client_args),
            retry_options=types.HttpRetryOptions(attempts=1),
        ),
    )


def collect_stream() -> list[str]:
    client = make_lab_client()
    chunks: list[str] = []
    try:
        for piece in client.models.generate_content_stream(
            model="lab-ok", contents="stream lab"
        ):
            if piece.text:
                chunks.append(piece.text)
    finally:
        client.close()
    return chunks


async def async_ping() -> str:
    client = make_lab_client()
    try:
        response = await client.aio.models.generate_content(
            model="lab-ok", contents="async lab"
        )
        return response.text or ""
    finally:
        await client.aio.aclose()
        client.close()

Если поток у вас рвётся, а unary запрос жив, не спешите менять тип IP. Сначала сравните путь: generateContent против streamGenerateContent, таймаут чтения, буфер прокси, не режет ли посредник text/event-stream. На макете посредник был тупым CONNECT и конечный поток прошёл. Боевой прокси с антивирусом и своим CA это уже другая история, её этот стенд не закрывает.

Живой WebSocket Live API в этот прогон не входил. Не переносите выводы CONNECT на другой транспорт.

07Как сверить прокси перед выбором в каталоге#

Клиенту 2.30.1 для этого гайда нужен HTTP-прокси с CONNECT и, если включена авторизация, Basic в URL или в Proxy-Authorization. Тип адреса (датацентр, резидент, мобайл) SDK не читает. Он читает хост, порт, схему и пароль. Страна выхода не лечит 407 и не добавляет Russia в список available-regions.

Практическая сверка до покупки плана, без обещания, что Google вас пустит:

1. В карточке провайдера должен быть именно HTTP CONNECT, не только SOCKS5. Если у плана один SOCKS-порт, этот гайд его не покрыл.
2. Авторизация login/password, которую можно положить в http://user:pass@host:port с кодированием спецсимволов. Клиентский сертификат прокси мы не тестировали.
3. Разрешение на HTTPS-цели. CONNECT на 443 к вашему origin. Корпоративный прокси, который пускает только белый список хостов, должен включать generativelanguage.googleapis.com, если вы целитесь в прод. Макет на 127.0.0.1 этого не доказывает.
4. После выдачи доступа у самого провайдера, не из сравнительной карточки, возьмите host:port. Карточки Top Proxy логин не выдают.
5. Повторите короткий стенд: CONNECT, TLS с нормальной CA, один unary запрос. Боевой ключ подключайте только когда 407 и сертификат уже молчат.
6. Отдельно прогоните чекер тем endpoint, который вам разрешён офертой. Совпадение IP на странице мой IP в браузере не доказывает маршрут SDK.

В каталоге отфильтруйте заявленный HTTP. Сравнение ниже это опубликованные русские карточки с HTTP в тексте, не тест этих провайдеров на нашем стенде и не обещание, что Gemini-аккаунт заработает.

SOAX — резидентские и мобильные прокси3.9219

SOAX продаёт ротационные резидентские и мобильные прокси на общем пуле кредитов; в тех же планах доступны датацентровые и ISP. Builder — от $200/мес + VAT (200 кредитов); Tier 1 на Builder — $3/GB.

85 стран

3,00 $

за 1 GB

Bright Data — резидентские и датацентровые прокси4.41439

Bright Data — крупная сеть прокси для сбора данных: резидентские, мобильные, ISP и датацентровые IP примерно в 195 странах. Резидентский трафик можно оплачивать по факту без подписки, серверные и ISP-адреса продаются пакетами от 10 IP. Для резидентских и мобильных прокси нужна проверка личности.

10 стран

5,00 $

за 1 GB

Decodo (Smartproxy) — резидентские и датацентровые прокси4.22106

Decodo (ранее Smartproxy) продаёт резидентские и датацентровые прокси. Резидентские: $4/GB без месячного плана, подписки от $3.75/GB за 3 GB. Оплата: карты, PayPal, Alipay, Google/Apple Pay, крипто (с оговорками).

83 страны

0,38 $

за 1 GB

Oxylabs — резидентские и датацентровые прокси4.0770

Oxylabs продаёт резидентские, датацентровые и мобильные прокси. Резидентские планы от $6/GB (5 GB / мес). Оплата: карты, банковский перевод, AliPay, PayPal.

10 стран

0,59 $

за 1 GB

Froxy — резидентские, мобильные и серверные4.1120

Froxy предлагает резидентские, мобильные, быстрые резидентские и серверные прокси с оплатой за трафик, месячными пакетами и платным тестом $1.99 / 3 дня / 100 МБ.

236 стран

5,50 $

за 1 GB

На сайт

Биллинг, лимит сессий и список разрешённых целей читайте в оферте выбранного плана. Прокси не покупает право на модель, регион и квоту. Если ключ живой, CONNECT зелёный, а Google отвечает 403, это уже продуктовая ошибка. Её чинят в кабинете Google, не сменой ASN.