Build a Self-Correcting AI Agent with Reflection and Retry Loops

1. What you'll build

A Python agent that writes a function, runs it against tests it can't see, hands the failures to a separate critic call, and retries with the critique until the tests pass or it hits an attempt cap. Two Claude calls per iteration, a deterministic check in between, structured outputs on both ends so nothing gets parsed out of Markdown.

2. Prerequisites

• Python 3.10 or newer. The SDK refuses older versions. Verified on 3.13.5.
• Anthropic Python SDK 1.2.0 (released 2026-08-27). It pulls in Pydantic 2.13.5, which you'll use for the response schemas.
• An API key from the Claude Console, exported as ANTHROPIC_API_KEY.
• Model: claude-opus-5. Thinking is on by default, and it supports structured outputs and the effort parameter. Expect 1 to 7 calls per run.
• Commands are for macOS/Linux. On Windows, activate the venv with .venv\Scripts\activate.

3. Set up the project

mkdir self-correcting-agent && cd self-correcting-agent
python3 -m venv .venv && source .venv/bin/activate
pip install "anthropic==1.2.0"
export ANTHROPIC_API_KEY="sk-ant-..."

4. Define the task and the success check

Create agent.py. The first block holds the task the generator sees and the tests it does not. The tests encode three details the task text leaves out: a bare number means seconds, units can be uppercase with padding, and garbage must raise ValueError. That gap is deliberate: in real work the tests are the spec, and the loop exists to close the gap without a human in the middle.

"""Self-correcting agent: generate -> check -> critique -> retry."""

import subprocess
import sys
from dataclasses import dataclass

import anthropic
from pydantic import BaseModel

MODEL = "claude-opus-5"
MAX_ATTEMPTS = 4
CHECK_TIMEOUT = 10

client = anthropic.Anthropic()

TASK = """Write a Python function parse_duration(s: str) -> int that converts a
duration string such as "1h30m", "45s", or "2h 15m 30s" into total seconds.
Supported units are h, m, and s. Whitespace between components is allowed.
Reject invalid input by raising ValueError."""

# The success check. The generator never sees it; the critic only sees what failed.
TEST_CODE = """
import sys
CASES = [("1h30m", 5400), ("45s", 45), ("2h 15m 30s", 8130),
         ("90", 90), (" 1H 30M ", 5400)]
failures = []
for s, want in CASES:
    try:
        got = parse_duration(s)
    except Exception as e:
        got = f"{type(e).__name__}: {e}"
    if got != want:
        failures.append(f"parse_duration({s!r}) -> {got!r}, want {want!r}")
for bad in ["", "abc", "1x"]:
    try:
        parse_duration(bad)
        failures.append(f"parse_duration({bad!r}) returned instead of raising ValueError")
    except ValueError:
        pass
    except Exception as e:
        failures.append(f"parse_duration({bad!r}) raised {type(e).__name__}, want ValueError")
print("\\n".join(failures) if failures else "ALL PASSED")
sys.exit(1 if failures else 0)
"""


class Draft(BaseModel):
    code: str
    notes: str


class Critique(BaseModel):
    root_cause: str
    fix_plan: list[str]


@dataclass
class CheckResult:
    passed: bool
    output: str


@dataclass
class Attempt:
    code: str
    check: CheckResult
    critique: Critique


def check(code: str) -> CheckResult:
    """Run the candidate plus the tests in a fresh interpreter."""
    try:
        proc = subprocess.run(
            [sys.executable, "-c", code + "\n" + TEST_CODE],
            capture_output=True, text=True, timeout=CHECK_TIMEOUT,
        )
    except subprocess.TimeoutExpired:
        return CheckResult(False, f"Timed out after {CHECK_TIMEOUT}s")
    output = (proc.stdout + proc.stderr).strip()
    return CheckResult(proc.returncode == 0, output[-3000:])

check() runs the candidate in a subprocess so a syntax error or infinite loop can't take the agent down. Whatever lands on stdout or stderr, tracebacks included, becomes the critic's evidence.

5. Write the generator and the critic

Append the two model calls. Both use client.messages.parse() with a Pydantic class as output_format; the SDK converts it to output_config.format on the wire and hands back a validated instance on response.parsed_output.

GENERATOR_SYSTEM = (
    "You write production-quality Python. Put the complete module source in "
    "`code`: plain Python, no Markdown fences, no example usage, no prints."
)

CRITIC_SYSTEM = (
    "You are a strict code reviewer. Diagnose why the code failed the check. "
    "Do not rewrite the code. Name the root cause and give minimal, concrete fix steps."
)


def generate(history: list[Attempt]) -> Draft:
    prompt = TASK
    if history:
        last = history[-1]
        lessons = "\n".join(f"- {a.critique.root_cause}" for a in history)
        steps = "\n".join(f"- {s}" for s in last.critique.fix_plan)
        prompt += (
            "\n\nYour previous attempt failed the acceptance check.\n\n"
            f"Previous code:\n{last.code}\n\n"
            f"Check output:\n{last.check.output}\n\n"
            f"Reviewer's fix plan:\n{steps}\n\n"
            f"Root causes found so far (do not repeat them):\n{lessons}\n\n"
            "Write a corrected version."
        )
    response = client.messages.parse(
        model=MODEL,
        max_tokens=16000,
        system=GENERATOR_SYSTEM,
        messages=[{"role": "user", "content": prompt}],
        output_format=Draft,
    )
    return parsed(response)


