Claude API Tool Use 실전 가이드

Claude가 외부 도구를 호출하고, 결과를 해석하여 자연어로 응답하는 전체 흐름을 구현한다.

Tool Use란

Claude API의 Tool Use(Function Calling)는 모델이 외부 함수를 호출할 수 있게 하는 기능이다. Claude가 직접 API를 호출하는 것이 아니라, 어떤 도구를 어떤 인자로 호출해야 하는지 JSON으로 응답하면 서버가 이를 실행하고 결과를 다시 Claude에 전달하는 구조다.

전체 흐름

plaintext

User → "내 삼성전자 수익률 알려줘"
  ↓
Claude API (tools 포함)
  ↓
Response: stop_reason="tool_use"
  content: [text("잔고를 조회하겠습니다"), tool_use("get_balance", {...})]
  ↓
서버: get_balance 실행 → KIS API 호출 → 결과 획득
  ↓
Claude API (tool_result 포함)
  ↓
Response: stop_reason="end_turn"
  content: [text("삼성전자 100주 보유, 수익률 +4.17%")]

도구 정의 (JSON Schema)

도구는 name, description, input_schema 세 필드로 정의한다.

python

tools = [
    {
        "name": "get_stock_price",
        "description": "특정 종목의 현재 시세를 조회합니다. "
                       "현재가, 전일대비, 등락률, 거래량을 반환합니다.",
        "input_schema": {
            "type": "object",
            "properties": {
                "symbol": {
                    "type": "string",
                    "description": "종목코드 (국내: 005930, 해외: AAPL)"
                },
                "market": {
                    "type": "string",
                    "enum": ["domestic", "overseas"],
                    "description": "시장 구분"
                }
            },
            "required": ["symbol", "market"]
        }
    },
    {
        "name": "get_balance",
        "description": "현재 계좌 잔고와 보유 종목을 조회합니다. "
                       "총 평가금액, 손익, 각 종목별 수량/평균가/현재가를 반환합니다.",
        "input_schema": {
            "type": "object",
            "properties": {
                "account_type": {
                    "type": "string",
                    "enum": ["live", "paper"],
                    "description": "조회할 계좌 유형. live=실전, paper=모의투자"
                }
            },
            "required": ["account_type"]
        }
    },
    {
        "name": "place_order",
        "description": "주식 주문을 실행합니다. "
                       "반드시 사용자 확인 후 실행해야 합니다.",
        "input_schema": {
            "type": "object",
            "properties": {
                "symbol": {"type": "string", "description": "종목코드"},
                "market": {"type": "string", "enum": ["domestic", "overseas"]},
                "side": {"type": "string", "enum": ["buy", "sell"]},
                "quantity": {"type": "integer", "description": "주문 수량"},
                "price": {"type": "number", "description": "주문 가격. 0이면 시장가"},
                "account_type": {"type": "string", "enum": ["live", "paper"]}
            },
            "required": ["symbol", "market", "side", "quantity", "price", "account_type"]
        }
    }
]

도구 설명 작성 베스트 프랙티스

도구가 무엇을 하는지, 무엇을 반환하는지 명확히 기술한다
각 파라미터의 description에 예시값을 포함한다 (e.g., "005930", "AAPL")
enum으로 가능한 값을 제한하여 Claude의 정확도를 높인다
위험한 도구(주문 등)에는 제약 조건을 명시한다 ("반드시 사용자 확인 후 실행")

tool_choice 옵션

옵션	설명	용도
`{"type": "auto"}`	Claude가 도구 사용 여부를 자율 판단 (기본값)	일반 대화
`{"type": "any"}`	반드시 하나 이상의 도구를 호출	데이터 조회 강제
`{"type": "tool", "name": "get_balance"}`	특정 도구를 반드시 호출	특정 작업 강제

python

# 잔고 조회를 강제하는 예시
response = await client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=4096,
    tools=tools,
    tool_choice={"type": "tool", "name": "get_balance"},
    messages=messages,
)

Agentic Loop 구현

Tool Use의 핵심은 루프다. Claude가 stop_reason="tool_use"를 반환하면, 서버가 도구를 실행하고 결과를 다시 보내는 것을 반복한다. 최종적으로 stop_reason="end_turn"이 나올 때까지 이 루프를 돌린다.

Python 구현

python

import anthropic
from app.chat.tool_executor import execute_tool
 
client = anthropic.AsyncAnthropic()
 
