Python

pip install claude-code-sdk
npm install -g @anthropic-ai/claude-code  # Обязательная зависимость

pip install ipython

# legal-agent.py
import asyncio
from claude_code_sdk import ClaudeSDKClient, ClaudeCodeOptions

async def main():
    async with ClaudeSDKClient(
        options=ClaudeCodeOptions(
            system_prompt="Вы юридический помощник. Выявляйте риски и предлагайте улучшения.",
            max_turns=2
        )
    ) as client:
        # Отправить запрос
        await client.query(
            "Проверьте эту статью договора на потенциальные проблемы: 'Сторона соглашается на неограниченную ответственность...'"
        )

        # Потоковая передача ответа
        async for message in client.receive_response():
            if hasattr(message, 'content'):
                # Печатать потоковое содержимое по мере поступления
                for block in message.content:
                    if hasattr(block, 'text'):
                        print(block.text, end='', flush=True)

if __name__ == "__main__":
    asyncio.run(main())

python legal-agent.py

await main()

import asyncio
from claude_code_sdk import ClaudeSDKClient, ClaudeCodeOptions

async def main():
    async with ClaudeSDKClient(
        options=ClaudeCodeOptions(
            system_prompt="Вы инженер по производительности",
            allowed_tools=["Bash", "Read", "WebSearch"],
            max_turns=5
        )
    ) as client:
        await client.query("Анализировать производительность системы")

        # Потоковые ответы
        async for message in client.receive_response():
            if hasattr(message, 'content'):
                for block in message.content:
                    if hasattr(block, 'text'):
                        print(block.text, end='', flush=True)

# Запуск как скрипт
asyncio.run(main())

# Или в IPython/Jupyter: await main()

from claude_code_sdk import query, ClaudeCodeOptions

async for message in query(
    prompt="Анализировать производительность системы",
    options=ClaudeCodeOptions(system_prompt="Вы инженер по производительности")
):
    if type(message).__name__ == "ResultMessage":
        print(message.result)

from claude_code_sdk import ClaudeCodeOptions

options = ClaudeCodeOptions(
    # Основная конфигурация
    system_prompt="Вы полезный помощник",
    append_system_prompt="Дополнительные системные инструкции",
    max_turns=5,
    model="claude-3-5-sonnet-20241022",
    max_thinking_tokens=8000,
    
    # Управление инструментами
    allowed_tools=["Bash", "Read", "Write"],
    disallowed_tools=["WebSearch"],
    
    # Управление сессиями
    continue_conversation=False,
    resume="session-uuid",
    
    # Окружение
    cwd="/path/to/working/directory",
    add_dirs=["/additional/context/dir"],
    settings="/path/to/settings.json",
    
    # Разрешения
    permission_mode="acceptEdits",  # "default", "acceptEdits", "plan", "bypassPermissions"
    permission_prompt_tool_name="mcp__approval_tool",
    
    # Интеграция MCP
    mcp_servers={
        "my_server": {
            "command": "npx",
            "args": ["-y", "@modelcontextprotocol/server-example"],
            "env": {"API_KEY": "your-key"}
        }
    },
    
    # Расширенные
    extra_args={"--verbose": None, "--custom-flag": "value"}
)

import asyncio
from claude_code_sdk import ClaudeSDKClient, ClaudeCodeOptions

async def advanced_agent():
    """Пример, демонстрирующий расширенные параметры конфигурации"""
    
    async with ClaudeSDKClient(
        options=ClaudeCodeOptions(
            # Пользовательский рабочий каталог и дополнительный контекст
            cwd="/project/root",
            add_dirs=["/shared/libs", "/common/utils"],
            
            # Конфигурация модели и мышления
            model="claude-3-5-sonnet-20241022",
            max_thinking_tokens=12000,
            
            # Расширенное управление инструментами
            allowed_tools=["Read", "Write", "Bash", "Grep"],
            disallowed_tools=["WebSearch", "Bash(rm*)"],
            
            # Пользовательские настройки и аргументы CLI
            settings='{"editor": "vim", "theme": "dark"}',
            extra_args={
                "--verbose": None,
                "--timeout": "300"
            }
        )
    ) as client:
        await client.query("Анализировать структуру кодовой базы")
        
        async for message in client.receive_response():
            if hasattr(message, 'content'):
                for block in message.content:
                    if hasattr(block, 'text'):
                        print(block.text, end='', flush=True)

