2059 단어
10 분
OpenAI Responses API SSE 리버싱 노트
AI 이 게시글은 AI가 생성한 콘텐츠를 포함합니다.
2026-06-20
리버싱 노트
태그 없음

OpenAI Responses API — SSE 스트리밍 내부 구조#

출처: OpenAI Python SDK openai-python (Stainless 생성) 범위: SSE (Server-Sent Events)만 다룸. WebSocket 모드 제외. 마지막 업데이트: 2026-06-20

1. 상위 수준 아키텍처#

flowchart TB
    subgraph User["User Code"]
        A["client.responses.create(stream=True)"]
    end
    subgraph HTTP["HTTP Layer"]
        B["POST /v1/responses<br/>Accept: text/event-stream<br/>body: { ..., 'stream': true }"]
    end
    subgraph SSE["SSE Decoding"]
        C["httpx.Response (청크 전송)"]
        D["Stream[ResponseStreamEvent]"]
        E["SSEDecoder.iter_bytes()"]
        F["ServerSentEvent.json()"]
        G["construct_type(discriminator='type')"]
    end
    subgraph Wrapper["SDK Wrapper"]
        H["ResponseStreamState"]
        I["ResponseStream (이터레이터)"]
    end
    A --> B --> C --> D
    C --> E --> F --> G --> D
    D --> H --> I

또는 편의 API 사용:

with client.responses.stream(input="...", model="gpt-4o") as stream:
    for event in stream:
        ...

내부적으로 ResponseStreamManager → ResponseStream을 사용한다.

2. SSE 와이어 포맷#

Assistants API(이벤트명 thread.message.delta 등으로 구분)와 달리, Responses API는 JSON 페이로드 내부에 type 필드로 이벤트 타입을 구분한다.

각 SSE 청크는 다음과 같다:

data: {"type":"response.created","response":{...},"sequence_number":0}


data: {"type":"response.in_progress","response":{...},"sequence_number":1}


data: {"type":"response.output_item.added","output_index":0,"item":{...},"sequence_number":2}


data: {"type":"response.content_part.added","output_index":0,"item_id":"...","content_index":0,"part":{...},"sequence_number":3}


data: {"type":"response.output_text.delta","output_index":0,"item_id":"...","content_index":0,"delta":"Hello","sequence_number":4}
...
  • Responses API SSE 이벤트에는 event: 라인이 설정되지 않는다.
  • JSON 본문 내부의 type 필드가 구분자(discriminator) 역할을 한다.
  • 모든 이벤트에는 순서 보장용 sequence_number: int가 있다.
  • 스트림은 연결 종료 시 끝난다(Responses에는 [DONE] 센티넬이 없음. Stream._iter_events의 [DONE] 체크는 레거시 Chat Completions 전용).

3. 전체 이벤트 목록 (53종)#

3.1 Response 생명주기 이벤트#

#type 값클래스필드
1response.createdResponseCreatedEventresponse: Response, sequence_number: int
2response.queuedResponseQueuedEventresponse: Response, sequence_number: int
3response.in_progressResponseInProgressEventresponse: Response, sequence_number: int
4response.completedResponseCompletedEventresponse: Response, sequence_number: int
5response.failedResponseFailedEventresponse: Response, sequence_number: int
6response.incompleteResponseIncompleteEventresponse: Response, sequence_number: int
7errorResponseErrorEventcode: Optional[str], message: str, param: Optional[str], sequence_number: int

3.2 출력 항목 — 구조적#

#type 값클래스필드
8response.output_item.addedResponseOutputItemAddedEventitem: ResponseOutputItem, output_index: int, sequence_number: int
9response.output_item.doneResponseOutputItemDoneEventitem: ResponseOutputItem, output_index: int, sequence_number: int
10response.content_part.addedResponseContentPartAddedEventpart: Part, content_index: int, item_id: str, output_index: int, sequence_number: int
11response.content_part.doneResponseContentPartDoneEventpart: Part, content_index: int, item_id: str, output_index: int, sequence_number: int

