스트리밍 입력
Claude Agent SDK의 두 가지 입력 모드를 이해하고 각각을 언제 사용할지 알아보기
개요
Claude Agent SDK는 에이전트와 상호작용하기 위한 두 가지 서로 다른 입력 모드를 지원합니다:
- 스트리밍 입력 모드: 지속적이고 대화형 세션
- 단일 메시지 입력: 세션 상태를 사용하고 재개하는 일회성 쿼리
스트리밍 입력 모드 (권장)
스트리밍 입력 모드는 Claude Agent SDK를 사용하는 선호되는 방식입니다. 에이전트의 기능에 대한 전체 액세스를 제공하고 풍부하고 대화형의 경험을 가능하게 합니다.
이를 통해 에이전트는 사용자 입력을 받아들이고, 중단을 처리하며, 권한 요청을 표시하고, 세션 관리를 처리하는 장기 실행 프로세스로 작동할 수 있습니다.
이점
스트리밍 입력 모드에서는 다음과 같은 기능을 갖춘 지속적인 세션에서 작업합니다:
- 이미지 업로드: 시각적 분석 및 이해를 위해 메시지에 직접 이미지를 첨부합니다
- 대기 중인 메시지: 순차적으로 처리되는 여러 메시지를 보내고 중단할 수 있습니다
- 도구 통합: 세션 중에 모든 도구 및 사용자 정의 MCP 서버에 대한 전체 액세스
- 실시간 피드백: 최종 결과뿐만 아니라 생성되는 응답을 실시간으로 확인합니다
- 컨텍스트 지속성: 여러 턴에 걸쳐 자연스럽게 대화 컨텍스트를 유지합니다
구현 예제
이 예제들은 작업 디렉토리에서 diagram.png라는 이미지를 읽습니다. 먼저 그곳에 하나를 생성하거나, 파일 이름을 변경하여 자신의 이미지를 가리키도록 합니다.
import { query, type SDKUserMessage } from "@anthropic-ai/claude-agent-sdk";
import { readFile } from "fs/promises";
async function* generateMessages(): AsyncGenerator<SDKUserMessage> {
// First message
yield {
type: "user",
message: {
role: "user",
content: "Analyze this codebase for security issues"
},
parent_tool_use_id: null
};
// Wait for conditions or user input
await new Promise((resolve) => setTimeout(resolve, 2000));
// Follow-up with image
yield {
type: "user",
message: {
role: "user",
content: [
{
type: "text",
text: "Review this architecture diagram"
},
{
type: "image",
source: {
type: "base64",
media_type: "image/png",
data: await readFile("diagram.png", "base64")
}
}
]
},
parent_tool_use_id: null
};
}
// Process streaming responses
for await (const message of query({
prompt: generateMessages(),
options: {
maxTurns: 10,
allowedTools: ["Read", "Grep"]
}
})) {
if (message.type === "result" && message.subtype === "success") {
console.log(message.result);
}
}
from claude_agent_sdk import (
ClaudeSDKClient,
ClaudeAgentOptions,
AssistantMessage,
TextBlock,
)
import asyncio
import base64
async def streaming_analysis():
async def message_generator():
# First message
yield {
"type": "user",
"message": {
"role": "user",
"content": "Analyze this codebase for security issues",
},
}
# Wait for conditions
await asyncio.sleep(2)
# Follow-up with image
with open("diagram.png", "rb") as f:
image_data = base64.b64encode(f.read()).decode()
yield {
"type": "user",
"message": {
"role": "user",
"content": [
{"type": "text", "text": "Review this architecture diagram"},
{
"type": "image",
"source": {
"type": "base64",
"media_type": "image/png",
"data": image_data,
},
},
],
},
}
# Use ClaudeSDKClient for streaming input
options = ClaudeAgentOptions(max_turns=10, allowed_tools=["Read", "Grep"])
async with ClaudeSDKClient(options) as client:
# Send streaming input
await client.query(message_generator())
# Process responses
async for message in client.receive_response():
if isinstance(message, AssistantMessage):
for block in message.content:
if isinstance(block, TextBlock):
print(block.text)
asyncio.run(streaming_analysis())
예제를 실행하면 TypeScript 버전은 완료될 때마다 각 응답을 출력합니다. Python 버전의 receive_response() 루프는 첫 번째 결과 메시지에서 끝나므로 보안 분석을 출력합니다. 두 응답을 모두 읽으려면 Python 참조의 대화 계속 예제에 표시된 대로 메시지당 하나의 query() 및 receive_response() 쌍을 사용합니다.
단일 메시지 입력
단일 메시지 입력은 더 간단하지만 더 제한적입니다.
단일 메시지 입력을 사용할 때
다음의 경우 단일 메시지 입력을 사용합니다:
- 일회성 응답이 필요한 경우
- 이미지 첨부 또는 중간 세션 제어 방법이 필요하지 않은 경우
- lambda 함수와 같은 상태 비저장 환경에서 작동해야 하는 경우
제한 사항
쿼리가 error_max_turns와 같은 오류 결과로 끝나면, 단일 메시지 query() 호출은 최종 결과 메시지를 생성한 후 실패 텍스트를 포함하는 오류를 발생시키므로, 코드가 계속 실행되어야 하는 경우 루프를 try 블록으로 감싸십시오. 결과 하위 유형에 대해서는 결과 처리를 참조하십시오.
구현 예제
import { query } from "@anthropic-ai/claude-agent-sdk";
// Simple one-shot query
// query() throws after an error result, such as error_max_turns
try {
for await (const message of query({
prompt: "Explain the authentication flow",
options: {
maxTurns: 5,
allowedTools: ["Read", "Grep"]
}
})) {
if (message.type === "result" && message.subtype === "success") {
console.log(message.result);
}
}
} catch (error) {
console.error(`Query failed: ${error}`);
}
// Continue conversation with session management
try {
for await (const message of query({
prompt: "Now explain the authorization process",
options: {
continue: true,
maxTurns: 5
}
})) {
if (message.type === "result" && message.subtype === "success") {
console.log(message.result);
}
}
} catch (error) {
console.error(`Query failed: ${error}`);
}
from claude_agent_sdk import query, ClaudeAgentOptions, ResultMessage
import asyncio
async def single_message_example():
# Simple one-shot query using query() function
# query() raises ResultError after an error result, such as error_max_turns
try:
async for message in query(
prompt="Explain the authentication flow",
options=ClaudeAgentOptions(max_turns=5, allowed_tools=["Read", "Grep"]),
):
if isinstance(message, ResultMessage) and message.subtype == "success":
print(message.result)
except Exception as e:
print(f"Query failed: {e}")
# Continue conversation with session management
try:
async for message in query(
prompt="Now explain the authorization process",
options=ClaudeAgentOptions(continue_conversation=True, max_turns=5),
):
if isinstance(message, ResultMessage) and message.subtype == "success":
print(message.result)
except Exception as e:
print(f"Query failed: {e}")
asyncio.run(single_message_example())
예제를 실행하면 각 쿼리는 최종 결과 텍스트를 출력합니다: 먼저 인증 설명, 그 다음 권한 부여 설명입니다.