Agent SDK 참조 - Python

pip install claude-agent-sdk

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

import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions

async def main():
    options = ClaudeAgentOptions(
        system_prompt="당신은 전문 Python 개발자입니다",
        permission_mode='acceptEdits',
        cwd="/home/user/project"
    )

    async for message in query(
        prompt="Python 웹 서버를 만들어주세요",
        options=options
    ):
        print(message)


asyncio.run(main())

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

from claude_agent_sdk import tool
from typing import Any

@tool("greet", "사용자에게 인사하기", {"name": str})
async def greet(args: dict[str, Any]) -> dict[str, Any]:
    return {
        "content": [{
            "type": "text",
            "text": f"안녕하세요, {args['name']}님!"
        }]
    }

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", "두 숫자 더하기", {"a": float, "b": float})
async def add(args):
    return {
        "content": [{
            "type": "text",
            "text": f"합계: {args['a'] + args['b']}"
        }]
    }

@tool("multiply", "두 숫자 곱하기", {"a": float, "b": float})
async def multiply(args):
    return {
        "content": [{
            "type": "text",
            "text": f"곱: {args['a'] * args['b']}"
        }]
    }

calculator = create_sdk_mcp_server(
    name="calculator",
    version="2.0.0",
    tools=[add, multiply]  # 데코레이트된 함수 전달
)

# Claude와 함께 사용
options = ClaudeAgentOptions(
    mcp_servers={"calc": calculator},
    allowed_tools=["mcp__calc__add", "mcp__calc__multiply"]
)

class ClaudeSDKClient:
    def __init__(self, options: ClaudeAgentOptions | 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 disconnect(self) -> None

async with ClaudeSDKClient() as client:
    await client.query("안녕하세요 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("프랑스의 수도는 어디인가요?")
        
        # 응답 처리
        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}")
        
        # 후속 질문 - Claude가 이전 컨텍스트를 기억함
        await client.query("그 도시의 인구는 얼마나 되나요?")
        
        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}")
        
        # 또 다른 후속 질문 - 여전히 동일한 대화에서
        await client.query("그곳의 유명한 랜드마크는 무엇인가요?")
        
        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())

import asyncio
from claude_agent_sdk import ClaudeSDKClient

async def message_stream():
    """메시지를 동적으로 생성합니다."""
    yield {"type": "text", "text": "다음 데이터를 분석해주세요:"}
    await asyncio.sleep(0.5)
    yield {"type": "text", "text": "온도: 25°C"}
    await asyncio.sleep(0.5)
    yield {"type": "text", "text": "습도: 60%"}
    await asyncio.sleep(0.5)
    yield {"type": "text", "text": "어떤 패턴을 보시나요?"}

async def main():
    async with ClaudeSDKClient() as client:
        # Claude에 입력 스트리밍
        await client.query(message_stream())
        
        # 응답 처리
        async for message in client.receive_response():
            print(message)
        
        # 동일한 세션에서 후속 조치
        await client.query("이 수치들에 대해 걱정해야 할까요?")
        
        async for message in client.receive_response():
            print(message)

asyncio.run(main())

import asyncio
from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions

async def interruptible_task():
    options = ClaudeAgentOptions(
        allowed_tools=["Bash"],
        permission_mode="acceptEdits"
    )
    
    async with ClaudeSDKClient(options=options) as client:
        # 장시간 실행되는 작업 시작
        await client.query("1부터 100까지 천천히 세어주세요")
        
        # 잠시 실행하도록 둠
        await asyncio.sleep(2)
        
        # 작업 중단
        await client.interrupt()
        print("작업이 중단되었습니다!")
        
        # 새 명령 전송
        await client.query("대신 그냥 안녕이라고 말해주세요")
        
        async for message in client.receive_response():
            # 새 응답 처리
            pass

asyncio.run(interruptible_task())

from claude_agent_sdk import (
    ClaudeSDKClient,
    ClaudeAgentOptions
)