asyncio.run(advanced_agent())

from claude_code_sdk import ClaudeSDKClient, ClaudeCodeOptions

async with ClaudeSDKClient() as client:
    # Текстовое сообщение
    await client.query("Анализировать этот код на предмет проблем безопасности")

    # Сообщение со ссылкой на изображение (изображение будет прочитано инструментом Read Claude)
    await client.query("Объясните, что показано на screenshot.png")

    # Несколько сообщений подряд
    messages = [
        "Сначала проанализируйте диаграмму архитектуры в diagram.png",
        "Теперь предложите улучшения на основе диаграммы",
        "Наконец, сгенерируйте код реализации"
    ]

    for msg in messages:
        await client.query(msg)
        async for response in client.receive_response():
            # Обработать каждый ответ
            pass

# SDK обрабатывает файлы изображений через встроенный инструмент Read Claude
# Поддерживаемые форматы: PNG, JPG, PDF и другие распространенные форматы

import asyncio
from claude_code_sdk import ClaudeSDKClient, ClaudeCodeOptions, query

# Метод 1: Использование ClaudeSDKClient для постоянных разговоров
async def multi_turn_conversation():
    async with ClaudeSDKClient() as client:
        # Первый запрос
        await client.query("Давайте рефакторим модуль платежей")
        async for msg in client.receive_response():
            # Обработать первый ответ
            pass

        # Продолжить в той же сессии
        await client.query("Теперь добавьте комплексную обработку ошибок")
        async for msg in client.receive_response():
            # Обработать продолжение
            pass

        # Контекст разговора поддерживается на протяжении всего времени

# Метод 2: Использование функции query с управлением сессиями
async def resume_session():
    # Продолжить самый последний разговор
    async for message in query(
        prompt="Теперь рефакторим это для лучшей производительности",
        options=ClaudeCodeOptions(continue_conversation=True)
    ):
        if type(message).__name__ == "ResultMessage":
            print(message.result)

    # Возобновить конкретную сессию
    async for message in query(
        prompt="Обновить тесты",
        options=ClaudeCodeOptions(
            resume="550e8400-e29b-41d4-a716-446655440000",
            max_turns=3
        )
    ):
        if type(message).__name__ == "ResultMessage":
            print(message.result)

# Запустить примеры
asyncio.run(multi_turn_conversation())

import asyncio
from claude_code_sdk import ClaudeSDKClient, ClaudeCodeOptions

async def specialized_agents():
    # Агент реагирования на инциденты SRE с потоковой передачей
    async with ClaudeSDKClient(
        options=ClaudeCodeOptions(
            system_prompt="Вы эксперт SRE. Диагностируйте проблемы систематически и предоставляйте действенные решения.",
            max_turns=3
        )
    ) as sre_agent:
        await sre_agent.query("API не работает, исследуйте")

        # Потоковая передача процесса диагностики
        async for message in sre_agent.receive_response():
            if hasattr(message, 'content'):
                for block in message.content:
                    if hasattr(block, 'text'):
                        print(block.text, end='', flush=True)

    # Агент юридического обзора с пользовательским промптом
    async with ClaudeSDKClient(
        options=ClaudeCodeOptions(
            append_system_prompt="Всегда включайте комплексную обработку ошибок и модульные тесты.",
            max_turns=2
        )
    ) as dev_agent:
        await dev_agent.query("Рефакторить эту функцию")

        # Собрать полный ответ
        full_response = []
        async for message in dev_agent.receive_response():
            if type(message).__name__ == "ResultMessage":
                print(message.result)

asyncio.run(specialized_agents())

import asyncio
from claude_code_sdk import ClaudeSDKClient, ClaudeCodeOptions

