رفتن به محتوا

مرجعِ Agent SDK — پایتون

Terminal window
pip install claude-agent-sdk

انتخاب بینِ query() و ClaudeSDKClient

Section titled “انتخاب بینِ query() و ClaudeSDKClient”

پایتون SDK دو راه برای تعامل با Claude Code فراهم می‌کند:

قابلیتquery()ClaudeSDKClient
نشستبه‌صورتِ پیش‌فرض یک نشستِ جدید می‌سازدهمان نشست را دوباره استفاده می‌کند
گفتگویک تبادلِ واحدچند تبادل در همان کانتکست
اتصالخودکار مدیریت می‌شودکنترلِ دستی
Streaming Input✅ پشتیبانی‌شده✅ پشتیبانی‌شده
Interrupts❌ پشتیبانی‌نشده✅ پشتیبانی‌شده
Hooks✅ پشتیبانی‌شده✅ پشتیبانی‌شده
Custom Tools✅ پشتیبانی‌شده✅ پشتیبانی‌شده
ادامه‌ی چتدستی از طریقِ continue_conversation یا resume✅ خودکار
مورد استفادهتسک‌های یک‌بارهگفتگوهای پیوسته

کِی از query() استفاده کنیم (تسک‌های یک‌باره)

Section titled “کِی از query() استفاده کنیم (تسک‌های یک‌باره)”

بهترین برای:

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

کِی از ClaudeSDKClient استفاده کنیم (گفتگوی پیوسته)

Section titled “کِی از ClaudeSDKClient استفاده کنیم (گفتگوی پیوسته)”

بهترین برای:

  • ادامه‌ی گفتگو - وقتی نیاز داری Claude کانتکست را به یاد بسپارد
  • پرسش‌های پیگیری - ساختن روی پاسخ‌های پیشین
  • اپلیکیشن‌های تعاملی - رابط‌های چت، REPLها
  • منطقِ پاسخ‌محور - وقتی اقدامِ بعدی به پاسخِ Claude بستگی دارد
  • کنترلِ نشست - مدیریتِ صریحِ چرخه‌ی حیاتِ گفتگو

به‌صورتِ پیش‌فرض برای هر تعامل با Claude Code یک نشستِ جدید می‌سازد. یک async iterator برمی‌گرداند که پیام‌ها را همان‌طور که می‌رسند yield می‌کند. هر فراخوانیِ query() بدونِ حافظه‌ای از تعامل‌های پیشین تازه شروع می‌شود، مگر اینکه continue_conversation=True یا resume را در ClaudeAgentOptions پاس بدهی. نشست‌ها را ببین.

async def query(
*,
prompt: str | AsyncIterable[dict[str, Any]],
options: ClaudeAgentOptions | None = None,
transport: Transport | None = None
) -> AsyncIterator[Message]
پارامترنوعتوضیح
promptstr | AsyncIterable[dict]پرامپتِ ورودی به‌صورتِ رشته یا async iterable برای حالتِ streaming
optionsClaudeAgentOptions | Noneآبجکتِ پیکربندیِ اختیاری (در صورتِ None، پیش‌فرض ClaudeAgentOptions())
transportTransport | Nonetransportِ سفارشیِ اختیاری برای ارتباط با پروسه‌ی CLI

یک AsyncIterator[Message] برمی‌گرداند که پیام‌های گفتگو را yield می‌کند.

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())

دکوراتور برای تعریفِ ابزارهای MCP با ایمنیِ نوع.

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]]
پارامترنوعتوضیح
namestrشناسه‌ی یکتای ابزار
descriptionstrتوضیحِ خوانا برای انسان درباره‌ی کاری که ابزار انجام می‌دهد
input_schematype | dict[str, Any]schemaِ تعریف‌کننده‌ی پارامترهای ورودیِ ابزار (پایین را ببین)
annotationsToolAnnotations | Noneannotationهای اختیاریِ ابزارِ MCP که نکته‌های رفتاری به clientها می‌دهند
  1. نگاشتِ نوعِ ساده (توصیه‌شده):

    {"text": str, "count": int, "enabled": bool}
  2. قالبِ JSON Schema (برای اعتبارسنجیِ پیچیده):

    {
    "type": "object",
    "properties": {
    "text": {"type": "string"},
    "count": {"type": "integer", "minimum": 0},
    },
    "required": ["text"],
    }

یک تابعِ دکوراتور که پیاده‌سازیِ ابزار را wrap می‌کند و یک نمونه‌ی SdkMcpTool برمی‌گرداند.

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']}!"}]}

از mcp.types دوباره export شده (به‌صورتِ from claude_agent_sdk import ToolAnnotations هم در دسترس است). همه‌ی فیلدها نکته‌های اختیاری‌اند؛ clientها نباید برای تصمیم‌های امنیتی به آن‌ها تکیه کنند.

فیلدنوعپیش‌فرضتوضیح
titlestr | NoneNoneعنوانِ خوانا برای انسان برای ابزار
readOnlyHintbool | NoneFalseاگر True باشد، ابزار محیطِ خود را تغییر نمی‌دهد
destructiveHintbool | NoneTrueاگر True باشد، ابزار ممکن است به‌روزرسانی‌های مخرب انجام دهد (فقط وقتی readOnlyHint برابرِ False است معنادار است)
idempotentHintbool | NoneFalseاگر True باشد، فراخوانی‌های مکررِ با همان آرگومان‌ها اثرِ اضافی ندارند (فقط وقتی readOnlyHint برابرِ False است معنادار است)
openWorldHintbool | NoneTrueاگر True باشد، ابزار با موجودیت‌های بیرونی تعامل دارد (مثلاً web search). اگر False باشد، دامنه‌ی ابزار بسته است (مثلاً یک ابزارِ حافظه)
from claude_agent_sdk import tool, ToolAnnotations
from typing import Any
@tool(
"search",
"Search the web",
{"query": str},
annotations=ToolAnnotations(readOnlyHint=True, openWorldHint=True),
)
async def search(args: dict[str, Any]) -> dict[str, Any]:
return {"content": [{"type": "text", "text": f"Results for: {args['query']}"}]}

یک سرورِ MCPِ in-process بساز که درونِ اپلیکیشنِ پایتونِ تو اجرا می‌شود.

def create_sdk_mcp_server(
name: str,
version: str = "1.0.0",
tools: list[SdkMcpTool[Any]] | None = None
) -> McpSdkServerConfig
پارامترنوعپیش‌فرضتوضیح
namestr-شناسه‌ی یکتای سرور
versionstr"1.0.0"رشته‌ی نسخه‌ی سرور
toolslist[SdkMcpTool[Any]] | NoneNoneفهرستِ توابعِ ابزار که با دکوراتورِ @tool ساخته شده‌اند

یک آبجکتِ McpSdkServerConfig برمی‌گرداند که می‌توان آن را به ClaudeAgentOptions.mcp_servers پاس داد.

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], # Pass decorated functions
)
# Use with Claude
options = ClaudeAgentOptions(
mcp_servers={"calc": calculator},
allowed_tools=["mcp__calc__add", "mcp__calc__multiply"],
)

نشست‌های گذشته را به‌همراه فراداده فهرست می‌کند. بر اساسِ دایرکتوریِ پروژه فیلتر کن یا نشست‌ها را در همه‌ی پروژه‌ها فهرست کن. همگام (synchronous)؛ بلافاصله برمی‌گردد.

def list_sessions(
directory: str | None = None,
limit: int | None = None,
include_worktrees: bool = True
) -> list[SDKSessionInfo]
پارامترنوعپیش‌فرضتوضیح
directorystr | NoneNoneدایرکتوری‌ای که نشست‌هایش فهرست شوند. وقتی حذف شود، نشست‌ها را در همه‌ی پروژه‌ها برمی‌گرداند
limitint | NoneNoneبیشینه‌ی تعدادِ نشست‌هایی که برگردانده می‌شوند
include_worktreesboolTrueوقتی directory درونِ یک مخزنِ git است، نشست‌ها را از همه‌ی مسیرهای worktree بگنجان
ویژگینوعتوضیح
session_idstrشناسه‌ی یکتای نشست
summarystrعنوانِ نمایشی: عنوانِ سفارشی، خلاصه‌ی خودکارتولیدشده، یا اولین پرامپت
last_modifiedintزمانِ آخرین تغییر، به میلی‌ثانیه از epoch
file_sizeint | Noneاندازه‌ی فایلِ نشست به بایت (None برای backendهای ذخیره‌سازیِ remote)
custom_titlestr | Noneعنوانِ نشست که کاربر تنظیم کرده
first_promptstr | Noneاولین پرامپتِ معنادارِ کاربر در نشست
git_branchstr | Noneشاخه‌ی git در پایانِ نشست
cwdstr | Noneدایرکتوریِ کاریِ نشست
tagstr | Noneبرچسبِ نشست که کاربر تنظیم کرده (tag_session() را ببین)
created_atint | Noneزمانِ ساختِ نشست، به میلی‌ثانیه از epoch

۱۰ نشستِ اخیرِ یک پروژه را چاپ کن. نتایج بر اساسِ last_modified نزولی مرتب می‌شوند، پس اولین آیتم تازه‌ترین است. برای جست‌وجو در همه‌ی پروژه‌ها directory را حذف کن.

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})")

پیام‌های یک نشستِ گذشته را بازیابی می‌کند. همگام؛ بلافاصله برمی‌گردد.

def get_session_messages(
session_id: str,
directory: str | None = None,
limit: int | None = None,
offset: int = 0
) -> list[SessionMessage]
پارامترنوعپیش‌فرضتوضیح
session_idstrالزامیشناسه‌ی نشستی که پیام‌هایش بازیابی شود
directorystr | NoneNoneدایرکتوریِ پروژه برای جست‌وجو. وقتی حذف شود، همه‌ی پروژه‌ها را جست‌وجو می‌کند
limitint | NoneNoneبیشینه‌ی تعدادِ پیام‌هایی که برگردانده می‌شوند
offsetint0تعدادِ پیام‌هایی که از ابتدا رد شوند
ویژگینوعتوضیح
typeLiteral["user", "assistant"]نقشِ پیام
uuidstrشناسه‌ی یکتای پیام
session_idstrشناسه‌ی نشست
messageAnyمحتوای خامِ پیام
parent_tool_use_idNoneرزرو برای استفاده‌ی آینده
from claude_agent_sdk import list_sessions, get_session_messages
sessions = list_sessions(limit=1)
if sessions:
messages = get_session_messages(sessions[0].session_id)
for msg in messages:
print(f"[{msg.type}] {msg.uuid}")

فراداده‌ی یک نشستِ واحد را بر اساسِ ID می‌خواند، بدونِ پویشِ کلِ دایرکتوریِ پروژه. همگام؛ بلافاصله برمی‌گردد.

def get_session_info(
session_id: str,
directory: str | None = None,
) -> SDKSessionInfo | None
پارامترنوعپیش‌فرضتوضیح
session_idstrالزامیUUIDِ نشستی که جست‌وجو شود
directorystr | NoneNoneمسیرِ دایرکتوریِ پروژه. وقتی حذف شود، همه‌ی دایرکتوری‌های پروژه را جست‌وجو می‌کند

یک SDKSessionInfo برمی‌گرداند، یا None اگر نشست پیدا نشود.

فراداده‌ی یک نشستِ واحد را بدونِ پویشِ دایرکتوریِ پروژه جست‌وجو کن. وقتی از یک اجرای پیشین یک session ID داری مفید است.

from claude_agent_sdk import get_session_info
info = get_session_info("550e8400-e29b-41d4-a716-446655440000")
if info:
print(f"{info.summary} (branch: {info.git_branch}, tag: {info.tag})")

یک نشست را با افزودنِ یک ورودیِ custom-title تغییرِ نام می‌دهد. فراخوانی‌های مکرر امن‌اند؛ تازه‌ترین عنوان برنده است. همگام.

def rename_session(
session_id: str,
title: str,
directory: str | None = None,
) -> None
پارامترنوعپیش‌فرضتوضیح
session_idstrالزامیUUIDِ نشستی که تغییرِ نام شود
titlestrالزامیعنوانِ جدید. پس از حذفِ فاصله‌ها باید ناخالی باشد
directorystr | NoneNoneمسیرِ دایرکتوریِ پروژه. وقتی حذف شود، همه‌ی دایرکتوری‌های پروژه را جست‌وجو می‌کند

اگر session_id یک UUIDِ معتبر نباشد یا title خالی باشد ValueError می‌دهد؛ اگر نشست پیدا نشود FileNotFoundError.

تازه‌ترین نشست را تغییرِ نام بده تا بعداً راحت‌تر پیدایش کنی. عنوانِ جدید در خواندن‌های بعدی در SDKSessionInfo.custom_title ظاهر می‌شود.

from claude_agent_sdk import list_sessions, rename_session
sessions = list_sessions(directory="/path/to/project", limit=1)
if sessions:
rename_session(sessions[0].session_id, "Refactor auth module")

یک نشست را برچسب می‌زند. برای پاک‌کردنِ برچسب None پاس بده. فراخوانی‌های مکرر امن‌اند؛ تازه‌ترین برچسب برنده است. همگام.

def tag_session(
session_id: str,
tag: str | None,
directory: str | None = None,
) -> None
پارامترنوعپیش‌فرضتوضیح
session_idstrالزامیUUIDِ نشستی که برچسب بخورد
tagstr | Noneالزامیرشته‌ی برچسب، یا None برای پاک‌کردن. پیش از ذخیره Unicode-sanitize می‌شود
directorystr | NoneNoneمسیرِ دایرکتوریِ پروژه. وقتی حذف شود، همه‌ی دایرکتوری‌های پروژه را جست‌وجو می‌کند

