세밀한 도구 스트리밍 (Fine-Grained Tool Streaming)¶
지연에 민감한 애플리케이션을 위해 서버 측 JSON 버퍼링 없이 도구 입력을 스트리밍해요.
세밀한 도구 스트리밍은 Claude가 도구의 입력을 생성하는 그대로, 서버 측 버퍼링이나 JSON 검증 없이 클라이언트로 전달해요. 이 버퍼링 단계를 건너뛰면 큰 파라미터(문서나 코드 블록 같은)의 첫 번째 조각까지 걸리는 시간이 줄어들어요. 조각들은 표준 도구 사용과 같은 Streaming messages 이벤트를 통해 도착해요.
세밀한 도구 스트리밍 사용하기¶
모든 모델이 Claude API, Amazon Bedrock, AWS의 Claude Platform, Google Cloud, Microsoft Foundry에서 세밀한 도구 스트리밍을 지원해요. 사용하려면 세밀한 스트리밍을 켜고 싶은 사용자 정의 도구에 eager_input_streaming을 true로 설정하고, 요청에서 스트리밍을 활성화하면 돼요.
eager_input_streaming 필드는 선택 사항이에요. true로 설정하면 그 도구에 세밀한 스트리밍이 켜지고, 생략하면 표준 버퍼링 스트리밍(API가 각 파라미터 값을 스트리밍하기 전에 버퍼링하고 검증하는 방식)을 사용해요. 예외가 하나 있는데, 레거시 fine-grained-tool-streaming-2025-05-14 beta 헤더를 보내는 요청이면 필드를 설정하지 않은 도구에 세밀한 스트리밍이 켜져요. 도구별 필드가 그 헤더를 대체하며, 명시적 false는 요청이 그 헤더를 여전히 보내더라도 해당 도구에 버퍼링 스트리밍을 유지해요. 레거시 헤더는 computer use나 browser use 도구셋 항목과 함께 쓸 수 없어요. 둘 다 보내는 요청은 API가 거부하므로, 헤더를 제거하고 필요한 사용자 정의 도구에 eager_input_streaming을 설정해야 해요. 필드 정의는 Tool reference를 참고하세요.
다음 예시는 make_file 도구에 세밀한 스트리밍을 켜고 Claude에게 긴 시를 요청해서, 도구 입력이 커서 스트리밍되는 걸 지켜볼 수 있게 한 거예요.
client = anthropic.Anthropic()
with client.messages.stream(
max_tokens=65536,
model="claude-opus-5",
tools=[
{
"name": "make_file",
"description": "Write text to a file",
"eager_input_streaming": True,
"input_schema": {
"type": "object",
"properties": {
"filename": {
"type": "string",
"description": "The filename to write text to",
},
"lines_of_text": {
"type": "array",
"description": "An array of lines of text to write to the file",
},
},
"required": ["filename", "lines_of_text"],
},
}
],
messages=[
{
"role": "user",
"content": "Can you write a long poem and make a file called poem.txt?",
}
],
) as stream:
for event in stream:
if event.type == "input_json":
print(event.partial_json, end="", flush=True)
final_message = stream.get_final_message()
print()
for block in final_message.content:
if block.type == "tool_use":
print(f"Complete tool input: {block.input}")
모든 탭이 make_file 도구에 세밀한 스트리밍을 켜요. SDK 탭은 각 입력 조각이 도착하는 즉시 출력하고, 스트림이 끝나면 누적된 완전한 입력을 출력해요. cURL 탭은 원시 이벤트 스트림을 보여주고, CLI 탭은 jq를 써서 조각만 출력해요. 출력된 조각들이 합쳐져 완전한 도구 입력이 되므로, Claude가 쓰는 대로 터미널에 시가 채워져요.
{"filename": "poem.txt", "lines_of_text": ["The Wanderer's Journey", "", "I.", "", "Beneath the vast and star-strewn sky,", "Where silver moonbeams softly lie,", ...
Complete tool input: {"filename": "poem.txt", "lines_of_text": ["The Wanderer's Journey", ...]}
eager_input_streaming이 없으면 API가 각 파라미터 값을 스트리밍하기 전에 버퍼링하고 검증하므로, 큰 파라미터는 Claude가 생성을 끝낼 때까지 아무것도 출력되지 않아요. 있으면 Claude가 파라미터를 시작하는 즉시 조각들이 도착하기 시작하고, 조각들은 보통 더 길며 단어 중간에서 끊기는 경우도 적어요.
도구 입력 델타 누적하기¶
누적 계약은 표준 도구 사용 스트리밍과 동일해서, 이 절은 eager_input_streaming이 있든 없든 적용돼요. 이벤트 형식은 Streaming messages의 Input JSON delta를 참고하세요. 세밀한 도구 스트리밍이 바꾸는 건 결과에 대해 가정할 수 있는 부분이에요. 서버가 조각들을 검증하지 않고 스트리밍하므로, 누적된 문자열이 유효한 JSON이 아닐 수 있어요.
tool_use 콘텐츠 블록이 스트리밍될 때, 처음 content_block_start 이벤트는 input: {}(빈 객체)를 담고 있어요. 이건 자리 표시자예요. 실제 입력은 각각 partial_json 문자열 조각을 담은 일련의 input_json_delta 이벤트로 도착해요. 완전한 입력을 조립하려면 이 조각들을 이어 붙이고 블록이 닫힐 때 그 결과를 파싱하면 돼요.
SDK가 누적 헬퍼를 제공하는 곳(앞 예시의 Python, TypeScript, Go, Java, Ruby 탭처럼)에서는 그게 이 작업을 대신 처리해요. 수동 패턴은 헬퍼가 없는 SDK, 또는 입력 조립 방식을 완전히 제어하고 싶을 때 써요.
누적 계약:
type: "tool_use"인content_block_start에서 빈 문자열로 초기화:input_json = ""type: "input_json_delta"인 각content_block_delta마다 이어 붙이기:input_json += event.delta.partial_jsoncontent_block_stop에서 누적된 문자열을 파싱
다음 SDK 예시처럼 파싱을 방어적으로 처리하세요. 파라미터 중간에서 max_tokens 때문에 응답이 멈출 수도 있어요. stop reason을 확인하고 더 높은 max_tokens로 요청을 다시 시도할지, 부분 입력을 복구할지 결정하세요.
초기 input: {}(객체)와 partial_json(문자열) 사이의 타입 불일치는 의도된 설계예요. 빈 객체는 콘텐츠 배열의 자리를 표시하고, 델타 문자열이 실제 값을 만들어요.
client = anthropic.Anthropic()
tool_inputs: dict[int, str] = {} # index -> accumulated JSON string
with client.messages.stream(
model="claude-opus-5",
max_tokens=1024,
tools=[
{
"name": "get_weather",
"description": "Get current weather for a city",
"eager_input_streaming": True,
"input_schema": {
"type": "object",
"properties": {"city": {"type": "string"}},
"required": ["city"],
},
}
],
messages=[{"role": "user", "content": "Weather in Paris?"}],
) as stream:
for event in stream:
match event.type:
case "content_block_start" if event.content_block.type == "tool_use":
tool_inputs[event.index] = ""
case "content_block_delta" if event.delta.type == "input_json_delta":
tool_inputs[event.index] += event.delta.partial_json
case "content_block_stop" if event.index in tool_inputs:
raw_input = tool_inputs[event.index]
try:
parsed = json.loads(raw_input)
except json.JSONDecodeError:
# The accumulated string is not guaranteed to be valid JSON.
# See "Handling invalid JSON in tool responses" on this page.
print(f"Invalid tool input: {raw_input}")
else:
print(f"Tool input: {parsed}")
도구 응답에서 잘못된 JSON 다루기¶
세밀한 도구 스트리밍에서는 도구 호출에 누적된 입력이 잘못되었거나 불완전한 JSON일 수 있어요. 그런 경우 도구를 실행할 수 없으니, 대신 실패를 Claude에게 알리면 돼요. 도구 결과의 content가 JSON일 필요는 없지만, 원시 문자열을 단일 키 아래 JSON 객체로 감싸면 잘못된 JSON을 받았다는 걸 Claude가 명확히 알 수 있고, 디버깅을 위해 원본 입력도 보존돼요.
이 래퍼를 문자열로 직렬화해서 tool result 콘텐츠 블록의 content로, is_error를 true로 설정해 반환하세요.
{
"type": "tool_result",
"tool_use_id": "toolu_01A09q90qw90lq917835lq9",
"is_error": true,
"content": "{\"INVALID_JSON\": \"<the unparseable input you received>\"}"
}
다음 단계¶
컨텍스트 윈도우가 어떻게 동작하는지, 확장 사고(extended thinking)와 도구 사용이 컨텍스트 윈도우에 어떻게 계산되는지, 대화가 길어질 때 컨텍스트를 어떻게 관리하는지 이해해요.
텍스트, 도구 사용, 확장 사고 델타를 포함해 server-sent events로 Messages API 응답을 점진적으로 스트리밍해요.
tool_use 블록을 파싱하고, tool_result 응답을 구성하며, is_error로 오류를 처리해요.
Anthropic 제공 도구 목록과 선택적 도구 정의 속성 참조.
이 페이지가 도움이 되었나요?