How to Build an AI Agent from Scratch: a Step-by-Step Tutorial

An AI agent isn’t a research-paper abstraction. It’s a program you can write in an evening and run in production within a week. The core of any agent is a loop: the LLM gets a task, calls tools, reads the results, and decides what to do next. Memory, guardrails, and multi-agent setups are just extensions on top of that loop.

In this tutorial we’ll build a real agent for analyzing GitHub repositories. We’ll start with 20 lines of code and work up to a multi-agent system with MCP integration, error handling, and deployment. Every step is working Python code built on the Claude Agent SDK.

What we’re building

An agent that analyzes GitHub repositories. Input: a repo URL. Output: a structured report covering:

• Project architecture (file structure, main components)
• Tech stack (languages, frameworks, dependencies)
• Code quality (tests, CI/CD, linters)
• Activity (commit frequency, open issues, contributors)
• Recommendations (what to improve, potential problems)

Why this use case: the GitHub API is accessible and well-documented. The task needs several tools (API calls, file reading, analysis). The result is practically useful — for code review, due diligence, or picking open-source dependencies.

Agent architecture

Every AI agent has three components tied together by a loop:

┌───────────────────────────────────────────┐
│              USER INPUT                    │
│   "Analyze github.com/org/repo"           │
└─────────────────┬─────────────────────────┘
                  ▼
┌───────────────────────────────────────────┐
│            AGENT LOOP                      │
│                                            │
│  ┌────────┐  ┌────────┐  ┌────────────┐  │
│  │ Think  │→ │  Act   │→ │  Observe   │  │
│  └────────┘  └────────┘  └────────────┘  │
│       ↑                        │          │
│       └────────────────────────┘          │
│                                            │
│  The LLM decides:                         │
│  - which tool to call                     │
│  - with what parameters                   │
│  - whether it has enough data to answer   │
└─────────────────┬─────────────────────────┘
                  │
        ┌─────────┼─────────┐
        ▼         ▼         ▼
   ┌─────────┐ ┌──────┐ ┌──────────┐
   │GitHub   │ │File  │ │ Shell    │
   │  API    │ │System│ │ Commands │
   └─────────┘ └──────┘ └──────────┘
                  │
                  ▼
┌───────────────────────────────────────────┐
│              MEMORY                        │
│  Short-term: current session context      │
│  Long-term: results from past analyses    │
└───────────────────────────────────────────┘
                  │
                  ▼
┌───────────────────────────────────────────┐
│         STRUCTURED REPORT                  │
│  Architecture, stack, quality, activity   │
└───────────────────────────────────────────┘

LLM — the agent’s brain. Receives the task, breaks it into steps, picks tools. Claude Sonnet 4.6 or Opus 4.8, depending on task complexity.

Tools — the agent’s hands. Functions for interacting with the outside world: HTTP requests to the GitHub API, reading files, running commands.

Memory — the agent’s memory. Context of the current conversation (short-term) and results from past analyses (long-term).

Agent Loop — the control loop. The LLM thinks → calls a tool → gets a result → thinks again. The cycle repeats until the agent decides the task is done.

The difference from a regular API call: the agent itself decides how many iterations it needs and which tools to call. Instead of our own if/else, the LLM plans and adapts.

Step 1: A minimal agent — 20 lines

Let’s start with the simplest possible agent. No custom tools — just the Claude Agent SDK and its built-in capabilities.

Installation

pip install claude-agent-sdk

The Claude Agent SDK (current version: 0.2.110) is a wrapper around the Claude Code CLI. The SDK ships with a bundled CLI binary, so you don’t need to install the claude CLI separately. You only need an Anthropic API key and Python 3.10+. The package is in alpha status — the API is stable, but breaking changes between minor versions are possible.

Minimal agent

import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions, ResultMessage

async def analyze_repo(repo_url: str) -> str:
    """Minimal agent: sends a prompt and returns the result."""
    options = ClaudeAgentOptions(
        allowed_tools=["Read", "Bash"],  # Allow file reading and shell access
    )

    async for message in query(
        prompt=f"""Analyze the GitHub repository: {repo_url}

        Use git clone to clone it, then explore the project structure.
        Determine the tech stack, architecture, and whether it has tests and CI/CD.
        Produce a structured report.""",
        options=options,
    ):
        if isinstance(message, ResultMessage) and message.subtype == "success":
            return message.result

    return "Analysis did not complete"