اگر session_id یک UUIDِ معتبر نباشد یا tag پس از sanitize خالی باشد ValueError می‌دهد؛ اگر نشست پیدا نشود FileNotFoundError.

یک نشست را برچسب بزن، سپس در خواندنِ بعدی بر اساسِ آن برچسب فیلتر کن. برای پاک‌کردنِ یک برچسبِ موجود None پاس بده.

from claude_agent_sdk import list_sessions, tag_session
# Tag a session
tag_session("550e8400-e29b-41d4-a716-446655440000", "needs-review")
# Later: find all sessions with that tag
for session in list_sessions(directory="/path/to/project"):
if session.tag == "needs-review":
print(session.summary)

یک نشستِ گفتگو را در طولِ چند تبادل نگه می‌دارد. این معادلِ پایتونیِ نحوه‌ی کارِ درونیِ تابعِ query()ِ تایپ‌اسکریپت است — یک آبجکتِ client می‌سازد که می‌تواند گفتگوها را ادامه دهد.

  • پیوستگیِ نشست: کانتکستِ گفتگو را در طولِ چند فراخوانیِ query() نگه می‌دارد
  • همان گفتگو: نشست پیام‌های پیشین را حفظ می‌کند
  • پشتیبانیِ Interrupt: می‌تواند اجرا را در میانه‌ی تسک متوقف کند
  • چرخه‌ی حیاتِ صریح: تو کنترل می‌کنی که نشست کِی شروع و کِی پایان یابد
  • جریانِ پاسخ‌محور: می‌تواند به پاسخ‌ها واکنش نشان دهد و پیگیری بفرستد
  • ابزارها و hookهای سفارشی: از ابزارهای سفارشی (ساخته‌شده با دکوراتورِ @tool) و hookها پشتیبانی می‌کند
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
متدتوضیح
__init__(options)client را با پیکربندیِ اختیاری مقداردهیِ اولیه کن
connect(prompt)به Claude با یک پرامپتِ اولیه یا استریمِ پیامِ اختیاری وصل شو
query(prompt, session_id)یک درخواستِ جدید در حالتِ streaming بفرست
receive_messages()همه‌ی پیام‌های Claude را به‌صورتِ یک async iterator دریافت کن
receive_response()پیام‌ها را تا و شاملِ یک ResultMessage دریافت کن
interrupt()سیگنالِ interrupt بفرست (فقط در حالتِ streaming کار می‌کند)
set_permission_mode(mode)permission mode را برای نشستِ فعلی تغییر بده
set_model(model)مدلِ نشستِ فعلی را تغییر بده. برای بازگشت به پیش‌فرض None پاس بده
rewind_files(user_message_id)فایل‌ها را به وضعیتشان در پیامِ کاربرِ مشخص‌شده بازگردان. نیازمندِ enable_file_checkpointing=True است. File checkpointing را ببین
get_mcp_status()وضعیتِ همه‌ی سرورهای MCPِ پیکربندی‌شده را بگیر. McpStatusResponse را برمی‌گرداند
reconnect_mcp_server(server_name)اتصال به سرورِ MCPی که ناموفق بود یا قطع شد را دوباره امتحان کن
toggle_mcp_server(server_name, enabled)یک سرورِ MCP را در میانه‌ی نشست فعال یا غیرفعال کن. غیرفعال‌کردن ابزارهایش را حذف می‌کند
stop_task(task_id)یک تسکِ پس‌زمینه‌ی در حالِ اجرا را متوقف کن. یک TaskNotificationMessage با وضعیتِ "stopped" در استریمِ پیام به دنبالش می‌آید
get_server_info()اطلاعاتِ سرور شاملِ session ID و قابلیت‌ها را بگیر
disconnect()از Claude قطع شو

می‌توان client را به‌عنوانِ یک async context manager برای مدیریتِ خودکارِ اتصال استفاده کرد:

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

مهم: هنگامِ پیمایش روی پیام‌ها، از break برای خروجِ زودهنگام پرهیز کن چون می‌تواند مشکلاتِ پاک‌سازیِ asyncio ایجاد کند. به‌جای آن بگذار پیمایش به‌طورِ طبیعی کامل شود یا از flagها برای ردیابیِ اینکه آنچه را که می‌خواستی پیدا کرده‌ای استفاده کن.

مثال - ادامه‌ی یک گفتگو

Section titled “مثال - ادامه‌ی یک گفتگو”
import asyncio
from claude_agent_sdk import ClaudeSDKClient, AssistantMessage, TextBlock, ResultMessage
async def main():
async with ClaudeSDKClient() as client:
# First question
await client.query("What's the capital of France?")
# Process response
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 question - the session retains the previous 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}")
# Another follow-up - still in the same conversation
await client.query("What are some famous landmarks there?")
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())

مثال - ورودیِ streaming با ClaudeSDKClient

Section titled “مثال - ورودیِ streaming با ClaudeSDKClient”
import asyncio
from claude_agent_sdk import ClaudeSDKClient
async def message_stream():
"""Generate messages dynamically."""
yield {
"type": "user",
"message": {"role": "user", "content": "Analyze the following data:"},
}
await asyncio.sleep(0.5)
yield {
"type": "user",
"message": {"role": "user", "content": "Temperature: 25°C, Humidity: 60%"},
}
await asyncio.sleep(0.5)
yield {
"type": "user",
"message": {"role": "user", "content": "What patterns do you see?"},
}
async def main():
async with ClaudeSDKClient() as client:
# Stream input to Claude
await client.query(message_stream())
# Process response
async for message in client.receive_response():
print(message)
# Follow-up in same session
await client.query("Should we be concerned about these readings?")
async for message in client.receive_response():
print(message)
asyncio.run(main())

مثال - استفاده از interruptها

Section titled “مثال - استفاده از interruptها”
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:
# Start a long-running task
await client.query("Count from 1 to 100 slowly, using the bash sleep command")
# Let it run for a bit
await asyncio.sleep(2)
# Interrupt the task
await client.interrupt()
print("Task interrupted!")
# Drain the interrupted task's messages (including its ResultMessage)
async for message in client.receive_response():
if isinstance(message, ResultMessage):
print(f"Interrupted task finished with subtype={message.subtype!r}")
# subtype is "error_during_execution" for interrupted tasks
# Send a new command
await client.query("Just say hello instead")
# Now receive the new response
async for message in client.receive_response():
if isinstance(message, ResultMessage) and message.subtype == "success":
print(f"New result: {message.result}")
asyncio.run(interruptible_task())

مثال - کنترلِ پیشرفته‌ی دسترسی

Section titled “مثال - کنترلِ پیشرفته‌ی دسترسی”
from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions
from claude_agent_sdk.types import (
PermissionResultAllow,
PermissionResultDeny,
ToolPermissionContext,
)
async def custom_permission_handler(
tool_name: str, input_data: dict, context: ToolPermissionContext
) -> PermissionResultAllow | PermissionResultDeny:
"""Custom logic for tool permissions."""
# Block writes to system directories
if tool_name == "Write" and input_data.get("file_path", "").startswith("/system/"):
return PermissionResultDeny(
message="System directory write not allowed", interrupt=True
)
# Redirect sensitive file operations
if tool_name in ["Write", "Edit"] and "config" in input_data.get("file_path", ""):
safe_path = f"./sandbox/{input_data['file_path']}"
return PermissionResultAllow(
updated_input={**input_data, "file_path": safe_path}
)
# Allow everything else
return PermissionResultAllow(updated_input=input_data)
async def main():
options = ClaudeAgentOptions(
can_use_tool=custom_permission_handler, allowed_tools=["Read", "Write", "Edit"]
)
async with ClaudeSDKClient(options=options) as client:
await client.query("Update the system config file")
async for message in client.receive_response():
# Will use sandbox path instead
print(message)
asyncio.run(main())

تعریفِ یک ابزارِ SDK MCP که با دکوراتورِ @tool ساخته شده.

@dataclass
class SdkMcpTool(Generic[T]):
name: str
description: str
input_schema: type[T] | dict[str, Any]
handler: Callable[[T], Awaitable[dict[str, Any]]]
annotations: ToolAnnotations | None = None
ویژگینوعتوضیح
namestrشناسه‌ی یکتای ابزار
descriptionstrتوضیحِ خوانا برای انسان
input_schematype[T] | dict[str, Any]schema برای اعتبارسنجیِ ورودی
handlerCallable[[T], Awaitable[dict[str, Any]]]تابعِ async که اجرای ابزار را رسیدگی می‌کند
annotationsToolAnnotations | Noneannotationهای اختیاریِ ابزارِ MCP (مثلِ readOnlyHint، destructiveHint، openWorldHint). از mcp.types

کلاسِ پایه‌ی انتزاعی برای پیاده‌سازی‌های transportِ سفارشی. از این برای ارتباط با پروسه‌ی Claude روی یک کانالِ سفارشی استفاده کن (مثلاً یک اتصالِ remote به‌جای یک subprocessِ محلی).

from abc import ABC, abstractmethod
from collections.abc import AsyncIterator
from typing import Any
class Transport(ABC):
@abstractmethod
async def connect(self) -> None: ...
@abstractmethod
async def write(self, data: str) -> None: ...
@abstractmethod
def read_messages(self) -> AsyncIterator[dict[str, Any]]: ...
@abstractmethod
async def close(self) -> None: ...
@abstractmethod
def is_ready(self) -> bool: ...
@abstractmethod
async def end_input(self) -> None: ...
متدتوضیح
connect()transport را وصل کن و برای ارتباط آماده شو
write(data)داده‌ی خام (JSON + newline) را روی transport بنویس
read_messages()async iterator که پیام‌های JSONِ parse‌شده را yield می‌کند
close()اتصال را ببند و منابع را پاک‌سازی کن
is_ready()اگر transport بتواند بفرستد و دریافت کند True برمی‌گرداند
end_input()استریمِ ورودی را ببند (مثلاً بستنِ stdin برای transportهای subprocess)

Import: from claude_agent_sdk import Transport

dataclassِ پیکربندی برای queryهای Claude Code.

