Messages API와 SDK

웹 대화창에서 제품 코드로 넘어가는 순간

앞 장까지의 프롬프트는 주로 사람이 Claude와 대화하는 장면을 다루었다. 이제부터는 상황이 달라진다. 사용자가 채팅창에 문장을 입력하는 대신, 여러분의 애플리케이션이 Claude에게 요청을 보내고, 응답을 받아 화면에 보여 주거나 데이터베이스에 저장하거나 다음 자동화 단계로 넘긴다. 이 전환점에서 가장 먼저 만나게 되는 것이 Messages API다.

Messages API를 한 문장으로 말하면, Claude와 주고받는 대화 턴을 개발자가 직접 구성해 모델을 호출하는 기본 인터페이스다. Anthropic 문서도 Messages API를 직접 모델 접근 방식으로 설명하며, 개발자가 대화 상태와 도구 루프를 직접 관리한다고 정리한다.¹ 여기서 중요한 단어는 직접이다. Claude.ai 웹 화면에서는 이전 대화가 자연스럽게 이어지는 것처럼 보이지만, API에서는 애플리케이션이 어떤 메시지를 보낼지, 어떤 규칙을 고정할지, 어떤 응답을 정상으로 볼지, 실패하면 어떻게 복구할지를 명시해야 한다.

이 장의 목표는 거창한 에이전트를 만드는 것이 아니다. 그보다 먼저 안정적인 첫 호출을 만들고, 요청과 응답의 구조를 정확히 읽고, 스트리밍과 오류 처리를 붙여 실제 서비스의 뼈대를 세우는 것이다. Anthropic Academy의 학습 자료도 API 시작, Messages API, 배치, 프롬프트 캐싱, 평가, 도구 사용 같은 흐름을 단계적으로 제시한다.² 이 장은 그중 가장 밑바닥에 해당하는 개발 기초를 다진다.

흐름은 단순하다.

사용자 또는 업무 이벤트
-> 애플리케이션 서버
-> Anthropic SDK
-> Messages API
-> Claude 모델
-> Message 응답
-> 파싱, 저장, 화면 출력, 후속 처리

이 구조를 제대로 이해하면 이후 장에서 다룰 tool use, RAG, 평가, 운영, 비용 최적화가 훨씬 쉬워진다. 반대로 이 구조를 대충 넘기면 나중에 같은 문제가 반복된다. 응답이 왜 잘렸는지 모른다. 비용이 왜 늘었는지 모른다. 레이트 리밋이 났을 때 무작정 재시도한다. 대화가 이어질 것이라고 착각해 이전 맥락을 보내지 않는다. API 통합의 실패는 대부분 모델이 똑똑하지 않아서가 아니라, 이 기본 접점이 느슨해서 생긴다.

SDK를 고르는 법: 직접 모델 호출부터 시작한다

Claude를 코드에서 호출하는 방법은 크게 두 갈래로 볼 수 있다. 하나는 이 장에서 다루는 Anthropic Client SDK다. Python, TypeScript를 비롯한 여러 언어에서 Messages API를 직접 호출하는 방식이다. 공식 SDK는 Python, TypeScript, Java, Go, Ruby, C#, PHP, CLI 등을 지원한다.³ 다른 하나는 다음 장에서 다룰 Agent SDK다. Agent SDK는 Claude Code의 도구 실행, 세션, 권한, MCP 연결 같은 에이전트 런타임을 프로그래밍 방식으로 쓰게 해 준다.¹⁰

처음에는 Client SDK로 시작하는 편이 좋다. 이유는 명확하다. 모델에 무엇을 보내고 무엇을 받는지 눈으로 확인할 수 있기 때문이다. 도구 실행과 에이전트 루프는 강력하지만, 기본 메시지 구조를 모르면 디버깅이 어려워진다. 초보 운전자가 먼저 핸들, 브레이크, 계기판을 익혀야 하는 것과 같다.

설치와 키 설정은 다음처럼 시작한다.

# Python
python -m venv .venv
source .venv/bin/activate
pip install anthropic python-dotenv

# TypeScript
npm install @anthropic-ai/sdk

# macOS 또는 Linux
export ANTHROPIC_API_KEY='sk-ant-...'
export CLAUDE_MODEL='claude-sonnet-4-6'

# Windows PowerShell
$env:ANTHROPIC_API_KEY='sk-ant-...'
$env:CLAUDE_MODEL='claude-sonnet-4-6'

모델 ID는 책에 적힌 예시를 그대로 외우는 값이 아니다. Claude 플랫폼은 모델명과 버전이 빠르게 바뀐다. 따라서 실제 운영 코드에서는 콘솔이나 공식 문서에 표시된 정확한 모델 ID를 설정값으로 관리해야 한다. 코드 곳곳에 모델명을 직접 박아 넣으면 모델 교체, 장애 대응, 비용 최적화, A/B 테스트가 모두 어려워진다.