async def custom_permission_handler(
    tool_name: str,
    input_data: dict,
    context: dict
):
    """도구 권한에 대한 사용자 정의 로직."""

    # 시스템 디렉토리에 대한 쓰기 차단
    if tool_name == "Write" and input_data.get("file_path", "").startswith("/system/"):
        return {
            "behavior": "deny",
            "message": "시스템 디렉토리 쓰기가 허용되지 않습니다",
            "interrupt": True
        }

    # 민감한 파일 작업 리디렉션
    if tool_name in ["Write", "Edit"] and "config" in input_data.get("file_path", ""):
        safe_path = f"./sandbox/{input_data['file_path']}"
        return {
            "behavior": "allow",
            "updatedInput": {**input_data, "file_path": safe_path}
        }

    # 나머지는 모두 허용
    return {
        "behavior": "allow",
        "updatedInput": 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("시스템 구성 파일을 업데이트해주세요")
        
        async for message in client.receive_response():
            # 대신 샌드박스 경로를 사용할 것입니다
            print(message)

asyncio.run(main())

@dataclass
class SdkMcpTool(Generic[T]):
    name: str
    description: str
    input_schema: type[T] | dict[str, Any]
    handler: Callable[[T], Awaitable[dict[str, Any]]]

@dataclass
class ClaudeAgentOptions:
    allowed_tools: list[str] = field(default_factory=list)
    max_thinking_tokens: int = 8000
    system_prompt: str | None = None
    mcp_servers: dict[str, McpServerConfig] | str | Path = field(default_factory=dict)
    permission_mode: PermissionMode | None = None
    continue_conversation: bool = False
    resume: str | None = None
    fork_session: bool = False
    max_turns: int | None = None
    disallowed_tools: list[str] = field(default_factory=list)
    model: str | None = None
    permission_prompt_tool_name: str | None = None
    cwd: 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)

SettingSource = Literal["user", "project", "local"]

# SDK v0.0.x처럼 모든 설정 로드
from claude_agent_sdk import query, ClaudeAgentOptions

async for message in query(
    prompt="이 코드를 분석해주세요",
    options=ClaudeAgentOptions(
        setting_sources=["user", "project", "local"]  # 모든 설정 로드
    )
):
    print(message)

# 프로젝트 설정만 로드, 사용자 및 로컬 무시
async for message in query(
    prompt="CI 검사 실행",
    options=ClaudeAgentOptions(
        setting_sources=["project"]  # .claude/settings.json만
    )
):
    print(message)

# 로컬 설정을 제외하여 CI에서 일관된 동작 보장
async for message in query(
    prompt="테스트 실행",
    options=ClaudeAgentOptions(
        setting_sources=["project"],  # 팀 공유 설정만
        permission_mode="bypassPermissions"
    )
):
    print(message)

# 모든 것을 프로그래밍 방식으로 정의 (기본 동작)
# 파일시스템 종속성 없음 - setting_sources는 기본적으로 None
async for message in query(
    prompt="이 PR을 검토해주세요",
    options=ClaudeAgentOptions(
        # setting_sources=None이 기본값이므로 지정할 필요 없음
        agents={ /* ... */ },
        mcp_servers={ /* ... */ },
        allowed_tools=["Read", "Grep", "Glob"]
    )
):
    print(message)

@dataclass
class AgentDefinition:
    description: str
    prompt: str
    tools: list[str] | None = None
    model: Literal["sonnet", "opus", "haiku", "inherit"] | None = None

PermissionMode = Literal[
    "default",           # 표준 권한 동작
    "acceptEdits",       # 파일 편집 자동 승인
    "plan",              # 계획 모드 - 실행 없음
    "bypassPermissions"  # 모든 권한 검사 우회 (주의해서 사용)
]

class McpSdkServerConfig(TypedDict):
    type: Literal["sdk"]
    name: str
    instance: Any  # MCP 서버 인스턴스

McpServerConfig = McpStdioServerConfig | McpSSEServerConfig | McpHttpServerConfig | McpSdkServerConfig

class McpStdioServerConfig(TypedDict):
    type: NotRequired[Literal["stdio"]]  # 하위 호환성을 위해 선택사항
    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]]

Message = UserMessage | AssistantMessage | SystemMessage | ResultMessage

@dataclass
class UserMessage:
    content: str | list[ContentBlock]

@dataclass
class AssistantMessage:
    content: list[ContentBlock]
    model: str

@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
    total_cost_usd: float | None = None
    usage: dict[str, Any] | None = None
    result: str | None = None

ContentBlock = TextBlock | ThinkingBlock | ToolUseBlock | ToolResultBlock

@dataclass
class TextBlock:
    text: str

@dataclass
class ThinkingBlock:
    thinking: str
    signature: str

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

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