@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
permission_prompt_tool_name: str | 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
debug_stderr: Any = sys.stderr # Deprecated
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)
max_thinking_tokens: int | None = None # Deprecated: use thinking instead
thinking: ThinkingConfig | None = None
effort: EffortLevel | None = None
enable_file_checkpointing: bool = False
session_store: SessionStore | None = None
session_store_flush: SessionStoreFlushMode = "batched"
ویژگینوعپیش‌فرضتوضیح
toolslist[str] | ToolsPreset | NoneNoneپیکربندیِ ابزارها. برای ابزارهای پیش‌فرضِ Claude Code از {"type": "preset", "preset": "claude_code"} استفاده کن
allowed_toolslist[str][]ابزارهایی که بدونِ پرامپت خودکار تأیید شوند. این Claude را به فقط همین ابزارها محدود نمی‌کند؛ ابزارهای فهرست‌نشده به permission_mode و can_use_tool سرریز می‌کنند. برای مسدودکردنِ ابزارها از disallowed_tools استفاده کن. Permissions را ببین
system_promptstr | SystemPromptPreset | NoneNoneپیکربندیِ system prompt. برای پرامپتِ سفارشی یک رشته پاس بده، یا برای system promptِ Claude Code از {"type": "preset", "preset": "claude_code"} استفاده کن. برای گسترشِ preset مقدارِ "append" را اضافه کن
mcp_serversdict[str, McpServerConfig] | str | Path{}پیکربندی‌های سرورِ MCP یا مسیرِ فایلِ پیکربندی
strict_mcp_configboolFalseوقتی True باشد، فقط از سرورهای پاس‌داده‌شده در mcp_servers استفاده کن و .mcp.jsonِ پروژه، تنظیماتِ کاربر، سرورهای MCPِ ارائه‌شده توسطِ plugin و connectorهای claude.ai را نادیده بگیر. به پرچمِ CLIِ --strict-mcp-config نگاشته می‌شود
permission_modePermissionMode | NoneNonepermission mode برای استفاده از ابزار
continue_conversationboolFalseتازه‌ترین گفتگو را ادامه بده
resumestr | NoneNonesession ID برای resume
max_turnsint | NoneNoneبیشینه‌ی نوبت‌های ایجنتیک (رفت‌وبرگشت‌های استفاده از ابزار)
max_budget_usdfloat | NoneNoneوقتی برآوردِ هزینه‌ی سمتِ client به این مقدارِ دلاری برسد، query را متوقف کن. در برابرِ همان برآوردِ total_cost_usd مقایسه می‌شود؛ برای ملاحظاتِ دقت ردیابی هزینه و مصرف را ببین
disallowed_toolslist[str][]ابزارهایی که deny شوند. یک نامِ خالی مثلِ "Bash" ابزار را از کانتکستِ Claude حذف می‌کند. یک قاعده‌ی scope‌دار مثلِ "Bash(rm *)" ابزار را در دسترس می‌گذارد و فراخوانی‌های مطابق را در هر permission mode، از جمله bypassPermissions، deny می‌کند. Permissions را ببین
enable_file_checkpointingboolFalseردیابیِ تغییرِ فایل را برای rewinding فعال کن. File checkpointing را ببین
modelstr | NoneNonealiasِ مدلِ Claude یا نامِ کاملِ مدل. مقادیرِ پذیرفته‌شده و IDهای مخصوصِ provider را ببین
fallback_modelstr | NoneNoneمدلِ fallback برای استفاده در صورتِ شکستِ مدلِ اصلی
betaslist[SdkBeta][]قابلیت‌های بتا که فعال شوند. برای گزینه‌های موجود SdkBeta را ببین
output_formatdict[str, Any] | NoneNoneقالبِ خروجی برای پاسخ‌های ساختاریافته (مثلاً {"type": "json_schema", "schema": {...}}). برای جزئیات Structured outputs را ببین
permission_prompt_tool_namestr | NoneNoneنامِ ابزارِ MCP برای پرامپت‌های دسترسی
cwdstr | Path | NoneNoneدایرکتوریِ کاریِ فعلی
cli_pathstr | Path | NoneNoneمسیرِ سفارشی به فایلِ اجراییِ Claude Code CLI
settingsstr | NoneNoneمسیرِ فایلِ تنظیمات
add_dirslist[str | Path][]دایرکتوری‌های اضافی که Claude می‌تواند به آن‌ها دسترسی داشته باشد
envdict[str, str]{}متغیرهای محیطی که روی محیطِ به‌ارث‌رسیده‌ی پروسه merge می‌شوند. برای متغیرهایی که CLIِ زیربنایی می‌خواند Environment variables و برای متغیرهای مرتبط با timeout بخشِ رسیدگی به پاسخ‌های کند یا متوقف‌شده‌ی API را ببین
extra_argsdict[str, str | None]{}آرگومان‌های CLIِ اضافی که مستقیماً به CLI پاس داده شوند
max_buffer_sizeint | NoneNoneبیشینه‌ی بایت هنگامِ buffer کردنِ stdoutِ CLI
debug_stderrAnysys.stderrمنسوخ - آبجکتِ فایل‌مانند برای خروجیِ debug. به‌جای آن از callbackِ stderr استفاده کن
stderrCallable[[str], None] | NoneNoneتابعِ callback برای خروجیِ stderr از CLI
can_use_toolCanUseTool | NoneNoneتابعِ callbackِ دسترسیِ ابزار. برای جزئیات انواعِ Permission را ببین
hooksdict[HookEvent, list[HookMatcher]] | NoneNoneپیکربندی‌های hook برای رهگیریِ eventها
userstr | NoneNoneشناسه‌ی کاربر
include_partial_messagesboolFalseeventهای streamingِ پیامِ جزئی را بگنجان. وقتی فعال باشد، پیام‌های StreamEvent yield می‌شوند
include_hook_eventsboolFalseeventهای چرخه‌ی حیاتِ hook را به‌عنوانِ آبجکت‌های HookEventMessage در استریمِ پیام بگنجان
fork_sessionboolFalseهنگامِ resume با resume، به‌جای ادامه‌ی نشستِ اصلی به یک session IDِ جدید fork کن
agentsdict[str, AgentDefinition] | NoneNoneساب‌ایجنت‌های تعریف‌شده به‌صورتِ برنامه‌نویسی‌شده
pluginslist[SdkPluginConfig][]plugin‌های سفارشی را از مسیرهای محلی بارگذاری کن. برای جزئیات Plugins را ببین
sandboxSandboxSettings | NoneNoneرفتارِ sandbox را به‌صورتِ برنامه‌نویسی‌شده پیکربندی کن. برای جزئیات Sandbox settings را ببین
setting_sourceslist[SettingSource] | NoneNone (پیش‌فرضِ CLI: همه‌ی منابع)کنترل کن که کدام تنظیماتِ فایل‌سیستم بارگذاری شوند. برای غیرفعال‌کردنِ تنظیماتِ user، project و local مقدارِ [] پاس بده. تنظیماتِ managed policy صرف‌نظر از این بارگذاری می‌شوند. Use Claude Code features را ببین
skillslist[str] | Literal["all"] | NoneNoneskillهای در دسترسِ نشست. برای فعال‌کردنِ هر skillِ کشف‌شده "all" پاس بده، یا فهرستی از نام‌های skill. وقتی تنظیم شود، SDK ابزارِ Skill را به‌صورتِ خودکار به allowed_tools اضافه می‌کند. اگر tools را هم پاس می‌دهی، "Skill" را در آن فهرست بگنجان. Skills را ببین
max_thinking_tokensint | NoneNoneمنسوخ - بیشینه‌ی توکن برای بلاک‌های thinking. به‌جای آن از thinking استفاده کن
thinkingThinkingConfig | NoneNoneرفتارِ extended thinking را کنترل می‌کند. بر max_thinking_tokens اولویت دارد
effortEffortLevel | NoneNoneسطحِ effort برای عمقِ thinking. تنظیمِ سطحِ effort را ببین
session_storeSessionStore | NoneNoneرونوشتِ transcriptهای نشست را به یک backendِ بیرونی بفرست تا هر host بتواند آن‌ها را resume کند. پایداریِ نشست‌ها در ذخیره‌سازیِ بیرونی را ببین
session_store_flushLiteral["batched", "eager"]"batched"کِی ورودی‌های transcriptِ رونوشت‌شده به session_store فلاش شوند. "batched" یک‌بار در هر نوبت یا وقتی buffer پر شود فلاش می‌کند؛ "eager" پس از هر frame یک فلاشِ پس‌زمینه تریگر می‌کند. وقتی session_store برابرِ None باشد نادیده گرفته می‌شود

رسیدگی به پاسخ‌های کند یا متوقف‌شده‌ی API

Section titled “رسیدگی به پاسخ‌های کند یا متوقف‌شده‌ی API”

subprocessِ CLI چند متغیرِ محیطی می‌خواند که timeoutهای API و تشخیصِ توقف (stall) را کنترل می‌کنند. آن‌ها را از طریقِ ClaudeAgentOptions.env پاس بده:

options = ClaudeAgentOptions(
env={
"API_TIMEOUT_MS": "120000",
"CLAUDE_CODE_MAX_RETRIES": "2",
"CLAUDE_ASYNC_AGENT_STALL_TIMEOUT_MS": "120000",
},
)
  • API_TIMEOUT_MS: timeoutِ هر-درخواست روی clientِ Anthropic، به میلی‌ثانیه. پیش‌فرض 600000. روی حلقه‌ی اصلی و همه‌ی ساب‌ایجنت‌ها اعمال می‌شود.
  • CLAUDE_CODE_MAX_RETRIES: بیشینه‌ی retryهای API. پیش‌فرض 10. هر retry پنجره‌ی API_TIMEOUT_MSِ خودش را می‌گیرد، پس بدترین حالتِ زمانِ کلی تقریباً API_TIMEOUT_MS × (CLAUDE_CODE_MAX_RETRIES + 1) به‌علاوه‌ی backoff است.
  • CLAUDE_ASYNC_AGENT_STALL_TIMEOUT_MS: watchdogِ توقف برای ساب‌ایجنت‌هایی که با run_in_background راه‌اندازی شده‌اند. پیش‌فرض 600000. روی هر stream event ریست می‌شود؛ هنگامِ توقف ساب‌ایجنت را abort می‌کند، تسک را failed علامت می‌زند، و خطا را با هر نتیجه‌ی جزئی به والد سرریز می‌کند. روی ساب‌ایجنت‌های همگام اعمال نمی‌شود.
  • CLAUDE_ENABLE_STREAM_WATCHDOG=1 با CLAUDE_STREAM_IDLE_TIMEOUT_MS: وقتی headerها رسیده‌اند اما بدنه‌ی پاسخ از streaming بازمی‌ایستد، درخواست را abort می‌کند. وقتی CLAUDE_ENABLE_STREAM_WATCHDOG تنظیم نشده باشد، پیش‌فرض روی Anthropic APIِ مستقیم سمتِ سرور کنترل می‌شود و روی سایرِ providerها خاموش است. CLAUDE_STREAM_IDLE_TIMEOUT_MS پیش‌فرضِ 300000 دارد و به همان مقدارِ کمینه clamp می‌شود. درخواستِ abort‌شده از مسیرِ retryِ عادی عبور می‌کند.

پیکربندی برای اعتبارسنجیِ خروجیِ ساختاریافته. این را به‌عنوانِ یک dict به فیلدِ output_format روی ClaudeAgentOptions پاس بده:

# Expected dict shape for output_format
{
"type": "json_schema",
"schema": {...}, # Your JSON Schema definition
}
فیلدالزامیتوضیح
typeبلهبرای اعتبارسنجیِ JSON Schema باید "json_schema" باشد
schemaبلهتعریفِ JSON Schema برای اعتبارسنجیِ خروجی

پیکربندی برای استفاده از system promptِ presetِ Claude Code با افزوده‌های اختیاری.

class SystemPromptPreset(TypedDict):
type: Literal["preset"]
preset: Literal["claude_code"]
append: NotRequired[str]
exclude_dynamic_sections: NotRequired[bool]
فیلدالزامیتوضیح
typeبلهبرای استفاده از یک system promptِ preset باید "preset" باشد
presetبلهبرای استفاده از system promptِ Claude Code باید "claude_code" باشد
appendخیردستورالعمل‌های اضافی برای الحاق به system promptِ preset
exclude_dynamic_sectionsخیرکانتکستِ هر-نشستی مثلِ دایرکتوریِ کاری، پرچمِ git-repo و مسیرهای auto-memory را از system prompt به اولین پیامِ کاربر منتقل می‌کند. استفاده‌ی مجددِ prompt-cache را بینِ کاربران و ماشین‌ها بهبود می‌دهد. Modify system prompts را ببین

کنترل می‌کند که SDK تنظیمات را از کدام منابعِ پیکربندیِ مبتنی‌بر فایل‌سیستم بارگذاری کند.

SettingSource = Literal["user", "project", "local"]
مقدارتوضیحمکان
"user"تنظیماتِ سراسریِ کاربر~/.claude/settings.json
"project"تنظیماتِ مشترکِ پروژه (تحتِ کنترلِ نسخه).claude/settings.json
"local"تنظیماتِ محلیِ پروژه (خارج از کنترلِ نسخه).claude/settings.local.json

وقتی setting_sources حذف شود یا None باشد، query() همان تنظیماتِ فایل‌سیستمِ Claude Code CLI را بارگذاری می‌کند: user، project و local. تنظیماتِ managed policy در همه‌ی حالت‌ها بارگذاری می‌شوند. برای ورودی‌هایی که صرف‌نظر از این گزینه خوانده می‌شوند و نحوه‌ی غیرفعال‌کردنشان، What settingSources does not control را ببین.

چرا از setting_sources استفاده کنیم

Section titled “چرا از setting_sources استفاده کنیم”

غیرفعال‌کردنِ تنظیماتِ فایل‌سیستم:

# Do not load user, project, or local settings from disk
from claude_agent_sdk import query, ClaudeAgentOptions
async for message in query(
prompt="Analyze this code",
options=ClaudeAgentOptions(
setting_sources=[]
),
):
print(message)

بارگذاریِ صریحِ همه‌ی تنظیماتِ فایل‌سیستم:

from claude_agent_sdk import query, ClaudeAgentOptions
async for message in query(
prompt="Analyze this code",
options=ClaudeAgentOptions(
setting_sources=["user", "project", "local"]
),
):
print(message)

بارگذاریِ فقط منابعِ تنظیماتِ مشخص:

# Load only project settings, ignore user and local
async for message in query(
prompt="Run CI checks",
options=ClaudeAgentOptions(
setting_sources=["project"] # Only .claude/settings.json
),
):
print(message)

محیط‌های تست و CI:

# Ensure consistent behavior in CI by excluding local settings
async for message in query(
prompt="Run tests",
options=ClaudeAgentOptions(
setting_sources=["project"], # Only team-shared settings
permission_mode="bypassPermissions",
),
):
print(message)

اپلیکیشن‌های فقط-SDK:

# Define everything programmatically.
# Pass [] to opt out of filesystem setting sources.
async for message in query(
prompt="Review this PR",
options=ClaudeAgentOptions(
setting_sources=[],
agents={...},
mcp_servers={...},
allowed_tools=["Read", "Grep", "Glob"],
),
):
print(message)

بارگذاریِ دستورالعمل‌های پروژه‌ی CLAUDE.md:

# Load project settings to include CLAUDE.md files
async for message in query(
prompt="Add a new feature following project conventions",
options=ClaudeAgentOptions(
system_prompt={
"type": "preset",
"preset": "claude_code", # Use Claude Code's system prompt
},
setting_sources=["project"], # Loads CLAUDE.md from project
allowed_tools=["Read", "Write", "Edit"],
),
):
print(message)

وقتی چند منبع بارگذاری شوند، تنظیمات با این اولویت merge می‌شوند (از بالاترین به پایین‌ترین):

  1. تنظیماتِ local (.claude/settings.local.json)
  2. تنظیماتِ project (.claude/settings.json)
  3. تنظیماتِ user (~/.claude/settings.json)

گزینه‌های برنامه‌نویسی‌شده مثلِ agents و allowed_tools بر تنظیماتِ فایل‌سیستمِ user، project و local غلبه می‌کنند. تنظیماتِ managed policy بر گزینه‌های برنامه‌نویسی‌شده اولویت دارند.

پیکربندی برای یک ساب‌ایجنت که به‌صورتِ برنامه‌نویسی‌شده تعریف شده.