Part는 다음 중 하나의 구분된 유니온:

  • ResponseOutputText (type=output_text)
  • ResponseOutputRefusal (type=refusal)
  • PartReasoningText (type=reasoning_text)

3.3 텍스트 콘텐츠#

#type 값클래스필드
12response.output_text.deltaResponseTextDeltaEventdelta: str, content_index: int, item_id: str, output_index: int, logprobs: List[Logprob], sequence_number: int
13response.output_text.doneResponseTextDoneEventtext: str, content_index: int, item_id: str, output_index: int, logprobs: List[Logprob], sequence_number: int
14response.output_text.annotation.addedResponseOutputTextAnnotationAddedEventannotation: object, annotation_index: int, content_index: int, item_id: str, output_index: int, sequence_number: int
logprob 구조:
class Logprob:
    token: str
    logprob: float
    top_logprobs: Optional[List[LogprobTopLogprob]]  # 최대 20개 대안


class LogprobTopLogprob:
    token: Optional[str]
    logprob: Optional[float]

3.4 거절(Refusal)#

#type 값클래스필드
15response.refusal.deltaResponseRefusalDeltaEventdelta: str, content_index: int, item_id: str, output_index: int, sequence_number: int
16response.refusal.doneResponseRefusalDoneEventrefusal: str, content_index: int, item_id: str, output_index: int, sequence_number: int

3.5 함수 호출(Function Calling)#

#type 값클래스필드
17response.function_call_arguments.deltaResponseFunctionCallArgumentsDeltaEventdelta: str, item_id: str, output_index: int, sequence_number: int
18response.function_call_arguments.doneResponseFunctionCallArgumentsDoneEventarguments: str, name: str, item_id: str, output_index: int, sequence_number: int

3.6 MCP (Model Context Protocol)#

#type 값클래스필드
19response.mcp_call.in_progressResponseMcpCallInProgressEventitem_id: str, output_index: int, sequence_number: int
20response.mcp_call_arguments.deltaResponseMcpCallArgumentsDeltaEventdelta: str (JSON 부분), item_id: str, output_index: int, sequence_number: int
21response.mcp_call_arguments.doneResponseMcpCallArgumentsDoneEventarguments: str (JSON 전체), item_id: str, output_index: int, sequence_number: int
22response.mcp_call.completedResponseMcpCallCompletedEventitem_id: str, output_index: int, sequence_number: int
23response.mcp_call.failedResponseMcpCallFailedEventitem_id: str, output_index: int, sequence_number: int
24response.mcp_list_tools.in_progressResponseMcpListToolsInProgressEventitem_id: str, output_index: int, sequence_number: int
25response.mcp_list_tools.completedResponseMcpListToolsCompletedEventitem_id: str, output_index: int, sequence_number: int
26response.mcp_list_tools.failedResponseMcpListToolsFailedEventitem_id: str, output_index: int, sequence_number: int

3.7 오디오#

#type 값클래스필드
27response.audio.deltaResponseAudioDeltaEventdelta: str (Base64 오디오 바이트), sequence_number: int
28response.audio.doneResponseAudioDoneEventsequence_number: int
29response.audio.transcript.deltaResponseAudioTranscriptDeltaEventdelta: str, sequence_number: int
30response.audio.transcript.doneResponseAudioTranscriptDoneEventsequence_number: int

3.8 웹 검색 도구#

#type 값클래스필드
31response.web_search_call.in_progressResponseWebSearchCallInProgressEventitem_id: str, output_index: int, sequence_number: int
32response.web_search_call.searchingResponseWebSearchCallSearchingEventitem_id: str, output_index: int, sequence_number: int
33response.web_search_call.completedResponseWebSearchCallCompletedEventitem_id: str, output_index: int, sequence_number: int

