SSE Parsing & Dual Adapter

Parse SSE by hand when the SDK isn't there

Sometimes you don't want the SDK in the loop — you're proxying, you're on a tiny container, you're calling the OAuth-only endpoint that the official SDK doesn't fully support. The SSE format is simple enough to parse directly: lines starting with data:, blank line between events, JSON payload after the prefix.

The dual-adapter pattern

Wrap both auth paths behind a single interface. The caller asks for a stream; the adapter picks API key or OAuth based on availability + recent failure history. The fallback is sticky for a session: once you flip from OAuth to API key (e.g., on a 401), stay on API key for the rest of the session. Otherwise you bounce.

Code

SSE parser — async, line-based·python

import httpx, json
from typing import AsyncIterator

async def stream_gemini_api_key(
    prompt: str, api_key: str, model: str = 'gemini-2.5-flash',
) -> AsyncIterator[str]:
    url = (
        f'https://generativelanguage.googleapis.com/v1beta/models/{model}'
        ':streamGenerateContent?alt=sse'
    )
    body = {'contents': [{'role': 'user', 'parts': [{'text': prompt}]}]}

    async with httpx.AsyncClient(timeout=120) as client:
        async with client.stream('POST', url,
            headers={'x-goog-api-key': api_key, 'Content-Type': 'application/json'},
            json=body,
        ) as resp:
            async for line in resp.aiter_lines():
                if not line.startswith('data: '):
                    continue
                chunk = json.loads(line[6:])
                for cand in chunk.get('candidates', []):
                    for part in cand.get('content', {}).get('parts', []):
                        if text := part.get('text'):
                            yield text

OAuth variant — same shape, different envelope·python

async def stream_gemini_oauth(
    prompt: str, access_token: str, project: str,
    model: str = 'gemini-2.5-flash',
) -> AsyncIterator[str]:
    url = 'https://cloudcode-pa.googleapis.com/v1internal:streamGenerateContent?alt=sse'
    body = {
        'model':   model,
        'project': project,
        'request': {
            'contents': [{'role': 'user', 'parts': [{'text': prompt}]}],
            'generationConfig': {},
        },
    }

    async with httpx.AsyncClient(timeout=120) as client:
        async with client.stream('POST', url,
            headers={
                'Authorization': f'Bearer {access_token}',
                'Content-Type':  'application/json',
            },
            json=body,
        ) as resp:
            async for line in resp.aiter_lines():
                if not line.startswith('data: '):
                    continue
                chunk = json.loads(line[6:])
                # OAuth wraps the response inside chunk['response']
                resp_obj = chunk.get('response', chunk)
                for cand in resp_obj.get('candidates', []):
                    for part in cand.get('content', {}).get('parts', []):
                        if text := part.get('text'):
                            yield text

Dual adapter — sticky fallback·python

class GeminiDualAdapter:
    def __init__(self, api_key: str, oauth_creds_path: str):
        self.api_key = api_key
        self.oauth_creds_path = oauth_creds_path
        self.fallback_active = False  # session-sticky

    async def stream(self, prompt: str):
        if not self.fallback_active:
            try:
                token   = get_access_token()  # from previous lesson
                project = load_code_assist(token)
                async for text in stream_gemini_oauth(
                    prompt, token, project,
                ):
                    yield text
                return
            except Exception as e:
                # Toast it visibly — never silent
                print(f'[OAuth failed: {e}. Falling back to API key, sticky for session.]')
                self.fallback_active = True

        # API key path
        async for text in stream_gemini_api_key(prompt, self.api_key):
            yield text

Exercise

Build the dual adapter from the third code block. Wire up both paths against your real credentials. Force a failure in the OAuth path (e.g., temporarily corrupt the credential file) and confirm: (a) fallback engages with a visible message, (b) subsequent calls in the same process stay on the API key path even after you fix the credentials.

Parse SSE by hand when the SDK isn't there

The dual-adapter pattern

Code

External links

Exercise

Progress

Comments 0