API 키는 절대 프론트엔드 코드, 모바일 앱 바이너리, Git 저장소, Docker 이미지, 로그에 넣지 않는다. API 키는 서버에서만 사용한다. 로컬 개발에서는 환경 변수를 쓰고, 운영에서는 클라우드의 Secret Manager, Vault, CI/CD secret 같은 안전한 저장소를 사용한다. 키가 노출되면 사용량 폭주와 데이터 유출이 동시에 발생할 수 있다.

첫 호출: 요청, 응답, 토큰 사용량까지 본다

가장 작은 성공 경험부터 만들자. 아래 Python 코드는 Claude에게 한 번 질문하고, 텍스트 응답과 종료 이유, 토큰 사용량을 함께 출력한다.

import os
import anthropic

client = anthropic.Anthropic()
model = os.getenv('CLAUDE_MODEL', 'claude-sonnet-4-6')

message = client.messages.create(
    model=model,
    max_tokens=800,
    system='You are a concise technical assistant. Answer in Korean.',
    messages=[
        {
            'role': 'user',
            'content': 'Claude Messages API를 처음 배우는 개발자에게 5문장으로 설명해줘.',
        }
    ],
)

text = ''.join(
    block.text
    for block in message.content
    if block.type == 'text'
)

print(text)
print('---')
print('message id:', message.id)
print('stop reason:', message.stop_reason)
print('input tokens:', message.usage.input_tokens)
print('output tokens:', message.usage.output_tokens)

여기서 초보자가 가장 자주 하는 실수는 message.content[0].text만 출력하고 끝내는 것이다. 물론 학습용으로는 충분하다. 하지만 운영 코드라면 stop_reason과 usage를 반드시 읽어야 한다. stop_reason은 Claude가 왜 생성을 멈췄는지 알려 주고, usage는 입력과 출력에 사용된 토큰 수를 알려 준다. Messages API의 응답은 단순 문자열이 아니라 Message 객체이며, content, model, stop_reason, usage 같은 필드를 함께 제공한다.⁴

TypeScript에서도 같은 흐름이다.

import Anthropic from '@anthropic-ai/sdk'

const client = new Anthropic()
const model = process.env.CLAUDE_MODEL ?? 'claude-sonnet-4-6'

const message = await client.messages.create({
  model,
  max_tokens: 800,
  system: 'You are a concise technical assistant. Answer in Korean.',
  messages: [
    {
      role: 'user',
      content: 'Claude Messages API를 처음 배우는 개발자에게 5문장으로 설명해줘.',
    },
  ],
})

const text = message.content
  .filter((block) => block.type === 'text')
  .map((block) => block.text)
  .join('')

console.log(text)
console.log('---')
console.log('message id:', message.id)
console.log('stop reason:', message.stop_reason)
console.log('input tokens:', message.usage.input_tokens)
console.log('output tokens:', message.usage.output_tokens)

이 두 예제의 구조를 몸에 익혀 두자. 앞으로 도구 사용, 문서 분석, 검색, 인용, 구조화 출력으로 확장하더라도 기본 호출은 거의 이 모양에서 출발한다.

요청 객체 해부하기: model, system, messages, max_tokens

Messages API 요청은 겉보기에는 짧지만, 각 필드의 책임이 분명하다. 책임을 섞으면 프롬프트가 길어지고, 유지보수가 어려워지고, 보안 검토도 힘들어진다.

system과 messages의 분리는 특히 중요하다. system은 서비스의 헌법에 가깝다. 예를 들어 고객지원 봇이라면 말투, 금지 사항, 답변 형식, 모르는 내용을 추측하지 말라는 원칙을 system에 둔다. 반면 messages에는 사용자의 실제 문의와 이전 대화가 들어간다.

나쁜 설계는 이런 식이다.

사용자 메시지 안에 역할, 정책, 말투, 출력 형식, 실제 질문을 모두 섞어 넣는다.

좋은 설계는 이렇게 나눈다.

message = client.messages.create(
    model=model,
    max_tokens=700,
    system=(
        'You are a customer support assistant. '
        'Answer in Korean. If the policy is unclear, say that escalation is needed. '
        'Never invent refund rules.'
    ),
    messages=[
        {
            'role': 'user',
            'content': '배송이 5일째 멈춰 있는데 환불 가능한가요?',
        }
    ],
)

이렇게 나누면 정책 변경이 쉬워진다. 고객지원 톤을 바꾸고 싶으면 system을 고치면 된다. 사용자의 문의 처리 로직은 그대로 둘 수 있다. 반대로 모든 규칙을 user message에 섞어두면, 나중에 프롬프트 템플릿을 찾기도 어렵고 테스트하기도 어렵다.