3.9 파일 검색 도구#

#type 값클래스필드
34response.file_search_call.in_progressResponseFileSearchCallInProgressEventitem_id: str, output_index: int, sequence_number: int
35response.file_search_call.searchingResponseFileSearchCallSearchingEventitem_id: str, output_index: int, sequence_number: int
36response.file_search_call.completedResponseFileSearchCallCompletedEventitem_id: str, output_index: int, sequence_number: int

3.10 코드 인터프리터 도구#

#type 값클래스필드
37response.code_interpreter_call.in_progressResponseCodeInterpreterCallInProgressEventitem_id: str, output_index: int, sequence_number: int
38response.code_interpreter_call.interpretingResponseCodeInterpreterCallInterpretingEventitem_id: str, output_index: int, sequence_number: int
39response.code_interpreter_call_code.deltaResponseCodeInterpreterCallCodeDeltaEventdelta: str, item_id: str, output_index: int, sequence_number: int
40response.code_interpreter_call_code.doneResponseCodeInterpreterCallCodeDoneEventcode: str, item_id: str, output_index: int, sequence_number: int
41response.code_interpreter_call.completedResponseCodeInterpreterCallCompletedEventitem_id: str, output_index: int, sequence_number: int

3.11 추론(Reasoning)#

#type 값클래스필드
42response.reasoning_text.deltaResponseReasoningTextDeltaEventdelta: str, content_index: int, item_id: str, output_index: int, sequence_number: int
43response.reasoning_text.doneResponseReasoningTextDoneEventtext: str, content_index: int, item_id: str, output_index: int, sequence_number: int
44response.reasoning_summary_part.addedResponseReasoningSummaryPartAddedEventpart: Part (summary_text), summary_index: int, item_id: str, output_index: int, sequence_number: int
45response.reasoning_summary_part.doneResponseReasoningSummaryPartDoneEventpart: Part (summary_text), summary_index: int, item_id: str, output_index: int, sequence_number: int
46response.reasoning_summary_text.deltaResponseReasoningSummaryTextDeltaEventdelta: str, summary_index: int, item_id: str, output_index: int, sequence_number: int
47response.reasoning_summary_text.doneResponseReasoningSummaryTextDoneEventtext: str, summary_index: int, item_id: str, output_index: int, sequence_number: int

3.12 이미지 생성#

#type 값클래스필드
48response.image_generation_call.in_progressResponseImageGenCallInProgressEventitem_id: str, output_index: int, sequence_number: int
49response.image_generation_call.generatingResponseImageGenCallGeneratingEventitem_id: str, output_index: int, sequence_number: int
50response.image_generation_call.partial_imageResponseImageGenCallPartialImageEventpartial_image_b64: str, partial_image_index: int (0부터 시작), item_id: str, output_index: int, sequence_number: int
51response.image_generation_call.completedResponseImageGenCallCompletedEventitem_id: str, output_index: int, sequence_number: int

3.13 사용자 정의 도구 호출#

#type 값클래스필드
52response.custom_tool_call_input.deltaResponseCustomToolCallInputDeltaEventdelta: str, item_id: str, output_index: int, sequence_number: int
53response.custom_tool_call_input.doneResponseCustomToolCallInputDoneEventinput: str, item_id: str, output_index: int, sequence_number: int

4. SSE 파싱 파이프라인 (와이어 → 타입화된 이벤트)#

flowchart LR
    subgraph Wire["HTTP Wire"]
        A["httpx.Response<br/>iter_bytes()"]
    end
    subgraph Decoder["SSE Decoder"]
        B["SSEDecoder.decode(line)<br/>RFC 8895"]
        C["ServerSentEvent<br/>.data = JSON 문자열"]
    end
    subgraph Dispatch["Type Dispatch"]
        D["construct_type()<br/>discriminator='type'"]
    end
    subgraph Output["Output"]
        E["ResponseCreatedEvent |<br/>ResponseTextDeltaEvent |<br/>... (53개 타입화된 모델)"]
    end
    A -- 바이트 --> B -- SSE 라인 --> C -- .json() dict --> D --> E