# Run it
result = asyncio.run(analyze_repo("https://github.com/anthropics/claude-code"))
print(result)

20 lines. The agent clones the repository, reads files, analyzes the structure, and generates a report. Claude decides on its own which files to read and in what order.

What’s happening under the hood

1. query() launches the Claude Code CLI with the given prompt
2. Claude receives the task and the list of allowed tools (Read, Bash)
3. The agent loop kicks in: Claude thinks → calls Bash(git clone ...) → reads files → thinks → calls Read(package.json) → analyzes → …
4. Once Claude decides it has gathered enough information, it generates a final answer
5. A ResultMessage with subtype == "success" contains the final report

The problem with this approach: the agent uses Bash for everything. There’s no structured access to the GitHub API — only curl through the shell. No typing, no control over requests. We’ll fix that in the next step.

Step 2: Adding tools

Custom tools give the agent structured access to external services. Instead of “run curl and parse the JSON,” you get a typed function with a description that the LLM calls by name.

The @tool decorator

The Claude Agent SDK uses the @tool decorator to define tools. Three arguments: name, description (the LLM reads this when picking a tool), and parameter schema.

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]:
    """Fetch repository metadata."""
    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]:
    """Fetch the repository's file tree."""
    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']
    # Cap at 200 files so we don't blow up the context
    if len(tree) > 200:
        tree = tree[:200] + [f"... and {len(tree) - 200} more files"]

    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]:
    """Fetch a file's content from the repository."""
    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')

    # Cap the size so it doesn't eat the whole context
    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]:
    """Fetch recent commits."""
    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]  # First line only
        lines.append(f"[{date}] {author}: {msg}")

    return {"content": [{"type": "text", "text": "\n".join(lines)}]}

Assembling an agent with tools

from claude_agent_sdk import create_sdk_mcp_server, ClaudeSDKClient, ClaudeAgentOptions

# Create an in-process MCP server with our tools
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=[
        # Listed explicitly — the wildcard mcp__github__* isn't reliably stable (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"""Analyze the repository {owner}/{repo}.

Steps:
1. Get repository metadata (get_repo_info)
2. Get the file tree (get_repo_tree)
3. Read key files: README, package.json / pyproject.toml / Cargo.toml, Dockerfile, CI configs
4. Get the last 20 commits (get_recent_commits)
5. Produce a structured report

Report format:
## Overview
## Tech Stack
## Architecture
## Code Quality
## Development Activity
## Recommendations""")

        result_parts = []
        async for message in client.receive_response():
            if isinstance(message, ResultMessage) and message.subtype == "success":
                return message.result

    return "Analysis did not complete"

Why an in-process MCP server

The Claude Agent SDK creates an MCP server directly inside your Python process. Tools are just regular async functions. Advantages over external servers:

• No subprocess — everything runs in one process, no IPC overhead
• Easier debugging — a standard Python debugger, breakpoints, logging
• Single deployment — one container instead of several processes
• Typing — Python type hints, IDE autocomplete for parameters

The tool description ("Get repository metadata: stars, forks, language, description") is the main lever. The LLM picks a tool based on its description: vague wording produces misses, specific wording produces hits.

Step 3: Adding memory

The agent from the previous step is stateless — every run starts from zero. If a user has analyzed 10 repositories, the agent doesn’t remember any of them. Memory solves two problems:

1. Short-term — context of the current session. The Claude Agent SDK handles this automatically through ClaudeSDKClient.
2. Long-term — results from past analyses, learned patterns, cache.

Short-term memory: an interactive session

ClaudeSDKClient supports a bidirectional conversation. Every query() within a single session sees the context of previous messages:

async def interactive_analysis():
    async with ClaudeSDKClient(options=options) as client:
        # First request — full analysis
        await client.query("Analyze the repository anthropics/claude-code")
        async for msg in client.receive_response():
            if isinstance(msg, ResultMessage) and msg.subtype == "success":
                print(msg.result)

        # Second request — the agent remembers the first one's context
        await client.query("Compare its architecture to what's typical for CLI tools")
        async for msg in client.receive_response():
            if isinstance(msg, ResultMessage) and msg.subtype == "success":
                print(msg.result)

        # Third request — going deeper
        await client.query("Which specific files handle the agent loop?")
        async for msg in client.receive_response():
            if isinstance(msg, ResultMessage) and msg.subtype == "success":
                print(msg.result)

Within a single async with ClaudeSDKClient(...) block, context accumulates. The agent remembers earlier requests, tool call results, and intermediate reasoning.

Long-term memory: saving results

For long-term memory, we add tools for writing to and reading from storage. The simplest option — JSON files:

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]:
    """Save an analysis result."""
    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"Analysis of {args['repo']} saved to 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]:
    """Search past analyses."""
    memory_file = MEMORY_DIR / "analyses.json"

    if not memory_file.exists():
        return {"content": [{"type": "text", "text": "Memory is empty — no analyses yet"}]}

    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"Nothing found for query '{args['query']}'"}]}

    results = []
    for m in matches[-5:]:  # Last 5 matches
        results.append(f"[{m['timestamp'][:10]}] {m['repo']}:\n{m['analysis'][:500]}")

    return {"content": [{"type": "text", "text": "\n\n---\n\n".join(results)}]}