class ClaudeSDKError(Exception):
    """Claude SDK의 기본 오류."""

class CLINotFoundError(CLIConnectionError):
    def __init__(self, message: str = "Claude Code를 찾을 수 없습니다", cli_path: str | None = None):
        """
        Args:
            message: 오류 메시지 (기본값: "Claude Code를 찾을 수 없습니다")
            cli_path: 찾을 수 없는 CLI의 선택적 경로
        """

class CLIConnectionError(ClaudeSDKError):
    """Claude Code에 연결하지 못했습니다."""

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

class CLIJSONDecodeError(ClaudeSDKError):
    def __init__(self, line: str, original_error: Exception):
        """
        Args:
            line: 파싱에 실패한 줄
            original_error: 원래 JSON 디코드 예외
        """
        self.line = line
        self.original_error = original_error

HookEvent = Literal[
    "PreToolUse",      # 도구 실행 전 호출
    "PostToolUse",     # 도구 실행 후 호출
    "UserPromptSubmit", # 사용자가 프롬프트를 제출할 때 호출
    "Stop",            # 실행을 중지할 때 호출
    "SubagentStop",    # 하위 에이전트가 중지할 때 호출
    "PreCompact"       # 메시지 압축 전 호출
]

HookCallback = Callable[
    [dict[str, Any], str | None, HookContext],
    Awaitable[dict[str, Any]]
]

@dataclass
class HookContext:
    signal: Any | None = None  # 미래: 중단 신호 지원

@dataclass
class HookMatcher:
    matcher: str | None = None        # 매칭할 도구 이름 또는 패턴 (예: "Bash", "Write|Edit")
    hooks: list[HookCallback] = field(default_factory=list)  # 실행할 콜백 목록

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]:
    """위험한 bash 명령을 검증하고 잠재적으로 차단합니다."""
    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': '위험한 명령이 차단되었습니다'
                }
            }
    return {}

async def log_tool_use(
    input_data: dict[str, Any],
    tool_use_id: str | None,
    context: HookContext
) -> dict[str, Any]:
    """감사를 위해 모든 도구 사용을 로그합니다."""
    print(f"사용된 도구: {input_data.get('tool_name')}")
    return {}

options = ClaudeAgentOptions(
    hooks={
        'PreToolUse': [
            HookMatcher(matcher='Bash', hooks=[validate_bash_command]),
            HookMatcher(hooks=[log_tool_use])  # 모든 도구에 적용
        ],
        'PostToolUse': [
            HookMatcher(hooks=[log_tool_use])
        ]
    }
)

async for message in query(
    prompt="이 코드베이스를 분석해주세요",
    options=options
):
    print(message)

{
    "description": str,      # 작업에 대한 짧은 (3-5단어) 설명
    "prompt": str,           # 에이전트가 수행할 작업
    "subagent_type": str     # 사용할 전문 에이전트 타입
}

{
    "result": str,                    # 하위 에이전트의 최종 결과
    "usage": dict | None,             # 토큰 사용량 통계
    "total_cost_usd": float | None,  # USD 총 비용
    "duration_ms": int | None         # 실행 시간 (밀리초)
}

{
    "command": str,                  # 실행할 명령
    "timeout": int | None,           # 선택적 타임아웃 (밀리초, 최대 600000)
    "description": str | None,       # 명확하고 간결한 설명 (5-10단어)
    "run_in_background": bool | None # 백그라운드에서 실행하려면 true로 설정
}

{
    "output": str,              # stdout과 stderr 출력 결합
    "exitCode": int,            # 명령의 종료 코드
    "killed": bool | None,      # 타임아웃으로 인해 명령이 종료되었는지 여부
    "shellId": str | None       # 백그라운드 프로세스용 셸 ID
}

{
    "file_path": str,           # 수정할 파일의 절대 경로
    "old_string": str,          # 교체할 텍스트
    "new_string": str,          # 교체할 텍스트
    "replace_all": bool | None  # 모든 발생을 교체 (기본값 False)
}

{
    "message": str,      # 확인 메시지
    "replacements": int, # 수행된 교체 수
    "file_path": str     # 편집된 파일 경로
}

{
    "file_path": str,     # 수정할 파일의 절대 경로
    "edits": [            # 편집 작업 배열
        {
            "old_string": str,          # 교체할 텍스트
            "new_string": str,          # 교체할 텍스트
            "replace_all": bool | None  # 모든 발생을 교체
        }
    ]
}

