ورودیِ استریمی

Understanding the two input modes for Claude Agent SDK and when to use each

مرورِ کلی

Claude Agent SDK از دو حالتِ ورودیِ متمایز برای تعامل با ایجنت‌ها پشتیبانی می‌کند:

حالتِ ورودیِ استریمی (پیش‌فرض و توصیه‌شده) — یک نشستِ پایدار و تعاملی
ورودیِ تک‌پیامی — query‌های تک‌مرحله‌ای که از وضعیتِ نشست و resume استفاده می‌کنند

این راهنما تفاوت‌ها، مزایا و موارد استفاده‌ی هر حالت را توضیح می‌دهد تا به تو کمک کند رویکردِ درست را برای برنامه‌ات انتخاب کنی.

حالتِ ورودیِ استریمی (توصیه‌شده)

حالتِ ورودیِ استریمی روشِ ترجیحی برای استفاده از Claude Agent SDK است. دسترسیِ کامل به قابلیت‌های ایجنت می‌دهد و تجربه‌های غنی و تعاملی را ممکن می‌کند.

این حالت به ایجنت اجازه می‌دهد به‌عنوانِ یک فرایندِ بلندعمر عمل کند که ورودیِ کاربر را می‌گیرد، وقفه‌ها را مدیریت می‌کند، درخواست‌های دسترسی را نمایان می‌کند، و مدیریتِ نشست را به‌عهده می‌گیرد.

چطور کار می‌کند

sequenceDiagram
    participant App as Your Application
    participant Agent as Claude Agent
    participant Tools as Tools/Hooks
    participant FS as Environment/<br/>File System

    App->>Agent: Initialize with AsyncGenerator
    activate Agent

    App->>Agent: Yield Message 1
    Agent->>Tools: Execute tools
    Tools->>FS: Read files
    FS-->>Tools: File contents
    Tools->>FS: Write/Edit files
    FS-->>Tools: Success/Error
    Agent-->>App: Stream partial response
    Agent-->>App: Stream more content...
    Agent->>App: Complete Message 1

    App->>Agent: Yield Message 2 + Image
    Agent->>Tools: Process image & execute
    Tools->>FS: Access filesystem
    FS-->>Tools: Operation results
    Agent-->>App: Stream response 2

    App->>Agent: Queue Message 3
    App->>Agent: Interrupt/Cancel
    Agent->>App: Handle interruption

    Note over App,Agent: Session stays alive
    Note over Tools,FS: Persistent file system<br/>state maintained

    deactivate Agent

مزایا

آپلودِ تصویر

تصاویر را مستقیم به پیام‌ها بچسبان تا برای تحلیل و درکِ بصری استفاده شوند

پیام‌های صف‌شده

چند پیام بفرست که به‌ترتیب پردازش شوند، با قابلیتِ وقفه

یکپارچگیِ ابزار

دسترسیِ کامل به همه‌ی ابزارها و سرورهای MCPِ سفارشی در طولِ نشست

بازخوردِ بی‌درنگ

پاسخ‌ها را هم‌زمان با تولید ببین، نه فقط نتایجِ نهایی را

پایداریِ کانتکست

کانتکستِ گفتگو را در طولِ چند نوبت به‌صورتِ طبیعی حفظ کن

مثالِ پیاده‌سازی

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 SDK، اگر تولیدکننده‌ی پیامت خطا بدهد — برای مثال وقتی فایلی که می‌خواند گم شده باشد — استریم با خطایی پایان می‌یابد که Claude Code process aborted by user است، نه خطای اصلی؛ پس وقتی این پیام را دیدی نخست کدِ داخلِ تولیدکننده‌ات را بررسی کن. این خطا ممکن است با یک خطِ بلندِ مینیفای‌شده از سورسِ بسته‌بندی‌شده‌ی SDK هم پیشاپیش بیاید، پس تا انتهای خروجی بخوان تا متنِ خطا را بیابی.

در Python SDK، خطای تولیدکننده در سطحِ debug لاگ می‌شود و نشست بدونِ راه‌انداختنِ خطا متوقف می‌ماند، پس اگر یک نشستِ استریمی بدونِ هیچ خروجی معلق ماند، لاگ‌گیریِ debug را فعال کن و تولیدکننده‌ات را بررسی کن.

ورودیِ تک‌پیامی

ورودیِ تک‌پیامی ساده‌تر اما محدودتر است.

چه زمانی از ورودیِ تک‌پیامی استفاده کنیم

از ورودیِ تک‌پیامی وقتی استفاده کن که:

به یک پاسخِ تک‌مرحله‌ای نیاز داری
به پیوستِ تصویر یا متدهای کنترلیِ میانِ نشست نیاز نداری
باید در یک محیطِ بدون‌حالت (stateless) کار کنی، مثلِ یک تابعِ lambda

محدودیت‌ها

اگر یک query با نتیجه‌ی خطا — مثلِ error_max_turns — پایان یابد، یک فراخوانیِ تک‌پیامیِ query() پس از برگرداندنِ پیامِ نتیجه‌ی نهایی، خطایی راه می‌اندازد که متنِ شکست را در بر دارد؛ پس اگر کدت لازم است ادامه پیدا کند، حلقه را در یک بلاکِ try بپیچ. برای انواعِ نتیجه به مدیریتِ نتیجه مراجعه کن.

مثالِ پیاده‌سازی

import { query } from "@anthropic-ai/claude-agent-sdk";

// Simple one-shot query
for await (const message of query({
  prompt: "Explain the authentication flow",
  options: {
    maxTurns: 1,
    allowedTools: ["Read", "Grep"]
  }
})) {
  if (message.type === "result" && message.subtype === "success") {
    console.log(message.result);
  }
}

// Continue conversation with session management
for await (const message of query({
  prompt: "Now explain the authorization process",
  options: {
    continue: true,
    maxTurns: 1
  }
})) {
  if (message.type === "result" && message.subtype === "success") {
    console.log(message.result);
  }
}

from claude_agent_sdk import query, ClaudeAgentOptions, ResultMessage
import asyncio


async def single_message_example():
    # Simple one-shot query using query() function
    async for message in query(
        prompt="Explain the authentication flow",
        options=ClaudeAgentOptions(max_turns=1, allowed_tools=["Read", "Grep"]),
    ):
        if isinstance(message, ResultMessage):
            print(message.result)

    # Continue conversation with session management
    async for message in query(
        prompt="Now explain the authorization process",
        options=ClaudeAgentOptions(continue_conversation=True, max_turns=1),
    ):
        if isinstance(message, ResultMessage):
            print(message.result)


asyncio.run(single_message_example())