Agent SDK Reference - Python

เอกสารอ้างอิง API ครบถ้วนสำหรับ Python Agent SDK รวมถึงทุก functions, types และ classes

การติดตั้ง

pip install claude-agent-sdk

การเลือกระหว่าง query() และ ClaudeSDKClient

Python SDK มีสองวิธีในการ interact กับ Claude Code:

Feature	query()	ClaudeSDKClient
Session	สร้าง session ใหม่โดยค่าเริ่มต้น	ใช้ session เดิมซ้ำ
Conversation	การแลกเปลี่ยนครั้งเดียว	การแลกเปลี่ยนหลายครั้งใน context เดียวกัน
Connection	จัดการโดยอัตโนมัติ	ควบคุมด้วยตนเอง
Streaming Input	✅ รองรับ	✅ รองรับ
Interrupts	❌ ไม่รองรับ	✅ รองรับ
Hooks	✅ รองรับ	✅ รองรับ
Custom Tools	✅ รองรับ	✅ รองรับ
Continue Chat	ด้วยตนเองผ่าน continue_conversation หรือ resume	✅ อัตโนมัติ
Use Case	งานครั้งเดียว	การสนทนาต่อเนื่อง

เมื่อไรใช้ query() (งานครั้งเดียว)

เหมาะสำหรับ:

• คำถามครั้งเดียวที่ไม่ต้องการประวัติการสนทนา
• งานอิสระที่ไม่ต้องการ context จากการแลกเปลี่ยนก่อนหน้า
• สคริปต์ automation ง่ายๆ
• เมื่อต้องการเริ่มต้นใหม่ทุกครั้ง

เมื่อไรใช้ ClaudeSDKClient (การสนทนาต่อเนื่อง)

เหมาะสำหรับ:

• การดำเนินการสนทนาต่อ - เมื่อต้องการให้ Claude จำ context
• คำถาม Follow-up - สร้างต่อจาก responses ก่อนหน้า
• Interactive applications - Chat interfaces, REPLs
• Response-driven logic - เมื่อขั้นตอนต่อไปขึ้นอยู่กับ response ของ Claude
• Session control - จัดการ conversation lifecycle อย่างชัดเจน

Functions

query()

สร้าง session ใหม่สำหรับการ interact กับ Claude Code แต่ละครั้งโดยค่าเริ่มต้น ส่งคืน async iterator ที่ yield messages เมื่อมาถึง

async def query(
    *,
    prompt: str | AsyncIterable[dict[str, Any]],
    options: ClaudeAgentOptions | None = None,
    transport: Transport | None = None
) -> AsyncIterator[Message]

Parameters

Parameter	Type	คำอธิบาย
prompt	str | AsyncIterable[dict]	Input prompt เป็น string หรือ async iterable สำหรับ streaming mode
options	ClaudeAgentOptions | None	Optional configuration object (ค่าเริ่มต้นเป็น ClaudeAgentOptions() ถ้าเป็น None)
transport	Transport | None	Optional custom transport สำหรับสื่อสารกับ CLI process

ตัวอย่าง

import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions


async def main():
    options = ClaudeAgentOptions(
        system_prompt="You are an expert Python developer",
        permission_mode="acceptEdits",
        cwd="/home/user/project",
    )

    async for message in query(prompt="Create a Python web server", options=options):
        print(message)


asyncio.run(main())

tool()

Decorator สำหรับกำหนด MCP tools ด้วย type safety

def tool(
    name: str,
    description: str,
    input_schema: type | dict[str, Any],
    annotations: ToolAnnotations | None = None
) -> Callable[[Callable[[Any], Awaitable[dict[str, Any]]]], SdkMcpTool[Any]]

ตัวอย่าง

from claude_agent_sdk import tool
from typing import Any


@tool("greet", "Greet a user", {"name": str})
async def greet(args: dict[str, Any]) -> dict[str, Any]:
    return {"content": [{"type": "text", "text": f"Hello, {args['name']}!"}]}

create_sdk_mcp_server()

สร้าง in-process MCP server ที่รันภายใน Python application ของคุณ

def create_sdk_mcp_server(
    name: str,
    version: str = "1.0.0",
    tools: list[SdkMcpTool[Any]] | None = None
) -> McpSdkServerConfig

ตัวอย่าง

from claude_agent_sdk import tool, create_sdk_mcp_server


@tool("add", "Add two numbers", {"a": float, "b": float})
async def add(args):
    return {"content": [{"type": "text", "text": f"Sum: {args['a'] + args['b']}"}]}