{
    "message": str,       # 성공 메시지
    "edits_applied": int, # 적용된 총 편집 수
    "file_path": str      # 편집된 파일 경로
}

{
    "file_path": str,       # 읽을 파일의 절대 경로
    "offset": int | None,   # 읽기를 시작할 줄 번호
    "limit": int | None     # 읽을 줄 수
}

{
    "content": str,         # 줄 번호가 있는 파일 내용
    "total_lines": int,     # 파일의 총 줄 수
    "lines_returned": int   # 실제로 반환된 줄 수
}

{
    
    "image": str,       # Base64 인코딩된 이미지 데이터
    "mime_type": str,   # 이미지 MIME 타입
    "file_size": int    # 파일 크기 (바이트)
}

{
    "file_path": str,  # 쓸 파일의 절대 경로
    "content": str     # 파일에 쓸 내용
}

{
    "message": str,        # 성공 메시지
    "bytes_written": int,  # 쓰여진 바이트 수
    "file_path": str       # 쓰여진 파일 경로
}

{
    "pattern": str,       # 파일을 매칭할 glob 패턴
    "path": str | None    # 검색할 디렉토리 (기본값은 cwd)
}

{
    "matches": list[str],  # 매칭되는 파일 경로 배열
    "count": int,          # 찾은 매치 수
    "search_path": str     # 사용된 검색 디렉토리
}

{
    "pattern": str,                    # 정규 표현식 패턴
    "path": str | None,                # 검색할 파일 또는 디렉토리
    "glob": str | None,                # 파일을 필터링할 glob 패턴
    "type": str | None,                # 검색할 파일 타입
    "output_mode": str | None,         # "content", "files_with_matches", 또는 "count"
    "-i": bool | None,                 # 대소문자 구분 없는 검색
    "-n": bool | None,                 # 줄 번호 표시
    "-B": int | None,                  # 각 매치 전에 표시할 줄 수
    "-A": int | None,                  # 각 매치 후에 표시할 줄 수
    "-C": int | None,                  # 전후에 표시할 줄 수
    "head_limit": int | None,          # 출력을 첫 N줄/항목으로 제한
    "multiline": bool | None           # 멀티라인 모드 활성화
}

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

{
    "files": list[str],  # 매치를 포함하는 파일들
    "count": int         # 매치가 있는 파일 수
}

{
    "notebook_path": str,                     # Jupyter 노트북의 절대 경로
    "cell_id": str | None,                    # 편집할 셀의 ID
    "new_source": str,                        # 셀의 새 소스
    "cell_type": "code" | "markdown" | None,  # 셀의 타입
    "edit_mode": "replace" | "insert" | "delete" | None  # 편집 작업 타입
}

{
    "message": str, # 성공 메시지
    "edit_type": "replaced" | "inserted" | "deleted",  # 수행된 편집 타입
    "cell_id": str | None,                       # 영향을 받은 셀 ID
    "total_cells": int                           # 편집 후 노트북의 총 셀 수
}

{
    "url": str,     # 콘텐츠를 가져올 URL
    "prompt": str   # 가져온 콘텐츠에 실행할 프롬프트
}

{
    "response": str,           # 프롬프트에 대한 AI 모델의 응답
    "url": str,                # 가져온 URL
    "final_url": str | None,   # 리디렉션 후 최종 URL
    "status_code": int | None  # HTTP 상태 코드
}

{
    "query": str,                        # 사용할 검색 쿼리
    "allowed_domains": list[str] | None, # 이 도메인의 결과만 포함
    "blocked_domains": list[str] | None  # 이 도메인의 결과는 절대 포함하지 않음
}

{
    "results": [
        {
            "title": str,
            "url": str,
            "snippet": str,
            "metadata": dict | None
        }
    ],
    "total_results": int,
    "query": str
}

{
    "todos": [
        {
            "content": str, # 작업 설명
            "status": "pending" | "in_progress" | "completed",  # 작업 상태
            "activeForm": str                            # 설명의 활성 형태
        }
    ]
}

{
    "message": str,  # 성공 메시지
    "stats": {
        "total": int,
        "pending": int,
        "in_progress": int,
        "completed": int
    }
}