messages 배열은 대화의 원장이다

messages 배열은 Claude와 주고받은 대화 턴을 순서대로 담는 원장이다. 각 항목에는 role과 content가 있다. 기본 role은 user와 assistant다. 사용자가 한 말은 user, Claude가 이전에 답한 말은 assistant로 넣는다.

Claude API는 기본적으로 stateless라고 이해하는 편이 안전하다. 즉, API 서버가 여러분의 이전 호출을 알아서 기억한다고 가정하면 안 된다. 대화를 이어가려면 애플리케이션이 이전 대화 중 필요한 부분을 다시 보내야 한다.⁴

history = [
    {
        'role': 'user',
        'content': 'Messages API가 뭐야?',
    },
    {
        'role': 'assistant',
        'content': 'Claude와 대화 턴을 주고받기 위한 API입니다.',
    },
    {
        'role': 'user',
        'content': '그럼 이전 대화는 자동으로 기억해?',
    },
]

message = client.messages.create(
    model=model,
    max_tokens=500,
    messages=history,
)

대화형 앱의 기본 루프는 다음과 같다.

사용자 입력을 history에 추가한다
-> Messages API를 호출한다
-> Claude 응답을 history에 추가한다
-> 다음 사용자 입력을 기다린다

이 루프는 간단하지만, 그대로 무한히 누적하면 입력 토큰이 계속 커진다. 그래서 운영 서비스에서는 최근 N개 메시지만 유지하거나, 오래된 대화를 요약하거나, 사실 정보는 RAG 저장소로 분리하거나, 고정 프롬프트는 캐싱하는 전략을 쓴다. 이 주제는 8장과 12장에서 더 깊게 다룬다. 지금은 한 가지만 기억하자. 대화 기억은 마법이 아니라 애플리케이션 상태 관리다.

content는 단순 문자열로 둘 수도 있고, content block 배열로 둘 수도 있다. 처음에는 문자열이면 충분하다. 하지만 이미지, 문서, 도구 사용으로 확장하면 content block 구조를 이해해야 한다.

messages = [
    {
        'role': 'user',
        'content': [
            {
                'type': 'text',
                'text': '아래 문서에서 계약 해지 조건만 추출해줘.',
            },
        ],
    }
]

content block은 Claude API가 텍스트뿐 아니라 이미지, 문서, 도구 호출 같은 다양한 입력과 출력을 하나의 메시지 구조 안에서 다루기 위한 기반이다. 이 장에서는 텍스트 블록만 사용하지만, 6장과 7장에서 이 구조가 본격적으로 중요해진다.

system prompt는 서비스의 고정 계약이다

프롬프트 엔지니어링 장에서 우리는 좋은 지시가 역할, 맥락, 제약, 출력 형식을 분리한다고 배웠다. API에서는 그 원칙이 더 중요해진다. Claude를 제품에 연결하면 프롬프트는 더 이상 개인 메모가 아니라 서비스 계약이 된다.

예를 들어 사내 문서 요약 도구를 만든다고 하자. 사용자는 매번 다른 문서를 올리고 다른 질문을 한다. 하지만 도구가 지켜야 할 규칙은 일정하다. 한국어로 답할 것, 문서에 없는 사실을 만들지 말 것, 불확실하면 불확실하다고 표시할 것, 개인정보가 있으면 원문 그대로 반복하지 말 것. 이런 규칙은 user message가 아니라 system에 들어가야 한다.

SYSTEM_PROMPT = '''
You are an internal document analysis assistant.
Answer in Korean.
Use only the information provided by the user.
If evidence is insufficient, say that the document does not contain enough information.
Do not reveal personal data unnecessarily.
Return concise bullet points only when there are three or more parallel items.
'''

message = client.messages.create(
    model=model,
    max_tokens=1000,
    system=SYSTEM_PROMPT,
    messages=[
        {
            'role': 'user',
            'content': '다음 회의록을 요약하고 결정 사항을 정리해줘. ...',
        }
    ],
)

system prompt를 코드 안에 직접 길게 쓰는 것은 학습용으로는 괜찮다. 운영에서는 별도 파일, 설정 저장소, 프롬프트 레지스트리로 분리하는 편이 좋다. 그래야 프롬프트 변경 이력을 남기고, 평가셋으로 회귀 테스트를 돌리고, 장애가 생겼을 때 어느 버전의 지시가 문제였는지 추적할 수 있다.

응답 객체를 읽는 법: content만 보지 말고 stop_reason을 본다

