도구 사용 패턴: 신뢰할 수 있는 에이전트-도구 인터페이스 구축

import anthropic

client = anthropic.Anthropic()

tools = [
    {
        "name": "search_products",
        "description": (
            "키워드로 제품 카탈로그를 검색한다. "
            "ID, 이름, 가격이 포함된 일치하는 제품 목록을 반환한다. "
            "사용자가 제품을 찾거나 탐색하고 싶을 때 사용한다."
        ),
        "input_schema": {
            "type": "object",
            "properties": {
                "query": {
                    "type": "string",
                    "description": "검색할 키워드"
                },
                "category": {
                    "type": "string",
                    "enum": ["electronics", "clothing", "food", "home", "all"],
                    "description": "필터링할 제품 카테고리. 지정하지 않으면 'all' 사용."
                },
                "max_results": {
                    "type": "integer",
                    "minimum": 1,
                    "maximum": 20,
                    "description": "반환할 결과 수. 기본값: 5"
                }
            },
            "required": ["query", "category"]
        }
    }
]

response = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    tools=tools,
    messages=[{"role": "user", "content": "10만원 이하 전자제품 찾아줘"}]
)

import json
from dataclasses import dataclass, asdict
from typing import Any, Optional


@dataclass
class ToolResult:
    success: bool
    data: Optional[Any] = None
    error: Optional[str] = None

    def to_content(self) -> str:
        return json.dumps(asdict(self), ensure_ascii=False)


def search_products(
    query: str,
    category: str,
    max_results: int = 5,
) -> ToolResult:
    try:
        raw_results = _query_database(query, category, limit=max_results)
        products = [
            {"id": r["product_id"], "name": r["title"], "price": r["price_usd"]}
            for r in raw_results
        ]
        return ToolResult(success=True, data={"products": products, "count": len(products)})
    except ConnectionError as e:
        return ToolResult(success=False, error=f"데이터베이스 사용 불가: {e}")
    except Exception as e:
        return ToolResult(success=False, error=f"검색 실패: {type(e).__name__}: {e}")


def _query_database(query, category, limit):
    return []

from concurrent.futures import ThreadPoolExecutor, as_completed


TOOL_REGISTRY = {
    "search_products": search_products,
}


def dispatch_tool(name: str, inputs: dict) -> ToolResult:
    handler = TOOL_REGISTRY.get(name)
    if not handler:
        return ToolResult(success=False, error=f"알 수 없는 도구: {name}")
    return handler(**inputs)


def process_tool_calls(response: anthropic.types.Message) -> list[dict]:
    tool_uses = [
        block for block in response.content
        if block.type == "tool_use"
    ]
    if not tool_uses:
        return []

    def execute(tool_use):
        result = dispatch_tool(tool_use.name, tool_use.input)
        return {
            "type": "tool_result",
            "tool_use_id": tool_use.id,
            "content": result.to_content(),
        }

    with ThreadPoolExecutor(max_workers=len(tool_uses)) as executor:
        futures = {executor.submit(execute, tu): tu for tu in tool_uses}
        results = []
        for future in as_completed(futures):
            results.append(future.result())

    return results


def run_agent(user_message: str) -> str:
    messages = [{"role": "user", "content": user_message}]

    while True:
        response = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=4096,
            tools=tools,
            messages=messages,
        )

        if response.stop_reason == "end_turn":
            for block in response.content:
                if hasattr(block, "text"):
                    return block.text
            return ""

        if response.stop_reason == "tool_use":
            tool_results = process_tool_calls(response)
            messages.append({"role": "assistant", "content": response.content})
            messages.append({"role": "user", "content": tool_results})
        else:
            break

    return ""

import signal


def timeout_handler(signum, frame):
    raise TimeoutError("도구 실행 시간 초과")


def safe_tool_call(
    name: str,
    inputs: dict,
    timeout_seconds: int = 30,
) -> ToolResult:
    """
    타임아웃으로 도구를 실행하고 모든 예외를 캐치한다.
    항상 ToolResult를 반환한다 — 절대 raise하지 않는다.
    """
    signal.signal(signal.SIGALRM, timeout_handler)
    signal.alarm(timeout_seconds)
    try:
        return dispatch_tool(name, inputs)
    except TimeoutError:
        return ToolResult(
            success=False,
            error=f"도구 '{name}'이 {timeout_seconds}초 후 시간 초과"
        )
    except Exception as e:
        return ToolResult(
            success=False,
            error=f"도구 '{name}'이 {type(e).__name__} 발생: {e}"
        )
    finally:
        signal.alarm(0)

MAX_TOOL_RESULT_CHARS = 8000


def validate_result(result: ToolResult, expected_keys: list[str]) -> ToolResult:
    if not result.success or not isinstance(result.data, dict):
        return result
    missing = [k for k in expected_keys if k not in result.data]
    if missing:
        return ToolResult(
            success=False,
            error=f"도구 응답에 예상 필드 누락: {missing}"
        )
    return result


def truncate_result(result: ToolResult) -> ToolResult:
    content = result.to_content()
    if len(content) <= MAX_TOOL_RESULT_CHARS:
        return result

    truncated_data = {
        "truncated": True,
        "chars_omitted": len(content) - MAX_TOOL_RESULT_CHARS,
        "content": content[:MAX_TOOL_RESULT_CHARS],
    }
    return ToolResult(
        success=result.success,
        data=truncated_data,
        error="결과 절단됨 — 컨텍스트 창에 비해 너무 큼",
    )


def safe_tool_call_validated(
    name: str,
    inputs: dict,
    expected_keys: list[str] | None = None,
) -> ToolResult:
    result = safe_tool_call(name, inputs)
    if expected_keys:
        result = validate_result(result, expected_keys)
    result = truncate_result(result)
    return result