@dataclass
class AgentDefinition:
description: str
prompt: str
tools: list[str] | None = None
disallowedTools: list[str] | None = None
model: str | None = None
skills: list[str] | None = None
memory: Literal["user", "project", "local"] | None = None
mcpServers: list[str | dict[str, Any]] | None = None
initialPrompt: str | None = None
maxTurns: int | None = None
background: bool | None = None
effort: EffortLevel | int | None = None
permissionMode: PermissionMode | None = None
فیلدالزامیتوضیح
descriptionبلهتوضیحِ زبانِ طبیعی درباره‌ی اینکه کِی از این ایجنت استفاده شود
promptبلهsystem promptِ ایجنت
toolsخیرآرایه‌ی نام‌های ابزارِ مجاز. اگر حذف شود، همه‌ی ابزارها به ارث می‌رسند
disallowedToolsخیرآرایه‌ی نام‌های ابزار که از مجموعه‌ی ابزارِ ایجنت حذف شوند
modelخیرoverride مدل برای این ایجنت. یک alias مثلِ "sonnet"، "opus"، "haiku" یا "inherit"، یا یک model IDِ کامل را می‌پذیرد. اگر حذف شود، از مدلِ اصلی استفاده می‌کند
skillsخیرفهرستِ نام‌های skill که هنگامِ راه‌اندازی پیش‌بارگذاری در کانتکستِ ایجنت شوند. skillهای فهرست‌نشده از طریقِ ابزارِ Skill قابلِ‌فراخوانی می‌مانند
memoryخیرمنبعِ حافظه برای این ایجنت: "user"، "project" یا "local"
mcpServersخیرسرورهای MCPِ در دسترسِ این ایجنت. هر ورودی یک نامِ سرور یا یک dictِ inlineِ {name: config} است
initialPromptخیروقتی این ایجنت به‌عنوانِ ایجنتِ thread اصلی اجرا می‌شود، به‌صورتِ خودکار به‌عنوانِ اولین نوبتِ کاربر submit می‌شود
maxTurnsخیربیشینه‌ی تعدادِ نوبت‌های ایجنتیک پیش از توقفِ ایجنت
backgroundخیراین ایجنت را هنگامِ فراخوانی به‌صورتِ یک تسکِ پس‌زمینه‌ی non-blocking اجرا کن
effortخیرسطحِ تلاشِ استدلال برای این ایجنت. یک سطحِ نام‌گذاری‌شده یا یک عددِ صحیح را می‌پذیرد. EffortLevel را ببین
permissionModeخیرpermission mode برای اجرای ابزار درونِ این ایجنت. PermissionMode را ببین

permission modeها برای کنترلِ اجرای ابزار.

PermissionMode = Literal[
"default", # Standard permission behavior
"acceptEdits", # Auto-accept file edits
"plan", # Planning mode - explore without editing
"dontAsk", # Deny anything not pre-approved instead of prompting
"bypassPermissions", # Bypass permission checks; explicit ask rules still prompt (use with caution)
]

سطوحِ effort برای هدایتِ عمقِ thinking.

EffortLevel = Literal[
"low", # Minimal thinking, fastest responses
"medium", # Moderate thinking
"high", # Deep reasoning
"xhigh", # Extended reasoning (Opus 4.8 and Opus 4.7; falls back to "high" on other models)
"max", # Maximum effort
]

aliasِ نوع برای توابعِ callbackِ دسترسیِ ابزار.

CanUseTool = Callable[
[str, dict[str, Any], ToolPermissionContext], Awaitable[PermissionResult]
]

این callback این‌ها را دریافت می‌کند:

  • tool_name: نامِ ابزاری که فراخوانی می‌شود
  • input_data: پارامترهای ورودیِ ابزار
  • context: یک ToolPermissionContext با اطلاعاتِ اضافی

یک PermissionResult برمی‌گرداند (یا PermissionResultAllow یا PermissionResultDeny).

اطلاعاتِ کانتکست که به callbackهای دسترسیِ ابزار پاس داده می‌شود.

@dataclass
class ToolPermissionContext:
signal: Any | None = None # Future: abort signal support
suggestions: list[PermissionUpdate] = field(default_factory=list)
blocked_path: str | None = None
decision_reason: str | None = None
title: str | None = None
display_name: str | None = None
description: str | None = None
فیلدنوعتوضیح
signalAny | Noneرزرو برای پشتیبانیِ آینده‌ی abort signal
suggestionslist[PermissionUpdate]پیشنهادهای به‌روزرسانیِ دسترسی از CLI. پرامپت‌های Bash شاملِ یک پیشنهاد با مقصدِ localSettings هستند، پس برگرداندنِ آن در updated_permissions قاعده را در .claude/settings.local.json می‌نویسد و در طولِ نشست‌ها پایدار می‌ماند.
blocked_pathstr | Noneمسیرِ فایلی که درخواستِ دسترسی را تریگر کرد، در صورتِ امکان. مثلاً وقتی یک فرمانِ Bash تلاش می‌کند به مسیری بیرونِ دایرکتوری‌های مجاز دسترسی پیدا کند
decision_reasonstr | Noneدلیلِ تریگرشدنِ این درخواستِ دسترسی. وقتی یک hookِ PreToolUse مقدارِ "ask" برگرداند، از permissionDecisionReasonِ آن فوروارد می‌شود
titlestr | Noneجمله‌ی کاملِ پرامپتِ دسترسی، مثلِ Claude wants to read foo.txt. وقتی موجود است به‌عنوانِ متنِ اصلیِ پرامپت استفاده کن
display_namestr | Noneعبارتِ اسمیِ کوتاه برای اقدامِ ابزار، مثلِ Read file، مناسب برای برچسبِ دکمه
descriptionstr | Noneزیرنویسِ خوانا برای انسان برای UIِ دسترسی

نوعِ union برای نتایجِ callbackِ دسترسی.

PermissionResult = PermissionResultAllow | PermissionResultDeny

نتیجه‌ای که نشان می‌دهد فراخوانیِ ابزار باید allow شود.

@dataclass
class PermissionResultAllow:
behavior: Literal["allow"] = "allow"
updated_input: dict[str, Any] | None = None
updated_permissions: list[PermissionUpdate] | None = None
فیلدنوعپیش‌فرضتوضیح
behaviorLiteral["allow"]"allow"باید “allow” باشد
updated_inputdict[str, Any] | NoneNoneورودیِ تغییریافته برای استفاده به‌جای اصلی
updated_permissionslist[PermissionUpdate] | NoneNoneبه‌روزرسانی‌های دسترسی که اعمال شوند

نتیجه‌ای که نشان می‌دهد فراخوانیِ ابزار باید deny شود.

@dataclass
class PermissionResultDeny:
behavior: Literal["deny"] = "deny"
message: str = ""
interrupt: bool = False
فیلدنوعپیش‌فرضتوضیح
behaviorLiteral["deny"]"deny"باید “deny” باشد
messagestr""پیامِ توضیح‌دهنده‌ی اینکه چرا ابزار deny شد
interruptboolFalseاینکه آیا اجرای فعلی interrupt شود

پیکربندی برای به‌روزرسانیِ دسترسی‌ها به‌صورتِ برنامه‌نویسی‌شده.

@dataclass
class PermissionUpdate:
type: Literal[
"addRules",
"replaceRules",
"removeRules",
"setMode",
"addDirectories",
"removeDirectories",
]
rules: list[PermissionRuleValue] | None = None
behavior: Literal["allow", "deny", "ask"] | None = None
mode: PermissionMode | None = None
directories: list[str] | None = None
destination: (
Literal["userSettings", "projectSettings", "localSettings", "session"] | None
) = None
فیلدنوعتوضیح
typeLiteral[...]نوعِ عملیاتِ به‌روزرسانیِ دسترسی
ruleslist[PermissionRuleValue] | Noneقواعد برای عملیاتِ add/replace/remove
behaviorLiteral["allow", "deny", "ask"] | Noneرفتار برای عملیاتِ مبتنی‌بر قاعده
modePermissionMode | Noneحالت برای عملیاتِ setMode
directorieslist[str] | Noneدایرکتوری‌ها برای عملیاتِ add/remove دایرکتوری
destinationLiteral[...] | Noneاینکه به‌روزرسانیِ دسترسی کجا اعمال شود

یک قاعده برای add، replace یا remove در یک به‌روزرسانیِ دسترسی.

@dataclass
class PermissionRuleValue:
tool_name: str
rule_content: str | None = None

پیکربندیِ ابزارهای preset برای استفاده از مجموعه‌ی ابزارِ پیش‌فرضِ Claude Code.

class ToolsPreset(TypedDict):
type: Literal["preset"]
preset: Literal["claude_code"]

رفتارِ extended thinking را کنترل می‌کند. یک union از سه پیکربندی:

ThinkingDisplay = Literal["summarized", "omitted"]
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
نوعفیلدهاتوضیح
adaptivetype, displayClaude خودش به‌صورتِ تطبیقی تصمیم می‌گیرد کِی فکر کند
enabledtype, budget_tokens, displaythinking را با یک بودجه‌ی توکنِ مشخص فعال کن
disabledtypethinking را غیرفعال کن

فیلدِ اختیاریِ display کنترل می‌کند که متنِ thinking به‌صورتِ "summarized" برگردانده شود یا "omitted". روی Claude Opus 4.7 و بالاتر، پیش‌فرضِ API برابرِ "omitted" است، پس برای دریافتِ محتوای thinking در خروجی‌های ThinkingBlock مقدارِ "summarized" را تنظیم کن.

چون این‌ها کلاس‌های TypedDict هستند، در زمانِ اجرا plain dict اند. یا آن‌ها را به‌صورتِ dict literal بساز یا کلاس را مثلِ یک constructor فراخوانی کن؛ هر دو یک dict تولید می‌کنند. فیلدها را با config["budget_tokens"] دسترسی بگیر، نه config.budget_tokens:

from claude_agent_sdk import ClaudeAgentOptions, ThinkingConfigEnabled
# Option 1: dict literal (recommended, no import needed)
options = ClaudeAgentOptions(thinking={"type": "enabled", "budget_tokens": 20000})
# Option 2: constructor-style (returns a plain dict)
config = ThinkingConfigEnabled(type="enabled", budget_tokens=20000)
print(config["budget_tokens"]) # 20000
# config.budget_tokens would raise AttributeError

نوعِ Literal برای قابلیت‌های بتای SDK.

SdkBeta = Literal["context-1m-2025-08-07"]

برای فعال‌کردنِ قابلیت‌های بتا با فیلدِ betas در ClaudeAgentOptions استفاده کن.

پیکربندی برای سرورهای SDK MCP که با create_sdk_mcp_server() ساخته شده‌اند.

class McpSdkServerConfig(TypedDict):
type: Literal["sdk"]
name: str
instance: Any # MCP Server instance

نوعِ union برای پیکربندی‌های سرورِ MCP.

McpServerConfig = (
McpStdioServerConfig | McpSSEServerConfig | McpHttpServerConfig | McpSdkServerConfig
)
class McpStdioServerConfig(TypedDict):
type: NotRequired[Literal["stdio"]] # Optional for backwards compatibility
command: str
args: NotRequired[list[str]]
env: NotRequired[dict[str, str]]
class McpSSEServerConfig(TypedDict):
type: Literal["sse"]
url: str
headers: NotRequired[dict[str, str]]
class McpHttpServerConfig(TypedDict):
type: Literal["http"]
url: str
headers: NotRequired[dict[str, str]]

پیکربندیِ یک سرورِ MCP، آن‌طور که توسطِ get_mcp_status() گزارش می‌شود. این union همه‌ی نوعِ transportهای McpServerConfig به‌علاوه‌ی یک نوعِ فقط-خروجیِ claudeai-proxy برای سرورهایی است که از طریقِ claude.ai پراکسی می‌شوند.

McpServerStatusConfig = (
McpStdioServerConfig
| McpSSEServerConfig
| McpHttpServerConfig
| McpSdkServerConfigStatus
| McpClaudeAIProxyServerConfig
)

