Lesson 7: The Three Layers of Agent Action

{
    "name": "read_file",
    "description": "Read the contents of a file at the given path. Use this when the user asks you to inspect, review, or summarise the contents of a local file.",
    "input_schema": {
        "type": "object",
        "properties": {
            "path": {
                "type": "string",
                "description": "Absolute or relative file path to read"
            },
            "encoding": {
                "type": "string",
                "description": "File encoding, defaults to utf-8",
                "default": "utf-8"
            }
        },
        "required": ["path"]
    }
}

pip install anthropic

"""
tool_demo.py
Demonstrates Layer 1: Tool Use with the Claude API.
The agent decides when to call `read_file` and `list_directory`
based on the user's natural-language request.
"""
 
import anthropic
import os
import json
 
# ── Tool definitions (JSON Schema) ─────────────────────────────────────────
 
TOOLS = [
    {
        "name": "read_file",
        "description": (
            "Read the contents of a local file. "
            "Use when the user asks to inspect, review, or summarise a file. "
            "Returns the file contents as a string."
        ),
        "input_schema": {
            "type": "object",
            "properties": {
                "path": {
                    "type": "string",
                    "description": "Path to the file (absolute or relative to cwd)",
                },
                "encoding": {
                    "type": "string",
                    "description": "File encoding. Default: utf-8",
                    "default": "utf-8",
                },
            },
            "required": ["path"],
        },
    },
    {
        "name": "list_directory",
        "description": (
            "List files and directories inside a given path. "
            "Use when the user wants to explore what is in a folder."
        ),
        "input_schema": {
            "type": "object",
            "properties": {
                "path": {
                    "type": "string",
                    "description": "Directory path to list",
                },
            },
            "required": ["path"],
        },
    },
]
 
# ── Tool execution (YOUR code runs here, not the LLM) ──────────────────────
 
def execute_tool(name: str, inputs: dict) -> str:
    """Execute a tool call returned by the LLM and return a string result."""
    if name == "read_file":
        path = inputs["path"]
        encoding = inputs.get("encoding", "utf-8")
        try:
            with open(path, encoding=encoding) as f:
                content = f.read()
            return content if content else "(empty file)"
        except FileNotFoundError:
            return f"Error: File not found at '{path}'"
        except Exception as e:
            return f"Error reading file: {e}"
 
    elif name == "list_directory":
        path = inputs["path"]
        try:
            entries = os.listdir(path)
            files = [e for e in entries if os.path.isfile(os.path.join(path, e))]
            dirs = [e + "/" for e in entries if os.path.isdir(os.path.join(path, e))]
            return "\n".join(sorted(dirs) + sorted(files))
        except FileNotFoundError:
            return f"Error: Directory not found at '{path}'"
        except Exception as e:
            return f"Error listing directory: {e}"
 
    return f"Error: Unknown tool '{name}'"
 
 
# ── Agentic loop ───────────────────────────────────────────────────────────
 
def run_agent(user_message: str) -> str:
    """
    Run a simple agentic loop:
    1. Send message + tools to Claude
    2. If Claude calls a tool, execute it and loop
    3. Return when Claude returns a plain text response
    """
    client = anthropic.Anthropic()  # reads ANTHROPIC_API_KEY from env
 
    messages = [{"role": "user", "content": user_message}]
    print(f"\n[User] {user_message}\n")
 
    while True:
        response = client.messages.create(
            model="claude-sonnet-4-5",
            max_tokens=1024,
            tools=TOOLS,
            messages=messages,
        )
 
        # If the model returned text with no tool calls, we are done
        if response.stop_reason == "end_turn":
            final_text = "".join(
                block.text for block in response.content if hasattr(block, "text")
            )
            print(f"[Assistant] {final_text}")
            return final_text
 
        # The model wants to call one or more tools
        if response.stop_reason == "tool_use":
            # Add the assistant's tool_use blocks to message history
            messages.append({"role": "assistant", "content": response.content})
 
            # Execute each tool call and collect results
            tool_results = []
            for block in response.content:
                if block.type == "tool_use":
                    print(f"  [Tool Call] {block.name}({json.dumps(block.input)})")
                    result = execute_tool(block.name, block.input)
                    print(f"  [Tool Result] {result[:200]}{'...' if len(result) > 200 else ''}")
                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": result,
                    })
 
            # Feed results back so the model can continue reasoning
            messages.append({"role": "user", "content": tool_results})
 
        else:
            # Unexpected stop reason — surface it and exit
            print(f"[Warning] Unexpected stop_reason: {response.stop_reason}")
            break
 
    return ""
 
 
