टूल उपयोग पैटर्न: विश्वसनीय एजेंट-टूल इंटरफेस बनाना

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": "₹1000 से कम के इलेक्ट्रॉनिक्स खोजें"}]
)

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}s के बाद टाइमआउट"
        )
    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