Claude API 완전 가이드: Tool Use, Vision, Streaming 실전 활용

import anthropic

client = anthropic.Anthropic(api_key="sk-ant-...")

# 기본 메시지
message = client.messages.create(
    model="claude-sonnet-4-6-20251001",
    max_tokens=1024,
    messages=[
        {"role": "user", "content": "안녕하세요!"}
    ]
)
print(message.content[0].text)

flowchart LR
    User[사용자 질문] --> Claude[Claude]
    Claude --> |tool_use 응답| App[앱 서버]
    App --> |실제 함수 실행| API[외부 API/DB]
    API --> |결과| App
    App --> |tool_result 전달| Claude
    Claude --> |최종 답변| User

import anthropic
import json

client = anthropic.Anthropic()

# 도구 정의
tools = [
    {
        "name": "get_weather",
        "description": "특정 도시의 현재 날씨를 조회합니다",
        "input_schema": {
            "type": "object",
            "properties": {
                "city": {
                    "type": "string",
                    "description": "도시명 (예: 서울, 부산)"
                },
                "unit": {
                    "type": "string",
                    "enum": ["celsius", "fahrenheit"],
                    "description": "온도 단위"
                }
            },
            "required": ["city"]
        }
    }
]

# 1단계: 도구 호출 요청
response = client.messages.create(
    model="claude-sonnet-4-6-20251001",
    max_tokens=1024,
    tools=tools,
    messages=[{"role": "user", "content": "서울 날씨 알려줘"}]
)

# 2단계: Claude가 도구 호출 요청을 반환
if response.stop_reason == "tool_use":
    tool_call = next(b for b in response.content if b.type == "tool_use")
    tool_name = tool_call.name        # "get_weather"
    tool_input = tool_call.input      # {"city": "서울", "unit": "celsius"}

    # 3단계: 실제 함수 실행
    def get_weather(city, unit="celsius"):
        # 실제로는 날씨 API 호출
        return {"temperature": 18, "condition": "맑음", "humidity": 55}

    result = get_weather(**tool_input)

    # 4단계: 결과를 Claude에게 전달
    final_response = client.messages.create(
        model="claude-sonnet-4-6-20251001",
        max_tokens=1024,
        tools=tools,
        messages=[
            {"role": "user", "content": "서울 날씨 알려줘"},
            {"role": "assistant", "content": response.content},
            {
                "role": "user",
                "content": [{
                    "type": "tool_result",
                    "tool_use_id": tool_call.id,
                    "content": json.dumps(result, ensure_ascii=False)
                }]
            }
        ]
    )
    print(final_response.content[0].text)
    # "서울은 현재 18°C이고 맑습니다. 습도는 55%입니다."

tools = [
    {"name": "get_stock_price", ...},
    {"name": "get_news", ...},
    {"name": "calculate", ...},
]

response = client.messages.create(
    model="claude-sonnet-4-6-20251001",
    max_tokens=2048,
    tools=tools,
    messages=[{"role": "user", "content": "삼성전자 주가와 최신 뉴스 요약해줘"}]
)

# Claude가 get_stock_price와 get_news를 동시에 호출할 수 있음
tool_calls = [b for b in response.content if b.type == "tool_use"]
# → 병렬 실행으로 레이턴시 절감

import base64
from pathlib import Path

# 방법 1: 파일 Base64
image_data = base64.standard_b64encode(Path("chart.png").read_bytes()).decode()

response = client.messages.create(
    model="claude-sonnet-4-6-20251001",
    max_tokens=1024,
    messages=[{
        "role": "user",
        "content": [
            {
                "type": "image",
                "source": {
                    "type": "base64",
                    "media_type": "image/png",
                    "data": image_data,
                },
            },
            {"type": "text", "text": "이 차트를 분석하고 주요 트렌드를 설명해줘"}
        ],
    }]
)

# 방법 2: URL
response = client.messages.create(
    model="claude-sonnet-4-6-20251001",
    max_tokens=1024,
    messages=[{
        "role": "user",
        "content": [
            {"type": "image", "source": {"type": "url", "url": "https://..."}},
            {"type": "text", "text": "이 이미지에서 텍스트를 추출해줘"}
        ]
    }]
)

# 스트리밍으로 토큰 순차 출력
with client.messages.stream(
    model="claude-sonnet-4-6-20251001",
    max_tokens=1024,
    messages=[{"role": "user", "content": "한국 AI 산업 동향 분석해줘"}]
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)

# FastAPI에서 SSE(Server-Sent Events)로 스트리밍
from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()

@app.post("/chat")
async def chat(message: str):
    async def generate():
        with client.messages.stream(
            model="claude-sonnet-4-6-20251001",
            max_tokens=2048,
            messages=[{"role": "user", "content": message}]
        ) as stream:
            for text in stream.text_stream:
                yield f"data: {text}

"
        yield "data: [DONE]

"

    return StreamingResponse(generate(), media_type="text/event-stream")

SYSTEM = '''당신은 ai.zip의 AI 가이드입니다.

역할:
- AI 모델, 도구, 기술에 대한 정확하고 실용적인 정보 제공
- 한국 개발자 관점에서 비용/성능 트레이드오프 분석
- 확실하지 않은 정보는 솔직하게 모른다고 말하기

형식:
- 마크다운 사용 (##, **, 표, 코드블록)
- 2,000자 이내로 간결하게
- 구체적인 수치와 예시 포함

금지:
- "물론이죠!", "좋은 질문이에요!" 같은 빈 말
- 확인되지 않은 가격 정보 단정적으로 말하기'''

response = client.messages.create(
    model="claude-sonnet-4-6-20251001",
    max_tokens=2048,
    system=SYSTEM,
    messages=[{"role": "user", "content": "GPT-4o랑 Claude 뭐가 달라?"}]
)

# 1. 짧은 태스크는 Haiku 사용 (5~10배 저렴)
classifier_response = client.messages.create(
    model="claude-haiku-4-5-20251001",  # 분류는 Haiku로
    max_tokens=10,
    messages=[{"role": "user", "content": f"분류: {text}
답: 긍정/부정/중립 중 하나만"}]
)

# 2. 캐시 활용 (동일 System Prompt 반복 시)
# Anthropic의 Prompt Caching으로 비용 90% 절감 가능
response = client.messages.create(
    model="claude-sonnet-4-6-20251001",
    max_tokens=1024,
    system=[{
        "type": "text",
        "text": LONG_SYSTEM_PROMPT,
        "cache_control": {"type": "ephemeral"}  # 5분 캐싱
    }],
    messages=[{"role": "user", "content": user_message}]
)

# 3. max_tokens 적절히 설정
# 필요 이상으로 크게 설정하면 불필요한 생성 → 비용 낭비