{
    "bash_id": str,       # 백그라운드 셸의 ID
    "filter": str | None  # 출력 줄을 필터링할 선택적 정규식
}

{
    "output": str, # 마지막 확인 이후의 새 출력
    "status": "running" | "completed" | "failed",       # 현재 셸 상태
    "exitCode": int | None # 완료 시 종료 코드
}

{
    "shell_id": str  # 종료할 백그라운드 셸의 ID
}

{
    "message": str,  # 성공 메시지
    "shell_id": str  # 종료된 셸의 ID
}

{
    "plan": str  # 사용자 승인을 위해 실행할 계획
}

{
    "message": str,          # 확인 메시지
    "approved": bool | None  # 사용자가 계획을 승인했는지 여부
}

{
    "server": str | None  # 리소스를 필터링할 선택적 서버 이름
}

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

{
    "server": str,  # MCP 서버 이름
    "uri": str      # 읽을 리소스 URI
}

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

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

class ConversationSession:
    """Claude와의 단일 대화 세션을 유지합니다."""
    
    def __init__(self, options: ClaudeCodeOptions = None):
        self.client = ClaudeSDKClient(options)
        self.turn_count = 0
    
    async def start(self):
        await self.client.connect()
        print("대화 세션을 시작합니다. Claude가 컨텍스트를 기억할 것입니다.")
        print("명령: 종료하려면 'exit', 현재 작업을 중지하려면 'interrupt', 새 세션을 위해 'new'")
        
        while True:
            user_input = input(f"\n[턴 {self.turn_count + 1}] 당신: ")
            
            if user_input.lower() == 'exit':
                break
            elif user_input.lower() == 'interrupt':
                await self.client.interrupt()
                print("작업이 중단되었습니다!")
                continue
            elif user_input.lower() == 'new':
                # 새로운 세션을 위해 연결 해제 후 재연결
                await self.client.disconnect()
                await self.client.connect()
                self.turn_count = 0
                print("새 대화 세션을 시작했습니다 (이전 컨텍스트가 지워졌습니다)")
                continue
            
            # 메시지 전송 - Claude가 이 세션의 모든 이전 메시지를 기억함
            await self.client.query(user_input)
            self.turn_count += 1
            
            # 응답 처리
            print(f"[턴 {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()  # 응답 후 새 줄
        
        await self.client.disconnect()
        print(f"대화가 {self.turn_count}턴 후에 종료되었습니다.")

async def main():
    options = ClaudeCodeOptions(
        allowed_tools=["Read", "Write", "Bash"],
        permission_mode="acceptEdits"
    )
    session = ConversationSession(options)
    await session.start()

# 예제 대화:
# 턴 1 - 당신: "hello.py라는 파일을 만들어주세요"
# 턴 1 - Claude: "hello.py 파일을 만들어드리겠습니다..."
# 턴 2 - 당신: "그 파일에 무엇이 들어있나요?"  
# 턴 2 - Claude: "방금 만든 hello.py 파일에는..." (기억함!)
# 턴 3 - 당신: "거기에 main 함수를 추가해주세요"
# 턴 3 - Claude: "hello.py에 main 함수를 추가하겠습니다..." (어떤 파일인지 알고 있음!)

asyncio.run(main())

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]:
    """실행 전에 모든 도구 사용을 로그합니다."""
    tool_name = input_data.get('tool_name', 'unknown')
    print(f"[PRE-TOOL] 사용 예정: {tool_name}")

    # 여기서 도구 실행을 수정하거나 차단할 수 있습니다
    if tool_name == "Bash" and "rm -rf" in str(input_data.get('tool_input', {})):
        return {
            'hookSpecificOutput': {
                'hookEventName': 'PreToolUse',
                'permissionDecision': 'deny',
                'permissionDecisionReason': '위험한 명령이 차단되었습니다'
            }
        }
    return {}

async def post_tool_logger(
    input_data: dict[str, Any],
    tool_use_id: str | None,
    context: HookContext
) -> dict[str, Any]:
    """도구 실행 후 결과를 로그합니다."""
    tool_name = input_data.get('tool_name', 'unknown')
    print(f"[POST-TOOL] 완료됨: {tool_name}")
    return {}