SYSTEM_PROMPT = """당신은 PersonalTrader의 AI 어시스턴트입니다.
사용자의 투자 관련 질문에 답변하기 위해 제공된 도구를 활용하세요.
- 계좌/시세 데이터를 조회할 때는 반드시 도구를 사용하세요
- 주문 실행 시 반드시 사용자에게 확인을 구하세요
- 금액은 한국 원화 형식(₩1,234,567)으로 표시하세요
- 수익률은 소수점 2자리까지 표시하세요"""
 
 
async def chat(
    user_message: str,
    history: list[dict],
    tools: list[dict],
) -> tuple[str, list[dict]]:
    """Tool Use 루프를 포함한 채팅 함수."""
    messages = history + [{"role": "user", "content": user_message}]
 
    while True:
        response = await client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=4096,
            system=SYSTEM_PROMPT,
            tools=tools,
            messages=messages,
        )
 
        # assistant 응답을 history에 추가
        messages.append({"role": "assistant", "content": response.content})
 
        # tool_use가 아니면 루프 종료
        if response.stop_reason != "tool_use":
            break
 
        # 도구 실행 및 결과 수집
        tool_results = []
        for block in response.content:
            if block.type == "tool_use":
                result = await execute_tool(block.name, block.input)
                tool_results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": result,
                })
 
        # tool_result를 user 메시지로 추가
        messages.append({"role": "user", "content": tool_results})
 
    # 최종 텍스트 추출
    final_text = "".join(
        block.text for block in response.content if block.type == "text"
    )
    return final_text, messages

도구 실행기 (Tool Executor)

python

import json
 
async def execute_tool(name: str, params: dict) -> str:
    """도구 이름과 파라미터로 실제 기능을 실행하고 JSON 문자열을 반환한다."""
 
    handlers = {
        "get_balance": _handle_get_balance,
        "get_stock_price": _handle_get_stock_price,
        "get_daily_prices": _handle_get_daily_prices,
        "get_agent_status": _handle_get_agent_status,
        "place_order": _handle_place_order,
    }
 
    handler = handlers.get(name)
    if not handler:
        return json.dumps({"error": f"Unknown tool: {name}"})
 
    try:
        result = await handler(params)
        return json.dumps(result, ensure_ascii=False, default=str)
    except Exception as e:
        return json.dumps({"error": str(e)})

messages 배열 구조

Claude API의 messages는 user와 assistant 역할이 교대하는 배열이다. Tool Use 과정에서 도구 결과는 user 역할로 추가된다.

python

messages = [
    # 1턴: 사용자 질문
    {"role": "user", "content": "내 잔고 알려줘"},
 
    # 2턴: Claude 응답 (도구 호출 포함)
    {"role": "assistant", "content": [
        {"type": "text", "text": "계좌 잔고를 조회하겠습니다."},
        {"type": "tool_use", "id": "toolu_abc", "name": "get_balance",
         "input": {"account_type": "live"}}
    ]},
 
    # 3턴: 도구 결과 (user 역할)
    {"role": "user", "content": [
        {"type": "tool_result", "tool_use_id": "toolu_abc",
         "content": '{"total_eval": 15000000, "holdings": [...]}'}
    ]},
 
    # 4턴: Claude 최종 응답
    {"role": "assistant", "content": [
        {"type": "text", "text": "현재 총 평가금액은 ₩15,000,000입니다..."}
    ]},
]

tool_result는 반드시 해당 tool_use의 id와 매칭되어야 한다. tool_use_id가 일치하지 않으면 API 에러가 발생한다.

병렬 도구 호출 (Parallel Tool Use)

Claude는 독립적인 도구를 한 번의 응답에서 여러 개 동시에 호출할 수 있다. 예를 들어 "삼성전자와 애플 현재가 비교해줘"라고 하면, 두 종목의 시세를 한 번에 요청한다.

python

# Claude 응답에 여러 tool_use 블록이 포함된 경우
# content: [
#   tool_use("get_stock_price", {"symbol": "005930", "market": "domestic"}),
#   tool_use("get_stock_price", {"symbol": "AAPL", "market": "overseas"})
# ]
 
import asyncio
 