@tool("multiply", "Multiply two numbers", {"a": float, "b": float})
async def multiply(args):
    return {"content": [{"type": "text", "text": f"Product: {args['a'] * args['b']}"}]}


calculator = create_sdk_mcp_server(
    name="calculator",
    version="2.0.0",
    tools=[add, multiply],
)

options = ClaudeAgentOptions(
    mcp_servers={"calc": calculator},
    allowed_tools=["mcp__calc__add", "mcp__calc__multiply"],
)

list_sessions()

แสดงรายการ sessions ที่ผ่านมาพร้อม metadata เป็น synchronous ส่งคืนทันที

def list_sessions(
    directory: str | None = None,
    limit: int | None = None,
    include_worktrees: bool = True
) -> list[SDKSessionInfo]

ตัวอย่าง

from claude_agent_sdk import list_sessions

for session in list_sessions(directory="/path/to/project", limit=10):
    print(f"{session.summary} ({session.session_id})")

get_session_messages()

ดึง messages จาก session ที่ผ่านมา

def get_session_messages(
    session_id: str,
    directory: str | None = None,
    limit: int | None = None,
    offset: int = 0
) -> list[SessionMessage]

get_session_info()

อ่าน metadata สำหรับ session เดียวโดย ID

def get_session_info(
    session_id: str,
    directory: str | None = None,
) -> SDKSessionInfo | None

rename_session()

เปลี่ยนชื่อ session โดยการต่อท้าย custom-title entry

def rename_session(
    session_id: str,
    title: str,
    directory: str | None = None,
) -> None

tag_session()

Tag session ส่ง None เพื่อลบ tag

def tag_session(
    session_id: str,
    tag: str | None,
    directory: str | None = None,
) -> None

Classes

ClaudeSDKClient

รักษา conversation session ข้ามการแลกเปลี่ยนหลายครั้ง นี่คือ Python equivalent ของวิธีที่ query() function ใน TypeScript SDK ทำงานภายใน

class ClaudeSDKClient:
    def __init__(self, options: ClaudeAgentOptions | None = None, transport: Transport | None = None)
    async def connect(self, prompt: str | AsyncIterable[dict] | None = None) -> None
    async def query(self, prompt: str | AsyncIterable[dict], session_id: str = "default") -> None
    async def receive_messages(self) -> AsyncIterator[Message]
    async def receive_response(self) -> AsyncIterator[Message]
    async def interrupt(self) -> None
    async def set_permission_mode(self, mode: str) -> None
    async def set_model(self, model: str | None = None) -> None
    async def rewind_files(self, user_message_id: str) -> None
    async def get_mcp_status(self) -> McpStatusResponse
    async def reconnect_mcp_server(self, server_name: str) -> None
    async def toggle_mcp_server(self, server_name: str, enabled: bool) -> None
    async def stop_task(self, task_id: str) -> None
    async def get_server_info(self) -> dict[str, Any] | None
    async def disconnect(self) -> None

Context Manager Support

async with ClaudeSDKClient() as client:
    await client.query("Hello Claude")
    async for message in client.receive_response():
        print(message)

ตัวอย่าง - การดำเนินการสนทนาต่อ

import asyncio
from claude_agent_sdk import ClaudeSDKClient, AssistantMessage, TextBlock, ResultMessage


async def main():
    async with ClaudeSDKClient() as client:
        # คำถามแรก
        await client.query("What's the capital of France?")

        async for message in client.receive_response():
            if isinstance(message, AssistantMessage):
                for block in message.content:
                    if isinstance(block, TextBlock):
                        print(f"Claude: {block.text}")

        # คำถาม Follow-up - session เก็บ context ก่อนหน้าไว้
        await client.query("What's the population of that city?")

        async for message in client.receive_response():
            if isinstance(message, AssistantMessage):
                for block in message.content:
                    if isinstance(block, TextBlock):
                        print(f"Claude: {block.text}")


asyncio.run(main())

ตัวอย่าง - การใช้ interrupts

import asyncio
from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions, ResultMessage


async def interruptible_task():
    options = ClaudeAgentOptions(allowed_tools=["Bash"], permission_mode="acceptEdits")

    async with ClaudeSDKClient(options=options) as client:
        await client.query("Count from 1 to 100 slowly, using the bash sleep command")

        await asyncio.sleep(2)
        await client.interrupt()
        print("Task interrupted!")

        async for message in client.receive_response():
            if isinstance(message, ResultMessage):
                print(f"Interrupted task finished with subtype={message.subtype!r}")


asyncio.run(interruptible_task())

Types

ClaudeAgentOptions