async def mcp_enabled_agent():
    # Юридический агент с доступом к документам и потоковой передачей
    # Примечание: Настройте ваши MCP серверы по мере необходимости
    mcp_servers = {
        # Пример конфигурации - раскомментируйте и настройте по мере необходимости:
        # "docusign": {
        #     "command": "npx",
        #     "args": ["-y", "@modelcontextprotocol/server-docusign"],
        #     "env": {"API_KEY": "your-key"}
        # }
    }

    async with ClaudeSDKClient(
        options=ClaudeCodeOptions(
            mcp_servers=mcp_servers,
            allowed_tools=["mcp__docusign", "mcp__compliance_db"],
            system_prompt="Вы корпоративный юрист, специализирующийся на обзоре договоров.",
            max_turns=4
        )
    ) as client:
        await client.query("Проверьте этот договор на риски соответствия")

        # Мониторинг использования инструментов и ответов
        async for message in client.receive_response():
            if hasattr(message, 'content'):
                for block in message.content:
                    if hasattr(block, 'type'):
                        if block.type == 'tool_use':
                            print(f"\n[Использование инструмента: {block.name}]\n")
                        elif hasattr(block, 'text'):
                            print(block.text, end='', flush=True)
                    elif hasattr(block, 'text'):
                        print(block.text, end='', flush=True)

            if type(message).__name__ == "ResultMessage":
                print(f"\n\nОбзор завершен. Общая стоимость: ${message.total_cost_usd:.4f}")

asyncio.run(mcp_enabled_agent())

import asyncio
from claude_code_sdk import ClaudeSDKClient, ClaudeCodeOptions

async def use_permission_prompt():
    """Пример использования пользовательского инструмента запроса разрешений"""

    # Конфигурация MCP сервера
    mcp_servers = {
        # Пример конфигурации - раскомментируйте и настройте по мере необходимости:
        # "security": {
        #     "command": "npx",
        #     "args": ["-y", "@modelcontextprotocol/server-security"],
        #     "env": {"API_KEY": "your-key"}
        # }
    }

    async with ClaudeSDKClient(
        options=ClaudeCodeOptions(
            permission_prompt_tool_name="mcp__security__approval_prompt",  # Изменено с permission_prompt_tool
            mcp_servers=mcp_servers,
            allowed_tools=["Read", "Grep"],
            disallowed_tools=["Bash(rm*)", "Write"],
            system_prompt="Вы аудитор безопасности"
        )
    ) as client:
        await client.query("Анализировать и исправить проблемы безопасности")

        # Мониторинг использования инструментов и разрешений
        async for message in client.receive_response():
            if hasattr(message, 'content'):
                for block in message.content:
                    if hasattr(block, 'type'):  # Добавлена проверка атрибута 'type'
                        if block.type == 'tool_use':
                            print(f"[Инструмент: {block.name}] ", end='')
                    if hasattr(block, 'text'):
                        print(block.text, end='', flush=True)

            # Проверить отказы в разрешениях в сообщениях об ошибках
            if type(message).__name__ == "ErrorMessage":
                if hasattr(message, 'error') and "Permission denied" in str(message.error):
                    print(f"\n⚠️ Разрешение отклонено: {message.error}")

# Пример реализации MCP сервера (Python)
# Это будет в коде вашего MCP сервера
async def approval_prompt(tool_name: str, input: dict, tool_use_id: str = None):
    """Пользовательский обработчик запроса разрешений"""
    # Ваша пользовательская логика здесь
    if "allow" in str(input):
        return json.dumps({
            "behavior": "allow",
            "updatedInput": input
        })
    else:
        return json.dumps({
            "behavior": "deny",
            "message": f"Разрешение отклонено для {tool_name}"
        })

asyncio.run(use_permission_prompt())

# Текстовый вывод по умолчанию с потоковой передачей
async with ClaudeSDKClient() as client:
    await client.query("Объясните файл src/components/Header.tsx")

    # Потоковая передача текста по мере поступления
    async for message in client.receive_response():
        if hasattr(message, 'content'):
            for block in message.content:
                if hasattr(block, 'text'):
                    print(block.text, end='', flush=True)
                    # Вывод потоков в реальном времени: Это React компонент, показывающий...