Claude 응답에서 가장 눈에 띄는 것은 텍스트다. 하지만 실제 제품은 텍스트만으로 굴러가지 않는다. 응답이 정상적으로 끝났는지, 중간에 잘렸는지, 도구 호출을 요구했는지, 토큰이 얼마나 들었는지를 함께 봐야 한다.

자주 만나는 stop_reason은 다음과 같다.

제품 코드에서는 end_turn만 성공으로 보고 나머지를 무시하면 안 된다. 특히 max_tokens는 조용히 품질 문제를 만든다. 사용자는 그럴듯한 답변을 받았다고 생각하지만, 실제로는 마지막 문단이나 결론이 잘렸을 수 있다.

def extract_text(message):
    return ''.join(
        block.text
        for block in message.content
        if block.type == 'text'
    )

message = client.messages.create(
    model=model,
    max_tokens=300,
    messages=[
        {
            'role': 'user',
            'content': '우리 회사의 AI 도입 보고서 목차를 자세히 작성해줘.',
        }
    ],
)

if message.stop_reason == 'max_tokens':
    print('응답이 출력 토큰 한도에 걸려 중단됐습니다. max_tokens를 늘리거나 작업을 나누세요.')
elif message.stop_reason != 'end_turn':
    print('추가 처리가 필요한 종료 상태:', message.stop_reason)

print(extract_text(message))

usage는 비용 최적화의 출발점이다. 어떤 기능에서 입력 토큰이 급증하는지, 어떤 프롬프트가 출력 토큰을 과도하게 쓰는지, 어떤 사용자가 반복 호출을 많이 하는지 알 수 있어야 한다. 단, 로그에 원문 프롬프트와 응답을 무조건 저장하는 것은 위험하다. 개인정보와 영업비밀이 섞일 수 있기 때문이다. 운영 로그에는 요청 ID, 모델, 토큰 수, 지연시간, 오류 유형, 프롬프트 버전처럼 진단에 필요한 메타데이터를 우선 남기고, 원문 저장은 별도 정책과 동의 절차를 둔다.

오류 처리: 재시도할 것과 즉시 실패할 것을 구분한다

AI API 오류는 일반 웹 API 오류보다 비용이 크다. 요청 하나가 긴 문서와 긴 출력을 포함할 수 있고, 스트리밍 중간에 실패할 수도 있으며, 재시도하면 다른 답변이 나올 수도 있다. 그래서 오류 처리는 단순히 try except로 감싸는 일이 아니다. 어떤 오류는 고쳐서 다시 보내야 하고, 어떤 오류는 기다렸다 다시 시도해야 하며, 어떤 오류는 사용자에게 명확히 안내해야 한다.

대표적인 분류는 다음과 같다. 공식 rate limit 문서는 한도를 넘으면 429 오류와 Retry-After 헤더를 반환한다고 설명한다.⁷ 오류 처리 자료들도 400, 401, 403, 429, 500, 529, 네트워크 오류를 서로 다른 방식으로 다룰 것을 권한다.⁸

아래 코드는 Python에서 재시도 가능한 오류와 즉시 실패해야 하는 오류를 나누는 기본 패턴이다.

import os
import time
import random
import anthropic
from anthropic import (
    Anthropic,
    APIConnectionError,
    APITimeoutError,
    APIStatusError,
    AuthenticationError,
    BadRequestError,
    InternalServerError,
    PermissionDeniedError,
    RateLimitError,
)

client = Anthropic(timeout=60.0)
model = os.getenv('CLAUDE_MODEL', 'claude-sonnet-4-6')


def retry_delay(attempt, base=1.0, cap=30.0):
    jitter = random.uniform(0, 0.5)
    return min(base * (2 ** attempt), cap) + jitter


def call_claude_with_retry(messages, system=None, max_tokens=800, max_retries=3):
    last_error = None

    for attempt in range(max_retries + 1):
        try:
            return client.messages.create(
                model=model,
                max_tokens=max_tokens,
                system=system,
                messages=messages,
            )

        except (AuthenticationError, PermissionDeniedError, BadRequestError) as exc:
            raise exc

        except RateLimitError as exc:
            last_error = exc
            retry_after = None
            if getattr(exc, 'response', None) is not None:
                retry_after = exc.response.headers.get('retry-after')
            delay = float(retry_after) if retry_after else retry_delay(attempt)

        except (APIConnectionError, APITimeoutError, InternalServerError) as exc:
            last_error = exc
            delay = retry_delay(attempt)

        except APIStatusError as exc:
            last_error = exc
            if exc.status_code and exc.status_code >= 500:
                delay = retry_delay(attempt)
            else:
                raise exc

        if attempt >= max_retries:
            break

        time.sleep(delay)

    raise RuntimeError(f'Claude call failed after {max_retries + 1} attempts') from last_error


