Интеграция Anthropic Tool Use Function Calling в приложение

Интеграция Anthropic Tool Use (Function Calling) в приложение

Tool Use — механизм, позволяющий Claude вызывать функции вашего приложения. Модель не выполняет код напрямую: она возвращает structured 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": "Отправить email клиенту",
        "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:
    """Полный agentic loop с 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 loop с 2–3 инструментами: 2–3 дня
• Параллельные вызовы + обработка ошибок: 1–2 дня
• Production-ready с логированием и retry: 1 неделя