Claude Code SDK

{
  "mcpServers": {
    "slack": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-slack"],
      "env": {"SLACK_TOKEN": "your-slack-token"}
    },
    "jira": {
      "command": "npx", 
      "args": ["-y", "@modelcontextprotocol/server-jira"],
      "env": {"JIRA_TOKEN": "your-jira-token"}
    },
    "database": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-postgres"],
      "env": {"DB_CONNECTION_STRING": "your-db-url"}
    }
  }
}

// вызов инструмента разрешен
{
  "behavior": "allow",
  "updatedInput": {...}, // обновленные входные данные, или просто верните обратно оригинальные входные данные
}

// вызов инструмента запрещен
{
  "behavior": "deny",
  "message": "..." // читаемая человеком строка, объясняющая, почему разрешение было отклонено
}

{
  "type": "result",
  "subtype": "success",
  "total_cost_usd": 0.003,
  "is_error": false,
  "duration_ms": 1234,
  "duration_api_ms": 800,
  "num_turns": 6,
  "result": "Текст ответа здесь...",
  "session_id": "abc123"
}

$ claude -p "Создайте приложение" --output-format stream-json

type SDKMessage =
  // Сообщение помощника
  | {
      type: "assistant";
      message: Message; // из Anthropic SDK
      session_id: string;
    }

  // Сообщение пользователя
  | {
      type: "user";
      message: MessageParam; // из Anthropic SDK
      session_id: string;
    }

  // Выдается как последнее сообщение
  | {
      type: "result";
      subtype: "success";
      duration_ms: float;
      duration_api_ms: float;
      is_error: boolean;
      num_turns: int;
      result: string;
      session_id: string;
      total_cost_usd: float;
    }

  // Выдается как последнее сообщение, когда мы достигли максимального количества оборотов
  | {
      type: "result";
      subtype: "error_max_turns" | "error_during_execution";
      duration_ms: float;
      duration_api_ms: float;
      is_error: boolean;
      num_turns: int;
      session_id: string;
      total_cost_usd: float;
    }

  // Выдается как первое сообщение в начале разговора
  | {
      type: "system";
      subtype: "init";
      apiKeySource: string;
      cwd: string;
      session_id: string;
      tools: string[];
      mcp_servers: {
        name: string;
        status: string;
      }[];
      model: string;
      permissionMode: "default" | "acceptEdits" | "bypassPermissions" | "plan";
    };

$ echo '{"type":"user","message":{"role":"user","content":[{"type":"text","text":"Объясните этот код"}]}}' | claude -p --output-format=stream-json --input-format=stream-json --verbose

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)