# Собрать все сообщения с метаданными
async with ClaudeSDKClient() as client:
    await client.query("Как работает слой данных?")

    messages = []
    result_data = None

    async for message in client.receive_messages():
        messages.append(message)

        # Захватить сообщение результата с метаданными
        if type(message).__name__ == "ResultMessage":
            result_data = {
                "result": message.result,
                "cost": message.total_cost_usd,
                "duration": message.duration_ms,
                "num_turns": message.num_turns,
                "session_id": message.session_id
            }
            break

    print(result_data)

import asyncio
from claude_code_sdk import ClaudeSDKClient

async def process_inputs():
    async with ClaudeSDKClient() as client:
        # Текстовый ввод
        await client.query("Объясните этот код")
        async for message in client.receive_response():
            # Обработать потоковый ответ
            pass

        # Ввод изображения (Claude автоматически использует инструмент Read)
        await client.query("Что на этой диаграмме? screenshot.png")
        async for message in client.receive_response():
            # Обработать анализ изображения
            pass

        # Множественные входы со смешанным содержимым
        inputs = [
            "Анализировать архитектуру в diagram.png",
            "Сравнить с лучшими практиками",
            "Сгенерировать улучшенную версию"
        ]

        for prompt in inputs:
            await client.query(prompt)
            async for message in client.receive_response():
                # Обработать каждый ответ
                pass

asyncio.run(process_inputs())

import asyncio
from claude_code_sdk import ClaudeSDKClient, ClaudeCodeOptions

async def investigate_incident(incident_description: str, severity: str = "medium"):
    """Автоматизированный агент реагирования на инциденты с потоковой передачей в реальном времени"""

    # Конфигурация MCP сервера для инструментов мониторинга
    mcp_servers = {
        # Пример конфигурации - раскомментируйте и настройте по мере необходимости:
        # "datadog": {
        #     "command": "npx",
        #     "args": ["-y", "@modelcontextprotocol/server-datadog"],
        #     "env": {"API_KEY": "your-datadog-key", "APP_KEY": "your-app-key"}
        # }
    }

    async with ClaudeSDKClient(
        options=ClaudeCodeOptions(
            system_prompt="Вы эксперт SRE. Диагностируйте проблемы систематически и предоставляйте действенные решения.",
            max_turns=6,
            allowed_tools=["Bash", "Read", "WebSearch", "mcp__datadog"],
            mcp_servers=mcp_servers
        )
    ) as client:
        # Отправить детали инцидента
        prompt = f"Инцидент: {incident_description} (Серьезность: {severity})"
        print(f"🚨 Расследование: {prompt}\n")
        await client.query(prompt)

        # Потоковая передача процесса расследования
        investigation_log = []
        async for message in client.receive_response():
            if hasattr(message, 'content'):
                for block in message.content:
                    if hasattr(block, 'type'):
                        if block.type == 'tool_use':
                            print(f"[{block.name}] ", end='')
                    if hasattr(block, 'text'):
                        text = block.text
                        print(text, end='', flush=True)
                        investigation_log.append(text)

            # Захватить финальный результат
            if type(message).__name__ == "ResultMessage":
                return {
                    'analysis': ''.join(investigation_log),
                    'cost': message.total_cost_usd,
                    'duration_ms': message.duration_ms
                }

# Использование
result = await investigate_incident("API платежей возвращает ошибки 500", "high")
print(f"\n\nРасследование завершено. Стоимость: ${result['cost']:.4f}")

import subprocess
import asyncio
import json
from claude_code_sdk import ClaudeSDKClient, ClaudeCodeOptions

