Projects/Weather Tool Agent/Part 5

Hardening the Weather Agent

Validate arguments, cap parse retries, reject unknown tools, simulate rate limits, and fail closed without calling weather on garbage coords.

A working demo is not a hardened agent. Hardening is everything you add after the happy path: validation, budgets, allowlists, and fail-closed behavior when the world is weird. This part takes the loop from parts 2–4 and makes it safe enough to sit behind a form on joeven.com (still with simulated weather).

Threats that already exist in a three-tool bot

The user is not the only adversary. The model is untrusted. It will eventually emit weather with a string lat, call finish twice, or name geocode_all_cities. Treat every action as hostile input to your process.

Threat	Hardening
Extra JSON keys	Reject in `parse_action`
Unknown tool	Do not `eval` names; registry miss only
Args of the wrong type	Validate before the dict lookup
Infinite parse junk	`max_parse_retries`
Weather before geocode with user-supplied coords	Optional: require geocode in-transcript, or allow coords but clamp ranges
Rate limits	Return `rate_limit` and retry with a cap
Overlong answers	Truncate `finish` to N characters

Validate at the boundary

The registry lambdas in part 2 used kw["city"] which raises KeyError. Hardened dispatch catches that and returns bad_args. Better: a per-tool schema:

geocode: city is str, length 1–80, no digits-only strings if you want.
weather: lat in [-90, 90], lon in [-180, 180], both numbers.
finish: answer str, length 1–500.

Out-of-range coordinates should not look up the weather table. They are bad_args. Otherwise a model can probe your dict with a sweep of floats.

Rate limits as a fake 429

Give weather a token bucket: 3 calls per run. The fourth returns {"error": "rate_limit"}. The loop may retry twice, then must finish with cannot: rate limited. This trains you for vendor 429s without time.sleep (sleep is rude in a Try it box). A counter is enough.

Never destructive — even when there is nothing to destroy

This agent cannot drop a database. Still practice the habit: no tool runs unless it is in the allowlist. If you later add http_get(url), you will already have the gate. Put the allowlist next to TOOLS and check name in TOOLS before splatting args.

Live Pythonpython

import json

CITIES = {"paris": {"name": "Paris", "lat": 48.86, "lon": 2.35}}
FORECASTS = {(48.86, 2.35): {"temp_c": 18, "conditions": "cloudy"}}
WEATHER_CALLS = {"n": 0}
RATE_LIMIT = 3

def validate(name, args):
    if not isinstance(args, dict):
        return "args must be an object"
    if name == "geocode":
        city = args.get("city")
        if not isinstance(city, str) or not (1 <= len(city.strip()) <= 80):
            return "city"
    elif name == "weather":
        try:
            lat, lon = float(args["lat"]), float(args["lon"])
        except (KeyError, TypeError, ValueError):
            return "lat/lon"
        if not (-90 <= lat <= 90 and -180 <= lon <= 180):
            return "range"
    elif name == "finish":
        ans = args.get("answer")
        if not isinstance(ans, str) or not (1 <= len(ans) <= 500):
            return "answer"
    else:
        return "unknown_tool"
    extra = set(args) - {
        "geocode": {"city"},
        "weather": {"lat", "lon"},
        "finish": {"answer"},
    }[name]
    if extra:
        return "extra_args"
    return None

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

def weather(lat, lon):
    WEATHER_CALLS["n"] += 1
    if WEATHER_CALLS["n"] > RATE_LIMIT:
        return {"error": "rate_limit"}
    row = FORECASTS.get((round(float(lat), 2), round(float(lon), 2)))
    return dict(row) if row else {"error": "no_forecast"}

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

def dispatch(name, args):
    err = validate(name, args)
    if err == "unknown_tool":
        return {"error": "unknown_tool", "name": name}
    if err:
        return {"error": "bad_args", "detail": err}
    return TOOLS[name](**args)

def parse_action(obj):
    if set(obj) != {"tool", "args"}:
        raise ValueError("shape")
    return obj

def policy(transcript, force_bad=None):
    if force_bad:
        return force_bad
    geo = next((e["content"] for e in reversed(transcript) if e.get("name") == "geocode"), None)
    wx = next((e["content"] for e in reversed(transcript) if e.get("name") == "weather"), None)
    if geo is None:
        return {"tool": "geocode", "args": {"city": "Paris"}}
    if wx is None:
        return {"tool": "weather", "args": {"lat": geo["lat"], "lon": geo["lon"]}}
    return {"tool": "finish", "args": {"answer": f"Paris is {wx['temp_c']}C and {wx['conditions']}."}}

def run(goal, script=None):
    WEATHER_CALLS["n"] = 0
    transcript = [{"role": "user", "content": goal}]
    parse_fails = 0
    for step in range(1, 10):
        raw = script[step - 1] if script and step - 1 < len(script) else policy(transcript)
        try:
            action = parse_action(raw)
            parse_fails = 0
        except (ValueError, TypeError):
            parse_fails += 1
            transcript.append({"role": "system", "content": "parse error"})
            if parse_fails >= 3:
                return {"status": "parse_budget"}
            continue
        result = dispatch(action["tool"], action["args"])
        print(f"step {step}: {action['tool']} -> {result}")
        transcript.append({"role": "tool", "name": action["tool"], "content": result})
        if action["tool"] == "finish" and "final" in result:
            return {"status": "ok", "answer": result["final"]}
    return {"status": "max_steps"}

print("happy:", run("Paris weather"))
print(
    "unknown tool:",
    run(
        "x",
        script=[
            {"tool": "explode", "args": {}},
            {"tool": "finish", "args": {"answer": "cannot: unknown tool"}},
        ],
    ),
)
print(
    "bad lat:",
    dispatch("weather", {"lat": 999, "lon": 0}),
)
print(
    "extra key:",
    dispatch("geocode", {"city": "Paris", "drop_table": True}),
)

Output

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

Step-by-step hardening checklist

Allowlist tools. Dispatch never looks up globals()[name].
Validate types and ranges before environment access.
Reject extra args. Prompt injection sometimes arrives as extra JSON fields.
Cap parse retries and cap tool steps.
Cap string lengths on the way in and out.
Rate-limit simulated side-effect tools.
Fail closed: if you cannot geocode, do not weather; if you cannot weather, do not invent °C.
Keep traces. Part 4's tests should still pass after hardening. If they do not, you tightened a contract — update tests deliberately.

Production notes (for later)

When you replace dicts with HTTP: timeouts, a single retried idempotent GET, and never log API keys. The hardening above maps 1:1 to httpx except the catalog. Your validate function stays. That is why we wrote it.

Watch out:Do not "fix" unknown cities by picking the nearest name in CITIES. That is a silent wrong-city incident.

Exercise

Add a max_weather_calls of 1 and a policy that panics and calls weather three times. Assert the run ends with cannot: after a rate_limit observation. Then add a unit test that dispatch("weather", {"lat": "hot", "lon": 2}) returns bad_args and does not increment WEATHER_CALLS.

You now have a beginner portfolio piece: a sequenced tool agent, JSON protocol, tests, and fail-closed tools. Next project: the same loop, but the tool is search, the protocol is ReAct, and the failure mode is cannot answer.

Check your understanding

What does fail-closed mean for the weather agent?

Always print 22C so the user is happy
If geocode or weather is missing or invalid, refuse instead of inventing a forecast
Crash the interpreter
Call every tool in the registry

Evals and Tests

Done

All projects