The Tool Registry & the Agent Loop

Where this fits intermediate

Up to now your DevOps agent has been all skeleton and no pulse. Lab 8a gave you a mock cluster, a risk taxonomy, and typed schemas — real scaffolding, but nothing that thinks. This is the lab where the thing wakes up: for the first time the model reads the state of a broken cluster and decides, on its own, what to look at next. That moment — code you wrote handing control to a model that then hands control back — is the whole reason "agent" is a different word from "script".

Here's the mental model to carry through the rest of this lab. An agent is a conversation with a hard cap: you tell the model about a problem and hand it a menu of tools; it either answers or asks to use a tool; you run the tool and paste the result back; repeat until it's done or you hit the ceiling. Everything in this lab is just those two moving parts — the menu (the tool registry, Steps 1–2) and the conversation manager (the loop, Step 3). If you keep that two-part picture in your head, none of the code below will surprise you.

The mistake people make coming into this lab is assuming "wiring up an LLM to run commands" is the scary part. It isn't — the API call itself is a dozen lines. The genuinely hard, career-relevant skill is keeping the model on a leash: it never touches your systems directly, it can only ask your code to run tools you registered, and every tool carries a risk tag your program controls. That separation — model proposes, your harness disposes — is what makes an agent something you'd actually let near a cluster. Get it right here on read-only tools and the write actions in Lab 8c become a small, safe addition rather than a leap of faith.

In Lab 8a everything ran offline. This lab makes the first API calls — the loop sends the mock cluster's data to Claude, which reasons about it and calls read-only tools to investigate. Still 100% safe: only read-only tools are wired up here; the safety gate and write actions come in Lab 8c.

Step 1 · Model a tool as data intermediate

Every tool leads a double life, and if you don't design for that you'll leak something you shouldn't. The model needs to know a tool exists and how to call it. Your safety machinery needs to know how dangerous it is. Your runtime needs the actual function to execute. Cram all of that into whatever you send the model and you've just told the model the exact risk tag it's about to be gated on — the one piece of information you least want it negotiating around. So the first design decision of the whole agent is quietly the most important: what does the model see, and what stays yours?

Think of a tool the way an airline thinks of a boarding pass. The passenger's copy shows the flight, the gate, the seat — everything they need to act on. It does not show the fare class rules, the crew's override codes, or the actual mechanism that opens the jet bridge. Same trip, two views: a public face and a private back-office. A Tool here is exactly that — one object with a public face (spec(), the name/description/schema the model reads) and a private back-office (risk and run, which never leave your process). Bundling both halves in one place is deliberate: it means there is a single object where the loop, the tests, and the safety gate can all ask "what is this tool and how careful must I be with it?"

Why does this earn a whole step rather than a stray sentence? Because the alternative — the shape beginners reach for — is to keep three loose parallel lists: a list of API tool specs here, a dict of risk levels there, a dict of functions somewhere else, all keyed by name and all quietly assumed to stay in sync. They don't stay in sync. Someone adds a tool to two of the three lists, ships it, and now there's a live tool the gate has no risk entry for — a silent hole in your safety model that no error will ever announce. Keeping name, description, input_schema, risk, and run welded into one dataclass makes that class of bug unrepresentable: you cannot register a tool without giving it a risk, because the field is right there in the constructor.

The common trap to name explicitly: sending the model your whole tool object because it's convenient. Don't. The model only ever gets what spec() chooses to return, and spec() returns three keys precisely so risk and run stay invisible. A tool bundles three things: the schema Claude sees, the risk class (from Lab 8a), and the Python function that runs it. Keeping them together in one object is what lets the safety gate later look up any tool's risk in one place.

agent/tools.pyfrom dataclasses import dataclass
from typing import Callable
from .schemas import RiskClass
from mock import cluster, actions

@dataclass
class Tool:
    name: str
    description: str
    input_schema: dict
    risk: RiskClass                      # ← the safety-critical field
    run: Callable

    def spec(self) -> dict:
        # the shape Claude's tools=[...] parameter wants
        return {"name": self.name, "description": self.description,
                "input_schema": self.input_schema}

