들어가며
Claude Agent SDK로 query()를 호출하면, Claude의 응답은 단순한 문자열이 아닙니다. 여러 종류의 메시지 객체가 순서대로 날아옵니다.
async for message in query(prompt="hello.py 만들어줘", options=options):
print(type(message)) # 이게 뭐가 올까?실행해보면 이런 타입들이 순서대로 출력됩니다:
<class 'AssistantMessage'> ← "파일을 작성하겠습니다" (텍스트)
<class 'AssistantMessage'> ← Write 도구 호출
<class 'AssistantMessage'> ← "완료했습니다" (텍스트)
<class 'ResultMessage'> ← 최종 결과 (비용, 턴 수, 종료 사유)이 글에서는 Claude가 보내는 모든 메시지 타입을 하나씩 뜯어보고, 실전에서 어떻게 처리하는지 정리합니다.
전체 메시지 구조 한눈에 보기
SDK가 정의하는 Message 타입은 6가지의 유니온입니다:
Message = (
UserMessage # 사용자 메시지
| AssistantMessage # Claude 응답
| SystemMessage # 시스템 이벤트
| ResultMessage # 최종 결과
| StreamEvent # 부분 스트리밍
| RateLimitEvent # 사용량 한도 이벤트
)그리고 AssistantMessage 안에는 콘텐츠 블록이 들어 있습니다:
ContentBlock = (
TextBlock # 텍스트
| ThinkingBlock # 사고 과정
| ToolUseBlock # 도구 호출
| ToolResultBlock # 도구 실행 결과
)이 두 계층을 이해하면 전체 구조가 보입니다.
query() 스트림
├── AssistantMessage
│ └── content: list[ContentBlock]
│ ├── TextBlock "파일을 작성하겠습니다"
│ ├── ThinkingBlock (사고 과정 — thinking 활성화 시)
│ ├── ToolUseBlock Write 도구 호출
│ └── ToolResultBlock 도구 실행 결과
├── SystemMessage 내부 상태 변화
├── RateLimitEvent 사용량 한도 변화
└── ResultMessage ← 항상 마지막1계층: 콘텐츠 블록 (ContentBlock)
AssistantMessage 안에 담기는 개별 조각들입니다. Claude의 한 번의 응답에 여러 블록이 섞여서 올 수 있습니다.
TextBlock — 텍스트 응답
가장 기본적인 블록입니다. Claude가 말하는 텍스트가 여기에 담깁니다.
@dataclass
class TextBlock:
text: str# 처리 예시
if isinstance(block, TextBlock):
print(block.text)
# → "안녕하세요! hello.py 파일을 작성하겠습니다."단순해 보이지만, 하나의 AssistantMessage에 TextBlock이 여러 개 올 수 있습니다. 도구 호출 전후로 텍스트가 나뉘어 오는 경우입니다.
ThinkingBlock — Claude의 사고 과정
ClaudeAgentOptions에서 thinking을 활성화하면, Claude가 답변을 생성하기 전에 내부적으로 생각하는 과정이 이 블록에 담깁니다.
@dataclass
class ThinkingBlock:
thinking: str # 사고 내용
signature: str # 검증용 서명# thinking 활성화
options = ClaudeAgentOptions(
thinking={"type": "enabled", "budget_tokens": 10000}
)
# 처리 예시
if isinstance(block, ThinkingBlock):
print(f"[사고 과정] {block.thinking[:100]}...")
# → "[사고 과정] 사용자가 hello.py를 만들어달라고 했다.
# 간단한 Hello World 스크립트를 작성하면..."핵심 포인트:
thinking이 비활성화(disabled)면 이 블록은 아예 오지 않음signature는 사고 과정의 무결성 검증용 — 일반적으로 무시해도 됨- 사고 과정은 응답 품질을 높이지만, 토큰을 소비하므로 비용과 트레이드오프
ToolUseBlock — 도구 호출 요청
Claude가 파일을 읽거나, 코드를 실행하거나, 웹을 검색하는 등 도구를 사용할 때 이 블록이 옵니다.
@dataclass
class ToolUseBlock:
id: str # 이 도구 호출의 고유 ID
name: str # 도구 이름 (Read, Write, Bash 등)
input: dict[str, Any] # 도구에 전달되는 입력값# 처리 예시
if isinstance(block, ToolUseBlock):
print(f"도구: {block.name}")
print(f"입력: {block.input}")
# → 도구: Write
# → 입력: {"file_path": "/app/workspace/hello.py", "content": "print('Hello!')"}주요 도구와 input 구조:
| 도구 | input 예시 |
|---|---|
Read | {"file_path": "/app/main.py"} |
Write | {"file_path": "/app/hello.py", "content": "..."} |
Edit | {"file_path": "/app/main.py", "old_string": "...", "new_string": "..."} |
Bash | {"command": "python hello.py"} |
Glob | {"pattern": "**/*.py"} |
Grep | {"pattern": "def main", "path": "/app"} |
핵심 포인트:
id는 이 호출을 식별하는 고유값 —ToolResultBlock의tool_use_id와 매칭됨- 하나의
AssistantMessage에 여러ToolUseBlock이 올 수 있음 (병렬 도구 호출)
ToolResultBlock — 도구 실행 결과
ToolUseBlock으로 요청된 도구가 실행된 후, 그 결과가 이 블록에 담깁니다.
@dataclass
class ToolResultBlock:
tool_use_id: str # 어떤 도구 호출의 결과인지
content: str | list[dict[str, Any]] | None = None # 실행 결과
is_error: bool | None = None # 에러 여부# 처리 예시
if isinstance(block, ToolResultBlock):
if block.is_error:
print(f"도구 실패: {block.content}")
else:
print(f"도구 성공: {block.content[:100]}...")핵심 포인트:
tool_use_id로 어떤ToolUseBlock에 대한 결과인지 매칭 가능is_error=True면 도구 실행이 실패한 것 (파일 없음, 명령 에러 등)content는 문자열 또는 복합 구조 — 도구에 따라 형식이 다름
콘텐츠 블록의 실제 흐름
Claude가 "hello.py 만들어줘"를 처리할 때, 하나의 AssistantMessage 안에 이런 블록들이 순서대로 담깁니다:
AssistantMessage.content = [
ThinkingBlock("사용자가 hello.py를 원한다..."), ← 사고 (선택적)
TextBlock("hello.py 파일을 작성하겠습니다."), ← 텍스트
ToolUseBlock(name="Write", input={...}), ← 도구 호출
]그리고 도구 실행 후 다음 AssistantMessage가 옵니다:
AssistantMessage.content = [
TextBlock("hello.py를 작성했습니다."), ← 결과 설명
]2계층: 메시지 타입 (Message)
AssistantMessage — Claude의 응답
가장 자주 마주치는 타입입니다. Claude가 보내는 모든 응답이 여기에 담깁니다.
@dataclass
class AssistantMessage:
content: list[ContentBlock] # 콘텐츠 블록 리스트
model: str # 사용된 모델
parent_tool_use_id: str | None = None # 서브 에이전트의 도구 호출 ID
error: AssistantMessageError | None = None # 에러 타입
usage: dict[str, Any] | None = None # 토큰 사용량
message_id: str | None = None # 메시지 고유 ID
stop_reason: str | None = None # 응답 중단 사유
session_id: str | None = None # 세션 ID
uuid: str | None = None # UUIDasync for message in query(prompt="hello.py 만들어줘", options=options):
if isinstance(message, AssistantMessage):
print(f"모델: {message.model}")
print(f"사용량: {message.usage}")
for block in message.content:
if isinstance(block, TextBlock):
print(f"텍스트: {block.text}")
elif isinstance(block, ToolUseBlock):
print(f"도구 호출: {block.name}({block.input})")error 필드의 가능한 값:
| 에러 | 의미 |
|---|---|
"authentication_failed" | 인증 실패 (토큰 만료 등) |
"billing_error" | 결제 오류 |
"rate_limit" | 사용량 한도 초과 |
"invalid_request" | 잘못된 요청 |
"server_error" | 서버 내부 오류 |
"unknown" | 알 수 없는 오류 |
stop_reason 필드의 주요 값:
| 값 | 의미 |
|---|---|
"end_turn" | 정상 종료 |
"max_tokens" | 최대 토큰 도달 |
"tool_use" | 도구 호출을 위해 중단 (이후 재개) |
ResultMessage — 최종 결과
항상 스트림의 마지막에 옵니다. 전체 실행에 대한 요약 정보를 담고 있습니다.
@dataclass
class ResultMessage:
subtype: str # 결과 타입
duration_ms: int # 전체 소요 시간 (ms)
duration_api_ms: int # API 호출 소요 시간 (ms)
is_error: bool # 에러로 종료되었는지
num_turns: int # 총 턴 수
session_id: str # 세션 ID
stop_reason: str | None = None # 종료 사유
total_cost_usd: float | None = None # 총 비용 (달러)
usage: dict[str, Any] | None = None # 토큰 사용량 요약
result: str | None = None # 최종 텍스트 결과
structured_output: Any = None # 구조화된 출력 (output_format 사용 시)
model_usage: dict[str, Any] | None = None # 모델별 사용량
permission_denials: list[Any] | None = None # 거부된 권한 요청 목록
errors: list[str] | None = None # 발생한 에러 목록
uuid: str | None = None # UUIDif isinstance(message, ResultMessage):
print(f"소요 시간: {message.duration_ms}ms")
print(f"API 시간: {message.duration_api_ms}ms")
print(f"총 턴 수: {message.num_turns}")
print(f"비용: ${message.total_cost_usd:.4f}")
print(f"종료 사유: {message.stop_reason}")
print(f"에러 여부: {message.is_error}")
if message.structured_output:
print(f"구조화 출력: {message.structured_output}")
if message.permission_denials:
print(f"권한 거부: {message.permission_denials}")핵심 포인트:
ResultMessage가 오면 스트림이 끝난 것 — 이후 메시지는 없음result는 Claude의 최종 텍스트 응답 (모든 턴의 결과를 합친 것)structured_output는output_format을 설정했을 때만 채워짐duration_msvsduration_api_ms: 전체 시간 vs 순수 API 호출 시간. 차이가 크면 도구 실행에 시간이 걸린 것permission_denials로 어떤 도구 호출이 거부되었는지 확인 가능
SystemMessage — 시스템 이벤트
SDK 내부의 상태 변화를 알려주는 메시지입니다. 직접 처리할 일은 많지 않지만, 서브 에이전트를 쓸 때 중요해집니다.
@dataclass
class SystemMessage:
subtype: str # 이벤트 종류
data: dict[str, Any] # 이벤트 데이터SystemMessage에는 3가지 특화된 서브클래스가 있습니다:
TaskStartedMessage — 서브태스크 시작
@dataclass
class TaskStartedMessage(SystemMessage):
task_id: str # 태스크 ID
description: str # 태스크 설명
uuid: str
session_id: str
tool_use_id: str | None = None
task_type: str | None = NoneTaskProgressMessage — 서브태스크 진행 중
@dataclass
class TaskProgressMessage(SystemMessage):
task_id: str
description: str
usage: TaskUsage # 현재까지의 토큰, 도구 사용, 시간
uuid: str
session_id: str
tool_use_id: str | None = None
last_tool_name: str | None = None # 마지막으로 사용한 도구TaskUsage는 이렇게 생겼습니다:
class TaskUsage(TypedDict):
total_tokens: int # 사용된 총 토큰
tool_uses: int # 도구 호출 횟수
duration_ms: int # 경과 시간TaskNotificationMessage — 서브태스크 완료/실패/중단
@dataclass
class TaskNotificationMessage(SystemMessage):
task_id: str
status: Literal["completed", "failed", "stopped"]
output_file: str # 결과 파일 경로
summary: str # 결과 요약
uuid: str
session_id: str
tool_use_id: str | None = None
usage: TaskUsage | None = None# 서브태스크 추적 예시
if isinstance(message, TaskStartedMessage):
print(f"태스크 시작: {message.description}")
elif isinstance(message, TaskProgressMessage):
print(f"태스크 진행: {message.usage['tool_uses']}개 도구 사용, "
f"{message.usage['total_tokens']} 토큰 소비")
elif isinstance(message, TaskNotificationMessage):
print(f"태스크 {message.status}: {message.summary}")핵심 포인트:
- 이 3가지는 모두
SystemMessage의 서브클래스 →isinstance(msg, SystemMessage)로도 잡힘 AgentDefinition으로 서브 에이전트를 정의했을 때 이 메시지들이 발생task_id로 여러 병렬 태스크를 구분 가능
UserMessage — 사용자 메시지
스트림에서 직접 올 일은 드물지만, 세션 재개(continue_conversation) 시 이전 대화의 사용자 메시지가 재생될 수 있습니다.
@dataclass
class UserMessage:
content: str | list[ContentBlock]
uuid: str | None = None
parent_tool_use_id: str | None = None
tool_use_result: dict[str, Any] | None = NoneStreamEvent — 부분 스트리밍
include_partial_messages=True로 설정하면, Claude가 토큰을 생성하는 도중의 부분 응답이 이 타입으로 옵니다.
@dataclass
class StreamEvent:
uuid: str
session_id: str
event: dict[str, Any] # Anthropic API의 원시 스트림 이벤트
parent_tool_use_id: str | None = Noneoptions = ClaudeAgentOptions(
include_partial_messages=True
)
async for message in query(prompt="설명해줘", options=options):
if isinstance(message, StreamEvent):
# 실시간 타이핑 효과
event_type = message.event.get("type")
if event_type == "content_block_delta":
delta = message.event.get("delta", {})
if delta.get("type") == "text_delta":
print(delta["text"], end="", flush=True)핵심 포인트:
- 기본값은
False— 활성화하지 않으면 이 타입은 아예 오지 않음 event는 Anthropic Messages API의 원시 스트림 이벤트 형식- 실시간 UI(채팅 인터페이스, 타이핑 효과)를 구현할 때 사용
- 최종 완성된 응답은 여전히
AssistantMessage로도 옴
RateLimitEvent — 사용량 한도 이벤트
Claude 구독 플랜의 사용량 한도 상태가 변할 때 발생합니다.
@dataclass
class RateLimitEvent:
rate_limit_info: RateLimitInfo
uuid: str
session_id: str
@dataclass
class RateLimitInfo:
status: Literal["allowed", "allowed_warning", "rejected"]
resets_at: int | None = None # 한도 초기화 Unix 타임스탬프
rate_limit_type: Literal[
"five_hour", "seven_day", "seven_day_opus",
"seven_day_sonnet", "overage"
] | None = None
utilization: float | None = None # 사용률 (0.0 ~ 1.0)
overage_status: str | None = None # 초과 사용 상태
overage_resets_at: int | None = None
overage_disabled_reason: str | None = None
raw: dict[str, Any] # 원시 데이터if isinstance(message, RateLimitEvent):
info = message.rate_limit_info
if info.status == "allowed_warning":
print(f"주의: 사용량 {info.utilization * 100:.0f}% 도달")
print(f"한도 초기화: {info.resets_at}")
elif info.status == "rejected":
print("사용량 한도 초과! 잠시 후 다시 시도하세요.")
print(f"초기화 시각: {info.resets_at}")status 값의 의미:
| 상태 | 의미 | 대응 |
|---|---|---|
"allowed" | 정상 범위 | 계속 사용 |
"allowed_warning" | 한도에 근접 | 사용자에게 경고 표시 |
"rejected" | 한도 초과 | 요청 중단, resets_at까지 대기 |
rate_limit_type 값의 의미:
| 타입 | 의미 |
|---|---|
"five_hour" | 5시간 단위 한도 |
"seven_day" | 7일 단위 한도 |
"seven_day_opus" | 7일 Opus 모델 전용 한도 |
"seven_day_sonnet" | 7일 Sonnet 모델 전용 한도 |
"overage" | 초과 사용량 (추가 과금) |
실전: 완전한 메시지 처리기 만들기
모든 타입을 한 곳에서 처리하는 예시입니다.
from claude_agent_sdk import (
query,
ClaudeAgentOptions,
AssistantMessage,
ResultMessage,
SystemMessage,
StreamEvent,
RateLimitEvent,
TaskStartedMessage,
TaskProgressMessage,
TaskNotificationMessage,
TextBlock,
ThinkingBlock,
ToolUseBlock,
ToolResultBlock,
)
async def process_query(prompt: str):
options = ClaudeAgentOptions(
permission_mode="bypassPermissions",
allowed_tools=["Read", "Write", "Edit", "Bash"],
cwd="/app/workspace",
)
async for message in query(prompt=prompt, options=options):
# ── Claude의 응답 ──
if isinstance(message, AssistantMessage):
if message.error:
print(f"[에러] {message.error}")
continue
for block in message.content:
if isinstance(block, TextBlock):
print(f"Claude: {block.text}")
elif isinstance(block, ThinkingBlock):
print(f"[사고] {block.thinking[:200]}...")
elif isinstance(block, ToolUseBlock):
print(f"[도구] {block.name}: {block.input}")
elif isinstance(block, ToolResultBlock):
status = "실패" if block.is_error else "성공"
print(f"[결과] ({status}) {str(block.content)[:200]}")
# ── 서브태스크 이벤트 ──
elif isinstance(message, TaskStartedMessage):
print(f"[태스크 시작] {message.description}")
elif isinstance(message, TaskProgressMessage):
print(f"[태스크 진행] 도구 {message.usage['tool_uses']}회 사용")
elif isinstance(message, TaskNotificationMessage):
print(f"[태스크 {message.status}] {message.summary}")
# ── 기타 시스템 메시지 ──
elif isinstance(message, SystemMessage):
print(f"[시스템] {message.subtype}")
# ── 사용량 한도 ──
elif isinstance(message, RateLimitEvent):
info = message.rate_limit_info
if info.status == "allowed_warning":
print(f"[경고] 사용량 {info.utilization * 100:.0f}%")
elif info.status == "rejected":
print(f"[한도 초과] {info.resets_at}에 초기화")
# ── 최종 결과 ── (항상 마지막)
elif isinstance(message, ResultMessage):
print(f"\n{'='*50}")
print(f"완료: {message.num_turns}턴, "
f"{message.duration_ms}ms, "
f"${message.total_cost_usd or 0:.4f}")
if message.is_error:
print(f"에러: {message.errors}")메시지 흐름 시각화
단순한 질문 ("피보나치 함수 알려줘")
AssistantMessage
└── TextBlock: "피보나치 함수는 다음과 같습니다..."
ResultMessage (1턴, 0.002$)파일 생성 ("hello.py 만들어줘")
AssistantMessage
├── TextBlock: "파일을 작성하겠습니다."
└── ToolUseBlock: Write(file_path="hello.py", content="...")
AssistantMessage
└── TextBlock: "hello.py를 작성했습니다."
ResultMessage (2턴, 0.005$)복잡한 작업 ("프로젝트 분석 후 버그 수정해줘")
AssistantMessage
├── ThinkingBlock: "프로젝트 구조를 먼저 파악해야..."
├── TextBlock: "프로젝트 구조를 살펴보겠습니다."
└── ToolUseBlock: Glob(pattern="**/*.py")
AssistantMessage
├── TextBlock: "main.py를 확인하겠습니다."
└── ToolUseBlock: Read(file_path="main.py")
AssistantMessage
├── ThinkingBlock: "42번째 줄에 off-by-one 에러가..."
├── TextBlock: "버그를 찾았습니다. 수정하겠습니다."
└── ToolUseBlock: Edit(file_path="main.py", old_string="...", new_string="...")
AssistantMessage
└── TextBlock: "수정 완료했습니다. 42번째 줄의..."
ResultMessage (4턴, 0.012$)서브 에이전트 포함 (agents 설정 시)
AssistantMessage
└── TextBlock: "코드 리뷰를 시작합니다."
TaskStartedMessage (task_id="abc", description="코드 리뷰")
TaskProgressMessage (tool_uses=3, total_tokens=5000)
TaskProgressMessage (tool_uses=7, total_tokens=12000)
TaskNotificationMessage (status="completed", summary="3개 이슈 발견")
AssistantMessage
└── TextBlock: "리뷰 결과, 3개 이슈를 발견했습니다..."
ResultMessage자주 하는 실수와 해결법
실수 1: AssistantMessage.content가 리스트인 걸 잊음
# 잘못된 코드
if isinstance(message, AssistantMessage):
print(message.content) # → [TextBlock(...), ToolUseBlock(...)]
# 올바른 코드
if isinstance(message, AssistantMessage):
for block in message.content:
if isinstance(block, TextBlock):
print(block.text)실수 2: ResultMessage를 무시함
# ResultMessage에는 중요한 정보가 있다
if isinstance(message, ResultMessage):
if message.is_error:
# 에러로 종료됨 — result 텍스트만 봤으면 모를 수 있음
print(f"에러: {message.errors}")
if message.permission_denials:
# 도구 호출이 거부됨 — 결과가 불완전할 수 있음
print(f"거부된 요청: {message.permission_denials}")실수 3: TaskNotificationMessage를 SystemMessage로만 처리
# TaskNotificationMessage는 SystemMessage의 서브클래스
# isinstance 순서가 중요하다!
# 잘못된 순서 — TaskNotificationMessage가 SystemMessage에 먼저 잡힘
if isinstance(message, SystemMessage):
print("시스템 메시지") # TaskNotificationMessage도 여기로 옴
elif isinstance(message, TaskNotificationMessage):
print("태스크 완료") # 여기에 도달하지 않음!
# 올바른 순서 — 서브클래스를 먼저 체크
if isinstance(message, TaskNotificationMessage):
print(f"태스크 {message.status}")
elif isinstance(message, SystemMessage):
print("기타 시스템 메시지")마치며
Claude Agent SDK의 메시지 구조를 정리하면 두 가지만 기억하면 됩니다:
AssistantMessage의content는 리스트다 — 항상for block in message.content로 순회ResultMessage는 항상 마지막에 온다 — 비용, 에러, 종료 사유를 여기서 확인
이 두 가지를 이해하면, 이후 훅 시스템, 권한 제어, 양방향 클라이언트 등 모든 고급 기능에서 메시지를 자유롭게 다룰 수 있습니다.
전체 소스 코드는 GitHub에서 확인할 수 있습니다:
👉 GitHub 레포지토리 링크
꿈꾸는 개발자