cd /news/artificial-intelligence/build-a-self-correcting-ai-agent-wit… · home topics artificial-intelligence article
[ARTICLE · art-116492] src=sourcefeed.dev ↗ pub= topic=artificial-intelligence verified=true sentiment=· neutral

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

Anthropic's Claude agent can be built into a self-correcting AI agent using a Python script that generates code, runs it against hidden tests, and retries with a critic's feedback until tests pass or an attempt cap is reached. The tutorial by Rachel Goldstein uses the Anthropic Python SDK 1.2.0, Pydantic 2.13.5, and the claude-opus-5 model, with a maximum of 4 attempts and a 10-second timeout per check. The agent's success check includes test cases that require handling bare numbers as seconds, uppercase units with padding, and raising ValueError for invalid input.

read8 min views2 publishedAug 31, 2026
Build a Self-Correcting AI Agent with Reflection and Retry Loops
Image: Sourcefeed (auto-discovered)

Add a critic call and a deterministic check so a Claude agent fixes its own output until the tests pass.

Rachel Goldstein

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 #

Python3.10 or newer. The SDK refuses older versions. Verified on 3.13.5.Anthropic Python SDK1.2.0 (released 2026-08-27). It pulls inPydantic2.13.5, which you'll use for the response schemas.- An API key from the Claude Console, exported asANTHROPIC_API_KEY

. - Model: claude-opus-5

. Thinking is on by default, and it supports structured outputs and theeffort

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."""

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

forpytest

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 thecode execution tool. TASK

and both system prompts repeat on every call. Mark the system prompt withcache_control

per theprompt cachingdocs and retries get cheaper.- Anthropic's Building effective agentscalls 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 runnerand keepcheck()

andcritique()

around it.

Sources & further reading #

Structured outputs— platform.claude.com - Effort— platform.claude.com - Models overview— platform.claude.com - Python SDK— platform.claude.com - Troubleshooting thinking— platform.claude.com - anthropic 1.2.0— pypi.org

Rachel Goldstein· Dev Tools Editor

Rachel has been embedded in the developer tooling ecosystem for nearly eight years, covering everything from IDE wars and package-manager drama to the quiet rise of AI-assisted coding. She has a soft spot for open-source maintainers and an unhealthy number of terminal emulators installed on a single laptop.

Discussion 0 #

No comments yet

Be the first to weigh in.

── more in #artificial-intelligence 4 stories · sorted by recency
── more on @anthropic 3 stories trending now
sponsored brought to you by zahid.host 4,200+ EU-deployed projects
reading about agents? ship yours in a single git push.

Run your AI side-project on zahid.host

EU-based hosting, git-push deploys, automatic HTTPS, no cold starts. Free tier with a custom domain — perfect for shipping the agent you just read about.

$git push zahid main
Live at https://your-agent.zahid.host
Get free account → Pricing
from €0/mo · no card required
LIVE [news/build-a-self-correct…] indexed:0 read:8min 2026-08-31 ·