디코더 주요 상세#

  • SSEDecoder는 빈 줄(\n\n, \r\r, \r\n\r\n)을 만날 때까지 라인을 누적한다.
  • 각 SSE 필드(event:, data:, id:, retry:)는 스펙에 따라 처리된다.
  • Responses API에서는 event: 필드가 없으므로 ServerSentEvent.event는 None이다.
  • 원본 JSON data:는 역직렬화되어 construct_type()에 전달된다.
  • ResponseStreamEvent는 Annotated[Union[...], PropertyInfo(discriminator="type")]이므로, construct_type이 "type" 키를 확인하여 올바른 서브클래스로 디스패치한다.

5. SDK 편의 API: ResponseStream / ResponseStreamState#

5.1 ResponseStreamManager (진입점)#

사용법:

with client.responses.stream(
    input="Hello", model="gpt-4o"
) as stream:  # ResponseStream[TextFormatT] 반환
    for event in stream:
        ...

매니저는:

  1. client.responses.create(stream=True, ...)를 호출하여 원본 Stream[ResponseStreamEvent]를 얻는다.
  2. ResponseStream(raw_stream=..., text_format=..., input_tools=...)으로 래핑한다.
  3. ResponseStreamEvent[TextFormatT]의 이터레이터로 노출한다.

5.2 ResponseStream → ResponseStreamState#

ResponseStream.__stream__()은 각 원본 이벤트를 ResponseStreamState.handle_event(sse_event)에 위임한다.

**handle_event()**는 각 원본 SSE 이벤트에 대해 1~2개의 이벤트를 생성:

원본 SSE type방출되는 이벤트
response.output_text.deltaResponseTextDeltaEvent (snapshot 필드 추가 — 지금까지 누적된 텍스트)
response.output_text.doneResponseTextDoneEvent (parsed 필드, text_format 설정 시)
response.function_call_arguments.deltaResponseFunctionCallArgumentsDeltaEvent (snapshot — 지금까지 누적된 인자)
response.completedResponseCompletedEvent (response가 완전히 파싱된 ParsedResponse)
그 외그대로 통과

5.3 accumulate_event() — 상태 머신#

ResponseStreamState는 __current_snapshot (부분 ParsedResponse)을 유지한다:

  • response.created → event.response에서 초기 스냅샷 생성
  • response.output_item.added → snapshot.output[]에 항목 추가 (function_call, message, 일반 타입 처리)
  • response.content_part.added → 인덱싱된 출력 메시지에 콘텐츠 파트 추가
  • response.output_text.delta → 일치하는 output_text 콘텐츠의 텍스트에 delta 연결
  • response.function_call_arguments.delta → 일치하는 함수 호출의 인자에 delta 연결
  • response.completed → parse_response()로 완전히 파싱된 response 저장 (구조화된 출력 파싱 포함)

5.4 편의 메서드#

  • stream.get_final_response() → 스트림 전체를 소비하고 누적된 ParsedResponse 반환
  • stream.until_done() → 스트림이 완전히 소비될 때까지 블로킹
  • stream.close() → 내부 httpx response 종료

5.5 ResponseStreamEvent[TextFormatT] (SDK 래퍼 타입)#

다음은 원본 생성 타입을 서브클래싱하거나 추가 필드로 래핑한다:

원본 타입SDK 래퍼추가 필드
ResponseTextDeltaEventlib.streaming.responses.ResponseTextDeltaEventsnapshot: str (누적 텍스트)
ResponseTextDoneEventlib.streaming.responses.ResponseTextDoneEventparsed: Optional[TextFormatT]
ResponseFunctionCallArgumentsDeltaEventlib.streaming.responses.ResponseFunctionCallArgumentsDeltaEventsnapshot: str (누적 인자)
ResponseCompletedEventlib.streaming.responses.ResponseCompletedEventresponse: ParsedResponse[TextFormatT]

