Claude Agent SDK for Python: Claude Code 내장하여 프로그래밍식 AI 에이전트 구축하기

import anyio
from claude_agent_sdk import query

async def main():
    async for message in query(prompt="/api/users 엔드포인트가 있는 Flask 앱 만들기"):
        print(message)

anyio.run(main)

from claude_agent_sdk import ClaudeAgentOptions, ClaudeSDKClient

options = ClaudeAgentOptions(
    system_prompt="당신은 시니어 Python 아키텍트입니다.",
    max_turns=5,
    permission_mode='autoApprove'
)

async with ClaudeSDKClient(options=options) as client:
    await client.query("이 저장소에 대한 마이크로서비스 아키텍처 설계")
    async for msg in client.receive_response():
        print(msg)

from claude_agent_sdk import tool, create_sdk_mcp_server, ClaudeAgentOptions, ClaudeSDKClient

@tool("deploy", "프로덕션에 앱 배포", {
    "environment": str,
    "branch": str
})
async def deploy_app(args):
    env = args["environment"]
    branch = args["branch"]
    # 여기에서 배포 로직 실행
    return {"status": f"{branch}를 {env}에 배포 완료"}

server = create_sdk_mcp_server(
    name="deployment-tools",
    tools=[deploy_app]
)

options = ClaudeAgentOptions(
    mcp_servers={"deploy": server},
    allowed_tools=["mcp__tools__deploy"]
)

options = ClaudeAgentOptions(
    allowed_tools=["Read", "Write", "Bash"],       # 자동 승인
    disallowed_tools=["WebFetch"],                   # 항상 차단
    permission_mode='acceptEdits',                   # 파일 편집 자동 승인
    can_use_tool=lambda tool_name, context: ...     # 사용자 지정 로직
)

from claude_agent_sdk import HookMatcher

async def block_destructive_commands(input_data, tool_use_id, context):
    if input_data["tool_name"] == "Bash":
        command = input_data["tool_input"].get("command", "")
        dangerous_patterns = ["rm -rf", "drop table", "> /dev/sda"]
        for pattern in dangerous_patterns:
            if pattern in command:
                return {
                    "hookSpecificOutput": {
                        "hookEventName": "PreToolUse",
                        "permissionDecision": "deny",
                        "permissionDecisionReason": f"위험 패턴 감지: {pattern}"
                    }
                }
    return {}

options = ClaudeAgentOptions(
    hooks={
        "PreToolUse": [
            HookMatcher(matcher="Bash", hooks=[block_destructive_commands])
        ]
    }
)

from claude_agent_sdk import query, ClaudeAgentOptions

async def review_code(pr_diff):
    options = ClaudeAgentOptions(
        system_prompt="""전문 코드 리뷰어입니다. 다음 사항 중점을 검사하세요:
        - 보안 취약점
        - 성능 문제  
        - 코드 스타일 및 모범 사례
        - 테스트 커버리지 빈자리""",
        allowed_tools=["Read", "Grep"],
        permission_mode='editFiles',
        max_turns=3
    )
    
    results = []
    async for message in query(
        prompt=f"다음 PR diff 리뷰:\n{pr_diff}",
        options=options
    ):
        results.append(str(message))
    
    return "\n".join(results)

from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions

async def self_heal_alert(alert_json):
    options = ClaudeAgentOptions(
        system_prompt="SRE 에이전트입니다. 프로덕션 문제 진단 및 복구.",
        allowed_tools=["Bash", "Read", "Edit", "code_run"],
        permission_mode='acceptEdits'
    )
    
    async with ClaudeSDKClient(options=options) as client:
        await client.query(f"알럿 조사: {alert_json}")
        async for msg in client.receive_response():
            log_action(msg)

from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions

class AnalyticsAgent:
    def __init__(self):
        self.options = ClaudeAgentOptions(
            system_prompt="""데이터 분석 어시스턴트입니다.
            자연어 요청을 SQL 쿼리, 데이터 시각화 및 
            인사이트로 변환하세요.""",
            allowed_tools=["code_run", "Read", "Write"],
            cwd="/data/reports/"
        )
    
    async def analyze(self, question: str) -> str:
        async with ClaudeSDKClient(options=self.options) as client:
            await client.query(question)
            return await client.receive_response()

pip install claude-agent-sdk

curl -fsSL https://claude.ai/install.sh | bash

export ANTHROPIC_API_KEY="sk-ant-your-key-here"

# hello_agent.py
from claude_agent_sdk import query

import anyio

async def main():
    async for message in query(
        prompt="반복 피보나치 수열 Python 함수 작성",
        options=ClaudeAgentOptions(max_turns=1)
    ):
        if hasattr(message, 'content'):
            for block in message.content:
                if hasattr(block, 'text'):
                    print(block.text)

anyio.run(main)

python hello_agent.py

# deploy_agent.py
from claude_agent_sdk import (
    tool, create_sdk_mcp_server, 
    ClaudeAgentOptions, ClaudeSDKClient
)

@tool("git_status", "현재 git 상태 보기", {})
async def git_status(_args):
    import subprocess
    result = subprocess.run(["git", "status", "--short"], 
                          capture_output=True, text=True)
    return {"content": [{"type": "text", "text": result.stdout}]}

server = create_sdk_mcp_server(
    name="git-tools", 
    tools=[git_status]
)

options = ClaudeAgentOptions(
    mcp_servers={"git": server},
    allowed_tools=["mcp__tools__git_status"]
)

import anyio

async def main():
    async with ClaudeSDKClient(options=options) as client:
        await client.query("현재 git 상태 표시 및 개선 권장")
        async for msg in client.receive_response():
            print(msg)

anyio.run(main)

from claude_agent_sdk import (
    CLINotFoundError,      # Claude Code 미설치
    CLIConnectionError,    # 연결 실패
    ProcessError,          # 서브프로세스 종료코드 이슈
    CLIJSONDecodeError     # 응답 파싱 실패
)

try:
    async for message in query(prompt="마이그레이션 실행"):
        process_result(message)
except CLINotFoundError:
    print("Claude Code가 설치되지 않았습니다. 실행: pip install claude-agent-sdk")
except CLIConnectionError as e:
    print(f"연결 오류: {e}. API 키를 확인하세요.")
except ProcessError as e:
    print(f"프로세스 실패, 종료코드 {e.exit_code}")
except CLIJSONDecodeError as e:
    print(f"파싱 오류: {e}")