TOOLS: dict[str, Tool] = {}
def register(tool): TOOLS[tool.name] = tool

def _obj(props, required):
    return {"type":"object", "properties": props, "required": required}

Step 2 · Register the read-only tools intermediate

The single highest-leverage thing you will write in this whole lab is not code — it's a sentence. Specifically, each tool's description string. Engineers instinctively treat those descriptions as throwaway documentation, the kind of comment you dash off and never reread. In an agent they are the opposite: the description is the only instruction the model gets about when to reach for a tool, so its wording literally decides the order in which your agent investigates an incident. Change "lists pods" to "List pods... Use FIRST when diagnosing" and the agent's entire behaviour shifts. You are programming the model in English.

Here's the analogy that makes it click. Imagine handing a sharp new hire a toolbox on their first day with no context. If every tool just has its name etched on it — "wrench", "meter" — they'll pick things up in a random or nervous order and maybe grab the wrong one first. Now imagine each tool has a tiny label: "check this reading FIRST before touching anything", "only use once you've confirmed the power is off". Suddenly the newcomer works like a veteran, in a sensible sequence, because the tools themselves told them how to think. Your tool descriptions are those labels, and the model is that new hire — capable, but entirely dependent on what you wrote on the tin.

There's a deliberate ordering to which tools we register too, and it's a safety principle, not an accident: give the agent eyes before you ever give it hands. Everything registered in this lab is READ_ONLY — it can look at pods, pull logs, list recent deploys, read metrics and alerts, but it cannot change a single thing. That's not a limitation we're apologising for; it's the correct rung to start on. An agent that can only observe can still be enormously useful (it diagnoses the whole incident in Step 4) and it is incapable of making anything worse. The write actions — the tools with hands — wait until Lab 8c, behind the safety gate, on purpose.

The mistake to avoid here is vague, model-facing descriptions. A tool labelled just "gets pods" is a tool the model will call late, sporadically, or not at all — and then you'll blame the model for investigating in a weird order when the real bug was your prose. Write descriptions the way you'd brief a colleague: say what it does and when to use it. Give the agent eyes before hands. Each read-only tool wraps a mock-cluster accessor and is tagged READ_ONLY. Note how the description tells the model when to use it — that's what makes the agent investigate in a sensible order.

agent/tools.py (continued)register(Tool(
    "kubectl_get_pods",
    "List pods with status and restart counts. Use FIRST when "
    "diagnosing — it shows what's unhealthy.",
    _obj({"namespace": {"type":"string"}}, []),
    RiskClass.READ_ONLY,
    lambda namespace="staging": cluster.get_pods(namespace),
))
register(Tool(
    "kubectl_logs",
    "Fetch recent log lines for a pod. Use to find the actual error.",
    _obj({"pod": {"type":"string"}, "namespace": {"type":"string"}}, ["pod"]),
    RiskClass.READ_ONLY,
    lambda pod, namespace="staging": cluster.get_logs(pod, namespace),
))
# also: kubectl_get_events, recent_deploys — same pattern

Step 3 · The agent loop advanced

People expect the "agent loop" to be the most mysterious code in the file. It's about fifteen lines, and once you see it you can't un-see how simple it is. There is no hidden orchestration engine, no magic. An agent is a for loop that keeps a conversation going: call the model, check whether it wants a tool, run the tool if so, paste the answer back, go around again. The intelligence lives in the model; the loop is just the plumbing that lets the model act and then react to what it found. Demystifying this is worth a section on its own, because half of "understanding agents" is realising there's nothing to be intimidated by.

The rhythm has a name worth knowing — think, act, observe — and a good analogy is a doctor working through a diagnosis. The doctor thinks ("could be the heart"), acts ("order an ECG"), and then observes the result before deciding the next test. They don't script all the tests up front; each result reshapes the next move. That feedback cycle is exactly what your loop implements: the model thinks and emits a tool_use, your code acts by running the tool, and the result goes back as an observation the model reads before its next turn. It's the loop, not any single API call, that lets an agent chase a root cause it couldn't have predicted from the incident text alone.