async def user_prompt_modifier(
    input_data: dict[str, Any],
    tool_use_id: str | None,
    context: HookContext
) -> dict[str, Any]:
    """사용자 프롬프트에 컨텍스트를 추가합니다."""
    original_prompt = input_data.get('prompt', '')

    # 모든 프롬프트에 타임스탬프 추가
    from datetime import datetime
    timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")

    return {
        'hookSpecificOutput': {
            'hookEventName': 'UserPromptSubmit',
            'updatedPrompt': f"[{timestamp}] {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("현재 디렉토리의 파일을 나열해주세요")
        
        async for message in client.receive_response():
            # 훅이 자동으로 도구 사용을 로그할 것입니다
            pass

asyncio.run(main())

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(
            "다양한 정렬 알고리즘으로 5개의 Python 파일을 만들어주세요"
        )
        
        # 실시간으로 진행 상황 모니터링
        files_created = []
        async for message in client.receive_messages():
            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"🔨 생성 중: {file_path}")
                    elif isinstance(block, ToolResultBlock):
                        print(f"✅ 도구 실행 완료")
                    elif isinstance(block, TextBlock):
                        print(f"💭 Claude가 말합니다: {block.text[:100]}...")
            
            # 최종 결과를 받았는지 확인
            if hasattr(message, 'subtype') and message.subtype in ['success', 'error']:
                print(f"\n🎯 작업 완료!")
                break

asyncio.run(monitor_progress())

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="setup.py가 있는 Python 프로젝트 구조를 만들어주세요",
        options=options
    ):
        if isinstance(message, AssistantMessage):
            for block in message.content:
                if isinstance(block, ToolUseBlock):
                    print(f"도구 사용: {block.name}")

asyncio.run(create_project())

from claude_agent_sdk import (
    query,
    CLINotFoundError,
    ProcessError,
    CLIJSONDecodeError
)

try:
    async for message in query(prompt="안녕하세요"):
        print(message)
except CLINotFoundError:
    print("Claude Code를 설치해주세요: npm install -g @anthropic-ai/claude-code")
except ProcessError as e:
    print(f"프로세스가 종료 코드 {e.exit_code}로 실패했습니다")
except CLIJSONDecodeError as e:
    print(f"응답 파싱에 실패했습니다: {e}")

from claude_agent_sdk import ClaudeSDKClient
import asyncio

async def interactive_session():
    async with ClaudeSDKClient() as client:
        # 초기 메시지 전송
        await client.query("날씨가 어떤가요?")
        
        # 응답 처리
        async for msg in client.receive_response():
            print(msg)
        
        # 후속 조치 전송
        await client.query("그것에 대해 더 자세히 말해주세요")
        
        # 후속 응답 처리
        async for msg in client.receive_response():
            print(msg)

asyncio.run(interactive_session())

from claude_agent_sdk import (
    ClaudeSDKClient,
    ClaudeAgentOptions,
    tool,
    create_sdk_mcp_server,
    AssistantMessage,
    TextBlock
)
import asyncio
from typing import Any

# @tool 데코레이터로 사용자 정의 도구 정의
@tool("calculate", "수학 계산 수행", {"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}"
            }]
        }
    except Exception as e:
        return {
            "content": [{
                "type": "text",
                "text": f"오류: {str(e)}"
            }],
            "is_error": True
        }

@tool("get_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}"
        }]
    }

async def main():
    # 사용자 정의 도구로 SDK MCP 서버 생성
    my_server = create_sdk_mcp_server(
        name="utilities",
        version="1.0.0",
        tools=[calculate, get_time]
    )

    # 서버로 옵션 구성
    options = ClaudeAgentOptions(
        mcp_servers={"utils": my_server},
        allowed_tools=[
            "mcp__utils__calculate",
            "mcp__utils__get_time"
        ]
    )
    
    # 대화형 도구 사용을 위해 ClaudeSDKClient 사용
    async with ClaudeSDKClient(options=options) as client:
        await client.query("123 * 456은 얼마인가요?")
        
        # 계산 응답 처리
        async for message in client.receive_response():
            if isinstance(message, AssistantMessage):
                for block in message.content:
                    if isinstance(block,TextBlock):
                        print(f"계산: {block.text}")
        
        # 시간 쿼리로 후속 조치
        await client.query("지금 몇 시인가요?")
        
        async for message in client.receive_response():
            if isinstance(message, AssistantMessage):
                for block in message.content:
                    if isinstance(block, TextBlock):
                        print(f"시간: {block.text}")

asyncio.run(main())