스트리밍 (Streaming)

Message를 만들 때 "stream": true로 설정하면, 응답을 서버 전송 이벤트(SSE, server-sent events)로 조금씩(증분) 내려받을 수 있어요. 텍스트는 물론 tool use, extended thinking 델타까지 포함해서요.

SDK에서 스트리밍하기

Python SDK와 TypeScript SDK는 여러 가지 방식으로 스트리밍을 지원해요. PHP SDK는 createStream()으로 스트리밍을 제공하고, Python SDK는 동기(sync)와 비동기(async) 스트림을 모두 쓸 수 있어요. 각 SDK의 자세한 방법은 해당 SDK 문서를 확인해 주세요.

ant messages create --stream --format jsonl \
  --model claude-opus-5 \
  --max-tokens 1024 \
  --message '{role: user, content: "Hello"}' \
  | jq -rj 'select(.delta.type? == "text_delta") | .delta.text'

client = anthropic.Anthropic()

with client.messages.stream(
    max_tokens=1024,
    messages=[{"role": "user", "content": "Hello"}],
    model="claude-opus-5",
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)

const client = new Anthropic();

await client.messages
  .stream({
    messages: [{ role: "user", content: "Hello" }],
    model: "claude-opus-5",
    max_tokens: 1024
  })
  .on("text", (text) => {
    console.log(text);
  });

AnthropicClient client = new();

var parameters = new MessageCreateParams
{
    Model = Model.ClaudeOpus5,
    MaxTokens = 1024,
    Messages = [new() { Role = Role.User, Content = "Hello" }]
};

await foreach (var msg in client.Messages.CreateStreaming(parameters))
{
    Console.Write(msg);
}

client := anthropic.NewClient()

stream := client.Messages.NewStreaming(context.TODO(), anthropic.MessageNewParams{
	Model:     anthropic.ModelClaudeOpus5,
	MaxTokens: 1024,
	Messages: []anthropic.MessageParam{
		anthropic.NewUserMessage(anthropic.NewTextBlock("Hello")),
	},
})

for stream.Next() {
	event := stream.Current()
	switch eventVariant := event.AsAny().(type) {
	case anthropic.ContentBlockDeltaEvent:
		switch deltaVariant := eventVariant.Delta.AsAny().(type) {
		case anthropic.TextDelta:
			fmt.Print(deltaVariant.Text)
		}
	}
}
if err := stream.Err(); err != nil {
	log.Fatal(err)
}

AnthropicClient client = AnthropicOkHttpClient.fromEnv();

MessageCreateParams params = MessageCreateParams.builder()
    .model(Model.CLAUDE_OPUS_5)
    .maxTokens(1024L)
    .addUserMessage("Hello")
    .build();

try (var streamResponse = client.messages().createStreaming(params)) {
    streamResponse.stream().forEach(event -> {
        event.contentBlockDelta().ifPresent(deltaEvent ->
            deltaEvent.delta().text().ifPresent(td ->
                System.out.print(td.text())
            )
        );
    });
}

$client = new Client();

$stream = $client->messages->createStream(
    maxTokens: 1024,
    messages: [
        ['role' => 'user', 'content' => 'Hello']
    ],
    model: 'claude-opus-5',
);

foreach ($stream as $message) {
    echo $message;
}

client = Anthropic::Client.new

stream = client.messages.stream(
  model: "claude-opus-5",
  max_tokens: 1024,
  messages: [{ role: "user", content: "Hello" }]
)

도착하는 텍스트를 일일이 처리할 필요가 없다면

텍스트가 도착하는 대로 처리할 필요가 없다면, SDK가 내부적으로는 스트리밍을 쓰되 완전한 Message 객체를 돌려주는 방식을 제공해요. 반환되는 객체는 .create()가 주는 것과 동일해요. 특히 max_tokens 값이 큰 요청에서 유용한데, 그런 경우 SDK가 HTTP 타임아웃을 피하려면 스트리밍이 필수거든요.

