Интеграция Claude Agent SDK для продакшен-агентов

Отметим: когда вашему сервису нужен AI-агент, который не просто отвечает на вопросы, а выполняет действия: ищет в базах, создаёт тикеты, бронирует переговорные. Обычный чат-бот с LLM тут не справится — нужна интеграция инструментов (tool use). Claude Agent SDK от Anthropic даёт готовый паттерн для таких агентов. Мы используем его в продакшене, и расскажем, как это работает.

Почему именно Claude Agent SDK?

Прямая интеграция через anthropic Python-клиент даёт контроль над каждым шагом. В отличие от LangChain, здесь нет лишних абстракций: вы сами определяете инструменты в формате Anthropic, пишете диспетчер вызовов и управляете историей. Для агентов с 3–5 инструментами это быстрее и легче. Плюс — нативная поддержка параллельных tool calls и streaming.

Что входит в работу?

Как работает агентный цикл?

Агентный цикл — это последовательность: пользователь → модель → вызов инструмента → результат → модель. Claude Agent SDK автоматизирует этот цикл. Модель решает, какой инструмент вызвать и с какими параметрами. Диспетчер execute_tool выполняет вызов и возвращает результат. История сообщений хранит все взаимодействия, включая результаты инструментов.

Прямая интеграция через Anthropic API

import anthropic import json client = anthropic.Anthropic() # Определение инструментов в формате Anthropic tools = [ { "name": "search_database", "description": "Поиск информации в корпоративной базе данных", "input_schema": { "type": "object", "properties": { "query": {"type": "string", "description": "Поисковый запрос"}, "table": {"type": "string", "enum": ["products", "orders", "customers"]}, "limit": {"type": "integer", "default": 10}, }, "required": ["query"], }, }, { "name": "create_ticket", "description": "Создать тикет в системе поддержки", "input_schema": { "type": "object", "properties": { "title": {"type": "string"}, "description": {"type": "string"}, "priority": {"type": "string", "enum": ["low", "medium", "high", "critical"]}, "customer_id": {"type": "string"}, }, "required": ["title", "description", "customer_id"], }, }, ] def execute_tool(tool_name: str, tool_input: dict) -> str: """Диспетчер вызовов инструментов""" handlers = { "search_database": lambda i: db.search(**i), "create_ticket": lambda i: helpdesk.create(**i), } handler = handlers.get(tool_name) if not handler: return f"Unknown tool: {tool_name}" try: result = handler(tool_input) return json.dumps(result, ensure_ascii=False) except Exception as e: return f"Error: {e}" def run_agent(user_message: str, system_prompt: str = None) -> str: """Агентный цикл с tool use""" messages = [{"role": "user", "content": user_message}] for iteration in range(10): response = client.messages.create( model="claude-opus-4-5", max_tokens=4096, system=system_prompt or "Ты — полезный ассистент с доступом к корпоративным инструментам.", tools=tools, messages=messages, ) # Добавляем ответ модели в историю messages.append({"role": "assistant", "content": response.content}) # Проверяем причину остановки if response.stop_reason == "end_turn": # Модель завершила работу — возвращаем текстовый ответ text_blocks = [b.text for b in response.content if b.type == "text"] return "\n".join(text_blocks) if response.stop_reason == "tool_use": # Обрабатываем вызовы инструментов tool_results = [] for block in response.content: if block.type == "tool_use": result = execute_tool(block.name, block.input) tool_results.append({ "type": "tool_result", "tool_use_id": block.id, "content": result, }) # Добавляем результаты инструментов messages.append({"role": "user", "content": tool_results}) return "Достигнут лимит итераций" 

Как работают параллельные tool calls?

def run_agent_with_parallel_tools(user_message: str) -> str: """Claude может вызывать несколько инструментов за один ход""" messages = [{"role": "user", "content": user_message}] while True: response = client.messages.create( model="claude-opus-4-5", max_tokens=4096, tools=tools, messages=messages, ) messages.append({"role": "assistant", "content": response.content}) if response.stop_reason == "end_turn": return next((b.text for b in response.content if b.type == "text"), "") # Все tool_use блоки в одном ответе — выполняем параллельно tool_use_blocks = [b for b in response.content if b.type == "tool_use"] if not tool_use_blocks: break import asyncio async def execute_parallel(): tasks = [ asyncio.to_thread(execute_tool, block.name, block.input) for block in tool_use_blocks ] return await asyncio.gather(*tasks) results = asyncio.run(execute_parallel()) tool_results = [ { "type": "tool_result", "tool_use_id": block.id, "content": result, } for block, result in zip(tool_use_blocks, results) ] messages.append({"role": "user", "content": tool_results}) return "" 

Streaming агент для низкой задержки

