SSE 파싱과 이중 어댑터

SDK가 없을 때 SSE 직접 파싱하기

프록시를 만들거나 작은 컨테이너에서 돌리거나, 공식 SDK가 완전히 지원하지 않는 OAuth 전용 엔드포인트를 호출할 때는 호출 반복에서 SDK를 빼고 싶을 수 있어. SSE 형식은 단순해서 직접 파싱할 수 있어. 줄은 data:로 시작하고, 이벤트 사이는 빈 줄로 나뉘며, 접두사 뒤에는 JSON 페이로드가 와.

이중 어댑터 패턴

두 인증 경로를 하나의 인터페이스 뒤에 감춰. 호출자가 스트림을 요청하면 어댑터가 가용성과 최근 실패 기록을 바탕으로 API 키나 OAuth를 골라. 폴백은 세션에 고정해야 해. 예를 들어 401 때문에 OAuth에서 API 키로 한 번 전환했다면 세션이 끝날 때까지 API 키를 유지해. 그렇지 않으면 두 경로 사이를 계속 오가게 돼.

Code

SSE 파서 — 비동기 줄 단위 처리·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 버전 — 같은 흐름, 다른 외피·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

이중 어댑터 — 세션 고정 폴백·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

세 번째 코드 블록처럼 이중 어댑터를 만들고 실제 자격 증명으로 두 경로를 모두 연결해. 자격 증명 파일을 잠시 망가뜨리는 식으로 OAuth 경로를 강제로 실패시킨 뒤, (a) 폴백이 보이는 메시지와 함께 시작되는지, (b) 자격 증명을 고쳐도 같은 프로세스의 이후 호출은 API 키 경로에 남는지 확인해.

SDK가 없을 때 SSE 직접 파싱하기

이중 어댑터 패턴

Code

External links

Exercise

Progress

댓글 0