def critique(code: str, check_output: str) -> Critique:
    response = client.messages.parse(
        model=MODEL,
        max_tokens=16000,
        system=CRITIC_SYSTEM,
        output_config={"effort": "medium"},  # short diagnosis; full depth not needed
        messages=[{
            "role": "user",
            "content": f"Task:\n{TASK}\n\nCode:\n{code}\n\nCheck output:\n{check_output}",
        }],
        output_format=Critique,
    )
    return parsed(response)


def parsed(response):
    if response.parsed_output is None:
        raise RuntimeError(f"No structured output (stop_reason={response.stop_reason})")
    return response.parsed_output

Two design choices matter. The critic is told not to rewrite the code, so its tokens go into diagnosis instead of a second draft the generator would have to reconcile. The generator gets every root cause found so far, not just the last, so a fix on attempt 3 doesn't reintroduce the bug from attempt 1.

max_tokens is 16000 because thinking tokens count against it; a cap sized for the JSON alone truncates on hard retries. output_config and output_format coexist: the SDK merges the schema into the config you pass.

6. Wire the retry loop

def run() -> str:
    history: list[Attempt] = []
    for n in range(1, MAX_ATTEMPTS + 1):
        draft = generate(history)
        result = check(draft.code)
        print(f"attempt {n}: {'PASS' if result.passed else 'FAIL'}")
        if result.passed:
            return draft.code
        print(result.output)
        if n == MAX_ATTEMPTS:
            break
        review = critique(draft.code, result.output)
        print(f"  root cause: {review.root_cause}")
        history.append(Attempt(draft.code, result, review))
    raise SystemExit(f"Gave up after {MAX_ATTEMPTS} attempts")


if __name__ == "__main__":
    code = run()
    with open("parse_duration.py", "w") as f:
        f.write(code)
    print("wrote parse_duration.py")

The cap is the safety valve: without it, a task the model can't solve, or a flaky check, burns tokens forever. Skipping the critique on the final failure saves one call nothing would consume.

7. Verify it works

python agent.py

A run where the first draft misses the hidden spec looks like this. The shape is fixed; the exception text and the root-cause line come from the model and will differ:

attempt 1: FAIL
parse_duration('90') -> 'ValueError: 90', want 90
parse_duration(' 1H 30M ') -> 'ValueError:  1H 30M ', want 5400
  root cause: Bare numbers and uppercase units are not handled
attempt 2: PASS
wrote parse_duration.py

claude-opus-5 sometimes infers the hidden cases and passes on attempt 1, which is fine. To force a retry, add a case the text doesn't imply, such as ("1.5h", 5400), to CASES.

Confirm the artifact is usable on its own:

python -c "from parse_duration import parse_duration; print(parse_duration('2h 15m 30s'))"

8130

8. Troubleshooting

TypeError: "Could not resolve authentication method. Expected one of api_key, auth_token, or credentials to be set. ..." The key isn't in the environment the script runs in. Export ANTHROPIC_API_KEY in the same shell you run python agent.py from; activating a venv doesn't carry it over from another terminal.

anthropic.BadRequestError: ... "thinking.type.enabled" is not supported for this model. Use "thinking.type.adaptive" and "output_config.effort" to control thinking behavior. You added thinking={"type": "enabled", "budget_tokens": ...} from an older example. Delete it. Thinking is already on for claude-opus-5; steer depth with output_config={"effort": ...} instead.

pydantic_core._pydantic_core.ValidationError: 1 validation error for Draft ... Invalid JSON: EOF while parsing a string ... [type=json_invalid The response hit max_tokens mid-JSON. Thinking spends from the same budget as the output, so raise max_tokens or drop the generator to output_config={"effort": "medium"}.

anthropic.RateLimitError after a few attempts The SDK already retries 429s twice with backoff. For a bigger MAX_ATTEMPTS, or several agents in parallel, construct the client with anthropic.Anthropic(max_retries=5) so a burst of retries doesn't abort the run.

9. Next steps

• Swap TEST_CODE for pytest on a real repo: write the draft to a temp file and run the suite as the check. For untrusted tasks, run the check in a container or the code execution tool.
• TASK and both system prompts repeat on every call. Mark the system prompt with cache_control per the prompt caching docs and retries get cheaper.
• Anthropic's Building effective agents calls this the evaluator-optimizer workflow and covers when it beats a single well-prompted call.
• With no deterministic check, use a rubric-scoring model call as the evaluator. Prefer the deterministic one; a judge that hallucinates a pass is worse than no loop.
• For tasks that need tools mid-generation, move the generator onto the SDK's tool runner and keep check() and critique() around it.