response = call_claude_with_retry(
    messages=[
        {
            'role': 'user',
            'content': '레이트 리밋 오류를 사용자에게 어떻게 안내하면 좋을까?',
        }
    ],
    system='Answer in Korean. Be concise and practical.',
)

print(extract_text(response))

이 코드의 핵심은 모든 오류를 똑같이 보지 않는 것이다. API 키가 틀린 401은 기다려도 해결되지 않는다. 모델명이 틀린 400도 마찬가지다. 반면 429, 일시적 5xx, 네트워크 오류는 잠시 기다리면 성공할 수 있다. 운영에서는 여기에 큐, 사용자별 한도, 회로 차단기, 대체 모델, 관리자 알림을 붙인다. 이 주제는 12장에서 다시 만난다.

TypeScript에서도 같은 원칙을 적용한다.

import Anthropic from '@anthropic-ai/sdk'

const client = new Anthropic({ timeout: 60_000 })
const model = process.env.CLAUDE_MODEL ?? 'claude-sonnet-4-6'

function delay(ms: number) {
  return new Promise((resolve) => setTimeout(resolve, ms))
}

function backoff(attempt: number) {
  const base = Math.min(1000 * 2 ** attempt, 30000)
  const jitter = Math.floor(Math.random() * 500)
  return base + jitter
}

export async function callClaudeWithRetry(userText: string, maxRetries = 3) {
  let lastError: unknown

  for (let attempt = 0; attempt <= maxRetries; attempt++) {
    try {
      return await client.messages.create({
        model,
        max_tokens: 800,
        system: 'Answer in Korean. Be concise and practical.',
        messages: [{ role: 'user', content: userText }],
      })
    } catch (err) {
      lastError = err

      if (err instanceof Anthropic.APIError) {
        if ([400, 401, 403, 404, 413].includes(err.status ?? 0)) {
          throw err
        }
      }

      if (attempt >= maxRetries) break
      await delay(backoff(attempt))
    }
  }

  throw lastError
}

실제 서비스에서는 재시도 횟수를 무한대로 두지 않는다. 사용자가 기다릴 수 있는 시간은 제한되어 있고, 재시도도 비용이다. 채팅 UI라면 몇 초 안에 처리 상태를 보여 주고, 백그라운드 작업이라면 큐에 넣고 완료 알림을 보내는 편이 낫다.

스트리밍: 총시간보다 체감시간을 줄인다

일반 호출은 Claude가 전체 응답을 생성할 때까지 기다렸다가 한 번에 결과를 돌려준다. 짧은 분류나 추출에는 괜찮다. 하지만 긴 설명, 코드 생성, 보고서 초안, 문서 요약에서는 사용자가 빈 화면을 오래 바라보게 된다. 스트리밍은 생성되는 텍스트를 조각 단위로 받아 화면에 바로 보여 주는 방식이다. Claude 스트리밍은 Server-Sent Events 기반으로 동작하며, SDK가 대부분의 저수준 이벤트 처리를 감싸 준다.⁵ SSE는 서버가 클라이언트로 이벤트를 밀어 보내는 단방향 HTTP 스트림이며, 텍스트 응답을 실시간으로 보여 주는 데 잘 맞는다.⁶

스트리밍을 써야 하는 경우는 분명하다.

• 사람이 화면 앞에서 기다리는 대화형 기능이다.

• 응답 길이가 길거나 예측하기 어렵다.

• 사용자가 중간에 읽기 시작하거나 생성을 중단할 수 있어야 한다.

• 진행 중인 결과를 다른 프로세스나 UI에 점진적으로 전달해야 한다.

반대로 완성된 JSON을 파싱해야 하거나, 데이터베이스에 원자적으로 저장해야 하거나, 대량 문서를 배치로 처리하는 경우에는 일반 호출이나 Batch API가 더 단순할 수 있다. 스트리밍은 사용자 경험을 좋게 만들지만, 파싱과 오류 처리는 더 복잡해진다.

Python에서 가장 쉬운 스트리밍 코드는 다음과 같다.

import os
import anthropic

client = anthropic.Anthropic()
model = os.getenv('CLAUDE_MODEL', 'claude-sonnet-4-6')

with client.messages.stream(
    model=model,
    max_tokens=1200,
    system='Answer in Korean. Explain clearly for junior developers.',
    messages=[
        {
            'role': 'user',
            'content': 'API 스트리밍이 사용자 경험을 개선하는 이유를 설명해줘.',
        }
    ],
) as stream:
    for text in stream.text_stream:
        print(text, end='', flush=True)

    final_message = stream.get_final_message()

print()
print('stop reason:', final_message.stop_reason)
print('input tokens:', final_message.usage.input_tokens)
print('output tokens:', final_message.usage.output_tokens)