스트림이 사용하는 복합 ResponseStreamEvent 타입 별칭:

ResponseStreamEvent = Annotated[Union[
    ResponseTextDeltaEvent,       # SDK 래퍼
    ResponseTextDoneEvent,        # SDK 래퍼
    ResponseFunctionCallArgumentsDeltaEvent,  # SDK 래퍼
    ResponseCompletedEvent,       # SDK 래퍼
    ResponseAudioDeltaEvent,      # 원본
    ... (그 외 모든 원본 이벤트)    # 원본
], PropertyInfo(discriminator="type")]

6. 백그라운드 모드 스트리밍#

responses.stream()에 background=True를 전달하면:

  1. API 요청에 "background": true가 포함된다.
  2. 서버는 즉시 response.created 이벤트로 response id를 반환한다.
  3. 스트림은 열린 상태로 유지되며, 백그라운드 처리 진행에 따라 비동기적으로 이벤트를 전달한다.
  4. 스트림을 중단하고 재개할 수 있다: response.created에서 response_id를 기록한 후, 나중에 responses.stream(response_id=id, starting_after=N)을 호출하여 시퀀스 번호 N부터 재개한다.

사용 패턴:

with client.responses.stream(input="...", model="gpt-4o", background=True) as stream:
    for event in stream:
        if event.type == "response.created":
            id = event.response.id
        if event.sequence_number == 10:
            break


# 나중에...
with client.responses.stream(response_id=id, starting_after=10) as stream:
    for event in stream:
        ...
    response = stream.get_final_response()

7. 데이터 흐름 요약#

flowchart TB
    subgraph HTTP["HTTP Layer"]
        A["HTTP POST /v1/responses<br/>{ stream: true }<br/>Content-Type: text/event-stream"]
    end
    subgraph SSE["SSE Decoder"]
        B["SSEDecoder<br/>빈 줄 단위 버퍼링<br/>event:/data: 파싱"]
    end
    subgraph Stream["Stream Layer"]
        C["Stream[ResponseStreamEvent]<br/>construct_type() 디스패치<br/>'type' discriminator 사용"]
    end
    subgraph Accumulator["Accumulator"]
        D["ResponseStreamState<br/>스냅샷 유지<br/>snapshot/parsed 추가"]
    end
    subgraph User["User Code"]
        E["for event in stream:"]
    end
    A --> B --> C --> D --> E

8. 와이어 레벨 예시 (합성)#

data: {"type":"response.created","response":{"id":"resp_abc123","object":"response","status":"queued","created_at":1718000000,"model":"gpt-4o","output":[],"tools":[],"tool_choice":"auto","parallel_tool_calls":true,"top_p":1,"temperature":1},"sequence_number":0}


data: {"type":"response.queued","response":{"id":"resp_abc123",...},"sequence_number":1}


data: {"type":"response.in_progress","response":{"id":"resp_abc123","status":"in_progress",...},"sequence_number":2}


data: {"type":"response.output_item.added","output_index":0,"item":{"id":"item_1","type":"message","role":"assistant","content":[],"status":"in_progress"},"sequence_number":3}


data: {"type":"response.content_part.added","output_index":0,"item_id":"item_1","content_index":0,"part":{"type":"output_text","text":"","annotations":[]},"sequence_number":4}


data: {"type":"response.output_text.delta","output_index":0,"item_id":"item_1","content_index":0,"delta":"Hello","sequence_number":5,"logprobs":[{"token":"Hello","logprob":-0.01}]}


data: {"type":"response.output_text.delta","output_index":0,"item_id":"item_1","content_index":0,"delta":" world","sequence_number":6,"logprobs":[...]}