async def execute_tools_parallel(tool_blocks: list) -> list[dict]:
    """여러 도구를 병렬로 실행한다."""
    tasks = [
        execute_tool(block.name, block.input)
        for block in tool_blocks
        if block.type == "tool_use"
    ]
    results = await asyncio.gather(*tasks, return_exceptions=True)
 
    tool_results = []
    for block, result in zip(
        [b for b in tool_blocks if b.type == "tool_use"], results
    ):
        if isinstance(result, Exception):
            content = json.dumps({"error": str(result)})
        else:
            content = result
        tool_results.append({
            "type": "tool_result",
            "tool_use_id": block.id,
            "content": content,
        })
    return tool_results

asyncio.gather를 사용하면 네트워크 호출이 동시에 진행되므로, 순차 실행 대비 응답 시간을 대폭 줄일 수 있다.

SSE 스트리밍

실시간 채팅 UI를 구현하려면 스트리밍이 필수다. Claude API는 Server-Sent Events(SSE) 방식으로 응답을 점진적으로 전달한다.

스트리밍 이벤트 흐름

plaintext

event: message_start
data: {"type":"message_start","message":{"id":"msg_...","role":"assistant",...}}
 
event: content_block_start
data: {"type":"content_block_start","index":0,
       "content_block":{"type":"text","text":""}}
 
event: content_block_delta
data: {"type":"content_block_delta","index":0,
       "delta":{"type":"text_delta","text":"잔고를"}}
 
event: content_block_delta
data: {"type":"content_block_delta","index":0,
       "delta":{"type":"text_delta","text":" 조회"}}
 
event: content_block_stop
data: {"type":"content_block_stop","index":0}
 
event: content_block_start
data: {"type":"content_block_start","index":1,
       "content_block":{"type":"tool_use","id":"toolu_...","name":"get_balance"}}
 
event: content_block_delta
data: {"type":"content_block_delta","index":1,
       "delta":{"type":"input_json_delta","partial_json":"{\"account_type\":"}}
 
event: content_block_delta
data: {"type":"content_block_delta","index":1,
       "delta":{"type":"input_json_delta","partial_json":" \"live\"}"}}
 
event: content_block_stop
data: {"type":"content_block_stop","index":1}
 
event: message_delta
data: {"type":"message_delta","delta":{"stop_reason":"tool_use"},
       "usage":{"output_tokens":89}}
 
event: message_stop
data: {"type":"message_stop"}

이벤트 타입 정리

이벤트	설명	주요 필드
`message_start`	메시지 시작	`message.id`, `message.role`
`content_block_start`	콘텐츠 블록 시작	`content_block.type` (text / tool_use)
`content_block_delta`	증분 데이터 전송	`delta.type` (text_delta / input_json_delta)
`content_block_stop`	블록 완료	`index`
`message_delta`	메시지 메타 업데이트	`stop_reason`, `usage`
`message_stop`	스트림 종료	—

input_json_delta는 부분 JSON 문자열이다. content_block_stop 시점에 전체 JSON을 파싱해야 한다.

Python 스트리밍 구현

python

async def chat_stream(
    user_message: str,
    history: list[dict],
    tools: list[dict],
):
    """스트리밍으로 응답을 반환하는 제너레이터."""
    messages = history + [{"role": "user", "content": user_message}]
 
    async with client.messages.stream(
        model="claude-sonnet-4-6",
        max_tokens=4096,
        system=SYSTEM_PROMPT,
        tools=tools,
        messages=messages,
    ) as stream:
        async for event in stream:
            if event.type == "content_block_start":
                if event.content_block.type == "text":
                    yield {"type": "text_start"}
                elif event.content_block.type == "tool_use":
                    yield {
                        "type": "tool_start",
                        "name": event.content_block.name,
                        "id": event.content_block.id,
                    }
 
            elif event.type == "content_block_delta":
                if event.delta.type == "text_delta":
                    yield {"type": "text_delta", "text": event.delta.text}
                elif event.delta.type == "input_json_delta":
                    yield {
                        "type": "input_delta",
                        "partial_json": event.delta.partial_json,
                    }
 
            elif event.type == "content_block_stop":
                yield {"type": "block_stop", "index": event.index}
 
        # 스트림 완료 후 최종 메시지 획득
        final_message = await stream.get_final_message()
        yield {"type": "message_complete", "message": final_message}

FastAPI SSE 엔드포인트

python

from fastapi import APIRouter
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
 
router = APIRouter()
 
class ChatRequest(BaseModel):
    message: str
    session_id: str
 
