Інтеграція Anthropic Claude Agent SDK для побудови агентів

Інтеграція Anthropic Claude Agent SDK

Claude Agent SDK (claude-agent-sdk) — офіційний Python SDK Anthropic для побудови агентів на основі Claude. Надає високорівневі абстракції над Anthropic API: управління життєвим циклом агента, інтеграція з інструментами, підтримка Model Context Protocol (MCP), потокова передача подій та вбудований 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-коментарів"}]
)

Багатоходовий діалог з історією

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 через затвердження

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 — еска

лація для фінансового офіцера

Workflow:

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 тиждень