def run_streaming_agent(user_message: str): """Агент со стримингом текста""" messages = [{"role": "user", "content": user_message}] while True: collected_content = [] tool_use_id = None tool_name = None tool_input_parts = [] with client.messages.stream( model="claude-opus-4-5", max_tokens=4096, tools=tools, messages=messages, ) as stream: for event in stream: if hasattr(event, "type"): if event.type == "content_block_start": if event.content_block.type == "tool_use": tool_use_id = event.content_block.id tool_name = event.content_block.name elif event.type == "content_block_delta": if hasattr(event.delta, "text"): print(event.delta.text, end="", flush=True) collected_content.append({"type": "text_delta", "text": event.delta.text}) elif hasattr(event.delta, "partial_json"): tool_input_parts.append(event.delta.partial_json) final_message = stream.get_final_message() messages.append({"role": "assistant", "content": final_message.content}) if final_message.stop_reason == "end_turn": break if tool_use_id: full_tool_input = json.loads("".join(tool_input_parts)) result = execute_tool(tool_name, full_tool_input) messages.append({ "role": "user", "content": [{"type": "tool_result", "tool_use_id": tool_use_id, "content": result}], }) 

Computer Use (бета) — управление компьютером

# Claude Computer Use — управление компьютером через скриншоты computer_use_tools = [ {"type": "computer_20241022", "name": "computer", "display_width_px": 1920, "display_height_px": 1080}, {"type": "bash_20241022", "name": "bash"}, {"type": "text_editor_20241022", "name": "str_replace_editor"}, ] response = client.messages.create( model="claude-opus-4-5", # Поддерживает computer use max_tokens=4096, tools=computer_use_tools, messages=[{"role": "user", "content": "Открой браузер, перейди на company.ru, найди раздел 'Контакты' и скопируй телефон."}], betas=["computer-use-2024-10-22"], ) 

Практический кейс: интеграция Claude в корпоративный портал

Из нашей практики: клиенту требовался AI-ассистент в корпоративный портал на Python/FastAPI без внешних фреймворков. Мы выбрали прямой Anthropic API, так как LangChain оказался избыточным для задачи с 5 инструментами. Инструменты включали: search_knowledge_base (векторный поиск по документам), get_employee_info (HR), create_it_ticket (ServiceDesk), get_meeting_rooms (бронирование), get_company_policies (нормативы). Результат: время внедрения 2 недели (против 4 с LangChain), кодовая база 450 строк, latency первого токена на 80 мс ниже. Заказчик сэкономил более $2.7k–3.9k на разработке — эти средства перенаправили на дополнительную функциональность.

Типичные ошибки при интеграции

• Неправильно спроектированные схемы инструментов (неоднозначные описания) — агент путается. В документации Anthropic рекомендуется использовать явные исключающие примеры в описании поля.
• Отсутствие обработки ошибок execute_tool — агент зависает в цикле.
• Игнорирование лимита итераций — утечка токенов.
• Неверный choice модели для Computer Use (нужна claude-opus-4-5).

Подробнее о настройке инструментов читайте в официальном репозитории Anthropic SDK.

Сроки и как заказать

Мониторинг и контроль расходов на токены

В production важно отслеживать стоимость каждого агентного вызова. Claude API возвращает usage с количеством токенов на вход и выход в каждом ответе. Мы встраиваем счётчик в run_agent:

def run_agent_tracked(user_message: str) -> dict: total_input_tokens = 0 total_output_tokens = 0 iterations = 0 messages = [{"role": "user", "content": user_message}] while iterations < 10: response = client.messages.create( model="claude-opus-4-5", max_tokens=4096, tools=tools, messages=messages, ) total_input_tokens += response.usage.input_tokens total_output_tokens += response.usage.output_tokens iterations += 1 # ... обработка tool use ... if response.stop_reason == "end_turn": break return { "result": "...", "tokens_in": total_input_tokens, "tokens_out": total_output_tokens, "iterations": iterations, } 

Средний агентный сеанс с 3 вызовами инструментов потребляет 2000–5000 input-токенов и 500–1500 output. Логируем эти данные в Prometheus и строим дашборды по стоимости на пользователя, на сценарий и на день. Это позволяет быстро выявить аномально длинные цепочки и оптимизировать промпты.

Дополнительная оптимизация: кэширование через prompt_caching (заголовок anthropic-beta: prompt-caching-2024-07-31) снижает стоимость повторяющихся системных промптов на 90%. Для инструментов с редкими описаниями (>1024 токенов) кэширование применяется автоматически и экономит до 40% токенов на многоходовых диалогах. Итоговая экономия на токенах при 10 000 агентных вызовов в день — существенная статья бюджета, которую стоит закладывать в архитектуру с первого дня.

Получите консультацию инженера — оценим ваш проект за 1 день. Опыт интеграции Claude Agent SDK подтверждён десятками успешных проектов. Свяжитесь с нами: мы поможем выбрать архитектуру и подготовить техническое задание.