stream.text_stream은 텍스트 조각만 간편하게 순회하게 해 준다. 화면에 바로 출력하는 챗 UI라면 이 방식이 가장 쉽다. 하지만 raw event를 다뤄야 하는 경우도 있다. 예를 들어 content block 시작과 종료를 구분하거나, 나중에 tool use 스트림을 처리해야 하는 경우다.

with client.messages.stream(
    model=model,
    max_tokens=1200,
    messages=[{'role': 'user', 'content': 'Python의 async와 await를 설명해줘.'}],
) as stream:
    for event in stream:
        if event.type == 'content_block_delta' and event.delta.type == 'text_delta':
            print(event.delta.text, end='', flush=True)

TypeScript에서는 이벤트 리스너 또는 async iterator 방식으로 처리할 수 있다.

import Anthropic from '@anthropic-ai/sdk'

const client = new Anthropic()
const model = process.env.CLAUDE_MODEL ?? 'claude-sonnet-4-6'

const stream = client.messages.stream({
  model,
  max_tokens: 1200,
  system: 'Answer in Korean. Explain clearly for junior developers.',
  messages: [
    {
      role: 'user',
      content: 'API 스트리밍이 사용자 경험을 개선하는 이유를 설명해줘.',
    },
  ],
})

stream.on('text', (text) => {
  process.stdout.write(text)
})

const finalMessage = await stream.finalMessage()
console.log()
console.log('stop reason:', finalMessage.stop_reason)
console.log('input tokens:', finalMessage.usage.input_tokens)
console.log('output tokens:', finalMessage.usage.output_tokens)

또는 raw event를 순회할 수도 있다.

const eventStream = client.messages.stream({
  model,
  max_tokens: 1200,
  messages: [{ role: 'user', content: 'TypeScript의 async iterator를 설명해줘.' }],
})

for await (const event of eventStream) {
  if (event.type === 'content_block_delta' && event.delta.type === 'text_delta') {
    process.stdout.write(event.delta.text)
  }
}

스트리밍을 화면에 연결할 때는 세 가지를 조심해야 한다. 첫째, 첫 토큰이 오기 전에도 로딩 표시를 보여 줘야 한다. 사용자는 1초만 멈춰도 앱이 죽었다고 느낄 수 있다. 둘째, 모든 토큰마다 React 상태를 갱신하면 렌더링이 과도해질 수 있다. 일정 간격으로 묶어 업데이트하는 편이 좋다. 셋째, 스트림 중간에 오류가 날 수 있다. 이미 받은 부분 응답을 버릴지, 부분 결과로 보여 줄지, 처음부터 재시도할지 정책을 정해야 한다.

공식 스트리밍 문서는 일부 모델과 버전에서 네트워크 문제, 타임아웃, 기타 오류로 중단된 스트림을 이어받는 기능을 언급한다.⁵ 하지만 모든 상황에서 자동 복구된다고 가정하면 안 된다. 서비스 설계에서는 최종 메시지를 받았는지, done 이벤트가 왔는지, 사용자가 중단했는지, 연결이 끊겼는지를 별도로 기록해야 한다.

부분 응답을 어떻게 파싱할 것인가

스트리밍의 함정은 응답이 아직 완성되지 않았다는 데 있다. 사용자가 보는 텍스트는 자연스럽게 흘러가지만, 프로그램 입장에서 중간 조각은 불완전한 데이터다. 특히 JSON, Markdown, 코드 블록을 다룰 때 주의가 필요하다.

파싱 전략은 목적에 따라 달라진다.

• 단순 채팅 UI라면 텍스트 조각을 누적해 화면에 보여 주고, 완료 후 최종 메시지의 usage와 stop_reason을 기록한다.

• JSON 응답이 필요하다면 스트리밍 중에는 진행 표시만 보여 주고, 전체 문자열을 버퍼에 모은 뒤 완료 시점에 파싱하고 스키마 검증을 한다.

• Markdown 렌더링은 문단 단위나 블록 단위로 묶어 처리한다. 글자 단위 렌더링은 깨진 목록, 닫히지 않은 코드 블록, 깜박이는 레이아웃을 만들 수 있다.

• 사용자가 중단 버튼을 누를 수 있다면 AbortController나 서버 측 취소 신호를 연결하고, 부분 응답의 저장 여부를 명확히 정한다.

Python에서 JSON 최종 파싱을 염두에 둔 스트리밍은 이렇게 쓸 수 있다.

import json

buffer = []

