pip install anthropic 

tools = [ { "name": "search_database", "description": "Пошук у корпоративній БД", "input_schema": { "type": "object", "properties": { "query": {"type": "string"}, "table": {"type": "string", "enum": ["products", "orders", "customers"]}, }, "required": ["query"], }, }, ] 

def execute_tool(name: str, inp: dict) -> str: handlers = { "search_database": lambda i: db.search(**i), } return handlers.get(name, lambda _: "Unknown")(inp) 

response = client.messages.create( model="claude-opus-4-5", tools=tools, messages=[{"role": "user", "content": "Знайди продукт X"}], ) print(response.content) 

def run_agent_tracked(msg): total_tokens = 0 # ... логіка з підрахунком токенів return {"result": "...", "tokens": total_tokens} 

import anthropic import json client = 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 "Досягнуто ліміт ітерацій" 

import asyncio def run_agent_parallel(user_message: str) -> str: 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_blocks = [b for b in response.content if b.type == "tool_use"] if not tool_use_blocks: break async def execute_parallel(): tasks = [asyncio.to_thread(execute_tool, b.name, b.input) for b in tool_use_blocks] return await asyncio.gather(*tasks) results = asyncio.run(execute_parallel()) tool_results = [ {"type": "tool_result", "tool_use_id": b.id, "content": r} for b, r in zip(tool_use_blocks, results) ] messages.append({"role": "user", "content": tool_results}) return "" 

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" and 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_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", max_tokens=4096, tools=computer_use_tools, messages=[{"role": "user", "content": "Відкрий браузер, перейди на company.ua, знайди розділ 'Контакти' і скопіюй телефон."}], betas=["computer-use-2024-10-22"], ) 

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 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 агентних викликах на день — суттєва стаття бюджету, яку варто закладати в архітектуру з першого дня.

Терміни та контакти

Докладніше про налаштування інструментів читайте в офіційному репозиторії Anthropic SDK. Гарантуємо стабільну роботу агента та супровід після запуску. Отримайте консультацію інженера — оцінимо ваш проєкт за 1 день. Досвід інтеграції Claude Agent SDK підтверджений десятками успішних проєктів.