# ── Entry point ────────────────────────────────────────────────────────────
 
if __name__ == "__main__":
    # Example 1: agent decides to call list_directory, then read_file
    run_agent("What Python files are in the current directory? Show me the contents of the first one you find.")
 
    # Example 2: agent answers directly (no tool needed)
    run_agent("What is the difference between a list and a tuple in Python?")

export ANTHROPIC_API_KEY="sk-ant-..."
python tool_demo.py

Review the pull request described in $ARGUMENTS (or the current branch if no PR number given).
 
Steps:
1. Run `git diff main...HEAD` to see all changes
2. For each changed file, use the Read tool to understand context
3. Check for:
   - Logic errors or off-by-one bugs
   - Missing error handling
   - Security issues (SQL injection, XSS, secrets in code)
   - Missing or outdated tests
   - Naming clarity
4. Produce a structured report:
   - LGTM items (things done well)
   - Suggestions (non-blocking improvements)
   - Required changes (blocking issues)
5. End with an overall assessment: APPROVE / REQUEST CHANGES / NEEDS DISCUSSION

Ship the feature described in $ARGUMENTS by executing the full development loop.
 
## Workflow
 
### Step 1 — Read the plan
- Read PLAN.md or .gsd/phases/current.md to understand the task
- If no plan exists, ask the user to describe the feature before continuing
 
### Step 2 — Implement
- Write the code changes required
- Follow all conventions in CLAUDE.md
- Do not create files that are not needed
 
### Step 3 — Test
- Run the test suite: `npm test` or `pytest` depending on the project
- If tests fail, fix them before proceeding
- If no tests exist for the new code, write them
 
### Step 4 — Self-review
- Read every file you changed
- Check for typos, missing imports, and debug statements left behind
 
### Step 5 — Commit
- Stage only the files you changed: `git add <specific files>`
- Write a conventional commit message: `feat(<scope>): <description>`
- Commit: `git commit -m "..."`
 
### Step 6 — Report
- Summarise what was done in 3-5 bullet points
- List any assumptions made
- Flag anything that needs human review

---
name: ship-feature
description: Ship a feature by running the full implement → test → commit loop
user-invocable: true
allowed-tools:
  - Read
  - Write
  - Edit
  - Bash
  - Glob
  - Grep
---
 
Ship the feature described in $ARGUMENTS by executing the full development loop.
 
[... same workflow steps as above ...]

# Project: PayFlow API
 
## Stack
- Python 3.12, FastAPI, SQLModel, PostgreSQL, Redis
- Testing: pytest + httpx (async)
- Package manager: uv (not pip, not poetry)
- Migrations: Alembic
- Lint/format: ruff (format + lint)
 
## Architecture Rules
- All business logic lives in `src/services/`, not in route handlers
- Route handlers in `src/api/` only: validate input, call a service, return response
- Database models in `src/models/`, Pydantic schemas in `src/schemas/`
- Never import from `src/api/` inside `src/services/` (one-way dependency)
 
## Coding Conventions
- All public functions must have type annotations
- No bare `except:` clauses — catch specific exceptions
- Never use `print()` in production code — use `structlog.get_logger()`
- All datetime values must be timezone-aware (use `datetime.now(UTC)`)
- Prefer `pathlib.Path` over `os.path`
 
## Testing Conventions
- Every service function must have at least one test
- Test files mirror source: `src/services/payment.py` → `tests/services/test_payment.py`
- Use factory-boy fixtures, never raw model construction in tests
- Assert on response *shape* and *status code*, not exact string messages
 
## Git Conventions
- Conventional commits: `feat`, `fix`, `refactor`, `test`, `docs`, `chore`
- Scope = module name: `feat(payment): add stripe webhook handler`
- Never commit secrets, `.env`, or migration files without review
 
## Available Custom Commands
- `/ship-feature <description>` — full implement → test → commit loop
- `/review-pr [PR number]` — structured PR review
- `/db-migrate <description>` — create and apply an Alembic migration
 
## Do Not
- Do not run `alembic upgrade head` in production — only in development
- Do not delete migration files
- Do not change `src/core/config.py` without asking

---
name: db-migrate
description: Create and apply a SQLAlchemy/Alembic database migration
user-invocable: true
allowed-tools:
  - Read
  - Write
  - Edit
  - Bash
  - Glob
---
 
Create and apply an Alembic database migration for: $ARGUMENTS
 
## Pre-flight Checks
1. Read `alembic/env.py` to understand the migration environment
2. Read `src/models/` to find relevant SQLModel models
3. Confirm the database URL is set: `echo $DATABASE_URL`
 