Updating the server

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,
    ],
)

Now the agent can save analyses and reference them later. We extend the prompt with an instruction:

After finishing an analysis, save the result via save_analysis.
Before analyzing a new repository, check search_memory —
this repository may have been analyzed before.

In production, replace JSON files with SQLite, PostgreSQL, or a vector store (for semantic search over past analyses). But the principle stays the same: tools for writing and reading.

Step 4: Multi-tool agent — the agent picks its own tools

So far we’ve given the agent explicit instructions: “first get_repo_info, then get_repo_tree.” A real agent should decide on its own which tools to call and in what order.

Expanding the toolset

Let’s add tools covering different aspects of the analysis:

@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]:
    """Fetch repository issues (excluding PRs — the GitHub API returns both types under /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's /issues endpoint returns PRs too — filter by the pull_request field
        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]:
    """Fetch contributors."""
    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]:
    """Fetch language breakdown."""
    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]:
    """Check for CI/CD configuration."""
    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)}]}

An agent that chooses on its own

The secret is in the prompt. Instead of listing steps, we describe the goal and quality criteria:

ANALYZER_PROMPT = """You are a senior engineer specializing in code review and architectural analysis.

Task: run a deep analysis of the repository {owner}/{repo}.

You have tools for working with the GitHub API. Use them at your discretion —
you decide what data to collect and in what order.

Criteria for a good analysis:
- Stack identified precisely (not "probably Python" but "Python 3.12, FastAPI 0.110, SQLAlchemy 2.0")
- Architecture described with concrete modules and how they relate
- Code quality assessed from facts (are there tests? CI? a linter? typing?)
- Activity — not "actively maintained" but "23 commits in the last month from 4 contributors"
- Recommendations that are concrete and actionable

If memory has a previous analysis of this repository, account for it: note what changed.

Format: Markdown with second-level headings."""

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 — see the note below
    )

    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 "Analysis did not complete"

Note on the wildcard: allowed_tools=["mcp__github__*"] is supposed to allow every tool on the server. In practice, wildcard patterns for MCP tools have been unreliable in a number of claude-agent-sdk versions (known issue) — if you hit problems, list the tools explicitly.

The LLM decides the order itself: it might start with get_repo_info, or jump straight to get_repo_tree if it needs the project structure first. It might skip get_issues entirely if the question is purely about architecture. It adapts to the task.

Step 5: Error handling and guardrails

A production agent without error handling is a ticking time bomb. The GitHub API returns 403 when you hit the rate limit. Networks fail. LLMs hallucinate. You need guardrails.

Hooks for control

The Claude Agent SDK supports hooks — functions called before and after every tool invocation:

from claude_agent_sdk import HookMatcher, HookContext
from datetime import datetime
import logging

logger = logging.getLogger("agent")

# Call counter for 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: logging + call limit."""
    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}")

    # Limit: max 50 calls per session
    if call_counter["total"] > 50:
        logger.warning("Tool call limit exceeded")
        return {
            "hookSpecificOutput": {
                "hookEventName": "PreToolUse",
                "permissionDecision": "deny",
                "permissionDecisionReason": "Tool call limit (50) exceeded. Generate a report from the data already collected.",
            }
        }

    # Block dangerous operations
    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"Dangerous command blocked: {command}",
                }
            }

    return {}


