JSON Action Loop and Retries

Parse {tool, args} from the policy, execute tools, retry bad JSON, and stop on finish or budget.

The loop is the product. Tools without a loop are a SDK. A loop without a parse contract is a chatbot with side effects. This part wires JSON actions, a transcript, and retries around the registry from part 2.

The action contract

Every policy response must parse as:

{"tool": "<name>", "args": { ... }}

Rules:

Top level is an object, not a list, not a string.
tool is a string.
args is an object (possibly empty). Missing args is a parse error, not "no arguments".
Extra keys are ignored or rejected — pick one and test it. This project rejects extra keys so the model cannot smuggle execute: true.

When the policy is a real LLM, it will wrap JSON in markdown fences, add "Sure!", or emit trailing commas. Your parser should (1) strip a ``json fence if present, (2) json.loads, (3) validate keys. If that fails, you do not call a tool. You append a role: system note: Parse error. Reply with JSON only.` and ask again. That is a parse retry, counted separately from tool steps so a confused model cannot spend the whole budget on prose.

Fake model that actually sequences

A good fake model is a state machine over the transcript:

If there is no geocode observation yet, emit geocode with a city extracted from the goal.
If geocode returned unknown_city, emit finish with a cannot: answer.
If geocode succeeded and there is no weather observation, emit weather with lat/lon from the last geocode result.
If weather failed, refuse.
If weather succeeded, finish with a sentence.

City extraction can be dumb: last capitalized word, or a regex, or a tiny list of known names found as substrings. Dumb is fine. The loop is the lesson.

Live Pythonpython

import json
import re

CITIES = {
    "paris": {"name": "Paris", "lat": 48.86, "lon": 2.35},
    "lyon": {"name": "Lyon", "lat": 45.76, "lon": 4.84},
}
FORECASTS = {
    (48.86, 2.35): {"temp_c": 18, "conditions": "cloudy"},
    (45.76, 4.84): {"temp_c": 21, "conditions": "sunny"},
}

def geocode(city):
    row = CITIES.get(str(city).strip().lower())
    if not row:
        return {"error": "unknown_city", "city": city}
    return dict(row)

def weather(lat, lon):
    row = FORECASTS.get((round(float(lat), 2), round(float(lon), 2)))
    if not row:
        return {"error": "no_forecast"}
    return dict(row)

def finish(answer):
    return {"final": str(answer).strip()}

TOOLS = {
    "geocode": lambda **kw: geocode(kw["city"]),
    "weather": lambda **kw: weather(kw["lat"], kw["lon"]),
    "finish": lambda **kw: finish(kw["answer"]),
}

def extract_city(goal):
    for name in CITIES:
        if name in goal.lower():
            return name.title()
    m = re.search(r"in ([A-Za-z]+)", goal)
    return m.group(1) if m else ""

def last_tool_result(transcript, name):
    for event in reversed(transcript):
        if event.get("role") == "tool" and event.get("name") == name:
            return event.get("content")
    return None

def fake_model(transcript):
    goal = transcript[0]["content"]
    geo = last_tool_result(transcript, "geocode")
    wx = last_tool_result(transcript, "weather")
    if geo is None:
        city = extract_city(goal) or "unknown"
        return {"tool": "geocode", "args": {"city": city}}
    if "error" in geo:
        return {"tool": "finish", "args": {"answer": "cannot: unknown city"}}
    if wx is None:
        return {"tool": "weather", "args": {"lat": geo["lat"], "lon": geo["lon"]}}
    if "error" in wx:
        return {"tool": "finish", "args": {"answer": "cannot: no forecast"}}
    ans = f"{geo['name']} is {wx['temp_c']}C and {wx['conditions']}."
    return {"tool": "finish", "args": {"answer": ans}}

def parse_action(raw):
    if isinstance(raw, dict):
        obj = raw
    else:
        text = str(raw).strip()
        fence = chr(96) * 3
        if text.startswith(fence):
            text = text.strip(chr(96))
            text = text.replace("json", "", 1).strip()
        obj = json.loads(text)
    if set(obj.keys()) != {"tool", "args"}:
        raise ValueError("action must have only tool and args")
    if not isinstance(obj["tool"], str) or not isinstance(obj["args"], dict):
        raise ValueError("types")
    return obj

def run_agent(goal, max_steps=8, max_parse_retries=3):
    transcript = [{"role": "user", "content": goal}]
    parse_fails = 0
    for step in range(1, max_steps + 1):
        try:
            action = parse_action(fake_model(transcript))
            parse_fails = 0
        except (ValueError, json.JSONDecodeError, TypeError) as exc:
            parse_fails += 1
            transcript.append({"role": "system", "content": f"parse error: {exc}"})
            if parse_fails >= max_parse_retries:
                return {"status": "parse_budget", "transcript": transcript}
            continue
        name, args = action["tool"], action["args"]
        if name not in TOOLS:
            result = {"error": "unknown_tool", "name": name}
        else:
            result = TOOLS[name](**args)
        print(f"step {step}: {name}({args}) -> {result}")
        transcript.append({"role": "tool", "name": name, "content": result, "args": args})
        if name == "finish" and "final" in result:
            return {"status": "ok", "answer": result["final"], "transcript": transcript}
    return {"status": "max_steps", "transcript": transcript}

for goal in ["What is the weather in Paris?", "Weather in Atlantis?"]:
    print("GOAL:", goal)
    out = run_agent(goal)
    print("STATUS:", out["status"], "ANSWER:", out.get("answer"))
    print("---")

Output

Run to execute this in your browser. Nothing is sent to a server.

Step-by-step: the loop you must be able to recite

Seed the transcript with the user goal as the first event. Everything the policy knows comes from this list.
Call the policy with the full transcript. In production this is an API call; here it is fake_model.
Parse. Dicts from a fake model still go through parse_action so the real LLM path is identical. If parse fails, append a system observation and continue without incrementing a successful tool step — but do count it toward max_parse_retries and max_steps so you cannot infinite-loop.
Dispatch. Registry lookup. Call with **args. Catch nothing from the tool if you already return error dicts; still guard TypeError if the model omits a key (kw["city"] vs kw.get).
Append a tool event with name, args, and result. Args belong in the trace even when they failed — that is how you debug.
Stop on finish with a final key. A finish that itself errors is not success; keep looping or refuse.
Budget. If the for ends, return max_steps. Never silently pick the last weather dict.

Retries are not hope

Retry parse errors with a protocol reminder. Retry timeouts by calling the same tool again (part 5). Do not retry unknown_city — that is information, not a blip. Retrying Atlantis will not create coordinates.

A simple rule: retry if the error code is in {"timeout", "parse", "rate_limit"}. Otherwise finish or change the plan (re-geocode with a stripped city).

Transcript hygiene

Keep tool results small. If a future weather API returns 400 JSON keys, store a projection: temp, conditions, place. The policy's context window is a budget. Logging the full HTTP body can live in a side channel, not in the prompt.

Note:When you swap fake_model for an LLM, wrap the API so it still returns a dict or a string. The loop above should not change.

Exercise

Break the fake model on purpose: make it return the string Sure! {"tool": "geocode", "args": {"city": "Paris"}} on the first call. Extend parse_action to find the first { and last } and parse that slice. Confirm Paris still works. Then confirm that a string with no braces hits parse_budget.

Check your understanding

What should happen when JSON parsing fails?

Call geocode anyway
Append a parse error to the transcript and retry until a parse budget is hit
Restart the Python process
Delete the user goal

Tools and Simulated Environment

Up next

Evals and Tests