async def audit_pr(pr_number: int):
    """Агент аудита безопасности для pull request с потоковой обратной связью"""
    # Получить diff PR
    pr_diff = subprocess.check_output(
        ["gh", "pr", "diff", str(pr_number)],
        text=True
    )

    async with ClaudeSDKClient(
        options=ClaudeCodeOptions(
            system_prompt="Вы инженер по безопасности. Проверьте этот PR на уязвимости, небезопасные паттерны и проблемы соответствия.",
            max_turns=3,
            allowed_tools=["Read", "Grep", "WebSearch"]
        )
    ) as client:
        print(f"🔍 Аудит PR #{pr_number}\n")
        await client.query(pr_diff)

        findings = []
        async for message in client.receive_response():
            if hasattr(message, 'content'):
                for block in message.content:
                    if hasattr(block, 'text'):
                        # Потоковая передача находок по мере их обнаружения
                        print(block.text, end='', flush=True)
                        findings.append(block.text)

            if type(message).__name__ == "ResultMessage":
                return {
                    'pr_number': pr_number,
                    'findings': ''.join(findings),
                    'metadata': {
                        'cost': message.total_cost_usd,
                        'duration': message.duration_ms,
                        'severity': 'high' if 'vulnerability' in ''.join(findings).lower() else 'medium'
                    }
                }

# Использование
report = await audit_pr(123)
print(f"\n\nАудит завершен. Серьезность: {report['metadata']['severity']}")
print(json.dumps(report, indent=2))

import asyncio
from claude_code_sdk import ClaudeSDKClient, ClaudeCodeOptions

async def legal_review():
    """Обзор юридических документов с постоянной сессией и потоковой передачей"""

    async with ClaudeSDKClient(
        options=ClaudeCodeOptions(
            system_prompt="Вы корпоративный юрист. Предоставляйте детальный юридический анализ.",
            max_turns=2
        )
    ) as client:
        # Многоэтапный обзор в той же сессии
        steps = [
            "Проверить contract.pdf на статьи об ответственности",
            "Проверить соответствие требованиям GDPR",
            "Сгенерировать исполнительное резюме рисков"
        ]

        review_results = []

        for step in steps:
            print(f"\n📋 {step}\n")
            await client.query(step)

            step_result = []
            async for message in client.receive_response():
                if hasattr(message, 'content'):
                    for block in message.content:
                        if hasattr(block, 'text'):
                            text = block.text
                            print(text, end='', flush=True)
                            step_result.append(text)

                if type(message).__name__ == "ResultMessage":
                    review_results.append({
                        'step': step,
                        'analysis': ''.join(step_result),
                        'cost': message.total_cost_usd
                    })

        # Резюме
        total_cost = sum(r['cost'] for r in review_results)
        print(f"\n\n✅ Юридический обзор завершен. Общая стоимость: ${total_cost:.4f}")
        return review_results

# Использование
results = await legal_review()

import asyncio
from claude_code_sdk import ClaudeSDKClient, ClaudeCodeOptions

# Всегда используйте менеджеры контекста
async with ClaudeSDKClient() as client:
    await client.query("Анализировать этот код")
    async for msg in client.receive_response():
        # Обработать потоковые сообщения
        pass

# Запустить несколько агентов одновременно
async with ClaudeSDKClient() as reviewer, ClaudeSDKClient() as tester:
    await asyncio.gather(
        reviewer.query("Проверить main.py"),
        tester.query("Написать тесты для main.py")
    )

# Обработка ошибок
from claude_code_sdk import CLINotFoundError, ProcessError

try:
    async with ClaudeSDKClient() as client:
        # Ваш код здесь
        pass
except CLINotFoundError:
    print("Установите CLI: npm install -g @anthropic-ai/claude-code")
except ProcessError as e:
    print(f"Ошибка процесса: {e}")

# Собрать полный ответ с метаданными
async def get_response(client, prompt):
    await client.query(prompt)
    text = []
    async for msg in client.receive_response():
        if hasattr(msg, 'content'):
            for block in msg.content:
                if hasattr(block, 'text'):
                    text.append(block.text)
        if type(msg).__name__ == "ResultMessage":
            return {'text': ''.join(text), 'cost': msg.total_cost_usd}

# В Jupyter используйте await прямо в ячейках
client = ClaudeSDKClient()
await client.connect()
await client.query("Анализировать data.csv")
async for msg in client.receive_response():
    print(msg)
await client.disconnect()

# Создайте переиспользуемые вспомогательные функции
async def stream_print(client, prompt):
    await client.query(prompt)
    async for msg in client.receive_response():
        if hasattr(msg, 'content'):
            for block in msg.content:
                if hasattr(block, 'text'):
                    print(block.text, end='', flush=True)