Интеграция Anthropic Claude Agent SDK для построения агентов

Интеграция Anthropic Claude Agent SDK

Claude Agent SDK (claude-agent-sdk) — официальный Python SDK Anthropic для построения агентов на основе Claude. Предоставляет высокоуровневые абстракции над Anthropic API: управление жизненным циклом агента, интеграция с инструментами, поддержка MCP (Model Context Protocol), стриминг событий и встроенный human-in-the-loop.

Установка и базовый агент

# pip install anthropic claude-agent-sdk
import anthropic
from claude_agent_sdk import Agent, AgentConfig, tool

client = anthropic.Anthropic()

# Определение инструментов через декоратор
@tool
def search_database(query: str, table: str = "products") -> str:
    """Поиск по базе данных компании.

    Args:
        query: Поисковый запрос
        table: Таблица для поиска (products, orders, customers)
    """
    results = db.search(query=query, table=table, limit=10)
    return results.to_json()

@tool
def create_support_ticket(
    customer_id: str,
    subject: str,
    description: str,
    priority: str = "normal",
) -> str:
    """Создать тикет в системе поддержки.

    Args:
        customer_id: ID клиента
        subject: Тема обращения
        description: Подробное описание
        priority: Приоритет (low, normal, high, critical)
    """
    ticket = helpdesk.create_ticket(
        customer_id=customer_id,
        subject=subject,
        description=description,
        priority=priority,
    )
    return f"Тикет #{ticket['id']} создан. URL: {ticket['url']}"

# Конфигурация и создание агента
config = AgentConfig(
    model="claude-opus-4-5",
    system_prompt="""Ты — агент клиентской поддержки компании TechCorp.
Помогай клиентам решать проблемы, используя доступные инструменты.
Всегда проверяй данные через инструменты — не полагайся на память.""",
    max_turns=10,
)

agent = Agent(
    client=client,
    config=config,
    tools=[search_database, create_support_ticket],
)

# Запуск агента
result = agent.run(
    messages=[{"role": "user", "content": "У клиента ID 12345 проблема с заказом №99876"}]
)
print(result.final_message)

Стриминг событий агента

import asyncio

async def run_agent_with_streaming():
    async for event in agent.astream(
        messages=[{"role": "user", "content": "Проанализируй последние 10 заказов клиента ID 12345"}]
    ):
        match event.type:
            case "text_delta":
                print(event.text, end="", flush=True)
            case "tool_use_start":
                print(f"\n[Инструмент: {event.tool_name}]")
            case "tool_result":
                print(f"[Результат получен, {len(event.content)} символов]")
            case "agent_turn_complete":
                print(f"\n[Завершено за {event.turn_count} ходов]")

asyncio.run(run_agent_with_streaming())

Интеграция с MCP (Model Context Protocol)

from claude_agent_sdk import Agent, MCPServerConfig

# MCP-серверы расширяют агента инструментами без явного определения
agent_with_mcp = Agent(
    client=client,
    config=config,
    mcp_servers=[
        MCPServerConfig(
            name="filesystem",
            command="npx",
            args=["-y", "@modelcontextprotocol/server-filesystem", "/workspace"],
        ),
        MCPServerConfig(
            name="postgres",
            command="npx",
            args=["-y", "@modelcontextprotocol/server-postgres"],
            env={"POSTGRES_URL": "postgresql://user:pass@localhost/db"},
        ),
        MCPServerConfig(
            name="github",
            command="npx",
            args=["-y", "@modelcontextprotocol/server-github"],
            env={"GITHUB_PERSONAL_ACCESS_TOKEN": "ghp_..."},
        ),
    ],
)

# Агент теперь может работать с файлами, PostgreSQL и GitHub
result = agent_with_mcp.run(
    messages=[{"role": "user", "content": "Прочитай файл config.yaml и создай задачи в GitHub Issues на основе TODO-комментариев"}]
)

Multi-turn диалог с историей

from claude_agent_sdk import ConversationSession

# Сессия сохраняет историю диалога
session = ConversationSession(
    agent=agent,
    session_id="customer_session_12345",
)

# Первый ход
response1 = session.send("Что такое статус моего заказа #99876?")

# Второй ход — агент помнит контекст
response2 = session.send("А можно перенести доставку на завтра?")

# Третий ход
response3 = session.send("Подтверди, пожалуйста")

print(session.get_history())

Human-in-the-loop через Approval

from claude_agent_sdk import Agent, ToolApprovalPolicy

class CustomApprovalPolicy(ToolApprovalPolicy):
    """Запрашивает подтверждение для деструктивных операций"""

    REQUIRES_APPROVAL = {"delete_order", "process_refund", "ban_customer"}

    async def should_approve(self, tool_name: str, tool_input: dict) -> bool:
        if tool_name not in self.REQUIRES_APPROVAL:
            return True  # Автоматически разрешаем безопасные инструменты

        # Уведомляем оператора
        await notify_operator(
            message=f"Требуется подтверждение: {tool_name}\nПараметры: {tool_input}",
            callback_url="/api/approve/{approval_id}",
        )

        # Ждём решения (таймаут 5 минут)
        approval = await wait_for_approval(timeout=300)
        return approval.approved

agent_with_approval = Agent(
    client=client,
    config=config,
    tools=[search_database, process_refund, ban_customer],
    approval_policy=CustomApprovalPolicy(),
)

Параллельные агенты

import asyncio

async def run_parallel_analysis(customer_ids: list[str]):
    """Параллельный анализ нескольких клиентов"""

    async def analyze_customer(customer_id: str):
        return await agent.arun(
            messages=[{
                "role": "user",
                "content": f"Проведи анализ активности клиента {customer_id} за последний месяц"
            }]
        )

    results = await asyncio.gather(*[
        analyze_customer(cid) for cid in customer_ids
    ], return_exceptions=True)

    return {
        cid: result.final_message if not isinstance(result, Exception) else str(result)
        for cid, result in zip(customer_ids, results)
    }

Практический кейс: агент финансового мониторинга

Задача: автоматический мониторинг транзакций для выявления подозрительной активности. Финансовый офицер тратил 3 часа ежедневно на ручной просмотр флагов от rule-based системы.

Инструменты агента:

• get_flagged_transactions — список флагов за период
• get_transaction_history — история транзакций клиента
• get_customer_profile — KYC-профиль
• check_external_sanctions — проверка по санкционным спискам
• create_sar_draft — создание черновика Suspicious Activity Report
• escalate_to_officer — эскалация для финансового офицера

Поток работы:

1. Агент получает список из 50–200 флагов ежедневно
2. Для каждого флага: анализирует контекст, историю, профиль
3. Делает вывод: ложное срабатывание / подозрительно / критично
4. Для критичных: создаёт черновик SAR и эскалирует
5. Ложные срабатывания закрывает автоматически

Результаты:

• Ложные срабатывания обработаны автоматически: 78%
• Время финансового офицера: 3 часа → 45 минут (только реальные случаи)
• Качество черновиков SAR (оценка compliance-офицера): 4.3/5.0
• Время реакции на критичные случаи: 4–8 часов → 15 минут

Сроки

• Базовый агент с инструментами: 3–5 дней
• MCP-интеграции: 1–3 дня на каждый сервер
• Human-in-the-loop с approval flow: 1 неделя
• Production-деплой с мониторингом: 1 неделя