.stream() 호출은 서버 전송 이벤트로 HTTP 연결을 계속 유지하고, 그다음 .get_final_message()(Python) 또는 .finalMessage()(TypeScript)가 모든 이벤트를 모아 완전한 Message 객체를 돌려줘요. Go에서는 스트림 루프 안에서 message.Accumulate(event)를 호출해 같은 완전한 Message를 만들고, Java에서는 MessageAccumulator.create()로 만든 다음 각 이벤트에 accumulator.accumulate(event)를 호출해요. C#에서는 스트림의 .Aggregate() 확장 메서드를 await 해서 완전한 Message를 얻거나, MessageContentAggregator를 .CollectAsync()에 넘겨 이벤트를 처리하면서 집계할 수 있어요. Ruby에서는 스트림의 .accumulated_message를 호출하고, PHP SDK에서는 스트림 이벤트를 직접 순회하며 응답을 누적해요.

이벤트 타입

각 서버 전송 이벤트는 이름 있는 이벤트 타입과 거기에 딸린 JSON 데이터를 담아요. 각 이벤트는 SSE 이벤트 이름(예: event: message_stop)을 가지면서, 데이터 안에도 그에 맞는 type을 포함해요.

각 스트림은 다음과 같은 흐름의 이벤트를 사용해요.

message_start: content가 빈 Message 객체를 담아요.
일련의 콘텐츠 블록. 각 블록은 content_block_start, 하나 이상의 content_block_delta 이벤트, content_block_stop 이벤트로 이뤄져요. 각 콘텐츠 블록은 최종 Message의 content 배열에서 자신의 위치를 가리키는 index를 가져요. 예외가 하나 있는데, 서버 측 폴백(server-side fallback) 응답에서는 각 모델 경계마다 fallback 콘텐츠 블록이 content_block_start와 content_block_stop 짝으로 도착하고, 그 사이에 델타는 없어요.
하나 이상의 message_delta 이벤트. 최종 Message 객체에 대한 최상위 변경을 나타내요.
마지막 message_stop 이벤트.

핑(ping) 이벤트

이벤트 스트림에는 ping 이벤트가 몇 개든 섞여 있을 수 있어요.

오류 이벤트

API는 가끔 이벤트 스트림 안에서 오류를 보내기도 해요. 예를 들어 사용량이 많은 시간대에는 overloaded_error를 받을 수 있는데, 비스트리밍 상황에서는 대개 HTTP 529에 해당해요.

기타 이벤트

버전 정책(versioning policy)에 따라 새로운 이벤트 타입이 추가될 수 있으니, 코드는 모르는 이벤트 타입을 우아하게 무시하도록 처리해야 해요.

콘텐츠 블록 델타 타입

각 content_block_delta 이벤트는 주어진 index의 content 블록을 갱신하는 타입의 delta를 담아요.

텍스트 델타

text 콘텐츠 블록 델타는 다음과 같은 형태예요.

입력 JSON 델타

tool_use 콘텐츠 블록의 델타는 블록의 input 필드에 대한 갱신에 해당해요. 최대한 세밀하게 지원하기 위해 델타는 *부분 JSON 문자열(partial JSON strings)*인 반면, 최종 tool_use.input은 항상 *객체(object)*예요.

덱타 문자열을 누적해 두고 content_block_stop 이벤트를 받으면 JSON을 한 번에 파싱하면 돼요. 이때 Pydantic 같은 라이브러리로 부분 JSON 파싱을 하거나, 누적된 증분 값을 접근하는 헬퍼를 제공하는 SDK를 쓰면 돼요.

tool_use 콘텐츠 블록 델타는 다음과 같은 형태예요.

참고: 현재 모델은 input에서 한 번에 완전한 key-value 속성 하나만 내보내는 걸 지원해요. 그래서 tool을 사용할 때는 모델이 작업하는 동안 스트리밍 이벤트 사이에 지연이 있을 수 있어요. input의 키와 값이 누적되면, 이 값들은 청크 단위의 부분 JSON으로 된 여러 content_block_delta 이벤트로 내보내져요. 이렇게 해서 앞으로 더 세밀한 모델에서도 그 형식을 자동으로 지원할 수 있게 되는 거예요.

씽킹(thinking) 델타

