Создание AI-агента на Python и Claude API: туториал

# Python 3.10+ обязателен
pip install anthropic claude-agent-sdk

# Создайте файл .env в корне проекта
ANTHROPIC_API_KEY=sk-ant-...

import asyncio
import os
from dotenv import load_dotenv
from claude_agent_sdk import ClaudeSDKClient, tool

load_dotenv()

# Определяем инструменты — простые async-функции

@tool("calculate", "Вычисляет математическое выражение", {"expression": str})
async def calculate(args: dict) -> dict:
    """Безопасное вычисление арифметики."""
    expression = args["expression"]
    # Только цифры и базовые операторы — никакого eval с произвольным кодом
    allowed = set("0123456789+-*/()., ")
    if not all(c in allowed for c in expression):
        return {"content": [{"type": "text", "text": "Недопустимые символы в выражении"}]}
    try:
        result = eval(expression)  # noqa: S307 — проверено выше
        return {"content": [{"type": "text", "text": str(result)}]}
    except Exception as e:
        return {"content": [{"type": "text", "text": f"Ошибка: {e}"}]}


@tool("get_current_date", "Возвращает текущую дату", {})
async def get_current_date(args: dict) -> dict:
    from datetime import datetime
    return {"content": [{"type": "text", "text": datetime.now().strftime("%Y-%m-%d")}]}


# Создаём клиент с системным промптом
client = ClaudeSDKClient(
    model="claude-sonnet-4-6",
    system_prompt="""Ты — умный ассистент с доступом к калькулятору и дате.
    Когда нужно что-то посчитать — используй инструмент calculate.
    Всегда отвечай на русском языке.""",
    tools=[calculate, get_current_date],
)


async def main():
    print("Агент запущен. Введите 'выход' для завершения.\n")
    while True:
        user_input = input("Вы: ").strip()
        if user_input.lower() in ("выход", "exit", "quit"):
            break

        print("Агент: ", end="", flush=True)
        async for message in client.query(user_input):
            if message.type == "text":
                print(message.text, end="", flush=True)
        print()  # Перевод строки после ответа


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

import aiohttp
import aiofiles
from pathlib import Path

@tool(
    "fetch_url",
    "Загружает содержимое веб-страницы по URL",
    {"url": str, "max_chars": int}
)
async def fetch_url(args: dict) -> dict:
    url = args["url"]
    max_chars = args.get("max_chars", 2000)

    # Базовая защита от SSRF — блокируем внутренние адреса
    if any(blocked in url for blocked in ["localhost", "127.0.0.1", "0.0.0.0", "169.254"]):
        return {"content": [{"type": "text", "text": "Доступ к внутренним адресам запрещён"}]}

    try:
        async with aiohttp.ClientSession() as session:
            async with session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as resp:
                text = await resp.text()
                return {"content": [{"type": "text", "text": text[:max_chars]}]}
    except Exception as e:
        return {"content": [{"type": "text", "text": f"Ошибка загрузки: {e}"}]}


@tool(
    "write_file",
    "Сохраняет текст в файл (только в папке ./output/)",
    {"filename": str, "content": str}
)
async def write_file(args: dict) -> dict:
    # Ограничиваем запись только в безопасную директорию
    output_dir = Path("./output")
    output_dir.mkdir(exist_ok=True)

    # Нормализуем путь и проверяем, что он внутри output/
    target = (output_dir / args["filename"]).resolve()
    if not str(target).startswith(str(output_dir.resolve())):
        return {"content": [{"type": "text", "text": "Запись за пределами output/ запрещена"}]}

    async with aiofiles.open(target, "w", encoding="utf-8") as f:
        await f.write(args["content"])

    return {"content": [{"type": "text", "text": f"Файл сохранён: {target}"}]}

Вы: Зайди на httpbin.org/json, извлеки данные и сохрани в файл result.json

Агент: [вызывает fetch_url → читает JSON → вызывает write_file]
Готово! Данные с httpbin.org/json сохранены в ./output/result.json

# Оркестратор делегирует задачи специализированным агентам
orchestrator = ClaudeSDKClient(
    model="claude-opus-4-6",  # Сильная модель для планирования
    system_prompt="Ты менеджер. Декомпозируй задачу и вызывай специализированных агентов.",
    tools=[call_search_agent, call_writer_agent, call_analyst_agent],
)

# Исполнители используют более дешёвую модель
search_agent = ClaudeSDKClient(
    model="claude-haiku-4-5",  # Быстро и дёшево для поиска
    system_prompt="Ты специалист по поиску информации.",
    tools=[fetch_url, search_web],
)

import json
from pathlib import Path

def save_session(session_id: str, messages: list):
    path = Path(f"./sessions/{session_id}.json")
    path.parent.mkdir(exist_ok=True)
    path.write_text(json.dumps(messages, ensure_ascii=False, indent=2))

def load_session(session_id: str) -> list:
    path = Path(f"./sessions/{session_id}.json")
    if path.exists():
        return json.loads(path.read_text())
    return []

# При инициализации клиента передайте историю
client = ClaudeSDKClient(
    model="claude-sonnet-4-6",
    system_prompt="...",
    initial_messages=load_session("user_123"),
)