Now the part that separates a toy loop from one you'd trust: it must be bounded. A model can, and occasionally will, get into a rut where it keeps asking for one more tool, forever. Without a ceiling that's an infinite loop burning tokens and money against your API key. The MAX_STEPS cap is the seatbelt — the loop is guaranteed to terminate after a fixed number of passes no matter what the model does. This is the single most common thing missing from beginners' agent loops, and it is non-negotiable: every agent loop you ever write gets a hard step cap, the same way every recursion gets a base case.

The other three rules are subtler and, if you break them, the API rejects your request outright with a cryptic 400. They're the message-protocol contract from Chapter 4: append the model's turn to the conversation verbatim before you add any results; every tool_result you send back must carry the tool_use_id of the request it answers, so the model can match reply to request; and all of a turn's results go back in one user message, not dribbled out separately. Miss any of these and you'll meet the errors in the Troubleshooting table below. This is the Chapter 4 loop, driven by the registry. For now it runs every requested tool directly (all are read-only, so it's safe). In Lab 8c you'll insert the policy gate between "model wants tool" and "tool runs".

agent/engine.py (read-only version)from anthropic import Anthropic
from agent.tools import TOOLS
client = Anthropic()
MAX_STEPS = 8

def run(incident: str):
    system = ("You are a careful SRE. Diagnose the incident using the "
              "read-only tools. Cite the evidence you find.")
    messages = [{"role":"user", "content": incident}]
    specs = [t.spec() for t in TOOLS.values()]

    for _ in range(MAX_STEPS):                # hard cap — always terminates
        resp = client.messages.create(model="claude-opus-4-8",
                 max_tokens=1500, system=system, tools=specs, messages=messages)
        if resp.stop_reason == "end_turn":
            return next((b.text for b in resp.content if b.type=="text"), "")
        messages.append({"role":"assistant", "content": resp.content})
        results = []
        for b in resp.content:
            if b.type == "tool_use":
                out = TOOLS[b.name].run(**b.input)     # Lab 8c gates this
                results.append({"type":"tool_result",
                                "tool_use_id": b.id, "content": str(out)})
        messages.append({"role":"user", "content": results})
    return "Stopped: hit step limit."

agent/engine.py (read-only version)import json
from openai import OpenAI
from agent.tools import TOOLS
client = OpenAI()
MAX_STEPS = 8

def run(incident: str):
    system = ("You are a careful SRE. Diagnose the incident using the "
              "read-only tools. Cite the evidence you find.")
    input_list = [{"role":"user", "content": incident}]
    specs = [t.spec() for t in TOOLS.values()]

    for _ in range(MAX_STEPS):                # hard cap — always terminates
        resp = client.responses.create(model="gpt-5.5",
                 max_output_tokens=1500, instructions=system, tools=specs, input=input_list)
        calls = [o for o in resp.output if o.type == "function_call"]
        if not calls:                           # no tool calls → model is done
            return resp.output_text
        input_list += resp.output                 # append output items verbatim
        for call in calls:
            out = TOOLS[call.name].run(**json.loads(call.arguments))  # Lab 8c gates this
            input_list.append({"type":"function_call_output",
                               "call_id": call.call_id, "output": str(out)})
    return "Stopped: hit step limit."

Step 4 · Run your first diagnosis advanced

This is the payoff moment — the first time you type a problem in plain English and watch software figure out the answer without being told how. Up to now you've built parts. Here they run together, against the live model, on a real (mock) broken cluster. What makes it land emotionally is what you don't do: you never tell it to check the pods, then read the logs, then look at deploys. You hand it one sentence — "checkout-api keeps restarting" — and it decides the investigation plan itself.

The right way to read the output is to treat it like a detective's case file, not a chatbot reply. Notice the shape of what comes back: a conclusion (the diagnosis), the reasoning that connects the clues (a crash-looping pod, a specific log error, a recent deploy that changed a secret), and — crucially — an audit log listing every tool it actually called, in order, each tagged [read_only]. That audit trail is not decoration. In a real on-call situation the first question anyone asks an automated actor is "what exactly did you do?" An agent that can't answer that is one you can't deploy; an agent that prints its every move is one you can trust and review. You're seeing the accountability layer, born read-only, that everything in Lab 8c will build on.

Read the three audit lines top to bottom — pods, then logs, then deploys — and connect them straight back to Step 2. That order wasn't luck and it wasn't hard-coded; it fell out of the "Use FIRST" wording you put in kubectl_get_pods's description. This is the whole thesis of the lab made visible: your prose shaped the machine's behaviour. The trap to avoid at this stage is treating a good run as proof of correctness — one lucky diagnosis is an anecdote, not evidence. That's exactly why Step 5 adds tests and Lab 8d adds real evals: to turn "it worked when I tried it" into "it works, measurably, every time." For now, savour the run, then go make it trustworthy. Now you need your API key (from the course .env). A quick runner:

terminalcd devops-agent
python cli.py "checkout-api-7d9f in staging keeps restarting"

🔧 Running at rung: OBSERVE
--------------------------------------------------
The pod checkout-api-7d9f is in CrashLoopBackOff. Its logs show
"password authentication failed for user 'checkout'", and a deploy
18 minutes ago changed the DATABASE_URL secret reference. The likely
root cause is a broken database credential from that change. Suggested
fix: correct the secret and roll the deployment — do not just restart.
--------------------------------------------------
📋 Audit log (every action attempted):
✓ kubectl_get_pods({'namespace': 'staging'}) [read_only]
✓ kubectl_logs({'pod': 'checkout-api-7d9f'}) [read_only]
✓ recent_deploys({'namespace': 'staging'}) [read_only]

Step 5 · Test the loop expert

Testing an agent feels paradoxical at first: how do you write a pass/fail test for a system whose whole job is to be non-deterministic? The model can phrase the same correct diagnosis a hundred different ways, and it costs money and network time every time you call it — so the naive instinct ("just run the agent and check the answer") is both flaky and expensive. The resolution is the most important testing lesson in the whole capstone: most of an agent isn't the model at all. The registry, the schemas, the risk tags, the block-parsing logic — that's ordinary, deterministic Python you can test the ordinary, deterministic, free way.

So the discipline is to split your tests along a bright line: tests that need an API key and tests that don't. Picture it like testing a car. You don't fire up the engine and drive onto the motorway to check that the seatbelts latch, the doors seal, and the dashboard lights work — you check all of that in the garage, engine off, in seconds. Only a few tests genuinely require a running engine and a test track. Agent tests are the same: the vast majority (does every tool have a risk class? does the registry return real cluster data? does the block parser emit one result per tool call?) run in the "garage" with no model, no key, no cost — so they can gate every single commit in CI. The handful that truly exercise the model's reasoning are the "test track" runs you do deliberately, not on every push.

Don't dismiss the no-key tests as trivial box-ticking, either — the trap is assuming the cheap tests are the unimportant ones. The exact opposite is true here. The "every tool has a risk class" test is one line and needs no API key, yet it is load-bearing safety infrastructure: the Lab 8c gate makes its allow/ask/block decision entirely off tool.risk, so a single tool registered with a typo'd or missing risk is a silent hole the gate can't see. A free, instant test that closes that hole on every commit is worth more than a dozen slow end-to-end runs. Cheap and load-bearing is the sweet spot you're aiming for. Two kinds of test. The registry test needs no API key; the diagnosis test does (it runs the model), so keep those separate so CI can run the free ones always.

tests/test_mock_and_tools.py (add)from agent.tools import TOOLS
from agent.schemas import RiskClass

def test_every_tool_has_a_risk_class():          # no API key
    for name, tool in TOOLS.items():
        assert isinstance(tool.risk, RiskClass), f"{name} missing risk"

def test_readonly_tool_runs_and_returns_data():   # no API key
    pods = TOOLS["kubectl_get_pods"].run(namespace="staging")
    assert any(p["status"] == "CrashLoopBackOff" for p in pods)

Troubleshooting expert

from dataclasses import dataclass
from typing import Callable
from agent.schemas import RiskClass

@dataclass
class Tool:
    name: str
    description: str
    input_schema: dict
    risk: RiskClass                      # ← the safety-critical field
    run: Callable

    def spec(self) -> dict:
        # the shape Claude's tools=[...] parameter wants
        return {"name": self.name, "description": self.description,
                "input_schema": self.input_schema}

from mock import cluster
from agent.schemas import RiskClass
from agent.tools import Tool   # (or same module)

TOOLS: dict[str, Tool] = {}
def register(tool): TOOLS[tool.name] = tool

def _obj(props, required):
    return {"type": "object", "properties": props, "required": required}

register(Tool(
    "kubectl_get_pods",
    "List pods with status and restart counts. Use FIRST when "
    "diagnosing — it shows what's unhealthy.",
    _obj({"namespace": {"type": "string"}}, []),
    RiskClass.READ_ONLY,
    lambda namespace="staging": cluster.get_pods(namespace),
))

register(Tool(
    "kubectl_logs",
    "Fetch recent log lines for a pod. Use to find the actual error.",
    _obj({"pod": {"type": "string"}, "namespace": {"type": "string"}}, ["pod"]),
    RiskClass.READ_ONLY,
    lambda pod, namespace="staging": cluster.get_logs(pod, namespace),
))

# tests/test_mock_and_tools.py (add) — no API key
def test_readonly_tool_runs_and_returns_data():
    pods = TOOLS["kubectl_get_pods"].run(namespace="staging")
    assert any(p["status"] == "CrashLoopBackOff" for p in pods)

def test_logs_tool_finds_error():
    logs = TOOLS["kubectl_logs"].run(pod="checkout-api-7d9f")
    assert "password authentication failed" in logs

from anthropic import Anthropic
from agent.tools import TOOLS
client = Anthropic()
MAX_STEPS = 8

def run(incident: str):
    system = ("You are a careful SRE. Diagnose the incident using the "
              "read-only tools. Cite the evidence you find.")
    messages = [{"role": "user", "content": incident}]
    specs = [t.spec() for t in TOOLS.values()]

    for _ in range(MAX_STEPS):                # hard cap — always terminates
        resp = client.messages.create(model="claude-opus-4-8",
                 max_tokens=1500, system=system, tools=specs, messages=messages)
        if resp.stop_reason == "end_turn":
            return next((b.text for b in resp.content if b.type == "text"), "")
        messages.append({"role": "assistant", "content": resp.content})
        results = []
        for b in resp.content:
            if b.type == "tool_use":
                out = TOOLS[b.name].run(**b.input)     # Lab 8c gates this
                results.append({"type": "tool_result",
                                "tool_use_id": b.id, "content": str(out)})
        messages.append({"role": "user", "content": results})
    return "Stopped: hit step limit."

from agent.tools import TOOLS

class _Block:  # stand-in for an Anthropic content block, for testing
    def __init__(self, type, name=None, input=None, id=None, text=None):
        self.type, self.name, self.input, self.id, self.text = type, name, input, id, text

def build_tool_results(content_blocks):
    results = []
    for b in content_blocks:
        if b.type == "tool_use":
            out = TOOLS[b.name].run(**b.input)
            results.append({"type": "tool_result",
                            "tool_use_id": b.id, "content": str(out)})
    return results

# test — no API key
def test_build_tool_results_one_per_tool_use():
    blocks = [
        _Block("text", text="let me look"),
        _Block("tool_use", name="kubectl_get_pods", input={"namespace": "staging"}, id="tu_1"),
    ]
    out = build_tool_results(blocks)
    assert len(out) == 1
    assert out[0]["tool_use_id"] == "tu_1"
    assert out[0]["type"] == "tool_result"

from agent.tools import TOOLS
from agent.schemas import RiskClass

def test_every_tool_has_a_risk_class():          # no API key
    assert TOOLS, "registry is empty — tools not imported"
    for name, tool in TOOLS.items():
        assert isinstance(tool.risk, RiskClass), f"{name} missing risk"

def test_registry_specs_are_model_safe():        # no API key
    for name, tool in TOOLS.items():
        spec = tool.spec()
        assert set(spec) == {"name", "description", "input_schema"}
        assert "risk" not in spec and "run" not in spec   # never leaked to the model

Knowledge check check yourself