@router.post("/api/chat/stream")
async def chat_stream_endpoint(req: ChatRequest):
    """SSE 스트리밍 채팅 엔드포인트."""
    history = await load_session_history(req.session_id)
 
    async def event_generator():
        async for chunk in chat_stream_with_tool_loop(
            req.message, history, tools
        ):
            yield f"data: {json.dumps(chunk, ensure_ascii=False)}\n\n"
        yield "data: [DONE]\n\n"
 
    return StreamingResponse(
        event_generator(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "X-Accel-Buffering": "no",
        },
    )

비용 최적화

Claude API는 토큰 단위로 과금된다. Tool Use를 사용하면 도구 정의, 시스템 프롬프트, 대화 이력이 매 요청마다 전송되므로 비용이 빠르게 증가할 수 있다. 이를 최적화하는 세 가지 핵심 전략이 있다.

프롬프트 캐싱 (Prompt Caching)

도구 정의와 시스템 프롬프트는 매 요청마다 동일하다. 이를 캐싱하면 읽기 비용이 90% 절감된다.

캐싱 비용 비교

항목	기본 비용	캐시 쓰기 (첫 요청)	캐시 읽기 (이후 요청)
Sonnet 4.6	$3/MTok	$3.75/MTok (x1.25)	$0.30/MTok (x0.1)
Haiku 4.5	$1/MTok	$1.25/MTok (x1.25)	$0.10/MTok (x0.1)

캐시 순서: tools -> system -> messages 순서로 prefix가 구성된다
5분 TTL: 마지막 사용으로부터 5분 후 만료, 사용 시마다 갱신
최소 토큰: Sonnet 4.6 기준 1,024 토큰 이상이어야 캐싱 가능

구현

python

async def create_cached_request(messages: list[dict]) -> dict:
    """도구 + 시스템 프롬프트를 캐싱하는 요청 구성."""
 
    # 도구 정의의 마지막 항목에 cache_control 추가
    cached_tools = list(tools)
    cached_tools[-1] = {
        **cached_tools[-1],
        "cache_control": {"type": "ephemeral"},
    }
 
    # 시스템 프롬프트에 cache_control 추가
    system = [
        {
            "type": "text",
            "text": SYSTEM_PROMPT,
            "cache_control": {"type": "ephemeral"},
        }
    ]
 
    # 대화 이력의 마지막 user 메시지에도 cache_control 추가
    cached_messages = list(messages)
    for i in range(len(cached_messages) - 1, -1, -1):
        if cached_messages[i]["role"] == "user":
            content = cached_messages[i]["content"]
            if isinstance(content, str):
                cached_messages[i]["content"] = [
                    {"type": "text", "text": content,
                     "cache_control": {"type": "ephemeral"}}
                ]
            elif isinstance(content, list):
                cached_messages[i]["content"][-1]["cache_control"] = {
                    "type": "ephemeral"
                }
            break
 
    response = await client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=4096,
        system=system,
        tools=cached_tools,
        messages=cached_messages,
    )
    return response

도구 정의 5개가 약 2,000 토큰이라면, 캐시 히트 시 매 요청마다 약 $0.006을 절약한다. 대화가 길어질수록 절감 효과가 커진다.

모델 티어링 (Model Tiering)

작업 복잡도에 따라 모델을 구분하여 비용을 절감한다.

모델	Input	Output	용도
Haiku 4.5	$1/MTok	$5/MTok	단순 조회, 포맷팅
Sonnet 4.6	$3/MTok	$15/MTok	분석, 대화, 도구 활용
Opus 4.6	$5/MTok	$25/MTok	복잡한 추론, 보고서

python

def select_model(user_message: str, tool_results: list | None = None) -> str:
    """메시지 내용에 따라 적절한 모델을 선택한다."""
 
    # 단순 조회 → Haiku
    simple_patterns = ["시세", "현재가", "잔고", "보유", "몇 주"]
    if any(p in user_message for p in simple_patterns) and not tool_results:
        return "claude-haiku-4-5"
 
    # 분석/추천 → Sonnet
    analysis_patterns = ["분석", "추천", "전략", "비교", "왜", "이유"]
    if any(p in user_message for p in analysis_patterns):
        return "claude-sonnet-4-6"
 
    # 기본값
    return "claude-sonnet-4-6"

추가 최적화 기법

도구 결과 압축: 도구가 반환하는 전체 데이터 중 Claude가 필요한 필드만 추출한다.

python