McpSdkServerConfigStatus شکلِ serializableِ McpSdkServerConfig است که فقط فیلدهای type ("sdk") و name (str) را دارد؛ instanceِ in-process حذف می‌شود. McpClaudeAIProxyServerConfig فیلدهای type ("claudeai-proxy"url (str) و id (str) را دارد.

پاسخ از ClaudeSDKClient.get_mcp_status(). فهرستِ وضعیتِ سرورها را زیرِ کلیدِ mcpServers wrap می‌کند.

class McpStatusResponse(TypedDict):
mcpServers: list[McpServerStatus]

وضعیتِ یک سرورِ MCPِ متصل، که در McpStatusResponse قرار دارد.

class McpServerStatus(TypedDict):
name: str
status: McpServerConnectionStatus # "connected" | "failed" | "needs-auth" | "pending" | "disabled"
serverInfo: NotRequired[McpServerInfo]
error: NotRequired[str]
config: NotRequired[McpServerStatusConfig]
scope: NotRequired[str]
tools: NotRequired[list[McpToolInfo]]
فیلدنوعتوضیح
namestrنامِ سرور
statusstrیکی از "connected"، "failed"، "needs-auth"، "pending" یا "disabled"
serverInfodict (اختیاری)نام و نسخه‌ی سرور ({"name": str, "version": str})
errorstr (اختیاری)پیامِ خطا اگر سرور نتوانست وصل شود
configMcpServerStatusConfig (اختیاری)پیکربندیِ سرور. همان شکلِ McpServerConfig (stdio، SSE، HTTP یا SDK)، به‌علاوه‌ی یک نوعِ claudeai-proxy برای سرورهای متصل از طریقِ claude.ai
scopestr (اختیاری)scopeِ پیکربندی
toolslist (اختیاری)ابزارهای ارائه‌شده توسطِ این سرور، هرکدام با فیلدهای name، description و annotations

پیکربندی برای بارگذاریِ plugin‌ها در SDK.

class SdkPluginConfig(TypedDict):
type: Literal["local"]
path: str
فیلدنوعتوضیح
typeLiteral["local"]باید "local" باشد (فعلاً فقط plugin‌های محلی پشتیبانی می‌شوند)
pathstrمسیرِ مطلق یا نسبی به دایرکتوریِ plugin

مثال:

plugins = [
{"type": "local", "path": "./my-plugin"},
{"type": "local", "path": "/absolute/path/to/plugin"},
]

برای اطلاعاتِ کامل درباره‌ی ساخت و استفاده از plugin‌ها، Plugins را ببین.

نوعِ union از همه‌ی پیام‌های ممکن.

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

پیامِ ورودیِ کاربر.

@dataclass
class UserMessage:
content: str | list[ContentBlock]
uuid: str | None = None
parent_tool_use_id: str | None = None
tool_use_result: dict[str, Any] | None = None
فیلدنوعتوضیح
contentstr | list[ContentBlock]محتوای پیام به‌صورتِ متن یا content block
uuidstr | Noneشناسه‌ی یکتای پیام
parent_tool_use_idstr | Nonetool use ID اگر این پیام یک پاسخِ نتیجه‌ی ابزار باشد
tool_use_resultdict[str, Any] | Noneداده‌ی نتیجه‌ی ابزار در صورتِ امکان

پیامِ پاسخِ assistant با content blockها.

@dataclass
class AssistantMessage:
content: list[ContentBlock]
model: str
parent_tool_use_id: str | None = None
error: AssistantMessageError | None = None
usage: dict[str, Any] | None = None
message_id: str | None = None
فیلدنوعتوضیح
contentlist[ContentBlock]فهرستِ content blockها در پاسخ
modelstrمدلی که پاسخ را تولید کرد
parent_tool_use_idstr | Nonetool use ID اگر این یک پاسخِ تودرتو باشد
errorAssistantMessageError | Noneنوعِ خطا اگر پاسخ به یک خطا برخورد کرد
usagedict[str, Any] | Noneمصرفِ توکنِ هر-پیامی (همان کلیدهای ResultMessage.usage)
message_idstr | NoneAPI message ID. چند پیام از یک نوبت همان ID را به اشتراک می‌گذارند

نوعِ خطاهای ممکن برای پیام‌های assistant.

AssistantMessageError = Literal[
"authentication_failed",
"billing_error",
"rate_limit",
"invalid_request",
"server_error",
"max_output_tokens",
"unknown",
]

پیامِ سیستم با فراداده.

@dataclass
class SystemMessage:
subtype: str
data: dict[str, Any]

پیامِ نتیجه‌ی نهایی با اطلاعاتِ هزینه و مصرف.

@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

فیلدِ subtype تعیین می‌کند کدام فیلدهای دیگر پر می‌شوند. یکی از "success"، "error_during_execution"، "error_max_turns"، "error_max_budget_usd" یا "error_max_structured_output_retries" است. dataclassِ پایتون همه‌ی نوع‌ها را در یک شکل flatten می‌کند، پس فیلدهایی که برای subtypeِ برگشتی صدق نمی‌کنند None اند.

چند فیلد وقتی گفتگو با یک خطا پایان می‌یابد جزئیاتِ تشخیصی حمل می‌کنند:

  • is_error: وقتی گفتگو در یک وضعیتِ خطا پایان یابد True است. روی subtypeهای error_* همیشه True است. روی subtype="success" وقتی True است که درخواستِ نهاییِ مدل ناموفق باشد، یعنی حلقه‌ی ایجنت کامل شده ولی آخرین فراخوانیِ API یک خطا برگردانده است.
  • api_error_status: کدِ وضعیتِ HTTPِ خطای پایان‌دهنده‌ی API. وقتی نوبت بدونِ آن پایان یابد None است. فقط روی subtype="success" پر می‌شود.
  • result: متنِ پیامِ نهاییِ assistant روی subtype="success"، یا None روی subtypeهای error_*. وقتی subtype="success" و is_error=True باشد، این رشته‌ی خطای API را در صورتِ موجودبودن نگه می‌دارد ولی می‌تواند خالی باشد، پس برای جزئیات api_error_status و محتوای AssistantMessageِ پیشین را بررسی کن.
  • errors: رشته‌های خطای سطحِ حلقه مثلِ پیامِ max-turns. فقط روی subtypeهای error_* پر می‌شود.

dictِ usage وقتی موجود باشد این کلیدها را در بر دارد:

کلیدنوعتوضیح
input_tokensintکلِ توکن‌های ورودیِ مصرف‌شده.
output_tokensintکلِ توکن‌های خروجیِ تولیدشده.
cache_creation_input_tokensintتوکن‌های استفاده‌شده برای ساختِ ورودی‌های جدیدِ cache.
cache_read_input_tokensintتوکن‌های خوانده‌شده از ورودی‌های موجودِ cache.

dictِ model_usage نام‌های مدل را به مصرفِ هر-مدلی نگاشت می‌کند. کلیدهای dictِ درونی از camelCase استفاده می‌کنند چون مقدار بدونِ تغییر از پروسه‌ی CLIِ زیربنایی پاس داده می‌شود، مطابقِ نوعِ تایپ‌اسکریپتِ ModelUsage:

کلیدنوعتوضیح
inputTokensintتوکن‌های ورودی برای این مدل.
outputTokensintتوکن‌های خروجی برای این مدل.
cacheReadInputTokensintتوکن‌های خواندنِ cache برای این مدل.
cacheCreationInputTokensintتوکن‌های ساختِ cache برای این مدل.
webSearchRequestsintدرخواست‌های web search که این مدل انجام داده.
costUSDfloatهزینه‌ی برآوردی به دلار برای این مدل، محاسبه‌شده سمتِ client. برای ملاحظاتِ هزینه ردیابی هزینه و مصرف را ببین.
contextWindowintاندازه‌ی context window برای این مدل.
maxOutputTokensintحدِ بیشینه‌ی توکنِ خروجی برای این مدل.

stream event برای به‌روزرسانی‌های جزئیِ پیام در حینِ streaming. فقط وقتی include_partial_messages=True در ClaudeAgentOptions باشد دریافت می‌شود. از طریقِ from claude_agent_sdk.types import StreamEvent ایمپورت کن.

@dataclass
class StreamEvent:
uuid: str
session_id: str
event: dict[str, Any] # The raw Claude API stream event
parent_tool_use_id: str | None = None
فیلدنوعتوضیح
uuidstrشناسه‌ی یکتای این event
session_idstrشناسه‌ی نشست
eventdict[str, Any]داده‌ی خامِ stream eventِ Claude API
parent_tool_use_idstr | Noneparent tool use ID اگر این event از یک ساب‌ایجنت باشد

وقتی وضعیتِ rate limit تغییر می‌کند منتشر می‌شود (مثلاً از "allowed" به "allowed_warning"). از این برای هشدار به کاربران پیش از رسیدن به یک حدِ سخت، یا برای back off وقتی وضعیت "rejected" است استفاده کن.

@dataclass
class RateLimitEvent:
rate_limit_info: RateLimitInfo
uuid: str
session_id: str
فیلدنوعتوضیح
rate_limit_infoRateLimitInfoوضعیتِ فعلیِ rate limit
uuidstrشناسه‌ی یکتای event
session_idstrشناسه‌ی نشست

وضعیتِ rate limit که توسطِ RateLimitEvent حمل می‌شود.

RateLimitStatus = Literal["allowed", "allowed_warning", "rejected"]
RateLimitType = Literal[
"five_hour", "seven_day", "seven_day_opus", "seven_day_sonnet", "overage"
]
@dataclass
class RateLimitInfo:
status: RateLimitStatus
resets_at: int | None = None
rate_limit_type: RateLimitType | None = None
utilization: float | None = None
overage_status: RateLimitStatus | None = None
overage_resets_at: int | None = None
overage_disabled_reason: str | None = None
raw: dict[str, Any] = field(default_factory=dict)
فیلدنوعتوضیح
statusRateLimitStatusوضعیتِ فعلی. "allowed_warning" یعنی نزدیک‌شدن به حد؛ "rejected" یعنی به حد رسیده‌ای
resets_atint | Nonetimestampِ یونیکس وقتی پنجره‌ی rate limit ریست می‌شود
rate_limit_typeRateLimitType | Noneاینکه کدام پنجره‌ی rate limit صدق می‌کند
utilizationfloat | Noneکسرِ rate limitِ مصرف‌شده (۰٫۰ تا ۱٫۰)
overage_statusRateLimitStatus | Noneوضعیتِ مصرفِ overageِ pay-as-you-go، در صورتِ امکان
overage_resets_atint | Nonetimestampِ یونیکس وقتی پنجره‌ی overage ریست می‌شود
overage_disabled_reasonstr | Noneچرا overage در دسترس نیست، اگر وضعیت "rejected" باشد
rawdict[str, Any]dictِ خامِ کامل از CLI، شاملِ فیلدهایی که بالا مدل نشده‌اند

وقتی یک تسکِ پس‌زمینه شروع می‌شود منتشر می‌شود. یک تسکِ پس‌زمینه هر چیزی است که بیرونِ نوبتِ اصلی ردیابی می‌شود: یک فرمانِ Bashِ پس‌زمینه‌شده، یک watchِ Monitor، یک ساب‌ایجنتِ ساخته‌شده از طریقِ ابزارِ Agent، یا یک ایجنتِ remote. فیلدِ task_type می‌گوید کدام است. این نام‌گذاری ربطی به تغییرِ نامِ ابزارِ Task به Agent ندارد.

@dataclass
class TaskStartedMessage(SystemMessage):
task_id: str
description: str
uuid: str
session_id: str
tool_use_id: str | None = None
task_type: str | None = None
فیلدنوعتوضیح
task_idstrشناسه‌ی یکتای تسک
descriptionstrتوضیحِ تسک
uuidstrشناسه‌ی یکتای پیام
session_idstrشناسه‌ی نشست
tool_use_idstr | Nonetool use IDِ مرتبط
task_typestr | Noneکدام نوع تسکِ پس‌زمینه: "local_bash" برای Bashِ پس‌زمینه و watchهای Monitor، "local_agent" یا "remote_agent"

داده‌ی توکن و زمان‌بندی برای یک تسکِ پس‌زمینه.

class TaskUsage(TypedDict):
total_tokens: int
tool_uses: int
duration_ms: int

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

@dataclass
class TaskProgressMessage(SystemMessage):
task_id: str
description: str
usage: TaskUsage
uuid: str
session_id: str
tool_use_id: str | None = None
last_tool_name: str | None = None
فیلدنوعتوضیح
task_idstrشناسه‌ی یکتای تسک
descriptionstrتوضیحِ وضعیتِ فعلی
usageTaskUsageمصرفِ توکن برای این تسک تا کنون
uuidstrشناسه‌ی یکتای پیام
session_idstrشناسه‌ی نشست
tool_use_idstr | Nonetool use IDِ مرتبط
last_tool_namestr | Noneنامِ آخرین ابزاری که تسک استفاده کرد

وقتی یک تسکِ پس‌زمینه کامل می‌شود، ناموفق می‌شود یا متوقف می‌شود منتشر می‌شود. تسک‌های پس‌زمینه شاملِ فرمان‌های Bashِ run_in_background، watchهای Monitor و ساب‌ایجنت‌های پس‌زمینه‌اند.

@dataclass
class TaskNotificationMessage(SystemMessage):
task_id: str
status: TaskNotificationStatus # "completed" | "failed" | "stopped"
output_file: str
summary: str
uuid: str
session_id: str
tool_use_id: str | None = None
usage: TaskUsage | None = None
فیلدنوعتوضیح
task_idstrشناسه‌ی یکتای تسک
statusTaskNotificationStatusیکی از "completed"، "failed" یا "stopped"
output_filestrمسیرِ فایلِ خروجیِ تسک
summarystrخلاصه‌ی نتیجه‌ی تسک
uuidstrشناسه‌ی یکتای پیام
session_idstrشناسه‌ی نشست
tool_use_idstr | Nonetool use IDِ مرتبط
usageTaskUsage | Noneمصرفِ توکنِ نهایی برای تسک

نوعِ union از همه‌ی content blockها.

ContentBlock = TextBlock | ThinkingBlock | ToolUseBlock | ToolResultBlock

content blockِ متنی.

@dataclass
class TextBlock:
text: str

content blockِ thinking (برای مدل‌هایی با قابلیتِ thinking).

@dataclass
class ThinkingBlock:
thinking: str
signature: str

blockِ درخواستِ استفاده از ابزار.

@dataclass
class ToolUseBlock:
id: str
name: str
input: dict[str, Any]

blockِ نتیجه‌ی اجرای ابزار.

@dataclass
class ToolResultBlock:
tool_use_id: str
content: str | list[dict[str, Any]] | None = None
is_error: bool | None = None

کلاسِ پایه‌ی استثنا برای همه‌ی خطاهای SDK.

class ClaudeSDKError(Exception):
"""Base error for Claude SDK."""

وقتی Claude Code CLI نصب نشده یا پیدا نشود raise می‌شود.

class CLINotFoundError(CLIConnectionError):
def __init__(
self, message: str = "Claude Code not found", cli_path: str | None = None
):
"""
Args:
message: Error message (default: "Claude Code not found")
cli_path: Optional path to the CLI that was not found
"""

وقتی اتصال به Claude Code ناموفق باشد raise می‌شود.

class CLIConnectionError(ClaudeSDKError):
"""Failed to connect to Claude Code."""

وقتی پروسه‌ی Claude Code ناموفق باشد raise می‌شود.

class ProcessError(ClaudeSDKError):
def __init__(
self, message: str, exit_code: int | None = None, stderr: str | None = None
):
self.exit_code = exit_code
self.stderr = stderr

وقتی parse کردنِ JSON ناموفق باشد raise می‌شود.

class CLIJSONDecodeError(ClaudeSDKError):
def __init__(self, line: str, original_error: Exception):
"""
Args:
line: The line that failed to parse
original_error: The original JSON decode exception
"""
self.line = line
self.original_error = original_error

برای راهنمای جامعِ استفاده از hookها با مثال و الگوهای رایج، راهنمای Hooks را ببین.

نوعِ eventهای hookِ پشتیبانی‌شده.

HookEvent = Literal[
"PreToolUse", # Called before tool execution
"PostToolUse", # Called after tool execution
"PostToolUseFailure", # Called when a tool execution fails
"UserPromptSubmit", # Called when user submits a prompt
"Stop", # Called when stopping execution
"SubagentStop", # Called when a subagent stops
"PreCompact", # Called before message compaction
"Notification", # Called for notification events
"SubagentStart", # Called when a subagent starts
"PermissionRequest", # Called when a permission decision is needed
]

تعریفِ نوع برای توابعِ callbackِ hook.

HookCallback = Callable[[HookInput, str | None, HookContext], Awaitable[HookJSONOutput]]

پارامترها:

  • input: ورودیِ hookِ قویاً تایپ‌شده با discriminated unionها بر اساسِ hook_event_name (HookInput را ببین)
  • tool_use_id: شناسه‌ی tool useِ اختیاری (برای hookهای مرتبط با ابزار)
  • context: کانتکستِ hook با اطلاعاتِ اضافی

یک HookJSONOutput برمی‌گرداند که ممکن است شامل این‌ها باشد:

  • decision: "block" برای مسدودکردنِ اقدام
  • systemMessage: پیامِ هشدار که به کاربر نشان داده می‌شود
  • hookSpecificOutput: داده‌ی خروجیِ مخصوصِ hook

اطلاعاتِ کانتکست که به callbackهای hook پاس داده می‌شود.

class HookContext(TypedDict):
signal: Any | None # Future: abort signal support

پیکربندی برای تطبیقِ hookها با eventها یا ابزارهای مشخص.

@dataclass
class HookMatcher:
matcher: str | None = (
None # Tool name or pattern to match (e.g., "Bash", "Write|Edit")
)
hooks: list[HookCallback] = field(
default_factory=list
) # List of callbacks to execute
timeout: float | None = (
None # Timeout in seconds for all hooks in this matcher (default: 60)
)

نوعِ union از همه‌ی نوعِ ورودی‌های hook. نوعِ واقعی به فیلدِ hook_event_name بستگی دارد.

HookInput = (
PreToolUseHookInput
| PostToolUseHookInput
| PostToolUseFailureHookInput
| UserPromptSubmitHookInput
| StopHookInput
| SubagentStopHookInput
| PreCompactHookInput
| NotificationHookInput
| SubagentStartHookInput
| PermissionRequestHookInput
)

فیلدهای پایه‌ای که در همه‌ی نوعِ ورودی‌های hook حضور دارند.

class BaseHookInput(TypedDict):
session_id: str
transcript_path: str
cwd: str
permission_mode: NotRequired[str]
فیلدنوعتوضیح
session_idstrشناسه‌ی نشستِ فعلی
transcript_pathstrمسیرِ فایلِ transcriptِ نشست
cwdstrدایرکتوریِ کاریِ فعلی
permission_modestr (اختیاری)permission modeِ فعلی

داده‌ی ورودی برای eventهای hookِ PreToolUse.

class PreToolUseHookInput(BaseHookInput):
hook_event_name: Literal["PreToolUse"]
tool_name: str
tool_input: dict[str, Any]
tool_use_id: str
agent_id: NotRequired[str]
agent_type: NotRequired[str]
فیلدنوعتوضیح
hook_event_nameLiteral["PreToolUse"]همیشه “PreToolUse”
tool_namestrنامِ ابزاری که می‌خواهد اجرا شود
tool_inputdict[str, Any]پارامترهای ورودیِ ابزار
tool_use_idstrشناسه‌ی یکتا برای این tool use
agent_idstr (اختیاری)شناسه‌ی ساب‌ایجنت، وقتی hook درونِ یک ساب‌ایجنت اجرا می‌شود حضور دارد
agent_typestr (اختیاری)نوعِ ساب‌ایجنت، وقتی hook درونِ یک ساب‌ایجنت اجرا می‌شود حضور دارد

داده‌ی ورودی برای eventهای hookِ PostToolUse.

class PostToolUseHookInput(BaseHookInput):
hook_event_name: Literal["PostToolUse"]
tool_name: str
tool_input: dict[str, Any]
tool_response: Any
tool_use_id: str
agent_id: NotRequired[str]
agent_type: NotRequired[str]
فیلدنوعتوضیح
hook_event_nameLiteral["PostToolUse"]همیشه “PostToolUse”
tool_namestrنامِ ابزاری که اجرا شد
tool_inputdict[str, Any]پارامترهای ورودی که استفاده شدند
tool_responseAnyپاسخ از اجرای ابزار
tool_use_idstrشناسه‌ی یکتا برای این tool use
agent_idstr (اختیاری)شناسه‌ی ساب‌ایجنت، وقتی hook درونِ یک ساب‌ایجنت اجرا می‌شود حضور دارد
agent_typestr (اختیاری)نوعِ ساب‌ایجنت، وقتی hook درونِ یک ساب‌ایجنت اجرا می‌شود حضور دارد

داده‌ی ورودی برای eventهای hookِ PostToolUseFailure. وقتی اجرای یک ابزار ناموفق باشد فراخوانی می‌شود.

class PostToolUseFailureHookInput(BaseHookInput):
hook_event_name: Literal["PostToolUseFailure"]
tool_name: str
tool_input: dict[str, Any]
tool_use_id: str
error: str
is_interrupt: NotRequired[bool]
agent_id: NotRequired[str]
agent_type: NotRequired[str]
فیلدنوعتوضیح
hook_event_nameLiteral["PostToolUseFailure"]همیشه “PostToolUseFailure”
tool_namestrنامِ ابزاری که ناموفق بود
tool_inputdict[str, Any]پارامترهای ورودی که استفاده شدند
tool_use_idstrشناسه‌ی یکتا برای این tool use
errorstrپیامِ خطا از اجرای ناموفق
is_interruptbool (اختیاری)اینکه آیا شکست ناشی از یک interrupt بوده
agent_idstr (اختیاری)شناسه‌ی ساب‌ایجنت، وقتی hook درونِ یک ساب‌ایجنت اجرا می‌شود حضور دارد
agent_typestr (اختیاری)نوعِ ساب‌ایجنت، وقتی hook درونِ یک ساب‌ایجنت اجرا می‌شود حضور دارد

داده‌ی ورودی برای eventهای hookِ UserPromptSubmit.

class UserPromptSubmitHookInput(BaseHookInput):
hook_event_name: Literal["UserPromptSubmit"]
prompt: str
فیلدنوعتوضیح
hook_event_nameLiteral["UserPromptSubmit"]همیشه “UserPromptSubmit”
promptstrپرامپتِ submit‌شده‌ی کاربر

داده‌ی ورودی برای eventهای hookِ Stop.

class StopHookInput(BaseHookInput):
hook_event_name: Literal["Stop"]
stop_hook_active: bool
فیلدنوعتوضیح
hook_event_nameLiteral["Stop"]همیشه “Stop”
stop_hook_activeboolاینکه آیا stop hook فعال است

داده‌ی ورودی برای eventهای hookِ SubagentStop.

class SubagentStopHookInput(BaseHookInput):
hook_event_name: Literal["SubagentStop"]
stop_hook_active: bool
agent_id: str
agent_transcript_path: str
agent_type: str
فیلدنوعتوضیح
hook_event_nameLiteral["SubagentStop"]همیشه “SubagentStop”
stop_hook_activeboolاینکه آیا stop hook فعال است
agent_idstrشناسه‌ی یکتای ساب‌ایجنت
agent_transcript_pathstrمسیرِ فایلِ transcriptِ ساب‌ایجنت
agent_typestrنوعِ ساب‌ایجنت

داده‌ی ورودی برای eventهای hookِ PreCompact.

class PreCompactHookInput(BaseHookInput):
hook_event_name: Literal["PreCompact"]
trigger: Literal["manual", "auto"]
custom_instructions: str | None
فیلدنوعتوضیح
hook_event_nameLiteral["PreCompact"]همیشه “PreCompact”
triggerLiteral["manual", "auto"]چه چیزی compaction را تریگر کرد
custom_instructionsstr | Noneدستورالعمل‌های سفارشی برای compaction

داده‌ی ورودی برای eventهای hookِ Notification.

class NotificationHookInput(BaseHookInput):
hook_event_name: Literal["Notification"]
message: str
title: NotRequired[str]
notification_type: str
فیلدنوعتوضیح
hook_event_nameLiteral["Notification"]همیشه “Notification”
messagestrمحتوای پیامِ نوتیفیکیشن
titlestr (اختیاری)عنوانِ نوتیفیکیشن
notification_typestrنوعِ نوتیفیکیشن

داده‌ی ورودی برای eventهای hookِ SubagentStart.

class SubagentStartHookInput(BaseHookInput):
hook_event_name: Literal["SubagentStart"]
agent_id: str
agent_type: str
فیلدنوعتوضیح
hook_event_nameLiteral["SubagentStart"]همیشه “SubagentStart”
agent_idstrشناسه‌ی یکتای ساب‌ایجنت
agent_typestrنوعِ ساب‌ایجنت

داده‌ی ورودی برای eventهای hookِ PermissionRequest. به hookها امکان می‌دهد تصمیم‌های دسترسی را به‌صورتِ برنامه‌نویسی‌شده رسیدگی کنند.

class PermissionRequestHookInput(BaseHookInput):
hook_event_name: Literal["PermissionRequest"]
tool_name: str
tool_input: dict[str, Any]
permission_suggestions: NotRequired[list[Any]]
فیلدنوعتوضیح
hook_event_nameLiteral["PermissionRequest"]همیشه “PermissionRequest”
tool_namestrنامِ ابزاری که دسترسی درخواست می‌کند
tool_inputdict[str, Any]پارامترهای ورودیِ ابزار
permission_suggestionslist[Any] (اختیاری)به‌روزرسانی‌های دسترسیِ پیشنهادی از CLI

نوعِ union برای مقادیرِ بازگشتیِ callbackِ hook.

HookJSONOutput = AsyncHookJSONOutput | SyncHookJSONOutput

خروجیِ hookِ همگام با فیلدهای کنترل و تصمیم.

class SyncHookJSONOutput(TypedDict):
# Control fields
continue_: NotRequired[bool] # Whether to proceed (default: True)
suppressOutput: NotRequired[bool] # Hide stdout from transcript
stopReason: NotRequired[str] # Message when continue is False
# Decision fields
decision: NotRequired[Literal["block"]]
systemMessage: NotRequired[str] # Warning message for user
reason: NotRequired[str] # Feedback for Claude
# Hook-specific output
hookSpecificOutput: NotRequired[HookSpecificOutput]

یک TypedDict که نامِ eventِ hook و فیلدهای مخصوصِ event را در بر دارد. شکل به مقدارِ hookEventName بستگی دارد. برای جزئیاتِ کاملِ فیلدهای موجود برای هر eventِ hook، کنترلِ اجرا با hookها را ببین.

یک discriminated union از نوعِ خروجی‌های مخصوصِ event. فیلدِ hookEventName تعیین می‌کند کدام فیلدها معتبرند.

class PreToolUseHookSpecificOutput(TypedDict):
hookEventName: Literal["PreToolUse"]
permissionDecision: NotRequired[Literal["allow", "deny", "ask", "defer"]]
permissionDecisionReason: NotRequired[str]
updatedInput: NotRequired[dict[str, Any]]
additionalContext: NotRequired[str]
class PostToolUseHookSpecificOutput(TypedDict):
hookEventName: Literal["PostToolUse"]
additionalContext: NotRequired[str]
updatedToolOutput: NotRequired[Any]
updatedMCPToolOutput: NotRequired[Any] # Deprecated: use updatedToolOutput, which works for all tools
class PostToolUseFailureHookSpecificOutput(TypedDict):
hookEventName: Literal["PostToolUseFailure"]
additionalContext: NotRequired[str]
class UserPromptSubmitHookSpecificOutput(TypedDict):
hookEventName: Literal["UserPromptSubmit"]
additionalContext: NotRequired[str]
class NotificationHookSpecificOutput(TypedDict):
hookEventName: Literal["Notification"]
additionalContext: NotRequired[str]
class SubagentStartHookSpecificOutput(TypedDict):
hookEventName: Literal["SubagentStart"]
additionalContext: NotRequired[str]
class PermissionRequestHookSpecificOutput(TypedDict):
hookEventName: Literal["PermissionRequest"]
decision: dict[str, Any]
HookSpecificOutput = (
PreToolUseHookSpecificOutput
| PostToolUseHookSpecificOutput
| PostToolUseFailureHookSpecificOutput
| UserPromptSubmitHookSpecificOutput
| NotificationHookSpecificOutput
| SubagentStartHookSpecificOutput
| PermissionRequestHookSpecificOutput
)

خروجیِ hookِ async که اجرای hook را به تعویق می‌اندازد.

class AsyncHookJSONOutput(TypedDict):
async_: Literal[True] # Set to True to defer execution
asyncTimeout: NotRequired[int] # Timeout in milliseconds

این مثال دو hook ثبت می‌کند: یکی که فرمان‌های bashِ خطرناک مثلِ rm -rf / را مسدود می‌کند، و دیگری که همه‌ی استفاده‌های ابزار را برای ممیزی لاگ می‌کند. hookِ امنیتی فقط روی فرمان‌های Bash اجرا می‌شود (از طریقِ matcher)، در حالی که hookِ لاگ‌کردن روی همه‌ی ابزارها اجرا می‌شود.

from claude_agent_sdk import query, ClaudeAgentOptions, HookMatcher, HookContext
from typing import Any
async def validate_bash_command(
input_data: dict[str, Any], tool_use_id: str | None, context: HookContext
) -> dict[str, Any]:
"""Validate and potentially block dangerous bash commands."""
if input_data["tool_name"] == "Bash":
command = input_data["tool_input"].get("command", "")
if "rm -rf /" in command:
return {
"hookSpecificOutput": {
"hookEventName": "PreToolUse",
"permissionDecision": "deny",
"permissionDecisionReason": "Dangerous command blocked",
}
}
return {}
async def log_tool_use(
input_data: dict[str, Any], tool_use_id: str | None, context: HookContext
) -> dict[str, Any]:
"""Log all tool usage for auditing."""
print(f"Tool used: {input_data.get('tool_name')}")
return {}
options = ClaudeAgentOptions(
hooks={
"PreToolUse": [
HookMatcher(
matcher="Bash", hooks=[validate_bash_command], timeout=120
), # 2 min for validation
HookMatcher(
hooks=[log_tool_use]
), # Applies to all tools (default 60s timeout)
],
"PostToolUse": [HookMatcher(hooks=[log_tool_use])],
}
)
async for message in query(prompt="Analyze this codebase", options=options):
print(message)

انواعِ ورودی/خروجیِ ابزار

Section titled “انواعِ ورودی/خروجیِ ابزار”

مستنداتِ schemaهای ورودی/خروجی برای همه‌ی ابزارهای توکارِ Claude Code. هرچند پایتون SDK این‌ها را به‌عنوانِ نوع export نمی‌کند، آن‌ها ساختارِ ورودی و خروجیِ ابزارها را در پیام‌ها نشان می‌دهند.

نامِ ابزار: Agent (پیش‌تر Task، که هنوز به‌عنوانِ alias پذیرفته می‌شود)

ورودی:

{
"description": str, # A short (3-5 word) description of the task
"prompt": str, # The task for the agent to perform
"subagent_type": str, # The type of specialized agent to use
}

خروجی:

{
"result": str, # Final result from the subagent
"usage": dict | None, # Token usage statistics
"total_cost_usd": float | None, # Estimated total cost in USD
"duration_ms": int | None, # Execution duration in milliseconds
}

نامِ ابزار: AskUserQuestion

در حینِ اجرا از کاربر پرسش‌های شفاف‌سازی می‌پرسد. برای جزئیاتِ استفاده رسیدگی به تأییدها و ورودیِ کاربر را ببین.

ورودی:

{
"questions": [ # Questions to ask the user (1-4 questions)
{
"question": str, # The complete question to ask the user
"header": str, # Very short label displayed as a chip/tag (max 12 chars)
"options": [ # The available choices (2-4 options)
{
"label": str, # Display text for this option (1-5 words)
"description": str, # Explanation of what this option means
}
],
"multiSelect": bool, # Set to true to allow multiple selections
}
],
"answers": dict[str, str | list[str]] | None,
# User answers populated by the permission system. Multi-select
# answers may be a list of labels or a comma-joined string
}

خروجی:

{
"questions": [ # The questions that were asked
{
"question": str,
"header": str,
"options": [{"label": str, "description": str}],
"multiSelect": bool,
}
],
"answers": dict[str, str], # Maps question text to answer string
# Multi-select answers are comma-separated
}

نامِ ابزار: Bash

ورودی:

{
"command": str, # The command to execute
"timeout": int | None, # Optional timeout in milliseconds (max 600000)
"description": str | None, # Clear, concise description (5-10 words)
"run_in_background": bool | None, # Set to true to run in background
}

خروجی:

{
"output": str, # Combined stdout and stderr output
"exitCode": int, # Exit code of the command
"killed": bool | None, # Whether command was killed due to timeout
"shellId": str | None, # Shell ID for background processes
}

نامِ ابزار: Monitor

یک اسکریپتِ پس‌زمینه اجرا می‌کند و هر خطِ stdout را به‌عنوانِ یک event به Claude تحویل می‌دهد تا بتواند بدونِ polling واکنش نشان دهد. Monitor از همان قواعدِ دسترسیِ Bash پیروی می‌کند. برای رفتار و در دسترس‌بودنِ provider مرجعِ ابزارِ Monitor را ببین.

ورودی:

{
"command": str, # Shell script; each stdout line is an event, exit ends the watch
"description": str, # Short description shown in notifications
"timeout_ms": int | None, # Kill after this deadline (default 300000, max 3600000)
"persistent": bool | None, # Run for the lifetime of the session; stop with TaskStop
}

خروجی:

{
"taskId": str, # ID of the background monitor task
"timeoutMs": int, # Timeout deadline in milliseconds (0 when persistent)
"persistent": bool | None, # True when running until TaskStop or session end
}

نامِ ابزار: Edit

ورودی:

{
"file_path": str, # The absolute path to the file to modify
"old_string": str, # The text to replace
"new_string": str, # The text to replace it with
"replace_all": bool | None, # Replace all occurrences (default False)
}

خروجی:

{
"message": str, # Confirmation message
"replacements": int, # Number of replacements made
"file_path": str, # File path that was edited
}

نامِ ابزار: Read

ورودی:

{
"file_path": str, # The absolute path to the file to read
"offset": int | None, # The line number to start reading from
"limit": int | None, # The number of lines to read
}

خروجی (فایل‌های متنی):

{
"content": str, # File contents with line numbers
"total_lines": int, # Total number of lines in file
"lines_returned": int, # Lines actually returned
}

خروجی (تصاویر):

{
"image": str, # Base64 encoded image data
"mime_type": str, # Image MIME type
"file_size": int, # File size in bytes
}

نامِ ابزار: Write

ورودی:

{
"file_path": str, # The absolute path to the file to write
"content": str, # The content to write to the file
}

خروجی:

{
"message": str, # Success message
"bytes_written": int, # Number of bytes written
"file_path": str, # File path that was written
}

نامِ ابزار: Glob

ورودی:

{
"pattern": str, # The glob pattern to match files against
"path": str | None, # The directory to search in (defaults to cwd)
}

خروجی:

{
"matches": list[str], # Array of matching file paths
"count": int, # Number of matches found
"search_path": str, # Search directory used
}

نامِ ابزار: Grep

ورودی:

{
"pattern": str, # The regular expression pattern
"path": str | None, # File or directory to search in
"glob": str | None, # Glob pattern to filter files
"type": str | None, # File type to search
"output_mode": str | None, # "content", "files_with_matches", or "count"
"-i": bool | None, # Case insensitive search
"-n": bool | None, # Show line numbers
"-B": int | None, # Lines to show before each match
"-A": int | None, # Lines to show after each match
"-C": int | None, # Lines to show before and after
"head_limit": int | None, # Limit output to first N lines/entries
"multiline": bool | None, # Enable multiline mode
}

خروجی (حالتِ content):

{
"matches": [
{
"file": str,
"line_number": int | None,
"line": str,
"before_context": list[str] | None,
"after_context": list[str] | None,
}
],
"total_matches": int,
}

خروجی (حالتِ files_with_matches):

{
"files": list[str], # Files containing matches
"count": int, # Number of files with matches
}

نامِ ابزار: NotebookEdit

ورودی:

{
"notebook_path": str, # Absolute path to the Jupyter notebook
"cell_id": str | None, # The ID of the cell to edit
"new_source": str, # The new source for the cell
"cell_type": "code" | "markdown" | None, # The type of the cell
"edit_mode": "replace" | "insert" | "delete" | None, # Edit operation type
}

خروجی:

{
"message": str, # Success message
"edit_type": "replaced" | "inserted" | "deleted", # Type of edit performed
"cell_id": str | None, # Cell ID that was affected
"total_cells": int, # Total cells in notebook after edit
}

نامِ ابزار: WebFetch

ورودی:

{
"url": str, # The URL to fetch content from
"prompt": str, # The prompt to run on the fetched content
}

خروجی:

{
"bytes": int, # Size of the fetched content in bytes
"code": int, # HTTP response code
"codeText": str, # HTTP response code text
"result": str, # Processed result from applying the prompt to the content
"durationMs": int, # Time to fetch and process the content, in milliseconds
"url": str, # URL that was fetched
}

نامِ ابزار: WebSearch

ورودی:

{
"query": str, # The search query to use
"allowed_domains": list[str] | None, # Only include results from these domains
"blocked_domains": list[str] | None, # Never include results from these domains
}

خروجی:

{
"query": str, # The search query
"results": list[str | {"tool_use_id": str, "content": list[{"title": str, "url": str}]}],
"durationSeconds": float, # Search duration in seconds
}

نامِ ابزار: TodoWrite

ورودی:

{
"todos": [
{
"content": str, # The task description
"status": "pending" | "in_progress" | "completed", # Task status
"activeForm": str, # Active form of the description
}
]
}

خروجی:

{
"message": str, # Success message
"stats": {"total": int, "pending": int, "in_progress": int, "completed": int},
}

نامِ ابزار: TaskCreate

ورودی:

{
"subject": str, # Short task title
"description": str, # Detailed task body
"activeForm": str | None, # Present-tense label shown while in progress
"metadata": dict | None, # Arbitrary caller metadata
}

خروجی:

{
"task": {"id": str, "subject": str}, # Created task with assigned ID
}

نامِ ابزار: TaskUpdate

ورودی:

{
"taskId": str, # ID of the task to patch
"status": Literal["pending", "in_progress", "completed", "deleted"] | None,
"subject": str | None,
"description": str | None,
"activeForm": str | None,
"addBlocks": list[str] | None, # Task IDs this task now blocks
"addBlockedBy": list[str] | None, # Task IDs that now block this task
"owner": str | None,
"metadata": dict | None,
}

خروجی:

{
"success": bool,
"taskId": str,
"updatedFields": list[str], # Names of fields that changed
"error": str | None,
"statusChange": {"from": str, "to": str} | None,
}

نامِ ابزار: TaskGet

ورودی:

{
"taskId": str, # ID of the task to read
}

خروجی:

{
"task": {
"id": str,
"subject": str,
"description": str,
"status": Literal["pending", "in_progress", "completed"],
"blocks": list[str],
"blockedBy": list[str],
} | None, # None when the ID is not found
}

نامِ ابزار: TaskList

ورودی:

{}

خروجی:

{
"tasks": [
{
"id": str,
"subject": str,
"status": Literal["pending", "in_progress", "completed"],
"owner": str | None,
"blockedBy": list[str],
}
],
}

نامِ ابزار: BashOutput

ورودی:

{
"bash_id": str, # The ID of the background shell
"filter": str | None, # Optional regex to filter output lines
}

خروجی:

{
"output": str, # New output since last check
"status": "running" | "completed" | "failed", # Current shell status
"exitCode": int | None, # Exit code when completed
}

نامِ ابزار: KillBash

ورودی:

{
"shell_id": str # The ID of the background shell to kill
}

خروجی:

{
"message": str, # Success message
"shell_id": str, # ID of the killed shell
}

نامِ ابزار: ExitPlanMode

ورودی:

{
"plan": str # The plan to run by the user for approval
}

خروجی:

{
"message": str, # Confirmation message
"approved": bool | None, # Whether user approved the plan
}

نامِ ابزار: ListMcpResourcesTool

ورودی:

{
"server": str | None # Optional server name to filter resources by
}

خروجی:

{
"resources": [
{
"uri": str,
"name": str,
"description": str | None,
"mimeType": str | None,
"server": str,
}
],
"total": int,
}

نامِ ابزار: ReadMcpResourceTool

ورودی:

{
"server": str, # The MCP server name
"uri": str, # The resource URI to read
}

خروجی:

{
"contents": [
{"uri": str, "mimeType": str | None, "text": str | None, "blob": str | None}
],
"server": str,
}

قابلیت‌های پیشرفته با ClaudeSDKClient

Section titled “قابلیت‌های پیشرفته با ClaudeSDKClient”

ساختِ یک رابطِ گفتگوی پیوسته

Section titled “ساختِ یک رابطِ گفتگوی پیوسته”
from claude_agent_sdk import (
ClaudeSDKClient,
ClaudeAgentOptions,
AssistantMessage,
TextBlock,
)
import asyncio
class ConversationSession:
"""Maintains a single conversation session with Claude."""
def __init__(self, options: ClaudeAgentOptions | None = None):
self.client = ClaudeSDKClient(options)
self.turn_count = 0
async def start(self):
await self.client.connect()
print("Starting conversation session. Claude will remember context.")
print(
"Commands: 'exit' to quit, 'interrupt' to stop current task, 'new' for new session"
)
while True:
user_input = input(f"\n[Turn {self.turn_count + 1}] You: ")
if user_input.lower() == "exit":
break
elif user_input.lower() == "interrupt":
await self.client.interrupt()
print("Task interrupted!")
continue
elif user_input.lower() == "new":
# Disconnect and reconnect for a fresh session
await self.client.disconnect()
await self.client.connect()
self.turn_count = 0
print("Started new conversation session (previous context cleared)")
continue
# Send message - the session retains all previous messages
await self.client.query(user_input)
self.turn_count += 1
# Process response
print(f"[Turn {self.turn_count}] Claude: ", end="")
async for message in self.client.receive_response():
if isinstance(message, AssistantMessage):
for block in message.content:
if isinstance(block, TextBlock):
print(block.text, end="")
print() # New line after response
await self.client.disconnect()
print(f"Conversation ended after {self.turn_count} turns.")
async def main():
options = ClaudeAgentOptions(
allowed_tools=["Read", "Write", "Bash"], permission_mode="acceptEdits"
)
session = ConversationSession(options)
await session.start()
# Example conversation:
# Turn 1 - You: "Create a file called hello.py"
# Turn 1 - Claude: "I'll create a hello.py file for you..."
# Turn 2 - You: "What's in that file?"
# Turn 2 - Claude: "The hello.py file I just created contains..." (remembers!)
# Turn 3 - You: "Add a main function to it"
# Turn 3 - Claude: "I'll add a main function to hello.py..." (knows which file!)
asyncio.run(main())

استفاده از Hookها برای تغییرِ رفتار

Section titled “استفاده از Hookها برای تغییرِ رفتار”
from claude_agent_sdk import (
ClaudeSDKClient,
ClaudeAgentOptions,
HookMatcher,
HookContext,
)
import asyncio
from typing import Any
async def pre_tool_logger(
input_data: dict[str, Any], tool_use_id: str | None, context: HookContext
) -> dict[str, Any]:
"""Log all tool usage before execution."""
tool_name = input_data.get("tool_name", "unknown")
print(f"[PRE-TOOL] About to use: {tool_name}")
# You can modify or block the tool execution here
if tool_name == "Bash" and "rm -rf" in str(input_data.get("tool_input", {})):
return {
"hookSpecificOutput": {
"hookEventName": "PreToolUse",
"permissionDecision": "deny",
"permissionDecisionReason": "Dangerous command blocked",
}
}
return {}
async def post_tool_logger(
input_data: dict[str, Any], tool_use_id: str | None, context: HookContext
) -> dict[str, Any]:
"""Log results after tool execution."""
tool_name = input_data.get("tool_name", "unknown")
print(f"[POST-TOOL] Completed: {tool_name}")
return {}
async def user_prompt_modifier(
input_data: dict[str, Any], tool_use_id: str | None, context: HookContext
) -> dict[str, Any]:
"""Add context to user prompts."""
original_prompt = input_data.get("prompt", "")
# Add a timestamp as additional context for Claude to see
from datetime import datetime
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
return {
"hookSpecificOutput": {
"hookEventName": "UserPromptSubmit",
"additionalContext": f"[Submitted at {timestamp}] Original prompt: {original_prompt}",
}
}
async def main():
options = ClaudeAgentOptions(
hooks={
"PreToolUse": [
HookMatcher(hooks=[pre_tool_logger]),
HookMatcher(matcher="Bash", hooks=[pre_tool_logger]),
],
"PostToolUse": [HookMatcher(hooks=[post_tool_logger])],
"UserPromptSubmit": [HookMatcher(hooks=[user_prompt_modifier])],
},
allowed_tools=["Read", "Write", "Bash"],
)
async with ClaudeSDKClient(options=options) as client:
await client.query("List files in current directory")
async for message in client.receive_response():
# Hooks will automatically log tool usage
pass
asyncio.run(main())

مانیتورینگِ پیشرفتِ بلادرنگ

Section titled “مانیتورینگِ پیشرفتِ بلادرنگ”
from claude_agent_sdk import (
ClaudeSDKClient,
ClaudeAgentOptions,
AssistantMessage,
ToolUseBlock,
ToolResultBlock,
TextBlock,
)
import asyncio
async def monitor_progress():
options = ClaudeAgentOptions(
allowed_tools=["Write", "Bash"], permission_mode="acceptEdits"
)
async with ClaudeSDKClient(options=options) as client:
await client.query("Create 5 Python files with different sorting algorithms")
# Monitor progress in real-time
async for message in client.receive_response():
if isinstance(message, AssistantMessage):
for block in message.content:
if isinstance(block, ToolUseBlock):
if block.name == "Write":
file_path = block.input.get("file_path", "")
print(f"Creating: {file_path}")
elif isinstance(block, ToolResultBlock):
print("Completed tool execution")
elif isinstance(block, TextBlock):
print(f"Claude says: {block.text[:100]}...")
print("Task completed!")
asyncio.run(monitor_progress())

عملیاتِ پایه‌ای فایل (با query)

Section titled “عملیاتِ پایه‌ای فایل (با query)”
from claude_agent_sdk import query, ClaudeAgentOptions, AssistantMessage, ToolUseBlock
import asyncio
async def create_project():
options = ClaudeAgentOptions(
allowed_tools=["Read", "Write", "Bash"],
permission_mode="acceptEdits",
cwd="/home/user/project",
)
async for message in query(
prompt="Create a Python project structure with setup.py", options=options
):
if isinstance(message, AssistantMessage):
for block in message.content:
if isinstance(block, ToolUseBlock):
print(f"Using tool: {block.name}")
asyncio.run(create_project())
from claude_agent_sdk import query, CLINotFoundError, ProcessError, CLIJSONDecodeError
try:
async for message in query(prompt="Hello"):
print(message)
except CLINotFoundError:
print(
"Claude Code CLI not found. Try reinstalling: pip install --force-reinstall claude-agent-sdk"
)
except ProcessError as e:
print(f"Process failed with exit code: {e.exit_code}")
except CLIJSONDecodeError as e:
print(f"Failed to parse response: {e}")
from claude_agent_sdk import ClaudeSDKClient
import asyncio
async def interactive_session():
async with ClaudeSDKClient() as client:
# Send initial message
await client.query("What's the weather like?")
# Process responses
async for msg in client.receive_response():
print(msg)
# Send follow-up
await client.query("Tell me more about that")
# Process follow-up response
async for msg in client.receive_response():
print(msg)
asyncio.run(interactive_session())

استفاده از ابزارهای سفارشی با ClaudeSDKClient

Section titled “استفاده از ابزارهای سفارشی با ClaudeSDKClient”
from claude_agent_sdk import (
ClaudeSDKClient,
ClaudeAgentOptions,
tool,
create_sdk_mcp_server,
AssistantMessage,
TextBlock,
)
import asyncio
from typing import Any
# Define custom tools with @tool decorator
@tool("calculate", "Perform mathematical calculations", {"expression": str})
async def calculate(args: dict[str, Any]) -> dict[str, Any]:
try:
result = eval(args["expression"], {"__builtins__": {}})
return {"content": [{"type": "text", "text": f"Result: {result}"}]}
except Exception as e:
return {
"content": [{"type": "text", "text": f"Error: {str(e)}"}],
"is_error": True,
}
@tool("get_time", "Get current time", {})
async def get_time(args: dict[str, Any]) -> dict[str, Any]:
from datetime import datetime
current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
return {"content": [{"type": "text", "text": f"Current time: {current_time}"}]}
async def main():
# Create SDK MCP server with custom tools
my_server = create_sdk_mcp_server(
name="utilities", version="1.0.0", tools=[calculate, get_time]
)
# Configure options with the server
options = ClaudeAgentOptions(
mcp_servers={"utils": my_server},
allowed_tools=["mcp__utils__calculate", "mcp__utils__get_time"],
)
# Use ClaudeSDKClient for interactive tool usage
async with ClaudeSDKClient(options=options) as client:
await client.query("What's 123 * 456?")
# Process calculation response
async for message in client.receive_response():
if isinstance(message, AssistantMessage):
for block in message.content:
if isinstance(block, TextBlock):
print(f"Calculation: {block.text}")
# Follow up with time query
await client.query("What time is it now?")
async for message in client.receive_response():
if isinstance(message, AssistantMessage):
for block in message.content:
if isinstance(block, TextBlock):
print(f"Time: {block.text}")
asyncio.run(main())

پیکربندی برای رفتارِ sandbox. از این برای فعال‌کردنِ command sandboxing و پیکربندیِ محدودیت‌های شبکه به‌صورتِ برنامه‌نویسی‌شده استفاده کن.

class SandboxSettings(TypedDict, total=False):
enabled: bool
autoAllowBashIfSandboxed: bool
excludedCommands: list[str]
allowUnsandboxedCommands: bool
network: SandboxNetworkConfig
ignoreViolations: SandboxIgnoreViolations
enableWeakerNestedSandbox: bool
ویژگینوعپیش‌فرضتوضیح
enabledboolFalseحالتِ sandbox را برای اجرای فرمان فعال کن
autoAllowBashIfSandboxedboolTrueوقتی sandbox فعال است فرمان‌های bash را خودکار تأیید کن
excludedCommandslist[str][]فرمان‌هایی که همیشه محدودیت‌های sandbox را دور می‌زنند (مثلاً ["docker"]). این‌ها به‌صورتِ خودکار و بدونِ دخالتِ مدل، بدونِ sandbox اجرا می‌شوند
allowUnsandboxedCommandsboolTrueبه مدل اجازه بده درخواستِ اجرای فرمان‌ها بیرونِ sandbox را بدهد. وقتی True باشد، مدل می‌تواند dangerouslyDisableSandbox را در ورودیِ ابزار تنظیم کند، که به سیستمِ دسترسی بازمی‌گردد
networkSandboxNetworkConfigNoneپیکربندیِ sandboxِ مخصوصِ شبکه
ignoreViolationsSandboxIgnoreViolationsNoneپیکربندی کن که کدام نقض‌های sandbox نادیده گرفته شوند
enableWeakerNestedSandboxboolFalseیک sandboxِ تودرتوی ضعیف‌تر را برای سازگاری فعال کن
from claude_agent_sdk import query, ClaudeAgentOptions, SandboxSettings
sandbox_settings: SandboxSettings = {
"enabled": True,
"autoAllowBashIfSandboxed": True,
"network": {"allowLocalBinding": True},
}
async for message in query(
prompt="Build and test my project",
options=ClaudeAgentOptions(sandbox=sandbox_settings),
):
print(message)

پیکربندیِ مخصوصِ شبکه برای حالتِ sandbox. این تنظیمات وقتی enabled در SandboxSettingsِ والد برابرِ True باشد، روی فرمان‌های Bashِ sandbox‌شده اعمال می‌شوند. آن‌ها ابزارِ WebFetch را محدود نمی‌کنند، که به‌جای آن از قواعدِ دسترسی استفاده می‌کند.

class SandboxNetworkConfig(TypedDict, total=False):
allowedDomains: list[str]
deniedDomains: list[str]
allowManagedDomainsOnly: bool
allowUnixSockets: list[str]
allowAllUnixSockets: bool
allowLocalBinding: bool
allowMachLookup: list[str]
httpProxyPort: int
socksProxyPort: int
ویژگینوعپیش‌فرضتوضیح
allowedDomainslist[str][]نام‌دامنه‌هایی که پروسه‌های sandbox‌شده می‌توانند به آن‌ها دسترسی داشته باشند
deniedDomainslist[str][]نام‌دامنه‌هایی که پروسه‌های sandbox‌شده نمی‌توانند به آن‌ها دسترسی داشته باشند. بر allowedDomains اولویت دارد
allowManagedDomainsOnlyboolFalseفقط managed-settings: وقتی در managed settings تنظیم شود، allowedDomains را از منابعِ تنظیماتِ غیر-managed نادیده بگیر. وقتی از طریقِ SDK options تنظیم شود اثری ندارد
allowUnixSocketslist[str][]مسیرهای Unix socket که پروسه‌ها می‌توانند به آن‌ها دسترسی داشته باشند (مثلاً Docker socket)
allowAllUnixSocketsboolFalseدسترسی به همه‌ی Unix socketها را مجاز کن
allowLocalBindingboolFalseبه پروسه‌ها اجازه بده به پورت‌های محلی bind شوند (مثلاً برای dev serverها)
allowMachLookuplist[str][]فقط macOS: نام‌های سرویسِ XPC/Mach که مجاز شوند. از یک wildcardِ انتهایی پشتیبانی می‌کند
httpProxyPortintNoneپورتِ HTTP proxy برای درخواست‌های شبکه
socksProxyPortintNoneپورتِ SOCKS proxy برای درخواست‌های شبکه

پیکربندی برای نادیده‌گرفتنِ نقض‌های مشخصِ sandbox.

class SandboxIgnoreViolations(TypedDict, total=False):
file: list[str]
network: list[str]
ویژگینوعپیش‌فرضتوضیح
filelist[str][]الگوهای مسیرِ فایل که نقض‌هایشان نادیده گرفته شوند
networklist[str][]الگوهای شبکه که نقض‌هایشان نادیده گرفته شوند

Fallbackِ دسترسی برای فرمان‌های بدونِ sandbox

Section titled “Fallbackِ دسترسی برای فرمان‌های بدونِ sandbox”

وقتی allowUnsandboxedCommands فعال باشد، مدل می‌تواند با تنظیمِ dangerouslyDisableSandbox: True در ورودیِ ابزار، درخواستِ اجرای فرمان‌ها بیرونِ sandbox را بدهد. این درخواست‌ها به سیستمِ دسترسیِ موجود بازمی‌گردند، یعنی handlerِ can_use_toolِ تو فراخوانی می‌شود و به تو امکان می‌دهد منطقِ مجوزدهیِ سفارشی پیاده کنی.

from claude_agent_sdk import (
query,
ClaudeAgentOptions,
HookMatcher,
PermissionResultAllow,
PermissionResultDeny,
ToolPermissionContext,
)
async def can_use_tool(
tool: str, input: dict, context: ToolPermissionContext
) -> PermissionResultAllow | PermissionResultDeny:
# Check if the model is requesting to bypass the sandbox
if tool == "Bash" and input.get("dangerouslyDisableSandbox"):
# The model is requesting to run this command outside the sandbox
print(f"Unsandboxed command requested: {input.get('command')}")
if is_command_authorized(input.get("command")):
return PermissionResultAllow()
return PermissionResultDeny(
message="Command not authorized for unsandboxed execution"
)
return PermissionResultAllow()
# Required: dummy hook keeps the stream open for can_use_tool
async def dummy_hook(input_data, tool_use_id, context):
return {"continue_": True}
async def prompt_stream():
yield {
"type": "user",
"message": {"role": "user", "content": "Deploy my application"},
}
async def main():
async for message in query(
prompt=prompt_stream(),
options=ClaudeAgentOptions(
sandbox={
"enabled": True,
"allowUnsandboxedCommands": True, # Model can request unsandboxed execution
},
permission_mode="default",
can_use_tool=can_use_tool,
hooks={"PreToolUse": [HookMatcher(matcher=None, hooks=[dummy_hook])]},
),
):
print(message)

این الگو به تو امکان می‌دهد:

  • ممیزیِ درخواست‌های مدل: وقتی مدل درخواستِ اجرای بدونِ sandbox می‌دهد لاگ کن
  • پیاده‌سازیِ allowlist: فقط به فرمان‌های مشخص اجازه‌ی اجرای بدونِ sandbox بده
  • افزودنِ ورک‌فلوهای تأیید: برای عملیاتِ ممتاز مجوزِ صریح بخواه