Как создать AI агента с нуля: пошаговый туториал
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 и встроенные возможности.
Установка
pip install claude-agent-sdk
Claude Agent SDK (актуальная версия: 0.2.110) — обёртка над Claude Code CLI. SDK поставляется с встроенным CLI-бинарём, отдельная установка claude CLI не нужна. Требуется только API-ключ Anthropic и Python 3.10+. Пакет находится в статусе alpha — API стабилен, но возможны breaking changes между minor-версиями.
Минимальный агент
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 сам решает, какие файлы читать и в каком порядке.
Что происходит под капотом
query()запускает Claude Code CLI с заданным промптом- Claude получает задачу и список разрешённых инструментов (
Read,Bash) - Включается agent loop: Claude думает → вызывает
Bash(git clone ...)→ читает файлы → думает → вызываетRead(package.json)→ анализирует → … - Когда Claude решает, что собрал достаточно информации — генерирует финальный ответ
ResultMessageсsubtype == "success"содержит итоговый отчёт
Проблема этого подхода: агент использует Bash для всего. Нет структурированного доступа к GitHub API — только curl через shell. Нет типизации, нет контроля над запросами. Исправим в следующем шаге.
Шаг 2: Добавляем инструменты
Кастомные инструменты — способ дать агенту структурированный доступ к внешним сервисам. Вместо «вызови curl и распарси JSON» — типизированная функция с описанием, которую LLM вызывает по имени.
Декоратор @tool
Claude Agent SDK использует декоратор @tool для определения инструментов. Три аргумента: имя, описание (LLM читает его при выборе инструмента), схема параметров.
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)}]}
Собираем агента с инструментами
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 решает две задачи:
- Short-term — контекст текущей сессии. Claude Agent SDK управляет этим автоматически через
ClaudeSDKClient. - Long-term — результаты прошлых анализов, выученные паттерны, кэш.
Short-term memory: интерактивная сессия
ClaudeSDKClient поддерживает bidirectional-разговор. Каждый query() в рамках одной сессии видит контекст предыдущих сообщений:
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-файлы:
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)}]}
Обновляем сервер
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». Настоящий агент должен сам решать, какие инструменты вызывать и в каком порядке.
Расширяем набор инструментов
Добавим инструменты, которые покрывают разные аспекты анализа:
@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)}]}
Агент с автономным выбором
Секрет — в промпте. Не перечисляем шаги, а описываем цель и критерии качества:
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) — при проблемах перечисляйте инструменты явно.
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 — функции, которые вызываются до и после каждого инструмента:
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 встраиваем прямо в инструменты:
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-конфигурацию
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])],
},
)
Три уровня защиты:
- Pre-tool hooks — блокируют опасные вызовы и лимитируют количество
- Retry в инструментах — обрабатывают transient-ошибки сети и API
- Post-tool hooks — логируют ошибки для анализа
Эти же принципы (изоляция, лимиты, логирование) применимы и к собственным MCP-серверам, которые вы разворачиваете отдельно от агента — подробнее в статье MCP production: создание и деплой кастомных серверов.
Четвёртый уровень — промпт. Добавьте в 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 серверы: что это, зачем нужны и как подключить.
Подключаем GitHub MCP-сервер
Вместо наших кастомных инструментов — официальный GitHub MCP-сервер. API-вызовы, пагинация, авторизация, обработка ошибок уже внутри.
Важно: пакет @modelcontextprotocol/server-github устарел (deprecated с апреля 2025). GitHub выпустил официальный сервер — github/github-mcp-server. Два способа подключения:
# Вариант 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:
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-серверы для стандартных интеграций:
# Кастомный сервер — наша аналитика и 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, альтернативные топологии — разобраны подробнее в статье Мультиагентные системы: архитектура и паттерны.
Реализация на CrewAI
CrewAI — фреймворк для мультиагентных систем (актуальная стабильная линейка на июнь 2026 — 1.14.x). Определяет агентов, задачи и стратегию координации. Архитектура строится на двух примитивах: Crews (команды агентов для задач) и Flows (управляемые пайплайны с condition/branching). В отличие от LangChain-зависимых решений, работает без внешних оркестрационных зависимостей:
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 вызываются как инструменты:
@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
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
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. Считайте токены.
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 минут — что-то пошло не так.
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
Полный код — финальная версия
Собранный агент — все шаги в одном файле. Скопируйте, установите зависимости, запустите.
"""
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())
Запуск
# Установка зависимостей (актуальные версии на июнь 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 система: как построить и когда нужна.
Scheduled analysis. Запускайте агента по расписанию для отслеживания изменений в ключевых зависимостях. Cron + агент = автоматический мониторинг open-source проектов.
Оценка качества. Добавьте LLM-as-Judge для автоматической оценки отчётов. Пусть второй LLM проверяет, что отчёт содержательный, конкретный и не содержит галлюцинаций.
Фундамент один: LLM + tools + loop. Всё остальное — инструменты, memory, мультиагентность, guardrails — слои поверх этого ядра. Начните с 20 строк, добавляйте сложность по мере необходимости.