# Как создать AI агента с нуля: пошаговый туториал > Пошаговое руководство: от минимального агента на 20 строк до мультиагентной системы. Claude Agent SDK, MCP, CrewAI, LangGraph, Google ADK. Полный рабочий код на Python. > Author: Roman Belov · Published: 2026-06-29 · Source: https://futurecraft.pro/ru/blog/how-to-create-ai-agent/ AI агент — не абстракция из исследовательских статей. Это программа, которую можно написать за вечер и запустить в production за неделю. Ядро любого агента — цикл: LLM получает задачу, вызывает инструменты, анализирует результаты, решает, что делать дальше. Memory, guardrails, мультиагентность — надстройки над этим циклом. В этом туториале построим реального агента для анализа GitHub-репозиториев. Начнём с 20 строк кода и дойдём до мультиагентной системы с MCP-интеграцией, error handling и deployment. Каждый шаг — рабочий код на Python с Claude Agent SDK. ## Что построим Агент для анализа GitHub-репозиториев. На входе — URL репозитория. На выходе — структурированный отчёт: - Архитектура проекта (структура файлов, основные компоненты) - Стек технологий (языки, фреймворки, зависимости) - Качество кода (наличие тестов, CI/CD, линтеров) - Активность (частота коммитов, открытые issues, contributors) - Рекомендации (что улучшить, потенциальные проблемы) Почему именно этот use case: GitHub API — доступный и документированный. Задача требует нескольких инструментов (API-запросы, чтение файлов, анализ). Результат полезен на практике — для code review, due diligence, выбора open-source зависимостей. ## Архитектура агента Любой AI агент состоит из трёх компонентов, соединённых циклом: ``` ┌───────────────────────────────────────────┐ │ USER INPUT │ │ "Проанализируй github.com/org/repo" │ └─────────────────┬─────────────────────────┘ ▼ ┌───────────────────────────────────────────┐ │ AGENT LOOP │ │ │ │ ┌────────┐ ┌────────┐ ┌────────────┐ │ │ │ Think │→ │ Act │→ │ Observe │ │ │ └────────┘ └────────┘ └────────────┘ │ │ ↑ │ │ │ └────────────────────────┘ │ │ │ │ LLM решает: │ │ - какой инструмент вызвать │ │ - с какими параметрами │ │ - достаточно ли данных для ответа │ └─────────────────┬─────────────────────────┘ │ ┌─────────┼─────────┐ ▼ ▼ ▼ ┌─────────┐ ┌──────┐ ┌──────────┐ │GitHub │ │File │ │ Shell │ │ API │ │System│ │ Commands │ └─────────┘ └──────┘ └──────────┘ │ ▼ ┌───────────────────────────────────────────┐ │ MEMORY │ │ Short-term: контекст текущей сессии │ │ Long-term: результаты прошлых анализов │ └───────────────────────────────────────────┘ │ ▼ ┌───────────────────────────────────────────┐ │ STRUCTURED REPORT │ │ Архитектура, стек, качество, активность │ └───────────────────────────────────────────┘ ``` **LLM** — мозг агента. Получает задачу, декомпозирует на шаги, выбирает инструменты. Claude Sonnet 4.6 или Opus 4.8 — зависит от сложности задачи. **Tools** — руки агента. Функции для взаимодействия с внешним миром: HTTP-запросы к GitHub API, чтение файлов, выполнение команд. **Memory** — память агента. Контекст текущего разговора (short-term) и результаты прошлых анализов (long-term). **Agent Loop** — управляющий цикл. LLM думает → вызывает инструмент → получает результат → думает снова. Цикл повторяется, пока агент не решит, что задача выполнена. Отличие от обычного вызова API: агент сам решает, сколько итераций нужно и какие инструменты вызвать. Вместо нашего `if/else` — LLM планирует и адаптируется. ## Шаг 1: Минимальный агент — 20 строк Начнём с простейшего агента. Никаких кастомных инструментов — только Claude Agent SDK и встроенные возможности. ### Установка ```bash pip install claude-agent-sdk ``` Claude Agent SDK (актуальная версия: `0.2.110`) — обёртка над [Claude Code CLI](/ru/blog/claude-code-complete-guide/). SDK поставляется с встроенным CLI-бинарём, отдельная установка `claude` CLI не нужна. Требуется только API-ключ Anthropic и Python 3.10+. Пакет находится в статусе alpha — API стабилен, но возможны breaking changes между minor-версиями. ### Минимальный агент ```python import asyncio from claude_agent_sdk import query, ClaudeAgentOptions, ResultMessage async def analyze_repo(repo_url: str) -> str: """Минимальный агент: отправляет промпт и получает результат.""" options = ClaudeAgentOptions( allowed_tools=["Read", "Bash"], # Разрешаем чтение файлов и shell ) async for message in query( prompt=f"""Проанализируй GitHub-репозиторий: {repo_url} Используй git clone для клонирования, затем изучи структуру проекта. Определи стек технологий, архитектуру, наличие тестов и CI/CD. Выдай структурированный отчёт.""", options=options, ): if isinstance(message, ResultMessage) and message.subtype == "success": return message.result return "Анализ не завершён" # Запуск result = asyncio.run(analyze_repo("https://github.com/anthropics/claude-code")) print(result) ``` 20 строк. Агент клонирует репозиторий, читает файлы, анализирует структуру, генерирует отчёт. Claude сам решает, какие файлы читать и в каком порядке. ### Что происходит под капотом 1. `query()` запускает Claude Code CLI с заданным промптом 2. Claude получает задачу и список разрешённых инструментов (`Read`, `Bash`) 3. Включается agent loop: Claude думает → вызывает `Bash(git clone ...)` → читает файлы → думает → вызывает `Read(package.json)` → анализирует → ... 4. Когда Claude решает, что собрал достаточно информации — генерирует финальный ответ 5. `ResultMessage` с `subtype == "success"` содержит итоговый отчёт Проблема этого подхода: агент использует `Bash` для всего. Нет структурированного доступа к GitHub API — только `curl` через shell. Нет типизации, нет контроля над запросами. Исправим в следующем шаге. ## Шаг 2: Добавляем инструменты Кастомные инструменты — способ дать агенту структурированный доступ к внешним сервисам. Вместо «вызови curl и распарси JSON» — типизированная функция с описанием, которую LLM вызывает по имени. ### Декоратор @tool Claude Agent SDK использует декоратор `@tool` для определения инструментов. Три аргумента: имя, описание (LLM читает его при выборе инструмента), схема параметров. ```python from claude_agent_sdk import tool from typing import Any import httpx GITHUB_API = "https://api.github.com" HEADERS = {"Accept": "application/vnd.github.v3+json"} @tool("get_repo_info", "Get repository metadata: stars, forks, language, description", {"owner": str, "repo": str}) async def get_repo_info(args: dict[str, Any]) -> dict[str, Any]: """Получить метаданные репозитория.""" async with httpx.AsyncClient() as client: resp = await client.get( f"{GITHUB_API}/repos/{args['owner']}/{args['repo']}", headers=HEADERS, ) data = resp.json() return {"content": [{"type": "text", "text": f""" Repository: {data['full_name']} Description: {data.get('description', 'N/A')} Language: {data.get('language', 'N/A')} Stars: {data['stargazers_count']} Forks: {data['forks_count']} Open Issues: {data['open_issues_count']} Created: {data['created_at']} Last Push: {data['pushed_at']} License: {data.get('license', {}).get('name', 'N/A')} """}]} @tool("get_repo_tree", "Get file tree of a repository", {"owner": str, "repo": str, "branch": str}) async def get_repo_tree(args: dict[str, Any]) -> dict[str, Any]: """Получить дерево файлов репозитория.""" async with httpx.AsyncClient() as client: resp = await client.get( f"{GITHUB_API}/repos/{args['owner']}/{args['repo']}/git/trees/{args['branch']}?recursive=1", headers=HEADERS, ) data = resp.json() tree = [item['path'] for item in data.get('tree', []) if item['type'] == 'blob'] # Ограничиваем до 200 файлов, чтобы не раздувать контекст if len(tree) > 200: tree = tree[:200] + [f"... и ещё {len(tree) - 200} файлов"] return {"content": [{"type": "text", "text": "\n".join(tree)}]} @tool("get_file_content", "Get content of a specific file from repository", {"owner": str, "repo": str, "path": str}) async def get_file_content(args: dict[str, Any]) -> dict[str, Any]: """Получить содержимое файла из репозитория.""" async with httpx.AsyncClient() as client: resp = await client.get( f"{GITHUB_API}/repos/{args['owner']}/{args['repo']}/contents/{args['path']}", headers=HEADERS, ) data = resp.json() import base64 content = base64.b64decode(data['content']).decode('utf-8') # Ограничиваем размер, чтобы не съесть весь контекст if len(content) > 10000: content = content[:10000] + "\n\n... [truncated]" return {"content": [{"type": "text", "text": content}]} @tool("get_recent_commits", "Get recent commits with messages and authors", {"owner": str, "repo": str, "count": int}) async def get_recent_commits(args: dict[str, Any]) -> dict[str, Any]: """Получить последние коммиты.""" async with httpx.AsyncClient() as client: resp = await client.get( f"{GITHUB_API}/repos/{args['owner']}/{args['repo']}/commits", headers=HEADERS, params={"per_page": min(args.get('count', 10), 30)}, ) commits = resp.json() lines = [] for c in commits: author = c['commit']['author']['name'] date = c['commit']['author']['date'][:10] msg = c['commit']['message'].split('\n')[0] # Только первая строка lines.append(f"[{date}] {author}: {msg}") return {"content": [{"type": "text", "text": "\n".join(lines)}]} ``` ### Собираем агента с инструментами ```python from claude_agent_sdk import create_sdk_mcp_server, ClaudeSDKClient, ClaudeAgentOptions # Создаём in-process MCP сервер с нашими инструментами github_server = create_sdk_mcp_server( name="github-analyzer", version="1.0.0", tools=[get_repo_info, get_repo_tree, get_file_content, get_recent_commits], ) options = ClaudeAgentOptions( mcp_servers={"github": github_server}, allowed_tools=[ # Перечисляем явно — wildcard mcp__github__* не работает стабильно (known issue) "mcp__github__get_repo_info", "mcp__github__get_repo_tree", "mcp__github__get_file_content", "mcp__github__get_recent_commits", ], ) async def analyze_repo(owner: str, repo: str) -> str: async with ClaudeSDKClient(options=options) as client: await client.query(f"""Проанализируй репозиторий {owner}/{repo}. Порядок действий: 1. Получи метаданные репозитория (get_repo_info) 2. Получи дерево файлов (get_repo_tree) 3. Прочитай ключевые файлы: README, package.json / pyproject.toml / Cargo.toml, Dockerfile, CI-конфиги 4. Получи последние 20 коммитов (get_recent_commits) 5. Сформируй структурированный отчёт Формат отчёта: ## Общая информация ## Стек технологий ## Архитектура ## Качество кода ## Активность разработки ## Рекомендации""") result_parts = [] async for message in client.receive_response(): if isinstance(message, ResultMessage) and message.subtype == "success": return message.result return "Анализ не завершён" ``` ### Почему in-process MCP сервер Claude Agent SDK создаёт MCP-сервер прямо в процессе Python-приложения. Инструменты — обычные async-функции. Преимущества перед внешними серверами: - **Без subprocess** — всё в одном процессе, нет накладных расходов на IPC - **Проще отладка** — стандартный Python-дебаггер, breakpoints, logging - **Одно развёртывание** — один контейнер вместо нескольких процессов - **Типизация** — Python type hints, IDE подсказывает параметры Описание инструмента (`"Get repository metadata: stars, forks, language, description"`) — главный рычаг. LLM выбирает инструмент по описанию: расплывчатое даёт промах, конкретное — попадание. ## Шаг 3: Добавляем memory Агент из предыдущего шага stateless — каждый запуск начинает с нуля. Если пользователь проанализировал 10 репозиториев, агент не помнит ни одного. Memory решает две задачи: 1. **Short-term** — контекст текущей сессии. Claude Agent SDK управляет этим автоматически через `ClaudeSDKClient`. 2. **Long-term** — результаты прошлых анализов, выученные паттерны, кэш. ### Short-term memory: интерактивная сессия `ClaudeSDKClient` поддерживает bidirectional-разговор. Каждый `query()` в рамках одной сессии видит контекст предыдущих сообщений: ```python async def interactive_analysis(): async with ClaudeSDKClient(options=options) as client: # Первый запрос — полный анализ await client.query("Проанализируй репозиторий anthropics/claude-code") async for msg in client.receive_response(): if isinstance(msg, ResultMessage) and msg.subtype == "success": print(msg.result) # Второй запрос — агент помнит контекст первого await client.query("Сравни его архитектуру с тем, что типично для CLI-инструментов") async for msg in client.receive_response(): if isinstance(msg, ResultMessage) and msg.subtype == "success": print(msg.result) # Третий запрос — углубление await client.query("Какие конкретно файлы отвечают за agent loop?") async for msg in client.receive_response(): if isinstance(msg, ResultMessage) and msg.subtype == "success": print(msg.result) ``` Внутри одного `async with ClaudeSDKClient(...)` контекст накапливается. Агент помнит предыдущие запросы, результаты вызовов инструментов, промежуточные рассуждения. ### Long-term memory: сохраняем результаты Для долговременной памяти добавим инструменты записи и чтения из хранилища. Простейший вариант — JSON-файлы: ```python import json from pathlib import Path MEMORY_DIR = Path("./agent_memory") MEMORY_DIR.mkdir(exist_ok=True) @tool("save_analysis", "Save repository analysis to memory for future reference", {"repo": str, "analysis": str, "tags": str}) async def save_analysis(args: dict[str, Any]) -> dict[str, Any]: """Сохранить результат анализа.""" memory_file = MEMORY_DIR / "analyses.json" analyses = [] if memory_file.exists(): analyses = json.loads(memory_file.read_text()) from datetime import datetime analyses.append({ "repo": args["repo"], "analysis": args["analysis"], "tags": args.get("tags", "").split(","), "timestamp": datetime.now().isoformat(), }) memory_file.write_text(json.dumps(analyses, indent=2, ensure_ascii=False)) return {"content": [{"type": "text", "text": f"Анализ {args['repo']} сохранён в memory"}]} @tool("search_memory", "Search previous analyses by keyword or repo name", {"query": str}) async def search_memory(args: dict[str, Any]) -> dict[str, Any]: """Поиск по прошлым анализам.""" memory_file = MEMORY_DIR / "analyses.json" if not memory_file.exists(): return {"content": [{"type": "text", "text": "Memory пуста — ещё не было анализов"}]} analyses = json.loads(memory_file.read_text()) query_lower = args["query"].lower() matches = [ a for a in analyses if query_lower in a["repo"].lower() or query_lower in a["analysis"].lower() or any(query_lower in tag.lower() for tag in a["tags"]) ] if not matches: return {"content": [{"type": "text", "text": f"Ничего не найдено по запросу '{args['query']}'"}]} results = [] for m in matches[-5:]: # Последние 5 совпадений results.append(f"[{m['timestamp'][:10]}] {m['repo']}:\n{m['analysis'][:500]}") return {"content": [{"type": "text", "text": "\n\n---\n\n".join(results)}]} ``` ### Обновляем сервер ```python github_server = create_sdk_mcp_server( name="github-analyzer", version="1.1.0", tools=[ get_repo_info, get_repo_tree, get_file_content, get_recent_commits, save_analysis, search_memory, ], ) ``` Теперь агент может сохранять анализы и обращаться к ним в будущем. Промпт дополняем инструкцией: ``` После завершения анализа — сохрани результат через save_analysis. Перед анализом нового репозитория — проверь search_memory, возможно этот репозиторий уже анализировался ранее. ``` В production вместо JSON-файлов — SQLite, PostgreSQL или vector store (для семантического поиска по прошлым анализам). Но принцип тот же: инструменты для записи и чтения. ## Шаг 4: Multi-tool agent — агент сам выбирает инструменты До сих пор мы давали агенту явные инструкции: «сначала get_repo_info, потом get_repo_tree». Настоящий агент должен сам решать, какие инструменты вызывать и в каком порядке. ### Расширяем набор инструментов Добавим инструменты, которые покрывают разные аспекты анализа: ```python @tool("get_issues", "Get open issues (excluding pull requests) with labels and comments count", {"owner": str, "repo": str, "state": str, "count": int}) async def get_issues(args: dict[str, Any]) -> dict[str, Any]: """Получить issues репозитория (без PR — GitHub API возвращает оба типа по /issues).""" async with httpx.AsyncClient() as client: resp = await client.get( f"{GITHUB_API}/repos/{args['owner']}/{args['repo']}/issues", headers=HEADERS, params={ "state": args.get("state", "open"), "per_page": min(args.get("count", 10), 30), }, ) resp.raise_for_status() issues = resp.json() lines = [] for issue in issues: # GitHub API /issues возвращает и PR — фильтруем их по наличию pull_request поля if "pull_request" in issue: continue labels = ", ".join(l["name"] for l in issue.get("labels", [])) lines.append( f"#{issue['number']} [{labels or 'no labels'}] " f"{issue['title']} ({issue['comments']} comments)" ) return {"content": [{"type": "text", "text": "\n".join(lines) or "No issues found"}]} @tool("get_contributors", "Get top contributors with commit counts", {"owner": str, "repo": str}) async def get_contributors(args: dict[str, Any]) -> dict[str, Any]: """Получить контрибьюторов.""" async with httpx.AsyncClient() as client: resp = await client.get( f"{GITHUB_API}/repos/{args['owner']}/{args['repo']}/contributors", headers=HEADERS, params={"per_page": 15}, ) contributors = resp.json() lines = [ f"{c['login']}: {c['contributions']} commits" for c in contributors ] return {"content": [{"type": "text", "text": "\n".join(lines)}]} @tool("get_languages", "Get programming languages breakdown with percentages", {"owner": str, "repo": str}) async def get_languages(args: dict[str, Any]) -> dict[str, Any]: """Получить распределение языков.""" async with httpx.AsyncClient() as client: resp = await client.get( f"{GITHUB_API}/repos/{args['owner']}/{args['repo']}/languages", headers=HEADERS, ) languages = resp.json() total = sum(languages.values()) lines = [ f"{lang}: {bytes_count / total * 100:.1f}%" for lang, bytes_count in sorted(languages.items(), key=lambda x: -x[1]) ] return {"content": [{"type": "text", "text": "\n".join(lines)}]} @tool("check_ci_config", "Check if CI/CD configuration exists and what system is used", {"owner": str, "repo": str}) async def check_ci_config(args: dict[str, Any]) -> dict[str, Any]: """Проверить наличие CI/CD конфигурации.""" ci_paths = [ ".github/workflows", ".gitlab-ci.yml", "Jenkinsfile", ".circleci/config.yml", ".travis.yml", "azure-pipelines.yml", ] found = [] async with httpx.AsyncClient() as client: for path in ci_paths: resp = await client.get( f"{GITHUB_API}/repos/{args['owner']}/{args['repo']}/contents/{path}", headers=HEADERS, ) if resp.status_code == 200: data = resp.json() if isinstance(data, list): found.append(f"{path}: {', '.join(f['name'] for f in data)}") else: found.append(path) if not found: return {"content": [{"type": "text", "text": "CI/CD configuration not found"}]} return {"content": [{"type": "text", "text": "CI/CD found:\n" + "\n".join(found)}]} ``` ### Агент с автономным выбором Секрет — в промпте. Не перечисляем шаги, а описываем цель и критерии качества: ```python ANALYZER_PROMPT = """Ты — senior-инженер, специализирующийся на code review и архитектурном анализе. Задача: провести глубокий анализ репозитория {owner}/{repo}. У тебя есть инструменты для работы с GitHub API. Используй их по своему усмотрению — ты решаешь, какие данные собрать и в каком порядке. Критерии хорошего анализа: - Стек определён точно (не "возможно Python", а "Python 3.12, FastAPI 0.110, SQLAlchemy 2.0") - Архитектура описана с конкретными модулями и их взаимосвязями - Качество кода оценено по фактам (есть тесты? CI? линтер? типизация?) - Активность — не "активно развивается", а "23 коммита за последний месяц от 4 контрибьюторов" - Рекомендации конкретные и actionable Если в memory есть прошлый анализ этого репозитория — учти его: отметь изменения. Формат: Markdown с заголовками второго уровня.""" async def smart_analyze(owner: str, repo: str) -> str: server = create_sdk_mcp_server( name="github-analyzer", version="2.0.0", tools=[ get_repo_info, get_repo_tree, get_file_content, get_recent_commits, get_issues, get_contributors, get_languages, check_ci_config, save_analysis, search_memory, ], ) opts = ClaudeAgentOptions( mcp_servers={"github": server}, allowed_tools=["mcp__github__*"], # Wildcard — см. примечание ниже ) async for message in query( prompt=ANALYZER_PROMPT.format(owner=owner, repo=repo), options=opts, ): if isinstance(message, ResultMessage) and message.subtype == "success": return message.result return "Анализ не завершён" ``` **Примечание по wildcard:** `allowed_tools=["mcp__github__*"]` теоретически разрешает все инструменты сервера. На практике wildcard-паттерн для MCP-инструментов нестабильно работает в ряде версий claude-agent-sdk ([known issue](https://github.com/anthropics/claude-code/issues/13077)) — при проблемах перечисляйте инструменты явно. LLM сам определяет порядок: может начать с `get_repo_info`, а может сразу пойти в `get_repo_tree`, если ему нужна структура проекта. Может пропустить `get_issues`, если вопрос только про архитектуру. Адаптируется к задаче. ## Шаг 5: Error handling и guardrails Production-агент без обработки ошибок — бомба замедленного действия. GitHub API возвращает 403 при превышении rate limit. Сеть падает. LLM галлюцинирует. Нужны guardrails. ### Hooks для контроля Claude Agent SDK поддерживает hooks — функции, которые вызываются до и после каждого инструмента: ```python from claude_agent_sdk import HookMatcher, HookContext from datetime import datetime import logging logger = logging.getLogger("agent") # Счётчик вызовов для cost control call_counter = {"total": 0, "by_tool": {}} async def pre_tool_guard( input_data: dict[str, Any], tool_use_id: str | None, context: HookContext ) -> dict[str, Any]: """Guardrail: логирование + лимит вызовов.""" tool_name = input_data.get("tool_name", "unknown") call_counter["total"] += 1 call_counter["by_tool"][tool_name] = call_counter["by_tool"].get(tool_name, 0) + 1 logger.info(f"[{call_counter['total']}] Tool: {tool_name}") # Лимит: максимум 50 вызовов за сессию if call_counter["total"] > 50: logger.warning("Tool call limit exceeded") return { "hookSpecificOutput": { "hookEventName": "PreToolUse", "permissionDecision": "deny", "permissionDecisionReason": "Превышен лимит вызовов инструментов (50). Сформируй ответ на основе собранных данных.", } } # Блокируем опасные операции if tool_name == "Bash": command = str(input_data.get("tool_input", {}).get("command", "")) dangerous = ["rm -rf", "sudo", "chmod 777", "> /dev/"] if any(d in command for d in dangerous): return { "hookSpecificOutput": { "hookEventName": "PreToolUse", "permissionDecision": "deny", "permissionDecisionReason": f"Опасная команда заблокирована: {command}", } } return {} async def post_tool_logger( input_data: dict[str, Any], tool_use_id: str | None, context: HookContext ) -> dict[str, Any]: """Логирование результатов.""" tool_name = input_data.get("tool_name", "unknown") # Проверяем наличие ошибки в результате result = input_data.get("tool_response", "") # поле называется tool_response в PostToolUse if "error" in str(result).lower() or "404" in str(result): logger.warning(f"Tool {tool_name} returned error: {str(result)[:200]}") return {} ``` ### Retry и timeout в инструментах Обработку ошибок GitHub API встраиваем прямо в инструменты: ```python import httpx from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type @retry( stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10), retry=retry_if_exception_type((httpx.HTTPStatusError, httpx.ConnectError)), ) async def github_request(url: str, params: dict | None = None) -> dict: """HTTP-запрос к GitHub API с retry и rate limit handling.""" async with httpx.AsyncClient(timeout=30.0) as client: resp = await client.get(url, headers=HEADERS, params=params) # Rate limit — ждём и ретраим if resp.status_code == 403: remaining = int(resp.headers.get("X-RateLimit-Remaining", 0)) if remaining == 0: reset_time = int(resp.headers.get("X-RateLimit-Reset", 0)) wait_seconds = max(reset_time - int(datetime.now().timestamp()), 1) logger.warning(f"Rate limit hit. Waiting {wait_seconds}s") await asyncio.sleep(min(wait_seconds, 60)) # Не ждём дольше минуты raise httpx.HTTPStatusError("Rate limit", request=resp.request, response=resp) resp.raise_for_status() return resp.json() ``` ### Собираем production-конфигурацию ```python options = ClaudeAgentOptions( mcp_servers={"github": github_server}, allowed_tools=["mcp__github__*"], # или явный список при проблемах с wildcard hooks={ "PreToolUse": [HookMatcher(hooks=[pre_tool_guard])], "PostToolUse": [HookMatcher(hooks=[post_tool_logger])], }, ) ``` Три уровня защиты: 1. **Pre-tool hooks** — блокируют опасные вызовы и лимитируют количество 2. **Retry в инструментах** — обрабатывают transient-ошибки сети и API 3. **Post-tool hooks** — логируют ошибки для анализа Эти же принципы (изоляция, лимиты, логирование) применимы и к собственным MCP-серверам, которые вы разворачиваете отдельно от агента — подробнее в статье [MCP production: создание и деплой кастомных серверов](/ru/blog/mcp-production-custom-servers/). Четвёртый уровень — промпт. Добавьте в system prompt: ``` Если инструмент вернул ошибку — попробуй другой подход. Не повторяй один и тот же вызов больше 2 раз. Если не удаётся получить данные — сообщи об этом в отчёте, а не выдумывай информацию. ``` ## Шаг 6: MCP-интеграция — подключаем готовые серверы До этого мы писали инструменты вручную. MCP (Model Context Protocol) — открытый стандарт: подключаешь готовый сервер и получаешь тысячи инструментов без написания кода. ### Что такое MCP MCP — открытый протокол для подключения AI-агентов к инструментам, данным и контексту. Один протокол, который работает с Claude, Cursor, VS Code, ChatGPT, Gemini. Спецификация актуальной версии: 2025-11-25. Де-факто стандарт для tool/context-интеграции: поддержан основными AI-платформами и развивается под governance umbrella Agentic AI Foundation (Linux Foundation, декабрь 2025 — Anthropic, Google, Microsoft, OpenAI, AWS в числе основателей). Для agent-to-agent коммуникации параллельно существует протокол **A2A** — MCP и A2A решают разные задачи и дополняют друг друга. Подробнее — в статье [MCP серверы: что это, зачем нужны и как подключить](/ru/blog/mcp-servers-explained/). ### Подключаем GitHub MCP-сервер Вместо наших кастомных инструментов — официальный GitHub MCP-сервер. API-вызовы, пагинация, авторизация, обработка ошибок уже внутри. **Важно:** пакет `@modelcontextprotocol/server-github` устарел (deprecated с апреля 2025). GitHub выпустил официальный сервер — `github/github-mcp-server`. Два способа подключения: ```bash # Вариант 1 — Docker (рекомендован) claude mcp add github -e GITHUB_PERSONAL_ACCESS_TOKEN=YOUR_TOKEN -- \ docker run -i --rm -e GITHUB_PERSONAL_ACCESS_TOKEN ghcr.io/github/github-mcp-server # Вариант 2 — удалённый HTTP-сервер (проще всего) claude mcp add-json github '{"type":"http","url":"https://api.githubcopilot.com/mcp","headers":{"Authorization":"Bearer YOUR_GITHUB_PAT"}}' ``` Для использования в Agent SDK через Docker: ```python import os from claude_agent_sdk import query, ClaudeAgentOptions, ResultMessage, SystemMessage, AssistantMessage async def analyze_with_mcp(owner: str, repo: str) -> str: options = ClaudeAgentOptions( mcp_servers={ "github": { "command": "docker", "args": ["run", "-i", "--rm", "-e", "GITHUB_PERSONAL_ACCESS_TOKEN", "ghcr.io/github/github-mcp-server"], "env": {"GITHUB_PERSONAL_ACCESS_TOKEN": os.environ["GITHUB_TOKEN"]}, } }, allowed_tools=[ "mcp__github__get_file_contents", "mcp__github__search_code", "mcp__github__list_issues", "mcp__github__list_commits", "mcp__github__search_repositories", ], ) async for message in query( prompt=f"""Проанализируй репозиторий {owner}/{repo}. Используй доступные GitHub-инструменты для сбора данных: - Прочитай ключевые файлы (README, конфигурации, package.json и т.д.) - Посмотри структуру репозитория - Изучи последние коммиты и issues Сформируй детальный отчёт: стек, архитектура, качество, активность, рекомендации.""", options=options, ): if isinstance(message, SystemMessage) and message.subtype == "init": mcp_info = message.data.get("mcp_servers", {}) print(f"MCP servers connected: {list(mcp_info.keys())}") if isinstance(message, AssistantMessage): for block in message.content: if hasattr(block, "name") and block.name.startswith("mcp__"): print(f" → MCP tool: {block.name}") if isinstance(message, ResultMessage) and message.subtype == "success": return message.result return "Анализ не завершён" ``` ### Комбинируем кастомные и MCP инструменты Сила подхода — комбинирование. Кастомные инструменты для бизнес-логики, MCP-серверы для стандартных интеграций: ```python # Кастомный сервер — наша аналитика и memory custom_server = create_sdk_mcp_server( name="analyzer", version="2.0.0", tools=[save_analysis, search_memory], ) options = ClaudeAgentOptions( mcp_servers={ # MCP-сервер для GitHub (внешний процесс через Docker) "github": { "command": "docker", "args": ["run", "-i", "--rm", "-e", "GITHUB_PERSONAL_ACCESS_TOKEN", "ghcr.io/github/github-mcp-server"], "env": {"GITHUB_PERSONAL_ACCESS_TOKEN": os.environ["GITHUB_TOKEN"]}, }, # Кастомный сервер (in-process) "analyzer": custom_server, }, allowed_tools=[ "mcp__github__*", # Все инструменты GitHub (wildcard, см. примечание выше) "mcp__analyzer__*", # Все наши инструменты ], ) ``` Два сервера работают параллельно. GitHub MCP-сервер запускается как subprocess через Docker (команда `docker run`). Кастомный `analyzer` — in-process. Агент вызывает инструменты обоих серверов прозрачно. ### Когда свои инструменты, когда MCP **MCP-сервер подходит**, когда: - Есть готовый сервер для нужного сервиса (GitHub, Slack, PostgreSQL, Notion) - Нужна стандартная интеграция без кастомной логики - Инструмент будет переиспользоваться в разных проектах **Кастомные инструменты подходят**, когда: - Нужна специфическая бизнес-логика (наш `save_analysis`) - Инструмент работает с внутренними системами компании - Нужен полный контроль над форматом ввода/вывода и обработкой ошибок - Критична производительность (in-process быстрее subprocess) ## Шаг 7: Мультиагентная система — supervisor + workers Один агент хорошо справляется с одной задачей. Для комплексного анализа — сборка из нескольких специализированных агентов. ### Архитектура: supervisor + workers ``` ┌───────────────────────────────────────────┐ │ SUPERVISOR AGENT │ │ Планирует, распределяет, собирает │ └──────┬──────────┬──────────┬──────────────┘ ▼ ▼ ▼ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ ARCH │ │ QUALITY │ │ ACTIVITY │ │ WORKER │ │ WORKER │ │ WORKER │ │ │ │ │ │ │ │Структура │ │Тесты, │ │Коммиты, │ │файлов, │ │CI/CD, │ │issues, │ │модули, │ │линтеры, │ │контриб- │ │паттерны │ │типизация │ │юторы │ └──────────┘ └──────────┘ └──────────┘ ``` **Supervisor** — основной агент. Получает задачу, разбивает на подзадачи, отправляет workers, собирает результаты в единый отчёт. **Workers** — специализированные агенты. Каждый фокусируется на одном аспекте анализа и имеет доступ только к нужным инструментам. Паттерны координации, разделение ответственности между supervisor и workers, альтернативные топологии — разобраны подробнее в статье [Мультиагентные системы: архитектура и паттерны](/ru/blog/multi-agent-architecture/). ### Реализация на CrewAI CrewAI — фреймворк для мультиагентных систем (актуальная стабильная линейка на июнь 2026 — 1.14.x). Определяет агентов, задачи и стратегию координации. Архитектура строится на двух примитивах: **Crews** (команды агентов для задач) и **Flows** (управляемые пайплайны с condition/branching). В отличие от LangChain-зависимых решений, работает без внешних оркестрационных зависимостей: ```python from crewai import Agent, Task, Crew, Process from crewai.tools import tool as crewai_tool import httpx # Инструменты для CrewAI — обёртки над GitHub API @crewai_tool("GitHub Repo Info") def get_repo_info_crew(owner_repo: str) -> str: """Get metadata for a GitHub repository. Input: 'owner/repo' format.""" owner, repo = owner_repo.split("/") resp = httpx.get(f"https://api.github.com/repos/{owner}/{repo}") data = resp.json() return f"Stars: {data['stargazers_count']}, Forks: {data['forks_count']}, Language: {data.get('language')}" @crewai_tool("GitHub File Tree") def get_tree_crew(owner_repo: str) -> str: """Get file tree of a GitHub repository. Input: 'owner/repo' format.""" owner, repo = owner_repo.split("/") resp = httpx.get(f"https://api.github.com/repos/{owner}/{repo}/git/trees/main?recursive=1") tree = [item["path"] for item in resp.json().get("tree", [])[:100]] return "\n".join(tree) @crewai_tool("GitHub File Content") def get_file_crew(owner_repo_path: str) -> str: """Get file content from GitHub. Input: 'owner/repo/path/to/file' format.""" parts = owner_repo_path.split("/", 2) owner, repo, path = parts[0], parts[1], parts[2] resp = httpx.get(f"https://api.github.com/repos/{owner}/{repo}/contents/{path}") import base64 return base64.b64decode(resp.json()["content"]).decode()[:5000] @crewai_tool("GitHub Commits") def get_commits_crew(owner_repo: str) -> str: """Get recent commits. Input: 'owner/repo' format.""" owner, repo = owner_repo.split("/") resp = httpx.get(f"https://api.github.com/repos/{owner}/{repo}/commits", params={"per_page": 20}) return "\n".join( f"[{c['commit']['author']['date'][:10]}] {c['commit']['message'].split(chr(10))[0]}" for c in resp.json() ) # Определяем агентов architect = Agent( role="Software Architect", goal="Analyze repository architecture, patterns, and module structure", backstory="Senior architect with 15 years of experience in distributed systems", tools=[get_tree_crew, get_file_crew], verbose=True, ) quality_engineer = Agent( role="Quality Engineer", goal="Assess code quality: tests, CI/CD, linting, type safety", backstory="QA lead who has built testing infrastructure for Fortune 500 companies", tools=[get_tree_crew, get_file_crew], verbose=True, ) activity_analyst = Agent( role="Activity Analyst", goal="Analyze development activity: commits, contributors, issues, velocity", backstory="Data analyst specializing in open-source project health metrics", tools=[get_repo_info_crew, get_commits_crew], verbose=True, ) # Определяем задачи с async_execution=True для параллельного запуска def create_tasks(owner: str, repo: str) -> list[Task]: arch_task = Task( description=f"Analyze architecture of {owner}/{repo}: file structure, main modules, design patterns, dependency graph", agent=architect, expected_output="Structured architecture report with module descriptions and dependency analysis", async_execution=True, ) quality_task = Task( description=f"Assess code quality of {owner}/{repo}: test coverage indicators, CI/CD setup, linting, type checking, code style", agent=quality_engineer, expected_output="Quality assessment with specific findings and scores", async_execution=True, ) activity_task = Task( description=f"Analyze development activity of {owner}/{repo}: commit frequency, contributor distribution, issue handling, development velocity", agent=activity_analyst, expected_output="Activity report with metrics and trends", async_execution=True, ) # Финальная задача собирает результаты трёх параллельных задач summary_task = Task( description=f"Consolidate all analysis results for {owner}/{repo} into a single structured report", agent=architect, expected_output="Complete repository analysis report", context=[arch_task, quality_task, activity_task], ) return [arch_task, quality_task, activity_task, summary_task] # Запускаем def analyze_multi_agent(owner: str, repo: str) -> str: crew = Crew( agents=[architect, quality_engineer, activity_analyst], tasks=create_tasks(owner, repo), process=Process.sequential, # async_execution на задачах даёт параллельность verbose=True, ) result = crew.kickoff() return str(result) ``` ### Supervisor на Claude Agent SDK Альтернативный подход — supervisor как Claude-агент, а workers вызываются как инструменты: ```python @tool("analyze_architecture", "Delegate architecture analysis to specialist agent", {"owner": str, "repo": str}) async def analyze_architecture(args: dict[str, Any]) -> dict[str, Any]: """Worker: анализ архитектуры.""" arch_server = create_sdk_mcp_server( name="arch-tools", tools=[get_repo_tree, get_file_content], ) async for msg in query( prompt=f"""Ты — software architect. Проанализируй архитектуру {args['owner']}/{args['repo']}. Определи: структуру модулей, паттерны проектирования, граф зависимостей. Ответ — структурированный отчёт, 300-500 слов.""", options=ClaudeAgentOptions( mcp_servers={"tools": arch_server}, allowed_tools=["mcp__tools__get_repo_tree", "mcp__tools__get_file_content"], ), ): if isinstance(msg, ResultMessage) and msg.subtype == "success": return {"content": [{"type": "text", "text": msg.result}]} return {"content": [{"type": "text", "text": "Architecture analysis failed"}]} # Supervisor имеет доступ к worker-инструментам supervisor_server = create_sdk_mcp_server( name="supervisor", tools=[analyze_architecture, analyze_quality, analyze_activity], # workers ) ``` Supervisor вызывает workers как инструменты. Каждый worker — отдельный `query()` со своими инструментами и промптом. Supervisor собирает результаты и формирует итоговый отчёт. ### Когда нужна мультиагентность **Нужна**, когда: - Задача декомпозируется на независимые подзадачи (параллельный анализ) - Разные подзадачи требуют разных промптов и инструментов - Нужна масштабируемость — добавить аспект анализа = добавить worker **Не нужна**, когда: - Задача решается одним агентом за разумное время - Подзадачи сильно связаны и требуют общего контекста - Бюджет ограничен (каждый worker — дополнительные вызовы LLM) ## Шаг 8: Deployment — production Агент работает локально. Для production нужен Docker, мониторинг и API-обёртка. ### Docker ```dockerfile FROM python:3.12-slim # Node.js нужен только если используете внешние MCP-серверы через npx/npm # claude-agent-sdk включает собственный CLI бинарь — отдельная установка не нужна # docker.io нужен только если агент запускает дочерние контейнеры (например, GitHub MCP-сервер через Docker). # ВНИМАНИЕ: для этого потребуется пробросить /var/run/docker.sock в контейнер, # что даёт процессу внутри root-эквивалентный доступ к хосту. Не использовать # в публично-доступных окружениях без дополнительной изоляции (rootless Docker, Sysbox). RUN apt-get update && apt-get install -y curl docker.io && \ apt-get clean && rm -rf /var/lib/apt/lists/* WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . . # Директория для memory RUN mkdir -p /app/agent_memory EXPOSE 8000 CMD ["uvicorn", "server:app", "--host", "0.0.0.0", "--port", "8000"] ``` ### API-обёртка на FastAPI ```python from fastapi import FastAPI, HTTPException from pydantic import BaseModel import asyncio app = FastAPI(title="GitHub Repo Analyzer Agent") class AnalyzeRequest(BaseModel): owner: str repo: str class AnalyzeResponse(BaseModel): status: str report: str tools_called: int duration_seconds: float @app.post("/analyze", response_model=AnalyzeResponse) async def analyze_endpoint(request: AnalyzeRequest): import time start = time.time() # Сбрасываем счётчик вызовов call_counter["total"] = 0 call_counter["by_tool"] = {} try: report = await smart_analyze(request.owner, request.repo) duration = time.time() - start return AnalyzeResponse( status="success", report=report, tools_called=call_counter["total"], duration_seconds=round(duration, 2), ) except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @app.get("/health") async def health(): return {"status": "ok"} ``` ### Мониторинг Три метрики, которые нужно отслеживать: **1. Стоимость.** Каждый `query()` — вызовы LLM. Считайте токены. ```python import os # Устанавливаем лимиты через environment MAX_TOKENS_PER_REQUEST = int(os.environ.get("MAX_TOKENS", "100000")) MAX_TOOL_CALLS = int(os.environ.get("MAX_TOOL_CALLS", "50")) ``` **2. Latency.** Типичный анализ репозитория — 30-120 секунд. Если дольше 5 минут — что-то пошло не так. ```python import asyncio async def analyze_with_timeout(owner: str, repo: str, timeout: int = 300) -> str: try: return await asyncio.wait_for( smart_analyze(owner, repo), timeout=timeout, ) except asyncio.TimeoutError: return f"Анализ {owner}/{repo} превысил таймаут {timeout}s" ``` **3. Качество.** Логируйте запросы и ответы. Периодически проверяйте, что отчёты содержательные, а не галлюцинации. ### Структура production-проекта ``` repo-analyzer/ ├── Dockerfile ├── docker-compose.yml ├── requirements.txt ├── server.py # FastAPI endpoint ├── agent/ │ ├── __init__.py │ ├── tools.py # @tool definitions │ ├── hooks.py # Pre/Post tool hooks │ ├── memory.py # Long-term memory │ ├── prompts.py # System prompts │ └── analyzer.py # Main agent logic ├── tests/ │ ├── test_tools.py │ └── test_agent.py └── agent_memory/ # Persistent storage ``` ## Полный код — финальная версия Собранный агент — все шаги в одном файле. Скопируйте, установите зависимости, запустите. ```python """ GitHub Repository Analyzer Agent Requires: pip install claude-agent-sdk httpx tenacity Environment: GITHUB_TOKEN, ANTHROPIC_API_KEY """ import asyncio import json import logging import os from datetime import datetime from pathlib import Path from typing import Any import httpx from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type from claude_agent_sdk import ( query, tool, create_sdk_mcp_server, ClaudeAgentOptions, ResultMessage, HookMatcher, HookContext, ) # ── Config ────────────────────────────────────────────── GITHUB_API = "https://api.github.com" GITHUB_TOKEN = os.environ.get("GITHUB_TOKEN", "") HEADERS = { "Accept": "application/vnd.github.v3+json", "Authorization": f"Bearer {GITHUB_TOKEN}" if GITHUB_TOKEN else "", } MEMORY_DIR = Path("./agent_memory") MEMORY_DIR.mkdir(exist_ok=True) logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s") logger = logging.getLogger("agent") # ── HTTP Client ───────────────────────────────────────── @retry( stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10), retry=retry_if_exception_type((httpx.HTTPStatusError, httpx.ConnectError)), ) async def github_get(path: str, params: dict | None = None) -> dict | list: async with httpx.AsyncClient(timeout=30.0) as client: resp = await client.get(f"{GITHUB_API}{path}", headers=HEADERS, params=params) if resp.status_code == 403: remaining = int(resp.headers.get("X-RateLimit-Remaining", 1)) if remaining == 0: wait_sec = max(int(resp.headers.get("X-RateLimit-Reset", 0)) - int(datetime.now().timestamp()), 1) logger.warning(f"Rate limit. Waiting {min(wait_sec, 60)}s") await asyncio.sleep(min(wait_sec, 60)) raise httpx.HTTPStatusError("Rate limit", request=resp.request, response=resp) resp.raise_for_status() return resp.json() # ── Tools ─────────────────────────────────────────────── @tool("get_repo_info", "Get repository metadata: stars, forks, language, description, license", {"owner": str, "repo": str}) async def get_repo_info(args: dict[str, Any]) -> dict[str, Any]: data = await github_get(f"/repos/{args['owner']}/{args['repo']}") text = ( f"Repository: {data['full_name']}\n" f"Description: {data.get('description', 'N/A')}\n" f"Language: {data.get('language', 'N/A')}\n" f"Stars: {data['stargazers_count']} | Forks: {data['forks_count']}\n" f"Open Issues: {data['open_issues_count']}\n" f"Created: {data['created_at'][:10]} | Last Push: {data['pushed_at'][:10]}\n" f"License: {data.get('license', {}).get('name', 'N/A') if data.get('license') else 'N/A'}\n" f"Default Branch: {data['default_branch']}" ) return {"content": [{"type": "text", "text": text}]} @tool("get_repo_tree", "Get full file tree of a repository", {"owner": str, "repo": str, "branch": str}) async def get_repo_tree(args: dict[str, Any]) -> dict[str, Any]: data = await github_get( f"/repos/{args['owner']}/{args['repo']}/git/trees/{args['branch']}", params={"recursive": "1"}, ) tree = [item["path"] for item in data.get("tree", []) if item["type"] == "blob"] if len(tree) > 200: tree = tree[:200] + [f"... и ещё {len(tree) - 200} файлов"] return {"content": [{"type": "text", "text": "\n".join(tree)}]} @tool("get_file_content", "Get content of a file from repository", {"owner": str, "repo": str, "path": str}) async def get_file_content(args: dict[str, Any]) -> dict[str, Any]: data = await github_get(f"/repos/{args['owner']}/{args['repo']}/contents/{args['path']}") import base64 content = base64.b64decode(data["content"]).decode("utf-8") if len(content) > 10_000: content = content[:10_000] + "\n\n... [truncated]" return {"content": [{"type": "text", "text": content}]} @tool("get_recent_commits", "Get recent commits with messages and authors", {"owner": str, "repo": str, "count": int}) async def get_recent_commits(args: dict[str, Any]) -> dict[str, Any]: commits = await github_get( f"/repos/{args['owner']}/{args['repo']}/commits", params={"per_page": min(args.get("count", 10), 30)}, ) lines = [ f"[{c['commit']['author']['date'][:10]}] {c['commit']['author']['name']}: {c['commit']['message'].split(chr(10))[0]}" for c in commits ] return {"content": [{"type": "text", "text": "\n".join(lines)}]} @tool("get_languages", "Get programming languages breakdown with percentages", {"owner": str, "repo": str}) async def get_languages(args: dict[str, Any]) -> dict[str, Any]: languages = await github_get(f"/repos/{args['owner']}/{args['repo']}/languages") total = sum(languages.values()) or 1 lines = [f"{lang}: {b / total * 100:.1f}%" for lang, b in sorted(languages.items(), key=lambda x: -x[1])] return {"content": [{"type": "text", "text": "\n".join(lines)}]} @tool("get_contributors", "Get top contributors with commit counts", {"owner": str, "repo": str}) async def get_contributors(args: dict[str, Any]) -> dict[str, Any]: contributors = await github_get( f"/repos/{args['owner']}/{args['repo']}/contributors", params={"per_page": 15}, ) lines = [f"{c['login']}: {c['contributions']} commits" for c in contributors] return {"content": [{"type": "text", "text": "\n".join(lines)}]} @tool("get_issues", "Get repository issues with labels (pull requests excluded)", {"owner": str, "repo": str, "state": str, "count": int}) async def get_issues(args: dict[str, Any]) -> dict[str, Any]: issues = await github_get( f"/repos/{args['owner']}/{args['repo']}/issues", params={"state": args.get("state", "open"), "per_page": min(args.get("count", 10), 30)}, ) # GitHub API /issues возвращает и PR — фильтруем по наличию pull_request поля lines = [ f"#{i['number']} [{', '.join(l['name'] for l in i.get('labels', [])) or '-'}] {i['title']}" for i in issues if "pull_request" not in i ] return {"content": [{"type": "text", "text": "\n".join(lines) or "No issues found"}]} @tool("save_analysis", "Save analysis result to long-term memory", {"repo": str, "analysis": str, "tags": str}) async def save_analysis(args: dict[str, Any]) -> dict[str, Any]: mem_file = MEMORY_DIR / "analyses.json" analyses = json.loads(mem_file.read_text()) if mem_file.exists() else [] analyses.append({ "repo": args["repo"], "analysis": args["analysis"], "tags": [t.strip() for t in args.get("tags", "").split(",")], "timestamp": datetime.now().isoformat(), }) mem_file.write_text(json.dumps(analyses, indent=2, ensure_ascii=False)) return {"content": [{"type": "text", "text": f"Analysis of {args['repo']} saved"}]} @tool("search_memory", "Search previous analyses by keyword or repo name", {"query": str}) async def search_memory(args: dict[str, Any]) -> dict[str, Any]: mem_file = MEMORY_DIR / "analyses.json" if not mem_file.exists(): return {"content": [{"type": "text", "text": "Memory is empty"}]} analyses = json.loads(mem_file.read_text()) q = args["query"].lower() matches = [a for a in analyses if q in a["repo"].lower() or q in a["analysis"].lower()] if not matches: return {"content": [{"type": "text", "text": f"Nothing found for '{args['query']}'"}]} results = [f"[{m['timestamp'][:10]}] {m['repo']}:\n{m['analysis'][:500]}" for m in matches[-5:]] return {"content": [{"type": "text", "text": "\n\n---\n\n".join(results)}]} # ── Hooks ─────────────────────────────────────────────── call_counter: dict[str, Any] = {"total": 0, "by_tool": {}} async def pre_tool_guard(input_data: dict[str, Any], tool_use_id: str | None, context: HookContext) -> dict[str, Any]: tool_name = input_data.get("tool_name", "unknown") call_counter["total"] += 1 call_counter["by_tool"][tool_name] = call_counter["by_tool"].get(tool_name, 0) + 1 logger.info(f"[{call_counter['total']}] → {tool_name}") if call_counter["total"] > 50: return { "hookSpecificOutput": { "hookEventName": "PreToolUse", "permissionDecision": "deny", "permissionDecisionReason": "Tool call limit (50) exceeded. Generate report from collected data.", } } return {} async def post_tool_logger(input_data: dict[str, Any], tool_use_id: str | None, context: HookContext) -> dict[str, Any]: tool_name = input_data.get("tool_name", "unknown") result = str(input_data.get("tool_response", "")) # поле tool_response в PostToolUse if "error" in result.lower() or "404" in result: logger.warning(f"Tool {tool_name} error: {result[:200]}") return {} # ── Agent ─────────────────────────────────────────────── SYSTEM_PROMPT = """Ты — senior software engineer, специализирующийся на code review и архитектурном анализе. Задача: глубокий анализ репозитория {owner}/{repo}. Порядок действий (адаптируй по ситуации): 1. Метаданные репозитория и языки 2. Дерево файлов — пойми структуру 3. Ключевые конфиги: README, package.json / pyproject.toml / Cargo.toml, Dockerfile, CI 4. Последние коммиты и контрибьюторы 5. Открытые issues Критерии отчёта: - Стек: конкретные версии, не "возможно X" - Архитектура: модули, слои, зависимости - Качество: тесты, CI/CD, линтер, типизация — факты, не предположения - Активность: цифры, не "активно развивается" - Рекомендации: конкретные, actionable После анализа — сохрани результат в memory (save_analysis). Перед анализом — проверь memory (search_memory), был ли анализ ранее. Формат: Markdown.""" async def analyze(owner: str, repo: str) -> str: server = create_sdk_mcp_server( name="github-analyzer", version="3.0.0", tools=[ get_repo_info, get_repo_tree, get_file_content, get_recent_commits, get_languages, get_contributors, get_issues, save_analysis, search_memory, ], ) call_counter["total"] = 0 call_counter["by_tool"] = {} options = ClaudeAgentOptions( mcp_servers={"gh": server}, allowed_tools=[ # При проблемах с wildcard — заменить на явный список инструментов "mcp__gh__get_repo_info", "mcp__gh__get_repo_tree", "mcp__gh__get_file_content", "mcp__gh__get_recent_commits", "mcp__gh__get_languages", "mcp__gh__get_contributors", "mcp__gh__get_issues", "mcp__gh__save_analysis", "mcp__gh__search_memory", ], hooks={ "PreToolUse": [HookMatcher(hooks=[pre_tool_guard])], "PostToolUse": [HookMatcher(hooks=[post_tool_logger])], }, ) try: result = "" async for message in query( prompt=SYSTEM_PROMPT.format(owner=owner, repo=repo), options=options, ): if isinstance(message, ResultMessage) and message.subtype == "success": result = message.result logger.info(f"Done. Tools called: {call_counter['total']}") return result or "Анализ не завершён" except Exception as e: logger.error(f"Agent error: {e}") return f"Ошибка анализа: {e}" # ── Entry point ───────────────────────────────────────── async def main(): import sys if len(sys.argv) < 2: print("Usage: python agent.py owner/repo") sys.exit(1) owner, repo = sys.argv[1].split("/") report = await analyze(owner, repo) print(report) if __name__ == "__main__": asyncio.run(main()) ``` ### Запуск ```bash # Установка зависимостей (актуальные версии на июнь 2026) pip install "claude-agent-sdk>=0.2.110" httpx tenacity # Переменные окружения export GITHUB_TOKEN="ghp_..." export ANTHROPIC_API_KEY="sk-ant-..." # Запуск python agent.py anthropics/claude-code ``` ## Что дальше Агент из этого туториала — рабочая основа. Направления развития: **Другие фреймворки.** Claude Agent SDK — правильный старт для Claude-центричных агентов. Для других сценариев: - **LangGraph 1.2** — stateful агенты с durable execution, HITL и persistence; лучший выбор для сложных graph-воркфлоу - **OpenAI Agents SDK 0.17** — provider-agnostic, поддерживает 100+ LLM, handoffs и agents-as-tools из коробки; нативный sandboxing для изолированного выполнения - **Pydantic AI 2.0** — type-safe агенты с structured outputs и встроенной поддержкой MCP; рекомендуется для Python-проектов с акцентом на типизацию - **Google ADK 2.0** — open-source code-first фреймворк от Google; официально поддерживает Python и TypeScript; Task API и graph workflow runtime (в 2.0 исполнитель переведён на граф-движок) - **Microsoft Agent Framework 1.9** — successor AutoGen + Semantic Kernel; unified path для Microsoft-экосистемы; нативная интеграция MCP и A2A протоколов - **Agno** (бывший Phidata) — лёгкий high-performance runtime для мультиагентных систем (agent teams, swarms); быстрее большинства фреймворков по latency; хорош для агентов реального времени **Больше источников данных.** Подключите npm/PyPI для анализа зависимостей на уязвимости. Добавьте SonarQube API для статического анализа. Интегрируйте Git blame для определения авторства проблемных участков. **Vector store для memory.** Замените JSON на pgvector или Qdrant. Семантический поиск по прошлым анализам: «найди репозитории с похожей архитектурой» вместо точного совпадения по строке. Подробнее — в [RAG система: как построить и когда нужна](/ru/blog/rag-system-guide/). **Scheduled analysis.** Запускайте агента по расписанию для отслеживания изменений в ключевых зависимостях. Cron + агент = автоматический мониторинг open-source проектов. **Оценка качества.** Добавьте LLM-as-Judge для автоматической оценки отчётов. Пусть второй LLM проверяет, что отчёт содержательный, конкретный и не содержит галлюцинаций. Конкретные eval-сетапы — в статье [Тестирование AI агентов: как убедиться что агент работает](/ru/blog/ai-agent-testing-evaluation/). Фундамент один: LLM + tools + loop. Всё остальное — инструменты, memory, мультиагентность, guardrails — слои поверх этого ядра. Начните с 20 строк, добавляйте сложность по мере необходимости.