with client.messages.stream(
    model=model,
    max_tokens=800,
    system='Return only a compact JSON object. No markdown.',
    messages=[
        {
            'role': 'user',
            'content': 'AI 도입 위험을 category, severity, reason 필드로 분류해줘.',
        }
    ],
) as stream:
    for text in stream.text_stream:
        buffer.append(text)
        print('.', end='', flush=True)

raw = ''.join(buffer)
data = json.loads(raw)
print(data)

이 예제는 학습용이다. 운영에서는 더 엄격한 스키마 검증이 필요하다. 프롬프트로 JSON을 요구해도 항상 완벽한 JSON이 나온다고 믿으면 안 된다. 구조화 출력과 스키마 검증은 4장과 9장에서 연결한 주제이며, 이후 장에서 다시 다룬다.

브라우저에서 직접 Claude를 호출하지 않는다

스트리밍을 웹 UI에 붙일 때 가장 위험한 실수는 브라우저에서 Claude API를 직접 호출하는 것이다. 그러면 API 키가 사용자에게 노출된다. 개발자 도구의 Network 탭, 번들 파일, 모바일 프록시만 봐도 키를 훔칠 수 있다.

정석은 백엔드 프록시 패턴이다.

브라우저
-> 여러분의 백엔드 API
-> Claude Messages API
-> 백엔드가 스트림을 다시 브라우저로 전달

백엔드는 사용자 인증을 확인하고, system prompt를 붙이고, 입력을 검증하고, Claude API를 호출하고, 결과를 브라우저로 흘려보낸다. 이때 브라우저에는 Claude API 키가 절대 내려가지 않는다. SSE를 그대로 프록시하거나, fetch의 ReadableStream으로 chunk를 읽어 UI에 표시할 수 있다.

TypeScript 백엔드에서 개념적으로는 다음 모양이 된다.

import Anthropic from '@anthropic-ai/sdk'

const client = new Anthropic()
const model = process.env.CLAUDE_MODEL ?? 'claude-sonnet-4-6'

export async function POST(req: Request) {
  const body = await req.json()
  const encoder = new TextEncoder()

  const readable = new ReadableStream({
    async start(controller) {
      const stream = client.messages.stream({
        model,
        max_tokens: 1000,
        system: 'You are a helpful assistant. Answer in Korean.',
        messages: [{ role: 'user', content: body.message }],
      })

      stream.on('text', (text) => {
        controller.enqueue(encoder.encode(text))
      })

      await stream.finalMessage()
      controller.close()
    },
  })

  return new Response(readable, {
    headers: {
      'Content-Type': 'text/plain; charset=utf-8',
      'Cache-Control': 'no-cache',
    },
  })
}

이 코드는 최소 예시다. 운영에서는 사용자 인증, 입력 길이 제한, 금칙어와 민감정보 필터링, 오류 이벤트, 타임아웃, 취소 처리, 로그, 비용 추적이 들어가야 한다. Nginx나 CDN 뒤에서 SSE를 쓴다면 버퍼링이 켜져 스트리밍이 한 번에 몰려 나오는 문제도 확인해야 한다. 스트리밍은 모델 호출보다 네트워크와 프록시 설정에서 더 많이 깨진다.

작은 CLI 챗봇으로 전체 흐름 묶기

이제 첫 호출, system prompt, 대화 이력, 스트리밍, 오류 처리를 하나의 작은 CLI로 묶어 보자. 아래 코드는 학습용이지만, 제품의 기본 구조와 닮아 있다.

import os
import time
import anthropic
from anthropic import RateLimitError, APIConnectionError, APITimeoutError

client = anthropic.Anthropic(timeout=60.0)
model = os.getenv('CLAUDE_MODEL', 'claude-sonnet-4-6')

SYSTEM = '''
You are a practical Claude API tutor.
Answer in Korean.
Keep answers concise unless the user asks for detail.
If you are uncertain, say so.
'''

history = []


def trim_history(messages, keep_last=8):
    return messages[-keep_last:]


def stream_reply(messages):
    for attempt in range(3):
        try:
            parts = []
            with client.messages.stream(
                model=model,
                max_tokens=900,
                system=SYSTEM,
                messages=messages,
            ) as stream:
                for text in stream.text_stream:
                    parts.append(text)
                    print(text, end='', flush=True)

                final_message = stream.get_final_message()

            print()
            return ''.join(parts), final_message

        except RateLimitError:
            wait = 2 ** attempt
            print(f'레이트 리밋입니다. {wait}초 후 재시도합니다.')
            time.sleep(wait)

        except (APIConnectionError, APITimeoutError):
            wait = 2 ** attempt
            print(f'연결 문제가 있습니다. {wait}초 후 재시도합니다.')
            time.sleep(wait)

    raise RuntimeError('응답을 가져오지 못했습니다.')


print('Claude CLI tutor입니다. 종료하려면 exit를 입력하세요.')