스트리밍을 켠 상태에서 thinking을 사용하면 thinking_delta 이벤트를 통해 씽킹 콘텐츠를 받아요. 이 델타들은 thinking 콘텐츠 블록의 thinking 필드에 해당해요.

씽킹 콘텐츠에서는 content_block_stop 이벤트 직전에 특별한 signature_delta 이벤트가 전송돼요. 이 서명은 씽킹 블록의 무결성을 검증하는 데 쓰여요.

씽킹 구성에서 display: "omitted"로 설정하면 thinking_delta 이벤트가 전송되지 않아요. 씽킹 블록은 열리고, signature_delta 하나만 받고, 닫혀요. 자세한 내용은 Controlling thinking display를 참고하세요.

일반적인 씽킹 델타는 다음과 같은 형태예요.

서명 델타는 다음과 같은 형태예요.

전체 HTTP 스트림 응답

스트리밍 모드에서는 클라이언트 SDK를 쓰는 걸 권장해요. 하지만 API를 직접 연동해 만들고 있다면 이 이벤트들을 직접 처리해야 해요.

스트림 응답은 다음으로 구성돼요.

message_start 이벤트
콘텐츠 블록이 여러 개일 수 있고, 각 블록은 다음을 포함해요.

content_block_start 이벤트
content_block_delta 이벤트가 여러 개일 수 있음
content_block_stop 이벤트

하나 이상의 message_delta 이벤트
message_stop 이벤트

응답 사이사이에 ping 이벤트가 섞여 있을 수도 있어요. 형식에 대한 더 자세한 내용은 Event types를 참고하세요.

기본 스트리밍 요청

tool use를 포함한 스트리밍 요청

이 요청은 Claude에게 tool을 사용해 날씨를 알려 달라고 하는 요청이에요.

thinking을 포함한 스트리밍 요청

이 요청은 스트리밍과 함께 thinking을 켜요. display: "summarized" 설정은 Claude의 추론 전체 사고 과정 대신 요약된 사고 내용을 스트리밍해요.

웹 검색 tool use를 포함한 스트리밍 요청

이 요청은 Claude에게 최신 날씨 정보를 위해 웹을 검색하라고 하는 요청이에요.

오류 복구

Claude 4.5 및 이전 모델

Claude 4.5 모델 및 그 이전 모델에서는 네트워크 문제, 타임아웃, 그 밖의 오류로 중단된 스트리밍 요청을, 중단된 지점부터 이어서 복구할 수 있어요. 이 방식은 전체 응답을 처음부터 다시 처리하지 않아도 되게 해 줘요.

기본 복구 전략은 다음과 같아요.

부분 응답 확보하기: 오류가 발생하기 전에 성공적으로 받은 모든 콘텐츠를 저장해요.
이어서 보낼 요청 구성하기: 부분 응답을 새 assistant 메시지의 시작 부분으로 넣어 새 API 요청을 만들어요.
스트리밍 재개하기: 중단된 지점부터 나머지 응답을 계속 받아요.

Claude 4.6 및 이후 모델

Claude 4.6 및 이후 모델에도 같은 저장 후 재개(capture-and-resume) 전략이 적용돼요. 다만 2단계가 달라져요. 부분 응답을 assistant 메시지에 넣는 대신, 모델에게 멈춘 지점부터 이어서 하라고 지시하는 user 메시지를 추가해요.

부분 응답 확보하기: 오류가 발생하기 전에 성공적으로 받은 모든 콘텐츠를 저장해요.
이어서 보낼 요청 구성하기: 부분 응답과 계속하라는 지시를 담은 user 메시지로 새 API 요청을 만들어요. 예를 들면 다음과 같아요.
스트리밍 재개하기: 중단된 지점부터 나머지 응답을 계속 받아요.

오류 복구 모범 사례

SDK 기능 활용하기: SDK가 내장한 메시지 누적과 오류 처리를 활용해요.
콘텐츠 타입 처리하기: 메시지는 여러 콘텐츠 블록(text, tool_use, thinking)을 담을 수 있다는 점을 기억하세요. tool use와 extended thinking 블록은 부분적으로 복구할 수 없어요. 가장 최근 텍스트 블록부터 스트리밍을 재개할 수 있답니다.

스트리밍 (Streaming)