data: {"type":"response.output_text.done","output_index":0,"item_id":"item_1","content_index":0,"text":"Hello world","sequence_number":7,"logprobs":[...]}


data: {"type":"response.content_part.done","output_index":0,"item_id":"item_1","content_index":0,"part":{"type":"output_text","text":"Hello world","annotations":[]},"sequence_number":8}


data: {"type":"response.output_item.done","output_index":0,"item":{"id":"item_1","type":"message","role":"assistant","content":[{"type":"output_text","text":"Hello world","annotations":[]}],"status":"completed"},"sequence_number":9}


data: {"type":"response.completed","response":{"id":"resp_abc123","status":"completed","output":[{"id":"item_1",...}],...},"sequence_number":10}

9. 구현 참조 (SDK 주요 파일)#

파일목적
src/openai/_streaming.pyStream[T], AsyncStream[T], SSEDecoder, SSEBytesDecoder, ServerSentEvent
src/openai/_response.pyAPIResponse, _parse() → Stream[T] 생성
src/openai/_base_client.pyrequest(), _process_response(), _process_response_data(), _make_sse_decoder(), make_request_options()
src/openai/types/responses/response_stream_event.pyResponseStreamEvent 구분된 유니온
src/openai/lib/streaming/responses/_responses.pyResponseStream, AsyncResponseStream, ResponseStreamManager, ResponseStreamState
src/openai/lib/streaming/responses/_events.pySDK 확장 래퍼 타입 + ResponseStreamEvent (라이브러리 버전)
src/openai/resources/responses/responses.pyResponses.create(), Responses.stream(), WebSocket 연결 코드

10. 이벤트 분류 다이어그램#

mindmap
  root(("Responses API SSE 이벤트<br/>(53종)"))
    생명주기
      response.created
      response.queued
      response.in_progress
      response.completed
      response.failed
      response.incomplete
      error
    출력 구조
      output_item.added
      output_item.done
      content_part.added
      content_part.done
    텍스트 콘텐츠
      output_text.delta
      output_text.done
      output_text.annotation.added
      refusal.delta
      refusal.done
    함수 호출
      function_call_arguments.delta
      function_call_arguments.done
    MCP
      mcp_call.in_progress
      mcp_call_arguments.delta
      mcp_call_arguments.done
      mcp_call.completed
      mcp_call.failed
      mcp_list_tools.in_progress
      mcp_list_tools.completed
      mcp_list_tools.failed
    오디오
      audio.delta
      audio.done
      audio.transcript.delta
      audio.transcript.done
    웹 검색
      web_search_call.in_progress
      web_search_call.searching
      web_search_call.completed
    파일 검색
      file_search_call.in_progress
      file_search_call.searching
      file_search_call.completed
    코드 인터프리터
      code_interpreter_call.in_progress
      code_interpreter_call.interpreting
      code_interpreter_call_code.delta
      code_interpreter_call_code.done
      code_interpreter_call.completed
    추론
      reasoning_text.delta
      reasoning_text.done
      reasoning_summary_part.added
      reasoning_summary_part.done
      reasoning_summary_text.delta
      reasoning_summary_text.done
    이미지 생성
      image_generation_call.in_progress
      image_generation_call.generating
      image_generation_call.partial_image
      image_generation_call.completed
    사용자 정의 도구 호출
      custom_tool_call_input.delta
      custom_tool_call_input.done

팁: SDK docstring에는 response.created만 문서화되어 있다. src/openai/types/responses/response_<name>_event.py의 타입 정의를 정식 출처로 사용하라. 모든 이벤트에는 type 리터럴(JSON의 type 필드와 일치), sequence_number: int, 그리고 이벤트별 필드가 있다.

OpenAI Responses API SSE Reversing Notes
Alice2 리버싱 노트
© 2026 concertypin. All Rights Reserved. / RSS / Sitemap
Powered by Astro & Fuwari