while True:
    user_text = input('nYou: ').strip()
    if user_text.lower() in {'exit', 'quit'}:
        break

    history.append({'role': 'user', 'content': user_text})
    request_messages = trim_history(history)

    print('Claude: ', end='', flush=True)
    assistant_text, final = stream_reply(request_messages)

    history.append({'role': 'assistant', 'content': assistant_text})

    if final.stop_reason == 'max_tokens':
        print('참고: 답변이 토큰 한도에 걸려 잘렸을 수 있습니다.')

    print(f'tokens in={final.usage.input_tokens}, out={final.usage.output_tokens}')

이 작은 프로그램에는 핵심 원칙이 모두 들어 있다. API 키는 환경 변수에서 읽는다. system prompt는 user message와 분리한다. history는 애플리케이션이 직접 관리한다. 스트리밍으로 체감 응답 속도를 높인다. 레이트 리밋과 연결 오류는 제한적으로 재시도한다. 최종 메시지에서 stop_reason과 usage를 읽는다.

다만 이 코드를 그대로 운영에 넣으면 안 된다. 운영에서는 사용자별 세션 저장소, 입력 검증, 개인정보 마스킹, 로그 정책, 예산 한도, 평가셋, 모니터링이 필요하다. 그래도 뼈대는 같다. API 통합은 복잡한 마법이 아니라, 작은 책임을 정확히 나누는 일이다.

공식 Quickstart를 읽는 법

Anthropic의 quickstart 예제들은 고객지원 에이전트, 금융 데이터 분석, 컴퓨터 사용 데모, 브라우저 도구, 자율 코딩 에이전트처럼 바로 실행 가능한 프로젝트 형태를 제공한다.⁹ 이런 예제를 볼 때는 코드 전체를 한꺼번에 이해하려고 하지 말고, 이 장에서 배운 렌즈로 나눠 읽으면 된다.

먼저 Claude를 호출하는 지점을 찾는다. client.messages.create인지, client.messages.stream인지 확인한다. 다음으로 system prompt와 messages가 어떻게 만들어지는지 본다. 그다음 응답을 어디에서 파싱하고, stop_reason과 usage를 남기는지 찾는다. 마지막으로 오류 처리와 재시도, 사용자 입력 검증, API 키 관리가 어디에 있는지 확인한다.

좋은 예제는 실행되는 것만으로 끝나지 않는다. 여러분의 제품에 들어갈 수 있는 구조를 보여 준다. 반대로 예제가 잘 돌아가도 API 키가 코드에 박혀 있거나, 오류 처리가 없거나, 대화 이력을 무한히 쌓는다면 운영 코드로 가져오기 전에 반드시 고쳐야 한다.

운영 코드로 가기 전 체크리스트

Messages API와 SDK를 배웠다면 이제 다음 질문을 스스로 던져야 한다.

• API 키가 서버 측 secret으로만 관리되는가?

• 모델명, max_tokens, temperature, timeout이 설정값으로 분리되어 있는가?

• system prompt와 user message의 책임이 분리되어 있는가?

• 대화 이력을 애플리케이션이 명시적으로 저장하고, 토큰 증가를 제어하는가?

• 응답에서 content뿐 아니라 stop_reason과 usage를 확인하는가?

• 400, 401, 403 같은 즉시 실패 오류와 429, 5xx, 네트워크 오류 같은 재시도 가능 오류를 구분하는가?

• 스트리밍 중간 실패, 사용자 취소, 최종 메시지 미수신을 처리하는가?

• 브라우저나 모바일 앱에 Claude API 키가 노출되지 않는가?

• 로그에 원문 개인정보나 비밀 정보가 과도하게 남지 않는가?

• 같은 프롬프트와 입력으로 회귀 테스트할 수 있는 최소 평가 샘플이 있는가?

이 체크리스트를 통과하면 Claude를 단순히 호출하는 수준을 넘어, 제품에 붙일 준비가 된 것이다. 다음 장에서는 이 기본 호출 위에 도구 사용, 에이전트, Claude Code, MCP를 얹는다. 그때부터 Claude는 답변만 하는 모델이 아니라, 파일을 읽고, 검색하고, 명령을 실행하고, 외부 시스템과 상호작용하는 작업자로 변한다. 하지만 그 모든 흐름의 바닥에는 여전히 이 장에서 배운 messages 배열, system prompt, content block, stop_reason, usage가 있다.

참고

¹Claude API Docs

²Anthropic Academy Claude API Development Guide

³Client SDKs

⁴Using the Messages API

⁵Streaming messages

⁶Claude Streaming API Guide

⁷Rate limits

⁸Claude API Error Handling

⁹Claude Quickstarts

¹⁰Agent SDK overview