def compress_tool_result(result: dict, tool_name: str) -> str:
    """도구 결과에서 Claude가 필요한 필드만 추출한다."""
    if tool_name == "get_balance":
        return json.dumps({
            "total": result["total_eval"],
            "profit_rate": result["profit_rate"],
            "holdings": [
                {"name": h["name"], "qty": h["quantity"],
                 "rate": h["profit_rate"]}
                for h in result["holdings"]
            ]
        }, ensure_ascii=False)
    return json.dumps(result, ensure_ascii=False)

max_tokens 제한: 응답 길이를 작업 유형에 맞게 제한한다. 단순 조회는 1,024, 분석은 4,096, 보고서는 8,192.

Batch API: 비실시간 작업(보고서 생성 등)은 Batch API를 사용하면 50% 할인이 적용된다.

대화 이력 관리

컨텍스트 윈도우 관리

긴 대화에서 토큰 한도를 초과하지 않도록 오래된 메시지부터 제거한다.

python

MAX_CONTEXT_TOKENS = 150_000  # Claude Sonnet 4.6 기준 여유 포함
 
async def trim_history(
    messages: list[dict],
    max_tokens: int = MAX_CONTEXT_TOKENS,
) -> list[dict]:
    """토큰 한도를 초과하면 오래된 메시지부터 제거한다."""
    total_tokens = estimate_tokens(messages)
 
    if total_tokens <= max_tokens:
        return messages
 
    # 가장 오래된 user/assistant 쌍부터 제거
    # tool_use/tool_result 쌍을 함께 제거하여 일관성을 유지한다
    trimmed = list(messages)
    while estimate_tokens(trimmed) > max_tokens and len(trimmed) > 2:
        trimmed.pop(0)
 
    return trimmed

tool_use와 tool_result는 반드시 쌍으로 존재해야 한다. 하나만 제거하면 API 에러가 발생하므로, 이력 정리 시 항상 쌍을 함께 제거해야 한다.

에러 처리

도구 실행 에러

도구 실행이 실패하면 is_error: True로 Claude에 알린다. Claude는 이를 받아 사용자에게 적절한 에러 메시지를 생성한다.

python

tool_results.append({
    "type": "tool_result",
    "tool_use_id": block.id,
    "content": json.dumps({"error": "KIS API 연결 실패: 토큰 만료"}),
    "is_error": True,
})

API 에러 핸들링

python

import anthropic
 
try:
    response = await client.messages.create(...)
except anthropic.RateLimitError:
    # 429: 레이트 리밋 → 지수 백오프 재시도
    await asyncio.sleep(retry_delay)
except anthropic.APIStatusError as e:
    if e.status_code == 529:
        # 과부하 → 잠시 대기 후 재시도
        await asyncio.sleep(30)
    else:
        raise

서비스 아키텍처

전체 시스템을 하나의 다이어그램으로 정리하면 다음과 같다.

plaintext

┌─────────────────────────────────────────────────┐
│  Frontend (Next.js)                             │
│  useChat() ──SSE──→ /api/chat/stream           │
└────────────────────────┬────────────────────────┘
                         │
┌────────────────────────▼────────────────────────┐
│  FastAPI Backend                                │
│  ┌──────────────────────────────────────────┐   │
│  │  ChatService                             │   │
│  │  ├─ model_selector → Haiku/Sonnet/Opus   │   │
│  │  ├─ tool_executor → KIS API / Agents     │   │
│  │  ├─ session_manager → PostgreSQL         │   │
│  │  └─ prompt_cache → tools + system 캐싱    │   │
│  └──────────────────────────────────────────┘   │
│                    │                             │
│  ┌─────────────────▼──────────────────────┐     │
│  │  Claude API (Anthropic)                │     │
│  │  messages.stream() + tools             │     │
│  └────────────────────────────────────────┘     │
│                    │                             │
│  ┌─────────────────▼──────────────────────┐     │
│  │  Tool Handlers                         │     │
│  │  ├─ get_balance → KISClient.account    │     │
│  │  ├─ get_stock_price → KISClient.dom/ov │     │
│  │  ├─ get_daily_prices → KISClient.dom   │     │
│  │  ├─ get_agent_status → Orchestrator    │     │
│  │  └─ place_order → KISClient.order      │     │
│  └────────────────────────────────────────┘     │
└─────────────────────────────────────────────────┘

Tool Use를 활용하면 Claude가 단순한 챗봇을 넘어 실시간 데이터를 조회하고, 외부 시스템과 상호작용하는 에이전트로 동작할 수 있다. 핵심은 Agentic Loop를 안정적으로 구현하고, 프롬프트 캐싱과 모델 티어링으로 비용을 관리하는 것이다.