세분화된 도구 스트리밍
지연 시간에 민감한 애플리케이션을 위해 서버 측 JSON 버퍼링 없이 도구 입력을 스트리밍합니다.
"Fine-grained tool streaming"(세분화된 도구 스트리밍)은 Claude가 도구의 입력을 생성하는 즉시 서버 측 버퍼링이나 JSON 검증 없이 클라이언트로 전달합니다. 버퍼링 단계를 건너뛰면 문서나 코드 블록과 같은 큰 매개변수의 첫 번째 조각이 도착하기까지의 시간이 줄어들며, 조각들은 표준 도구 사용과 동일한 메시지 스트리밍 이벤트를 통해 도착합니다.
세분화된 도구 스트리밍 사용 방법
모든 모델은 Claude API, Amazon Bedrock, Claude Platform on AWS, Google Cloud, Microsoft Foundry에서 세분화된 도구 스트리밍을 지원합니다. 이를 사용하려면 세분화된 스트리밍을 활성화하려는 사용자 정의 도구에서 eager_input_streaming을 true로 설정하고, 요청에서 스트리밍을 활성화하세요.
eager_input_streaming 필드는 선택 사항입니다. 이를 true로 설정하면 해당 도구에 대해 세분화된 스트리밍이 켜지고, 생략하면 API가 각 매개변수 값을 스트리밍하기 전에 버퍼링하고 검증하는 표준 버퍼링 스트리밍이 적용됩니다. 예외는 레거시 fine-grained-tool-streaming-2025-05-14 베타 헤더를 여전히 보내는 요청으로, 이 경우 필드를 설정하지 않은 도구에 대해 세분화된 스트리밍이 켜집니다. 도구별 필드는 해당 헤더를 대체하며, 명시적으로 false를 지정하면 요청이 여전히 헤더를 보내더라도 해당 도구에 대해 버퍼링 스트리밍이 유지됩니다. 레거시 헤더는 컴퓨터 사용 또는 브라우저 사용 도구 세트 항목과 함께 사용할 수 없습니다. API는 둘 다 보내는 요청을 거부하므로, 헤더를 제거하고 필요한 사용자 정의 도구에 eager_input_streaming을 설정하세요. 필드 정의는 도구 참조를 참조하세요.
다음 예제는 make_file 도구에 대해 세분화된 스트리밍을 켜고 Claude에게 긴 시를 요청하여, 도구 입력이 스트리밍되는 것을 지켜볼 수 있을 만큼 충분히 크게 만듭니다:
client = anthropic.Anthropic()
with client.messages.stream(
max_tokens=65536,
model="claude-opus-5-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 사용 여부와 관계없이 적용됩니다. 이벤트 형식은 메시지 스트리밍의 입력 JSON 델타를 참조하세요. 세분화된 도구 스트리밍은 결과에 대해 가정할 수 있는 것을 바꿉니다. 서버가 조각들을 검증하지 않고 스트리밍하므로, 누적된 문자열이 유효한 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에서 멈출 수도 있습니다. 중지 이유를 확인하고 더 높은 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-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:
# 누적된 문자열이 유효한 JSON이라는 보장은 없습니다.
# 이 페이지의 "도구 응답에서 잘못된 JSON 처리하기"를 참조하세요.
print(f"Invalid tool input: {raw_input}")
else:
print(f"Tool input: {parsed}")도구 응답에서 유효하지 않은 JSON 처리하기
세분화된 도구 스트리밍을 사용하면 도구 호출에 대해 누적된 입력이 유효하지 않거나 불완전한 JSON일 수 있습니다. 그런 경우 도구를 실행할 수 없으므로, 대신 실패를 Claude에게 보고하세요. 도구 결과의 content가 반드시 JSON일 필요는 없지만, 원시 문자열을 단일 키 아래의 JSON 객체로 감싸면 유효하지 않은 JSON을 받았다는 사실이 Claude에게 명확해지고, 디버깅을 위해 원본 입력이 보존됩니다:
{
"INVALID_JSON": "<the unparseable input you received>"
}문자열로 직렬화한 래퍼를 is_error가 true로 설정된 도구 결과 콘텐츠 블록의 content로 반환하세요:
{
"type": "tool_result",
"tool_use_id": "toolu_01A09q90qw90lq917835lq9",
"is_error": true,
"content": "{\"INVALID_JSON\": \"<the unparseable input you received>\"}"
}다음 단계
컨텍스트 윈도우가 어떻게 작동하는지, 확장 사고와 도구 사용이 어떻게 계산에 포함되는지, 대화가 길어짐에 따라 컨텍스트를 어떻게 관리하는지 이해하세요.
텍스트, 도구 사용, 확장 사고 델타를 포함하여 서버 전송 이벤트로 Messages API 응답을 점진적으로 스트리밍하세요.
tool_use 블록을 파싱하고, tool_result 응답을 형식화하며, is_error로 오류를 처리하세요.
Anthropic 제공 도구 디렉터리 및 선택적 도구 정의 속성에 대한 참조입니다.
Was this page helpful?