async def post_tool_logger(
    input_data: dict[str, Any], tool_use_id: str | None, context: HookContext
) -> dict[str, Any]:
    """Log results."""
    tool_name = input_data.get("tool_name", "unknown")
    # Check for errors in the result
    result = input_data.get("tool_response", "")  # this field is named tool_response in 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 and timeouts in tools

We build GitHub API error handling directly into the tools:

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 request to the GitHub API with retry and rate-limit handling."""
    async with httpx.AsyncClient(timeout=30.0) as client:
        resp = await client.get(url, headers=HEADERS, params=params)

        # Rate limit — wait and retry
        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))  # Never wait longer than a minute
                raise httpx.HTTPStatusError("Rate limit", request=resp.request, response=resp)

        resp.raise_for_status()
        return resp.json()

Assembling the production configuration

options = ClaudeAgentOptions(
    mcp_servers={"github": github_server},
    allowed_tools=["mcp__github__*"],  # or an explicit list if the wildcard misbehaves
    hooks={
        "PreToolUse": [HookMatcher(hooks=[pre_tool_guard])],
        "PostToolUse": [HookMatcher(hooks=[post_tool_logger])],
    },
)

Three layers of protection:

1. Pre-tool hooks — block dangerous calls and cap the count
2. Retry in tools — handle transient network and API errors
3. Post-tool hooks — log errors for later analysis

A fourth layer is the prompt. Add this to your system prompt:

If a tool returns an error, try a different approach.
Don't repeat the same call more than twice.
If you can't get the data, say so in the report —
don't make information up.

Step 6: MCP integration — connecting ready-made servers

Up to this point we’ve been writing tools by hand. MCP (Model Context Protocol) is an open standard: connect a ready-made server and get thousands of tools without writing any code.

What MCP is

MCP is an open protocol for connecting AI agents to tools, data, and context. One protocol that works across Claude, Cursor, VS Code, ChatGPT, and Gemini. Current spec version: 2025-11-25. It’s the de facto standard for tool/context integration, supported by every major AI platform and developed under the governance of the Agentic AI Foundation (Linux Foundation, December 2025 — with Anthropic, Google, Microsoft, OpenAI, and AWS among the founding members). For agent-to-agent communication there’s a separate protocol, A2A — MCP and A2A solve different problems and complement each other. For more detail, see MCP Servers: What They Are, Why You Need Them, and How to Connect One.

Connecting the GitHub MCP server

Instead of our custom tools, we can use the official GitHub MCP server. API calls, pagination, authorization, and error handling are already built in.

Important: the @modelcontextprotocol/server-github package is deprecated (since April 2025). GitHub ships an official server instead — github/github-mcp-server. Two ways to connect it:

# Option 1 — Docker (recommended)
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

# Option 2 — remote HTTP server (simplest)
claude mcp add-json github '{"type":"http","url":"https://api.githubcopilot.com/mcp","headers":{"Authorization":"Bearer YOUR_GITHUB_PAT"}}'

To use it from the Agent SDK via 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"""Analyze the repository {owner}/{repo}.

Use the available GitHub tools to gather data:
- Read key files (README, config files, package.json, etc.)
- Look at the repository structure
- Review recent commits and issues