Configuration dataclass สำหรับ Claude Code queries

@dataclass
class ClaudeAgentOptions:
    tools: list[str] | ToolsPreset | None = None
    allowed_tools: list[str] = field(default_factory=list)
    system_prompt: str | SystemPromptPreset | None = None
    mcp_servers: dict[str, McpServerConfig] | str | Path = field(default_factory=dict)
    strict_mcp_config: bool = False
    permission_mode: PermissionMode | None = None
    continue_conversation: bool = False
    resume: str | None = None
    max_turns: int | None = None
    max_budget_usd: float | None = None
    disallowed_tools: list[str] = field(default_factory=list)
    model: str | None = None
    fallback_model: str | None = None
    betas: list[SdkBeta] = field(default_factory=list)
    output_format: dict[str, Any] | None = None
    cwd: str | Path | None = None
    cli_path: str | Path | None = None
    settings: str | None = None
    add_dirs: list[str | Path] = field(default_factory=list)
    env: dict[str, str] = field(default_factory=dict)
    extra_args: dict[str, str | None] = field(default_factory=dict)
    max_buffer_size: int | None = None
    stderr: Callable[[str], None] | None = None
    can_use_tool: CanUseTool | None = None
    hooks: dict[HookEvent, list[HookMatcher]] | None = None
    user: str | None = None
    include_partial_messages: bool = False
    include_hook_events: bool = False
    fork_session: bool = False
    agents: dict[str, AgentDefinition] | None = None
    setting_sources: list[SettingSource] | None = None
    sandbox: SandboxSettings | None = None
    plugins: list[SdkPluginConfig] = field(default_factory=list)
    thinking: ThinkingConfig | None = None
    effort: EffortLevel | None = None
    enable_file_checkpointing: bool = False
    session_store: SessionStore | None = None
    session_store_flush: SessionStoreFlushMode = "batched"

PermissionMode

PermissionMode = Literal[
    "default",          # พฤติกรรม permission มาตรฐาน
    "acceptEdits",      # ยอมรับการแก้ไขไฟล์โดยอัตโนมัติ
    "plan",             # Planning mode - สำรวจโดยไม่แก้ไข
    "dontAsk",          # ปฏิเสธสิ่งที่ไม่ได้รับการ pre-approved แทนการแจ้ง
    "bypassPermissions",# ข้ามการตรวจสอบ permission
]

EffortLevel

EffortLevel = Literal[
    "low",    # การคิดน้อยที่สุด responses เร็วที่สุด
    "medium", # การคิดปานกลาง
    "high",   # การใช้เหตุผลลึก
    "xhigh",  # การใช้เหตุผลขยาย
    "max",    # ความพยายามสูงสุด
]

ThinkingConfig

ควบคุมพฤติกรรม extended thinking:

class ThinkingConfigAdaptive(TypedDict):
    type: Literal["adaptive"]
    display: NotRequired[ThinkingDisplay]

class ThinkingConfigEnabled(TypedDict):
    type: Literal["enabled"]
    budget_tokens: int
    display: NotRequired[ThinkingDisplay]

class ThinkingConfigDisabled(TypedDict):
    type: Literal["disabled"]

ThinkingConfig = ThinkingConfigAdaptive | ThinkingConfigEnabled | ThinkingConfigDisabled

Message Types

Message

Union type ของ messages ที่เป็นไปได้ทั้งหมด

Message = (
    UserMessage
    | AssistantMessage
    | SystemMessage
    | ResultMessage
    | StreamEvent
    | RateLimitEvent
)

ResultMessage

Final result message พร้อมข้อมูลค่าใช้จ่ายและการใช้งาน

@dataclass
class ResultMessage:
    subtype: str
    duration_ms: int
    duration_api_ms: int
    is_error: bool
    num_turns: int
    session_id: str
    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
    model_usage: dict[str, Any] | None = None
    permission_denials: list[Any] | None = None
    deferred_tool_use: DeferredToolUse | None = None
    errors: list[str] | None = None
    api_error_status: int | None = None
    uuid: str | None = None

field subtype เป็นหนึ่งใน: "success", "error_during_execution", "error_max_turns", "error_max_budget_usd" หรือ "error_max_structured_output_retries"

dict usage มี:

• input_tokens: Token input ทั้งหมดที่ใช้
• output_tokens: Token output ทั้งหมดที่สร้าง
• cache_creation_input_tokens: Tokens ที่ใช้สร้าง cache entries ใหม่
• cache_read_input_tokens: Tokens ที่อ่านจาก cache entries ที่มีอยู่