export ANTHROPIC_API_KEY=your-key-here

1from claude_agent_sdk import Agent, Tool
2 
3def query_database(sql: str) -> str:
4    """Execute a read-only SQL query and return results as CSV."""
5    import sqlite3
6    conn = sqlite3.connect("app.db")
7    cursor = conn.execute(sql)
8    rows = cursor.fetchall()
9    headers = [desc[0] for desc in cursor.description]
10    conn.close()
11    lines = [",".join(headers)]
12    for row in rows:
13        lines.append(",".join(str(v) for v in row))
14    return "\n".join(lines)
15 
16db_tool = Tool(
17    name="query_database",
18    description="Run a read-only SQL query against the application database",
19    input_schema={
20        "type": "object",
21        "properties": {
22            "sql": {"type": "string", "description": "SQL SELECT query to execute"}
23        },
24        "required": ["sql"]
25    },
26    handler=query_database,
27)
28 
29agent = Agent(custom_tools=[db_tool])
30result = agent.run("How many users signed up in the last 7 days?")

pip install claude-agent-sdk python-dotenv

1import os
2import json
3from dataclasses import dataclass
4from claude_agent_sdk import Agent, Hook, HookAction
5 
6@dataclass
7class ReviewComment:
8    file: str
9    line: int
10    severity: str  # "critical", "warning", "suggestion"
11    message: str
12 
13class CodeReviewAgent:
14    """An agent that reviews git diffs and produces structured feedback."""
15 
16    SYSTEM_PROMPT = """You are a senior code reviewer. You will be given a git diff
17to review. Your job:
18 
191. Use Bash to run `git diff` and get the changes
202. Use Read and Grep to understand the context around each change
213. Identify issues in these categories:
22   - Bugs or logic errors (critical)
23   - Performance problems (warning)
24   - Style and readability (suggestion)
25   - Missing error handling (warning)
26   - Security concerns (critical)
27 
284. Produce your review as a JSON array of objects with these fields:
29   - file: the file path
30   - line: the approximate line number
31   - severity: "critical", "warning", or "suggestion"
32   - message: a clear explanation of the issue and how to fix it
33 
34Return ONLY the JSON array, no other text."""
35 
36    def __init__(self):
37        self.agent = Agent(
38            system_prompt=self.SYSTEM_PROMPT,
39            allowed_tools=["Read", "Grep", "Glob", "Bash"],
40            max_turns=30,
41            hooks=[
42                Hook(event="PreToolUse", handler=self._restrict_bash),
43            ],
44        )
45 
46    def _restrict_bash(self, tool_name: str, tool_input: dict) -> HookAction:
47        """Only allow read-only bash commands."""
48        if tool_name != "Bash":
49            return HookAction.ALLOW
50 
51        command = tool_input.get("command", "")
52        allowed_prefixes = ["git diff", "git log", "git show", "git status", "cat ", "wc "]
53 
54        if any(command.strip().startswith(p) for p in allowed_prefixes):
55            return HookAction.ALLOW
56 
57        return HookAction.BLOCK(
58            reason=f"Only git and read-only commands are allowed. Got: {command}"
59        )
60 
61    def review(self, diff_target: str = "HEAD~1") -> list[ReviewComment]:
62        """Review changes in the given diff range."""
63        result = self.agent.run(
64            f"Review the code changes in `git diff {diff_target}`. "
65            f"Focus on bugs, security issues, and maintainability."
66        )
67 
68        # Parse structured output
69        try:
70            comments_data = json.loads(result.output)
71            return [ReviewComment(**c) for c in comments_data]
72        except (json.JSONDecodeError, TypeError) as e:
73            print(f"Warning: Could not parse structured output: {e}")
74            print(f"Raw output:\n{result.output}")
75            return []
76 
77    def format_review(self, comments: list[ReviewComment]) -> str:
78        """Format review comments grouped by severity."""
79        if not comments:
80            return "No issues found."
81        lines = [f"## Code Review — {len(comments)} issue(s)\n"]
82        for sev in ["critical", "warning", "suggestion"]:
83            group = [c for c in comments if c.severity == sev]
84            if group:
85                lines.append(f"\n### [{sev.upper()}] ({len(group)})\n")
86                for c in group:
87                    lines.append(f"**{c.file}:{c.line}** — {c.message}\n")
88        return "\n".join(lines)
89 
90if __name__ == "__main__":
91    reviewer = CodeReviewAgent()
92    comments = reviewer.review("HEAD~1")
93    print(reviewer.format_review(comments))

1## Code Review — 3 issue(s) found
2 
3### [CRITICAL] (1)
4 
5**src/auth.py:42**
6  SQL query built with string concatenation is vulnerable to injection.
7  Use parameterized queries: `cursor.execute("SELECT * FROM users WHERE id = ?", (user_id,))`
8 
9### [WARNING] (1)
10 
11**src/api/routes.py:118**
12  The `except Exception` clause silently swallows all errors.
13  At minimum, log the exception. Consider catching specific exception types.
14 
15### [SUGGESTION] (1)
16 
17**src/utils/helpers.py:7**
18  Function `do_stuff` is 85 lines long with 6 levels of nesting.
19  Extract the inner logic into well-named helper functions.

1from fastapi import FastAPI, BackgroundTasks
2from claude_agent_sdk import Agent
3 
4app = FastAPI()
5 
6@app.post("/review")
7async def review_endpoint(repo_url: str, diff_ref: str, background_tasks: BackgroundTasks):
8    """Trigger a code review. Returns immediately; posts results when done."""
9    background_tasks.add_task(run_review, repo_url, diff_ref)
10    return {"status": "queued"}
11 
12async def run_review(repo_url: str, diff_ref: str):
13    agent = Agent(system_prompt="You are a code reviewer...", max_turns=30)
14    result = agent.run(f"Clone {repo_url} and review changes in {diff_ref}")
15    post_results(result.output)  # Post to GitHub, Slack, etc.

1from claude_agent_sdk import Agent, AgentError, RateLimitError, ContextOverflowError
2import time
3 
4def run_with_resilience(task: str, max_retries: int = 3) -> str:
5    for attempt in range(max_retries):
6        try:
7            agent = Agent(max_turns=50)
8            return agent.run(task).output
9        except RateLimitError:
10            wait = min(2 ** attempt * 10, 120)
11            print(f"Rate limited. Waiting {wait}s... (attempt {attempt + 1})")
12            time.sleep(wait)
13        except ContextOverflowError:
14            return Agent(max_turns=50, compaction_strategy="summarize").run(task).output
15        except AgentError as e:
16            if attempt == max_retries - 1:
17                raise
18    raise RuntimeError("Max retries exceeded")

1# Spans emitted by the SDK:
2#   agent.run       — total duration, final status
3#   agent.turn      — each reasoning + tool cycle
4#   agent.tool_call — tool name, input/output size, duration
5#   agent.llm_call  — model, tokens in/out, latency