Интеграция Claude Agent SDK
Claude Agent SDK — более ранняя версия официального SDK Anthropic для построения агентов на базе моделей Claude (дополнение к разделу id=177). Текущий рекомендованный SDK использует anthropic Python-клиент с нативной поддержкой tool_use, а для сложных агентных систем — интеграцию с LangGraph или OpenAI Agents SDK.
Прямая интеграция через 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) без внешних фреймворков.
Выбор прямого API: команда оценила LangChain как избыточный для задачи с 5 инструментами. Прямой Anthropic API + простой агентный цикл — достаточно.
Инструменты:
- search_knowledge_base (векторный поиск по внутренним документам)
- get_employee_info (HR справочник)
- create_it_ticket (ServiceDesk интеграция)
- get_meeting_rooms (бронирование переговорных)
- get_company_policies (нормативные документы)
Результаты:
- Время внедрения: 2 недели (vs 4 недели оценка с LangChain)
- Кодовая база: 450 строк vs 900 у LangChain-версии в PoC
- Latency первого токена: на 80ms ниже (без LangChain overhead)
Сроки
- Базовый агент с 3–5 инструментами: 3–5 дней
- Streaming + production error handling: 3–5 дней
- Computer Use интеграция: 1–2 недели
- Интеграция в web-приложение: 1 неделя