Інтеграція Anthropic Tool Use Function Calling у застосунок

Інтеграція Anthropic Tool Use (Function Calling) у додаток

Tool Use — механізм, що дозволяє Claude викликати функції у вашому додатку. Модель не виконує код безпосередньо: вона повертає структурований JSON з назвою інструмента та аргументами, а ваш код виконує інструмент і повертає результат. Це основа для побудови агентів з доступом до реальних даних та дій.

Базовий цикл Tool Use

import anthropic
import json
from typing import Any

client = anthropic.Anthropic()

# Визначення інструментів
TOOLS = [
    {
        "name": "search_database",
        "description": "Пошук клієнтів у CRM за параметрами",
        "input_schema": {
            "type": "object",
            "properties": {
                "query": {"type": "string", "description": "Пошуковий запит"},
                "limit": {"type": "integer", "description": "Кількість результатів", "default": 10},
                "status": {
                    "type": "string",
                    "enum": ["active", "churned", "trial"],
                    "description": "Фільтр за статусом"
                }
            },
            "required": ["query"]
        }
    },
    {
        "name": "send_email",
        "description": "Надіслати електронний лист клієнту",
        "input_schema": {
            "type": "object",
            "properties": {
                "to": {"type": "string"},
                "subject": {"type": "string"},
                "body": {"type": "string"},
            },
            "required": ["to", "subject", "body"]
        }
    }
]

# Диспетчер інструментів
def execute_tool(tool_name: str, tool_input: dict) -> Any:
    if tool_name == "search_database":
        return search_crm(**tool_input)
    elif tool_name == "send_email":
        return send_email_via_smtp(**tool_input)
    raise ValueError(f"Unknown tool: {tool_name}")

def run_agent(user_message: str) -> str:
    """Повний агентський цикл з tool use"""
    messages = [{"role": "user", "content": user_message}]

    while True:
        response = client.messages.create(
            model="claude-sonnet-4-5",
            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 ""

        # Обробка tool_use блоків
        tool_results = []
        for block in response.content:
            if block.type == "tool_use":
                print(f"Calling tool: {block.name}({block.input})")
                result = execute_tool(block.name, block.input)

                tool_results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": json.dumps(result, ensure_ascii=False),
                })

        messages.append({"role": "assistant", "content": response.content})
        messages.append({"role": "user", "content": tool_results})

Паралельні виклики інструментів

Claude може викликати кілька інструментів в одній відповіді — використовуйте asyncio для паралельного виконання:

import asyncio

async def execute_tool_async(tool_name: str, tool_input: dict) -> tuple[str, Any]:
    """Асинхронне виконання інструмента"""
    result = await asyncio.to_thread(execute_tool, tool_name, tool_input)
    return tool_name, result

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

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

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

        # Виконати всі інструменти паралельно
        tool_use_blocks = [b for b in response.content if b.type == "tool_use"]

        results = await asyncio.gather(*[
            execute_tool_async(b.name, b.input)
            for b in tool_use_blocks
        ])

        tool_results = [
            {
                "type": "tool_result",
                "tool_use_id": block.id,
                "content": json.dumps(result, ensure_ascii=False),
            }
            for block, (_, result) in zip(tool_use_blocks, results)
        ]

        messages.append({"role": "assistant", "content": response.content})
        messages.append({"role": "user", "content": tool_results})

Примусовий виклик інструмента

# tool_choice="required" — Claude повинен викликати принаймні один інструмент
# tool_choice={"type": "tool", "name": "..."} — викликати конкретний

response = client.messages.create(
    model="claude-sonnet-4-5",
    max_tokens=1024,
    tools=TOOLS,
    tool_choice={"type": "tool", "name": "search_database"},
    messages=[{"role": "user", "content": "Знайти клієнтів зі статусом trial"}],
)

Обробка помилок інструментів

# Передайте помилки назад до моделі — вона адаптує поведінку
try:
    result = execute_tool(block.name, block.input)
    tool_results.append({
        "type": "tool_result",
        "tool_use_id": block.id,
        "content": json.dumps(result),
    })
except Exception as e:
    tool_results.append({
        "type": "tool_result",
        "tool_use_id": block.id,
        "content": f"Error: {str(e)}",
        "is_error": True,  # Модель знає про помилку та може виправити
    })

Практичний кейс: CRM помічник

Завдання: менеджери по продажам витратили 20–30 хвилин на підготовку до зустрічей (пошук у CRM, останні контакти, відкриті завдання).

Інструменти: get_customer_info, get_deal_history, get_recent_activities, create_task, get_calendar.

Результат: підготовка до зустрічі — 2 хв через діалог з помічником.

Терміни

• Базовий tool use цикл з 2–3 інструментами: 2–3 дні
• Паралельні виклики + обробка помилок: 1–2 дні
• Production-ready з логуванням та retry: 1 тиждень