## Generate the Migration
1. Identify what changed (new table, new column, index, constraint)
2. Run: `alembic revision --autogenerate -m "<description>"`
3. Read the generated migration file in `alembic/versions/`
4. IMPORTANT: Review the autogenerated migration — autogenerate misses:
   - Non-nullable column additions to tables with existing rows
   - Enum type changes
   - Custom constraints with server defaults
5. Fix any issues found in the migration file
 
## Apply the Migration (development only)
1. Run: `alembic upgrade head`
2. Confirm success: `alembic current`
 
## Write a Regression Test
1. In `tests/test_migrations.py`, add a test that:
   - Applies the migration
   - Verifies the schema change is present
   - Downgrades (`alembic downgrade -1`)
   - Verifies the schema change is reversed
 
## Report
List:
- Migration file created (path + revision ID)
- Tables/columns affected
- Whether a data migration is needed (flag for human review if so)

"""
exercises/url_checker_tool.py
Layer 1 exercise: define and use a custom `check_url` tool.
"""
 
import anthropic
import requests
 
# Step 1: Define the tool schema
CHECK_URL_TOOL = {
    "name": "check_url",
    "description": (
        "Check whether a given URL is reachable. "
        "Returns the HTTP status code and response time in milliseconds. "
        "Use this when the user wants to verify if a website or API endpoint is up."
    ),
    "input_schema": {
        "type": "object",
        "properties": {
            "url": {
                "type": "string",
                "description": "The full URL to check (must include http:// or https://)",
            },
            "timeout_seconds": {
                "type": "number",
                "description": "Request timeout in seconds. Default: 5",
                "default": 5,
            },
        },
        "required": ["url"],
    },
}
 
 
# Step 2: Implement the tool execution
def check_url(url: str, timeout_seconds: float = 5.0) -> dict:
    """Execute the check_url tool."""
    import time
    start = time.monotonic()
    try:
        response = requests.get(url, timeout=timeout_seconds, allow_redirects=True)
        elapsed_ms = int((time.monotonic() - start) * 1000)
        return {
            "status": "reachable",
            "http_status_code": response.status_code,
            "response_time_ms": elapsed_ms,
            "final_url": response.url,
        }
    except requests.exceptions.ConnectionError:
        return {"status": "unreachable", "error": "Connection refused or DNS failure"}
    except requests.exceptions.Timeout:
        return {"status": "timeout", "error": f"No response within {timeout_seconds}s"}
    except Exception as e:
        return {"status": "error", "error": str(e)}
 
 
# Step 3: Wire it into an agent loop
def run(user_message: str) -> None:
    import json
    client = anthropic.Anthropic()
    messages = [{"role": "user", "content": user_message}]
 
    while True:
        response = client.messages.create(
            model="claude-sonnet-4-5",
            max_tokens=512,
            tools=[CHECK_URL_TOOL],
            messages=messages,
        )
 
        if response.stop_reason == "end_turn":
            for block in response.content:
                if hasattr(block, "text"):
                    print(block.text)
            break
 
        if response.stop_reason == "tool_use":
            messages.append({"role": "assistant", "content": response.content})
            results = []
            for block in response.content:
                if block.type == "tool_use":
                    print(f"Checking: {block.input.get('url')}")
                    result = check_url(**block.input)
                    results.append({
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": json.dumps(result),
                    })
            messages.append({"role": "user", "content": results})
 
 
if __name__ == "__main__":
    run("Are these URLs up? https://anthropic.com and https://this-domain-does-not-exist-xyz.com")

pip install anthropic requests
python exercises/url_checker_tool.py

Check whether the URLs listed in $ARGUMENTS are reachable.
 
For each URL:
1. Use the Bash tool to run: `curl -o /dev/null -s -w "%{http_code} %{time_total}" --max-time 5 <URL>`
2. Report: URL | Status | Response Time
3. Flag any URL that returns a non-2xx status code or times out
 
Present results as a Markdown table.
Format: | URL | Status Code | Response Time | Notes |

# URL Checker Project
 
## Purpose
A Python utility that checks whether URLs are reachable, returns HTTP status codes
and response times.
 
## Stack
- Python 3.12
- `requests` library for HTTP calls
- `anthropic` library for AI-powered analysis
- `pytest` for testing
 
## Conventions
- All URL-checking logic lives in `src/checker.py`, not in scripts or tests
- Always set a timeout on HTTP calls (default: 5 seconds)
- Return structured dicts from checker functions, never raise exceptions to callers
- Test files in `tests/`, named `test_<module>.py`
 
## Running Tests
```bash
pytest tests/ -v