Produce a detailed report: stack, architecture, quality, activity, recommendations.""",
        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 "Analysis did not complete"

Combining custom and MCP tools

The real power is in combining both. Custom tools for business logic, MCP servers for standard integrations:

# Custom server — our analytics and memory
custom_server = create_sdk_mcp_server(
    name="analyzer",
    version="2.0.0",
    tools=[save_analysis, search_memory],
)

options = ClaudeAgentOptions(
    mcp_servers={
        # MCP server for GitHub (external process via 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"]},
        },
        # Custom server (in-process)
        "analyzer": custom_server,
    },
    allowed_tools=[
        "mcp__github__*",          # All GitHub tools (wildcard, see the note above)
        "mcp__analyzer__*",        # All our own tools
    ],
)

Both servers run in parallel. The GitHub MCP server starts as a subprocess via Docker (the docker run command). The custom analyzer server runs in-process. The agent calls tools from both servers transparently.

When to write your own tools vs. use MCP

An MCP server fits when:

• A ready-made server already exists for the service you need (GitHub, Slack, PostgreSQL, Notion)
• You need a standard integration with no custom logic
• The tool will be reused across different projects

Custom tools fit when:

• You need specific business logic (our save_analysis)
• The tool talks to your company’s internal systems
• You need full control over the input/output format and error handling
• Performance matters (in-process is faster than subprocess)

For a deeper look at building and running your own MCP servers in production, see MCP Servers in Production: Building Custom Servers.

Step 7: A multi-agent system — supervisor + workers

A single agent handles a single task well. For a comprehensive analysis, assemble several specialized agents — see Multi-Agent Architecture: Patterns and Trade-offs for a broader treatment of these designs.

Architecture: supervisor + workers

┌───────────────────────────────────────────┐
│             SUPERVISOR AGENT               │
│  Plans, distributes, aggregates           │
└──────┬──────────┬──────────┬──────────────┘
       ▼          ▼          ▼
┌──────────┐ ┌──────────┐ ┌──────────┐
│ ARCH     │ │ QUALITY  │ │ ACTIVITY │
│ WORKER   │ │ WORKER   │ │ WORKER   │
│          │ │          │ │          │
│File      │ │Tests,    │ │Commits,  │
│structure,│ │CI/CD,    │ │issues,   │
│modules,  │ │linters,  │ │contribu- │
│patterns  │ │typing    │ │tors      │
└──────────┘ └──────────┘ └──────────┘

Supervisor — the main agent. Receives the task, breaks it into subtasks, dispatches workers, aggregates results into one report.

Workers — specialized agents. Each focuses on one aspect of the analysis and has access only to the tools it needs.

Implementation with CrewAI

CrewAI is a framework for multi-agent systems (current stable line as of June 2026 — 1.14.x). It defines agents, tasks, and coordination strategy. Its architecture rests on two primitives: Crews (teams of agents for tasks) and Flows (managed pipelines with condition/branching). Unlike LangChain-dependent solutions, it works without external orchestration dependencies:

from crewai import Agent, Task, Crew, Process
from crewai.tools import tool as crewai_tool
import httpx

# Tools for CrewAI — wrappers around the 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()
    )

# Define the agents
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,
)

# Define tasks with async_execution=True to run in parallel
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,
    )
    # Final task consolidates the results of the three parallel tasks
    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]

# Run it
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 on tasks gives us parallelism
        verbose=True,
    )

    result = crew.kickoff()
    return str(result)

A supervisor built on the Claude Agent SDK

An alternative approach: the supervisor is a Claude agent, and workers are called as tools:

@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: architecture analysis."""
    arch_server = create_sdk_mcp_server(
        name="arch-tools",
        tools=[get_repo_tree, get_file_content],
    )

    async for msg in query(
        prompt=f"""You are a software architect. Analyze the architecture of {args['owner']}/{args['repo']}.
        Determine: module structure, design patterns, dependency graph.
        Reply with a structured report, 300-500 words.""",
        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"}]}


# The supervisor has access to worker tools
supervisor_server = create_sdk_mcp_server(
    name="supervisor",
    tools=[analyze_architecture, analyze_quality, analyze_activity],  # workers
)

The supervisor calls workers as tools. Each worker is a separate query() with its own tools and prompt. The supervisor aggregates the results and produces the final report.

When you need multi-agent setups

You need it when:

• The task decomposes into independent subtasks (parallel analysis)
• Different subtasks need different prompts and tools
• You need scalability — adding an analysis dimension means adding a worker

You don’t need it when:

• One agent can solve the task in a reasonable amount of time
• Subtasks are tightly coupled and need shared context
• The budget is tight (every worker means additional LLM calls)

Step 8: Deployment — going to production

The agent works locally. For production, you need Docker, monitoring, and an API wrapper.

Docker

FROM python:3.12-slim

# Node.js is only needed if you use external MCP servers via npx/npm
# claude-agent-sdk ships its own CLI binary — no separate install needed
# docker.io is only needed if the agent spawns child containers (e.g. the GitHub MCP server via Docker).
# WARNING: this requires mounting /var/run/docker.sock into the container,
# which gives the process inside it root-equivalent access to the host. Do not use
# in publicly reachable environments without additional isolation (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 . .

# Directory for memory
RUN mkdir -p /app/agent_memory

EXPOSE 8000

CMD ["uvicorn", "server:app", "--host", "0.0.0.0", "--port", "8000"]

API wrapper with 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()

    # Reset the call counter
    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"}

Monitoring

Three metrics you need to track:

1. Cost. Every query() is an LLM call. Track tokens.

import os

# Set limits via environment variables
MAX_TOKENS_PER_REQUEST = int(os.environ.get("MAX_TOKENS", "100000"))
MAX_TOOL_CALLS = int(os.environ.get("MAX_TOOL_CALLS", "50"))

2. Latency. A typical repository analysis takes 30-120 seconds. If it takes longer than 5 minutes, something’s wrong.

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"Analysis of {owner}/{repo} exceeded the {timeout}s timeout"

3. Quality. Log requests and responses. Periodically check that reports are substantive, not hallucinated.

Production project structure

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

Full code — final version

The assembled agent — every step in one file. Copy it, install dependencies, run it.

"""
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"... and {len(tree) - 200} more files"]
    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's /issues endpoint returns PRs too — filter by the pull_request field
    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", ""))  # this field is named tool_response in PostToolUse
    if "error" in result.lower() or "404" in result:
        logger.warning(f"Tool {tool_name} error: {result[:200]}")
    return {}

# ── Agent ───────────────────────────────────────────────

SYSTEM_PROMPT = """You are a senior software engineer specializing in code review and architectural analysis.

Task: run a deep analysis of the repository {owner}/{repo}.

Suggested order (adapt as needed):
1. Repository metadata and languages
2. File tree — understand the structure
3. Key configs: README, package.json / pyproject.toml / Cargo.toml, Dockerfile, CI
4. Recent commits and contributors
5. Open issues

Report criteria:
- Stack: concrete versions, not "probably X"
- Architecture: modules, layers, dependencies
- Quality: tests, CI/CD, linter, typing — facts, not assumptions
- Activity: numbers, not "actively maintained"
- Recommendations: concrete, actionable

After the analysis, save the result to memory (save_analysis).
Before starting, check memory (search_memory) for a prior analysis.

Format: 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=[
            # If the wildcard misbehaves, replace with an explicit tool list
            "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 "Analysis did not complete"

    except Exception as e:
        logger.error(f"Agent error: {e}")
        return f"Analysis error: {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())

Running it

# Install dependencies (current versions as of June 2026)
pip install "claude-agent-sdk>=0.2.110" httpx tenacity

# Environment variables
export GITHUB_TOKEN="ghp_..."
export ANTHROPIC_API_KEY="sk-ant-..."

# Run
python agent.py anthropics/claude-code

What’s next

The agent from this tutorial is a working foundation. Directions to take it further:

Other frameworks. The Claude Agent SDK is the right starting point for Claude-centric agents. For other scenarios:

• LangGraph 1.2 — stateful agents with durable execution, HITL, and persistence; the best choice for complex graph-based workflows
• OpenAI Agents SDK 0.17 — provider-agnostic, supports 100+ LLMs, handoffs and agents-as-tools out of the box; native sandboxing for isolated execution
• Pydantic AI 2.0 — type-safe agents with structured outputs and built-in MCP support; recommended for Python projects that care about typing
• Google ADK 2.0 — open-source, code-first framework from Google; officially supports Python and TypeScript; Task API and a graph workflow runtime (ADK 2.0 moved the executor to a graph engine)
• Microsoft Agent Framework 1.9 — successor to AutoGen and Semantic Kernel; a unified path for the Microsoft ecosystem; native MCP and A2A protocol integration
• Agno (formerly Phidata) — a lightweight, high-performance runtime for multi-agent systems (agent teams, swarms); faster than most frameworks on latency; a good fit for real-time agents

More data sources. Connect npm/PyPI to check dependencies for vulnerabilities. Add the SonarQube API for static analysis. Integrate git blame to attribute problematic sections of code.

Vector store for memory. Replace JSON with pgvector or Qdrant. Semantic search over past analyses: “find repositories with a similar architecture” instead of exact string matches. For more detail, see RAG Systems: How to Build One and When You Need It.

Scheduled analysis. Run the agent on a schedule to track changes in key dependencies. Cron + agent equals automatic monitoring of open-source projects.

Quality evaluation. Add an LLM-as-Judge step to automatically grade reports. Have a second LLM verify that the report is substantive, specific, and free of hallucinations — see AI Agent Testing and Evaluation for concrete evaluation setups.

The foundation is always the same: LLM + tools + loop. Everything else — tools, memory, multi-agent setups, guardrails — is a layer on top of that core. Start with 20 lines, and add complexity as you actually need it.