Functions, Modules & OOP
This is where code becomes a system. The capstone splits into modules (agent/, mock/); tools are dataclasses; risk levels are enums; the stateful agent and vector store are classes. Learn these and the whole capstone architecture reads cleanly.
- an Anthropic API key (
ANTHROPIC_API_KEY) +pip install anthropic
Learning objectives
- Use
*args/**kwargsand unpacking — how tools get called. - Split code into modules and import between them — the capstone layout.
- Read type hints so function signatures tell you what goes in and out.
- Write classes for stateful things (the agent, the vector store).
- Use dataclasses and enums — the Tool registry and RiskClass.
1 · Functions, deeper intermediate
From P1 you know def, parameters, defaults, and return. Two more essentials: functions are values (you can store and pass them), and they can return multiple things (as a tuple).
pythondef get_weather(city):
return f"sunny in {city}"
# a function is a value — store it, pass it around
fn = get_weather
fn("Paris") # 'sunny in Paris'
dispatch = {"weather": get_weather} # map a name -> a function
dispatch["weather"]("Oslo") # call it by name
# return several values as a tuple
def answer(q):
return "the reply", ["chunk1", "chunk2"]
text, sources = answer("q") # unpack the two return values
# a lambda: a tiny one-line function (no name)
run = lambda city, unit="C": f"18°{unit} in {city}"
This block shows a big idea: in Python a function is itself a value, just like a number or a string. You can put it in a variable, store it in a dict, and pass it around — then call it later. It also shows returning several results at once.
fn = get_weather— no parentheses, so we are not calling the function; we are giving it a second name.fn("Paris")then runs the same function.dispatch = {"weather": get_weather}stores the function under a name in a dictionary.dispatch["weather"]("Oslo")looks it up by that string and calls it. This "name → function" table is exactly how an agent picks which tool to run.return "the reply", ["chunk1", "chunk2"]returns two things as a tuple.text, sources = answer("q")unpacks them into two variables in one line.lambda city, unit="C": ...is a tiny nameless function written on one line — handy when a function is so small that giving it a fulldefwould be overkill.
What the output means: Nothing prints on its own here; each line evaluates to a value (e.g. fn("Paris") is 'sunny in Paris'). In a notebook the last expression of a cell shows automatically.
Try this: Add a second entry to dispatch, like "echo": lambda s: s, then call dispatch["echo"]("hi"). You just extended a dispatch table without touching any of the existing code — that is the whole point of storing functions by name.
DISPATCH = {"get_weather": get_weather} then DISPATCH[block.name](...) in Ch 4. Returning (text, hits) is how Ch 3's answer() works. lambda is how each capstone Tool's run= is defined.2 · *args, **kwargs & unpacking intermediate → essential
The ** operator "spreads" a dict into keyword arguments. This is the exact mechanism the agent uses to call a tool with the arguments the model chose.
pythondef scale(name, replicas, namespace="staging"):
return f"{namespace}/{name} -> {replicas}"
args = {"name": "web", "replicas": 3}
scale(**args) # same as scale(name="web", replicas=3)
# a function that accepts any keyword args
def complete(messages, **kw): # kw is a dict of the extras
print(kw) # {'model': 'claude-opus-4-8', 'effort': 'high'}
complete([], model="claude-opus-4-8", effort="high")
The ** operator "spreads" a dictionary into named arguments of a function. This is the single trick that lets an agent call any tool with whatever arguments the model chose — without knowing them in advance.
args = {"name": "web", "replicas": 3}is a plain dict of argument values.scale(**args)is the magic: the**unpacks the dict so it becomesscale(name="web", replicas=3). The dict keys must match the parameter names.def complete(messages, **kw):does the reverse — the**kwcollects any extra keyword arguments the caller passes into a dict namedkw. So callingcomplete([], model="claude-opus-4-8", effort="high")makeskw == {'model': 'claude-opus-4-8', 'effort': 'high'}.
What the output means: print(kw) shows the collected extras as a dictionary: {'model': 'claude-opus-4-8', 'effort': 'high'}.
Try this: Add another argument, e.g. complete([], model="x", temperature=0.2), and watch temperature appear in the printed dict. This is why the agent's tool.run(**block.input) works no matter what fields a tool needs.
scale with {name:'web', replicas:3}", the agent runs tool.run(**block.input) — the ** turns that dict into the function's arguments. One line, and any tool can be called generically.tool.run(**block.input) is the heart of tool execution in Ch 4 and the capstone engine. **kw pass-through is in the Ch 1 client wrapper and Ch 6 traced_call.3 · Modules & imports essential
A module is just a .py file. import lets one file use another's functions/classes. A folder with an __init__.py is a package.
Requires: pip install anthropic pydantic
python# import a whole module, use dotted names
import json
json.load(f)
# import specific names from a module
from anthropic import Anthropic
from pydantic import BaseModel, Field
# import from your OWN package (the capstone does this)
from agent.schemas import RiskClass # from agent/schemas.py
from mock import cluster, actions # from mock/cluster.py, mock/actions.py
A module is just a .py file, and import lets one file borrow the functions and classes defined in another. This is how the capstone is split into agent/ and mock/ folders instead of one giant file.
import jsonbrings in the whole module; you then reach inside it with a dot:json.load(f). The name stays prefixed, so it's obvious whereloadcame from.from anthropic import Anthropicpulls out just one name so you can writeAnthropicdirectly instead ofanthropic.Anthropic.from agent.schemas import RiskClassimports from your own package. The dotted pathagent.schemasmeans "the fileschemas.pyinside the folderagent/". A folder becomes an importable package when it contains an__init__.pyfile.
Try this: If you ever hit ModuleNotFoundError, it almost always means you ran Python from the wrong folder, or a package folder is missing its __init__.py. Check both before assuming the code is broken.
ModuleNotFoundError almost always means you're running from the wrong directory, or a folder is missing its __init__.py. The capstone's troubleshooting tables (Lab 8a) cover this exactly — cd into the project root and ensure __init__.py exists.agent/ imports from mock/, the loop imports from agent.policy import evaluate, tests import both (Lab 8a layout). The shared llmkit.py is a module every lab imports.4 · Type hints intermediate
Type hints annotate what a function takes and returns. Python doesn't enforce them, but they make code readable and power tools like Pydantic. This course uses them throughout.
pythonfrom typing import Optional, Literal
def get_logs(pod: str, lines: int = 20) -> str:
# pod is a str, lines is an int (default 20), returns a str
return "..."
names: list[str] = ["a", "b"] # a list of strings
counts: dict[str, int] = {"a": 1} # dict: str keys, int values
ref: Optional[str] = None # str OR None
level: Literal["low", "high"] = "low" # only these exact values
Type hints are labels that say what kind of value each variable or argument holds. Python does not enforce them at runtime — they are documentation for humans and fuel for tools like editors and Pydantic. They make a function's contract readable at a glance.
def get_logs(pod: str, lines: int = 20) -> str:reads as:podshould be a string,linesan integer (defaulting to 20), and the function returns a string (that's what-> strmeans).names: list[str]means "a list whose items are strings";counts: dict[str, int]means "a dict with string keys and integer values".Optional[str]means "a string orNone" — use it when a value might be missing.Literal["low", "high"]means the value must be exactly one of those spelled-out choices, nothing else.
Try this: Change level to "medium". Python won't complain when you run it (hints aren't enforced), but an editor or Pydantic model using that Literal would flag it — which is exactly how the course keeps structured LLM output valid.
Literal[...] is what makes structured output safe — priority: Literal["low","medium","high"] in the Ch 2 Ticket and fix_risk: RiskClass in the capstone Diagnosis. Return hints like -> Sentiment | None appear in the Ch 2 classifier.5 · Classes (OOP) intermediate
A class bundles data + behavior. Use one when something has state that changes over time — like an agent that remembers a conversation, or a vector store that holds indexed chunks.
pythonclass Agent:
def __init__(self): # runs when you create one
self.messages = [] # state: this instance's history
def chat(self, text): # a method — self is "this instance"
self.messages.append({"role": "user", "content": text})
return f"({len(self.messages)} messages so far)"
a = Agent() # create an instance (__init__ runs)
a.chat("hi") # call a method
a.chat("again") # state persists on 'a'
A class is a blueprint that bundles data + behaviour together. You use one when a thing has state that changes over time — like an agent that remembers its conversation. Each object you build from the class (an instance) keeps its own private copy of that state.
def __init__(self):is the constructor — it runs automatically the moment you create an instance, and its job is to set up starting state. Here it makes an emptyself.messages = [].selfmeans "this particular instance". Storing data onselfis what lets oneAgentremember things independently of any otherAgent.def chat(self, text):is a method — a function that belongs to the class and receivesselfso it can read and change that instance's state.a = Agent()builds an instance (running__init__). Each call toa.chat(...)appends to a's own list, so the count grows across calls.
What the output means: a.chat("again") returns "(2 messages so far)" because the two messages accumulated on this one instance — the state persisted between calls.
Try this: Create a second agent b = Agent() and call b.chat("hi"). Its count starts at 1, proving each instance has its own messages list.
__init__ is the setup method; self refers to the specific instance so each object keeps its own state.
class Agent that remembers turns is Ch 4 Lab 4.5 and the capstone. class VectorStore holds the embedded chunks and has a .search() method in Ch 3 Lab 3.2. class Hybrid wraps BM25 in Lab 3.5.6 · Dataclasses intermediate
A dataclass is a class that's mostly just fields — Python writes the boilerplate __init__ for you. Perfect for a "record" like a Tool.
pythonfrom dataclasses import dataclass
from typing import Callable
@dataclass
class Tool:
name: str
description: str
risk: str
run: Callable # a function is a valid field!
t = Tool("kubectl_get", "list pods", "read_only", lambda: "pods...")
t.name # 'kubectl_get'
t.run() # 'pods...'
A dataclass is a shortcut for a class that mostly just holds a few fields. You list the fields with type hints and Python auto-writes the tedious __init__ (and a readable printout) for you. It's perfect for a plain "record" — like describing one tool.
@dataclassis a decorator: a line above the class that rewrites it, adding the constructor automatically from the fields below.name: str,description: str,risk: str,run: Callabledeclare four fields.Callablemeans "a function" — so a field can literally hold a function you'll call later.t = Tool("kubectl_get", "list pods", "read_only", lambda: "pods...")creates one in the order the fields were declared — no__init__written by hand.t.run()calls the stored function. Bundling the behaviour (run) together with its metadata (name, risk) in one object is what makes a tool registry tidy.
What the output means: t.name is 'kubectl_get' and t.run() returns 'pods...' — the lambda we stored gets executed on demand.
Try this: Add a field enabled: bool = True after run (fields with defaults must come last) and create a tool without passing it. The dataclass fills in the default for you.
@dataclass class Tool — name + description + risk + the run function, all in one record. Bundling the risk class with the tool is what lets the safety gate look it up.7 · Enums intermediate
An enum is a fixed set of named values. Use one when a value must be one of a known list — like risk levels. Typo-proof and readable.
Setup to run this snippet
class _Any:
'''stands in for any undefined demo value; supports call/attr/index/
iteration and basic arithmetic (as 0.7) so demo snippets run.'''
def __call__(self, *a, **k): return _Any()
def __getattr__(self, k): return _Any()
def __getitem__(self, k): return _Any()
def __iter__(self): return iter([])
def __len__(self): return 0
def __contains__(self, o): return True
def __enter__(self, *a): return _Any()
def __exit__(self, *a): return False
def __float__(self): return 0.7
def __int__(self): return 1
def __lt__(self, o): return True
def __gt__(self, o): return False
def __le__(self, o): return True
def __ge__(self, o): return False
def __add__(self, o): return o
def __radd__(self, o): return o
def __bool__(self): return True
def __repr__(self): return 'demo'
def __str__(self): return 'demo'
risk = _Any()pythonfrom enum import Enum, IntEnum
class RiskClass(str, Enum):
READ_ONLY = "read_only"
IRREVERSIBLE = "irreversible"
RiskClass.READ_ONLY # the member
RiskClass.READ_ONLY.value # 'read_only'
risk == RiskClass.IRREVERSIBLE # safe comparison — no typos possible
# IntEnum members are ordered — useful for "rungs"
class Rung(IntEnum):
OBSERVE = 1
ACT = 3
Rung.ACT > Rung.OBSERVE # True
An enum (enumeration) is a fixed, named set of allowed values. Reach for one when a value must be one of a known short list — like a risk level. It's typo-proof: you refer to RiskClass.READ_ONLY, not the loose string "read_only" that's easy to misspell.
class RiskClass(str, Enum):defines the set. Inheriting fromstrtoo means each member also behaves like its string value, which is handy for JSON and comparisons.READ_ONLY = "read_only"defines one member.RiskClass.READ_ONLYis the member object;RiskClass.READ_ONLY.valueis its underlying string'read_only'.risk == RiskClass.IRREVERSIBLEis a safe comparison — if you fat-finger the name, Python raises an error instead of silently comparing against a wrong string.class Rung(IntEnum):makes members that are also integers, so they can be ordered.Rung.ACT > Rung.OBSERVEisTruebecause 3 > 1 — useful for ranking permission levels.
What the output means: Rung.ACT > Rung.OBSERVE evaluates to True; enums as ordered rungs let the safety gate compare "how powerful is this action" numerically.
Try this: Compare two RiskClass members with < — you'll get a TypeError, because a plain Enum has no order. That's exactly why the ordered levels use IntEnum instead.
RiskClass and Rung are the two enums the entire capstone safety model is built on (Lab 8a, Lab 8c). The gate is literally _MATRIX[rung][risk] — enums as dict keys. The "irreversible never allowed" safety test compares enum members.8 · The mutable-default trap must-know gotcha
One Python quirk bites nearly everyone building agents: a mutable default argument (like [] or {}) is created once and shared across all calls. This causes conversation histories to leak between agent instances — a nasty, confusing bug.
python# ✗ WRONG — the default list is shared across every call!
def add_message(msg, history=[]): # created ONCE, reused
history.append(msg)
return history
add_message("a") # ['a']
add_message("b") # ['a', 'b'] ← 'a' leaked in! not a fresh list
# ✓ RIGHT — use None as the default, create inside
def add_message(msg, history=None):
if history is None:
history = [] # a fresh list every call
history.append(msg)
return history
This is one of Python's most famous traps, and it bites people building agents. A default argument is created only once — when the function is first defined — not fresh on each call. So a mutable default like [] is secretly shared by every call that relies on it.
def add_message(msg, history=[]):looks innocent, but that[]is built a single time. Every call that doesn't pass its ownhistoryreuses the same list.- So
add_message("a")gives['a'], but the next calladd_message("b")gives['a', 'b']— the earlier'a'leaked in because it was never a fresh list. - The fix: use
history=Noneas the default, thenif history is None: history = []inside the function. Now a brand-new list is created on every call that needs one.
What the output means: The wrong version accumulates across calls (['a'] then ['a', 'b']); the fixed version returns a clean ['a'] then ['b'].
Try this: Run the wrong version three times in a row and watch the list keep growing. This is the exact reason the course's agent sets self.messages = [] inside __init__ (runs per instance) rather than as a shared default.
Agent() would share one messages list — one user's conversation bleeding into another's. That's why the course's stateful agent creates self.messages = [] inside __init__ (which runs per instance), never as a class-level default.9 · *args and packing/unpacking, in full intermediate
P2 showed **kwargs for calling tools. Here's the complete picture — *args collects positional arguments, and */** unpack collections into calls. This is how the agent passes "all tool results in one message" and how variadic helpers work.
Uses the client / objects set up earlier in this lesson, and needs ANTHROPIC_API_KEY set. Run the earlier blocks first.
python# *args: accept any number of positional arguments (collected into a tuple)
def combine(*blocks):
return list(blocks)
combine("a", "b", "c") # ['a', 'b', 'c']
# unpack a list INTO positional args with *
results = [{"type":"tool_result"}, {"type":"tool_result"}]
# NewUserMessage(*results) would pass each result as a separate argument
# merge dicts with ** (build a request from a base + overrides)
base = {"model": "claude-opus-4-8", "max_tokens": 1024}
call = {**base, "max_tokens": 4096} # override just one key
# {'model': 'claude-opus-4-8', 'max_tokens': 4096}
# combine: forward extra kwargs to another function
def complete(messages, **kw):
return client.messages.create(messages=messages, **kw) # pass-through
This rounds out packing and unpacking. * deals with positional values (order matters, no names), and ** deals with keyword values (name=value). Collecting is "packing"; spreading them back into a call is "unpacking".
def combine(*blocks):— the*packs any number of positional arguments into a single tuple namedblocks.combine("a","b","c")makesblocks == ("a","b","c").NewUserMessage(*results)would unpack a list back out: each item becomes a separate positional argument — how the agent sends several tool results in one message.{**base, "max_tokens": 4096}builds a new dict frombaseand then overrides one key. This is the clean way to make a request variant without mutating the original.def complete(messages, **kw):thenclient.messages.create(messages=messages, **kw)shows pass-through: collect extra kwargs, then forward them onward unchanged.
What the output means: combine("a","b","c") is ['a','b','c']; the merged dict is {'model': 'claude-opus-4-8', 'max_tokens': 4096} — same model, bigger token budget.
Try this: Build {**base, "model": "claude-haiku"} and confirm base itself is unchanged afterward. Spreading always makes a copy, which is why it's safe for overrides.
NewUserMessage(*tool_results) sends all results in one turn (Ch 4). {**base, ...} merging builds request variants. **kw pass-through is the Ch 1 wrapper and Ch 6 traced_call.10 · Inheritance & super() intermediate → advanced
A subclass reuses and extends a parent class. You'll meet this when the SDK gives you a base class to subclass — like the memory-tool handler or a custom exception.
pythonclass Store:
def __init__(self, name):
self.name = name
self.items = []
def add(self, x):
self.items.append(x)
# VectorStore IS-A Store, plus embedding behavior
class VectorStore(Store):
def __init__(self, name, model):
super().__init__(name) # run the parent's __init__ first
self.model = model # then add our own state
def add(self, x):
super().add(x) # reuse parent behavior…
print(f"embedded & stored {x}") # …and extend it
vs = VectorStore("docs", "minilm")
vs.add("chunk1") # uses both
Inheritance lets one class build on another instead of copying code. The child (subclass) automatically gets the parent's methods, and can add new ones or tweak existing ones. Read it as an "is-a" relationship: a VectorStore is a Store with extra powers.
class VectorStore(Store):— the(Store)means "inherit everything fromStore". Without writing them,VectorStorealready hasStore's attributes and methods.super().__init__(name)calls the parent's constructor first, so the shared setup (self.name,self.items = []) still happens. Then the child adds its own state:self.model = model.def add(self, x):overrides the parent'sadd. Inside,super().add(x)reuses the parent's version (append to the list), and then the child extends it with an extraprint.
What the output means: vs.add("chunk1") both stores the item (parent behaviour) and prints embedded & stored chunk1 (the child's addition) — reuse plus extension.
Try this: Comment out the super().__init__(name) line and create a VectorStore. You'll get an error the first time self.items is used, because the parent's setup never ran. That's what super() was doing for you.
BetaAbstractMemoryTool, referenced in the tool docs). Custom exceptions (P4) subclass Exception. The pattern of "reuse the parent, add specifics" is how you'd extend the capstone's VectorStore or Agent.11 · __repr__, @property, static & class methods intermediate
Four class features that make your objects pleasant to use and debug.
pythonclass Diagnosis:
def __init__(self, cause, confidence):
self.cause = cause
self.confidence = confidence
# __repr__: what you see when you print the object (great for debugging)
def __repr__(self):
return f"Diagnosis(cause={self.cause!r}, conf={self.confidence:.2f})"
# @property: a method you access like an attribute (computed, read-only)
@property
def is_confident(self):
return self.confidence >= 0.8
# @staticmethod: a plain function grouped under the class (no self)
@staticmethod
def from_dict(d):
return Diagnosis(d["cause"], d["confidence"])
d = Diagnosis("db auth", 0.91)
print(d) # Diagnosis(cause='db auth', conf=0.91) ← thanks to __repr__
d.is_confident # True ← accessed like an attribute, no ()
Four small class features that make your objects nice to use and debug. "Dunder" methods (named with double underscores like __repr__) are hooks Python calls for you at special moments.
def __repr__(self):defines what shows when youprintthe object or view it in a REPL. Without it you'd get an unhelpful<Diagnosis object at 0x...>; with it you get a readable summary.{self.confidence:.2f}formats the number to 2 decimals.@propertyturns a method into something you read like a plain attribute —d.is_confidentwith no parentheses. It's computed on the fly and read-only, great for derived values like "is this confident enough?".@staticmethodmarks a function that lives under the class for tidiness but takes noself— it doesn't use instance state.Diagnosis.from_dict(d)is a common pattern: a helper that builds an instance from raw data.
What the output means: print(d) shows Diagnosis(cause='db auth', conf=0.91) thanks to __repr__; d.is_confident is True and is accessed without () because it's a property.
Try this: Add () after d.is_confident and run it — you'll get a TypeError because a property returns the value directly, not a callable. Properties look like data, not methods.
BaseModel (P4) auto-generates a useful __repr__ and validates on construction — which is why the course prefers Pydantic models over hand-written classes for data. But knowing @property and __repr__ helps you read any codebase.12 · Abstract base classes — defining an interface advanced
An ABC declares "any subclass must implement these methods." It's how SDKs give you a contract to fill in — like a memory backend where you supply the storage but the SDK drives the calls.
pythonfrom abc import ABC, abstractmethod
class MemoryBackend(ABC):
@abstractmethod
def read(self, path): ... # subclasses MUST implement
@abstractmethod
def write(self, path, data): ...
# a concrete file-based implementation
class FileMemory(MemoryBackend):
def read(self, path):
with open(path) as f: return f.read()
def write(self, path, data):
with open(path, "w") as f: f.write(data)
# MemoryBackend() -> TypeError: can't instantiate abstract class
mem = FileMemory() # OK — it implements both methods
An abstract base class (ABC) defines a contract: it lists methods that any subclass is required to implement, but provides no working body itself. It's how an SDK hands you a shape to fill in — you supply the storage, the SDK calls your methods.
class MemoryBackend(ABC):inherits fromABC, marking it abstract — a template, not something you use directly.@abstractmethodabovedef read(...)anddef write(...)means "every subclass must provide these". The...body is just a placeholder; there's no real implementation here.class FileMemory(MemoryBackend):is a concrete subclass that actually implementsreadandwriteusing real files.- Trying
MemoryBackend()directly raisesTypeError: can't instantiate abstract class— Python enforces the contract.FileMemory()works because it filled in both required methods.
What the output means: mem = FileMemory() succeeds; instantiating the abstract MemoryBackend itself would fail with a TypeError.
Try this: Delete FileMemory's write method and try to create one. Python refuses, listing write as still abstract — the ABC guarantees you can't forget part of the interface.
BetaAbstractMemoryTool and implement view/create/str_replace/etc.; the tool runner calls them (referenced in Ch 4 tool concepts). The capstone's audit backend follows the same "define an interface, swap the implementation" idea when you go from in-memory to durable storage.Worked example · a mini tool registry putting it together
Dataclass + enum + type hints + dispatch + **kwargs + __repr__ — a runnable miniature of the capstone's tool system, safe to run with no API key.
mini_registry.pyfrom dataclasses import dataclass
from enum import Enum
from typing import Callable
class Risk(str, Enum):
READ_ONLY = "read_only"
WRITE = "write"
@dataclass
class Tool:
name: str
risk: Risk
run: Callable
def __repr__(self):
return f"Tool({self.name}, {self.risk.value})"
REGISTRY: dict[str, Tool] = {}
def register(tool): REGISTRY[tool.name] = tool
register(Tool("get_pods", Risk.READ_ONLY, lambda ns="staging": f"pods in {ns}"))
register(Tool("scale", Risk.WRITE, lambda name, n: f"scaled {name} to {n}"))
def call(name, allow_writes=False, **kwargs): # a tiny policy gate
tool = REGISTRY[name]
if tool.risk == Risk.WRITE and not allow_writes:
return f"BLOCKED: {tool.name} is a write and writes are off"
return tool.run(**kwargs) # unpack args into the tool
if __name__ == "__main__":
print(REGISTRY["get_pods"]) # Tool(get_pods, read_only)
print(call("get_pods", ns="prod")) # pods in prod
print(call("scale", name="web", n=3)) # BLOCKED (writes off)
print(call("scale", allow_writes=True, name="web", n=3)) # scaled web to 3
Tool(get_pods, read_only)
pods in prod
BLOCKED: scale is a write and writes are off
scaled web to 3
This is everything in P3 working together: a dataclass, an enum, type hints, a registry dict, **kwargs, and __repr__ — a runnable miniature of the capstone's tool system. It answers "which tools exist, and may I run this one right now?"
class Risk(str, Enum)fixes the allowed risk levels;@dataclass class Toolbundles a tool'sname, itsrisk, and therunfunction into one record (with a friendly__repr__).REGISTRY: dict[str, Tool] = {}is the lookup table, andregister(tool)adds a tool under its name. Two tools get registered — a read-only one and a write one.def call(name, allow_writes=False, **kwargs):is the policy gate. It finds the tool, and if the tool is aWRITEwhileallow_writesis off, it returns aBLOCKEDmessage instead of running it.return tool.run(**kwargs)is the payoff line: the**kwargsunpacks whatever arguments were passed straight into the stored function — the same generic-call trick from section 2, now inside a real gate.if __name__ == "__main__":runs the demo only when you execute this file directly, so importing it elsewhere stays quiet.
What the output means: The four prints show: the tool's __repr__; a successful read (pods in prod); a blocked write (BLOCKED: scale ...); and finally the same write allowed through with allow_writes=True giving scaled web to 3.
Try this: Add a third tool with Risk.READ_ONLY and call it through call(...) with no allow_writes — it runs freely because only writes are gated. You've just rebuilt the shape of the capstone's safety policy.
agent/tools.py + agent/policy.py — a dataclass tool with a risk enum, a registry, and a gate that blocks writes. The real one adds more risk levels and rungs, but the structure is exactly this.Exercises expert
Exercise P3.1 — tool dispatch
Context: An agent picks an action by name, and the cleanest dispatch is a dict mapping names to functions — the agent's dispatch table in miniature.
Your task: Make a dict mapping add and mul to two small functions, then call the right one by looking up a name string.
Requirements:
- A dict mapping two names to functions
- Two small functions (add and multiply)
- Look up by a name string and call the result
💡 Hint: Store the functions as dict values and call table[name](...); this is how an agent routes a chosen tool.
Solution
ops = {"add": lambda a,b: a+b, "mul": lambda a,b: a*b}
ops["add"](2, 3) # 5Exercise P3.2 — a Tool dataclass
Context: A tool is naturally a dataclass carrying a callable, so its behaviour lives right next to its metadata — the shape the registry builds on.
Your task: Define a @dataclass Tool with name, risk, and run; create one whose run is a lambda returning a string, and call t.run().
Requirements:
- A dataclass with name, risk, and run fields
- run holds a callable (a lambda)
- Instantiate with a string-returning lambda
- Call
t.run()and get the string
💡 Hint: Give the dataclass a run field typed as a callable and assign a lambda; invoke it with t.run().
Exercise P3.3 — a risk enum + comparison
Context: Classifying risk with an Enum lets a policy function decide danger by identity — here flagging only the irreversible member, a preview of the approval gate.
Your task: Define a RiskClass enum with three members and a function is_dangerous(risk) returning True only for the irreversible member.
Requirements:
- An enum with three risk members
- A function taking a risk member
- Returns True only for the irreversible member
- Returns False for the other two
💡 Hint: Compare the argument against the irreversible enum member; identity comparison is cleaner and safer than matching a raw string.
Solution
Setup to run this snippet
class _RiskClass_t:
_ = None
def __getattr__(self, k): return 'demo'
RiskClass = _RiskClass_t()def is_dangerous(risk):
return risk == RiskClass.IRREVERSIBLE🎯 Interview practice interview
The interview questions this topic gets asked — worked, with code. For the full pattern catalog see A9 · Big Tech AI-engineering patterns.
The classic OOP design question: O(1) get/put via a dict + ordering. OrderedDict makes it clean; the from-scratch dict+DLL version is in D7.
pythonfrom collections import OrderedDict
class LRUCache:
def __init__(self, cap):
self.cap = cap
self.d = OrderedDict()
def get(self, key):
if key not in self.d: return -1
self.d.move_to_end(key) # mark most-recent
return self.d[key]
def put(self, key, val):
if key in self.d: self.d.move_to_end(key)
self.d[key] = val
if len(self.d) > self.cap:
self.d.popitem(last=False) # evict least-recent
An LRU (Least Recently Used) cache keeps a fixed number of items and, when full, throws away the one untouched for the longest. This is the classic object-oriented design interview question; OrderedDict makes it clean because it remembers insertion/access order.
self.d = OrderedDict()stores the items in an order-aware dict;self.capis the maximum size.get: if the key is missing, return-1. Otherwiseself.d.move_to_end(key)bumps it to the "most recently used" end, then returns its value. Touching an item makes it recent — that's the whole idea.put: if the key already exists, move it to the recent end; then set the value.if len(self.d) > self.cap:means we just overflowed, soself.d.popitem(last=False)evicts from the front — the least-recently-used item.
What the output means: After exceeding capacity, the item you accessed or added least recently is the one dropped. Both get and put run in O(1) average time.
Try this: Create c = LRUCache(2), then put(1,1), put(2,2), get(1), put(3,3). Key 2 is evicted, because get(1) made 1 more recent than 2.
LLD rounds love "design the classes for X." Enums for suits/ranks, a frozen dataclass for a card, a class for the deck.
pythonfrom dataclasses import dataclass
from enum import Enum
class Suit(Enum): SPADE=1; HEART=2; DIAMOND=3; CLUB=4
@dataclass(frozen=True)
class Card:
rank: int # 2..14
suit: Suit
class Deck:
def __init__(self):
self.cards = [Card(r, s) for s in Suit for r in range(2, 15)]
"Design the classes for X" is a staple of low-level-design interviews. A deck of cards is the friendly version: it shows off enums for fixed categories, a frozen dataclass for an immutable value, and a normal class to hold the collection.
class Suit(Enum): SPADE=1; HEART=2; ...— an enum is the right tool because there are exactly four suits and no others should ever exist.@dataclass(frozen=True)onCardmakes each card immutable — once created, itsrankandsuitcan't change. Frozen dataclasses are also hashable, so cards can live in sets or act as dict keys.rank: intwith the comment# 2..14means 2 through 14 (11-14 being Jack/Queen/King/Ace);suit: Suitreuses the enum as the field's type.self.cards = [Card(r, s) for s in Suit for r in range(2, 15)]is a nested comprehension: for each suit, for each rank 2..14, build a card — producing all 52 cards in one line.
What the output means: A fresh Deck() holds 52 unique Card objects (4 suits × 13 ranks). Because cards are frozen, two Card(14, Suit.SPADE) values are interchangeable.
Try this: Add a deal(self, n) method that pops n cards off self.cards and returns them. Notice you never need to guard suits against typos — the Suit enum already made that impossible.
🪜 Practice ladder beginner → industry
Six graded exercises, easy to real-world. Try each before opening its solution.
Context: A dataclass models a small record with type hints and gives you a readable repr for free — the lesson's mini tool registry starts here.
Your task: Model a tool as a @dataclass with name: str and risk: str, create one, and print it.
Requirements:
- A
@dataclasswith two typed fields - name and risk are strings
- Instantiate one tool
- Print it to show the auto-generated repr
💡 Hint: Decorate with @dataclass and add two annotated fields; printing the instance shows the free repr.
Show solution
Dataclass + type hints from sections 4 and 6. Runnable:
from dataclasses import dataclass
@dataclass
class Tool:
name: str
risk: str
t = Tool("get_pods", "read_only")
print(t) # Tool(name='get_pods', risk='read_only')
Context: Free-text risk invites typos and invalid values. An Enum makes the set explicit, and inheriting from str keeps the value JSON-friendly.
Your task: Replace the free-text risk with an Enum (READ_ONLY, WRITE) inheriting from str, Enum so the value is JSON-friendly, and print the value.
Requirements:
- An Enum with READ_ONLY and WRITE members
- Inherits from
str, Enum - The tool's risk field is the Enum type
- Print the member's
.value
💡 Hint: class Risk(str, Enum) gives members whose .value is the plain string, so they serialize cleanly.
Show solution
The lesson's class Risk(str, Enum). Runnable:
from enum import Enum
from dataclasses import dataclass
class Risk(str, Enum):
READ_ONLY = "read_only"
WRITE = "write"
@dataclass
class Tool:
name: str
risk: Risk
t = Tool("scale", Risk.WRITE)
print(t.name, t.risk.value) # scale write
Context: A tool registry maps names to tools and dispatches calls, and a policy gate blocks write tools unless writes are explicitly allowed — the core of the P3 worked example.
Your task: Build the registry: REGISTRY + register + call(name, allow_writes=False, **kwargs) that blocks WRITE tools unless writes are allowed and passes args through with **kwargs.
Requirements:
- A registry dict and a register function
- Tools carry a callable to run
- call dispatches by name
- WRITE tools are blocked unless
allow_writesis True - Arguments pass through via
**kwargs
💡 Hint: Look the tool up by name, check its risk against allow_writes before running, and forward **kwargs to the tool's callable.
Show solution
The core of the P3 worked example: dispatch + a policy gate + **kwargs. Runnable:
from dataclasses import dataclass
from enum import Enum
from typing import Callable
class Risk(str, Enum):
READ_ONLY = "read_only"; WRITE = "write"
@dataclass
class Tool:
name: str; risk: Risk; run: Callable
REGISTRY: dict[str, Tool] = {}
def register(tool): REGISTRY[tool.name] = tool
register(Tool("get_pods", Risk.READ_ONLY, lambda ns="staging": f"pods in {ns}"))
register(Tool("scale", Risk.WRITE, lambda dep, n: f"scaled {dep} to {n}"))
def call(tool_name, allow_writes=False, **kwargs):
tool = REGISTRY[tool_name]
if tool.risk == Risk.WRITE and not allow_writes:
return f"BLOCKED: {tool.name} is a write and writes are off"
return tool.run(**kwargs)
print(call("get_pods", ns="prod"))
print(call("scale", dep="web", n=3))
print(call("scale", allow_writes=True, dep="web", n=3))
Context: The mutable-default-argument trap — def f(x, seen=[]) — silently shares one list across calls. The fix is the None-sentinel idiom that makes a fresh list each call.
Your task: Write register_all(tools, into=None) that appends to a registry list using the None-sentinel idiom so calls don't share state.
Requirements:
- Default the accumulator to
None, not a list - Create a fresh list inside when it's None
- Append the tools to it and return it
- Two separate calls produce independent lists
💡 Hint: Guard with if into is None: into = [] at the top; never put a list literal in the default, or every call shares it.
Show solution
Section 8's mutable-default trap, fixed the canonical way. Runnable — the two calls stay independent:
def register_all(tools, into=None):
if into is None: # fresh list per call, NOT a shared default
into = []
into.extend(tools)
return into
a = register_all(["get_pods"])
b = register_all(["scale"])
print(a) # ['get_pods'] -- not polluted by b
print(b) # ['scale']
Context: An abstract base class defines an interface: subclasses must implement the abstract method, and instantiating an incomplete one fails — enforcing the contract.
Your task: Use abc.ABC to define a Tool interface with an abstract run(), then two concrete tools, and show an incomplete subclass fails to instantiate.
Requirements:
- A
Tool(ABC)with an@abstractmethod run() - Two concrete subclasses implement run
- The concrete tools instantiate and run
- A subclass missing run raises
TypeErroron instantiation
💡 Hint: Mark run with @abstractmethod; Python refuses to instantiate any subclass that hasn't implemented it.
Show solution
Abstract base class enforcing an interface. Runnable:
from abc import ABC, abstractmethod
class Tool(ABC):
name: str
@abstractmethod
def run(self, **kwargs) -> str: ...
class GetPods(Tool):
name = "get_pods"
def run(self, ns="staging"): return f"pods in {ns}"
print(GetPods().run(ns="prod"))
try:
class Broken(Tool):
name = "broken" # forgot run()
Broken()
except TypeError as e:
print("cannot instantiate:", type(e).__name__)
Context: The chapter's patterns cohere into the capstone's tool system: a dataclass + enum tool, dispatch through a registry, and a policy gate that returns a structured result with an allowed flag.
Your task: Combine dataclass + enum + dispatch + policy into a registry a real agent loop calls: run(name, allow_writes, **kwargs) returning a result dict with an allowed flag; assert a write is blocked by default.
Requirements:
- A dataclass tool with an Enum risk and a callable
- A registry class with register and run
- run returns a dict including an
allowedflag and the result - WRITE tools are blocked when writes are off
- Assert a write is blocked by default
💡 Hint: Return {"tool": ..., "allowed": bool, "result": ...} so the agent loop can branch on allowed; default allow_writes False and assert the block.
Show solution
The P3 patterns cohering into the capstone's tool system shape. Runnable:
from dataclasses import dataclass
from enum import Enum
from typing import Callable
class Risk(str, Enum):
READ_ONLY = "read_only"; WRITE = "write"
@dataclass
class Tool:
name: str; risk: Risk; run: Callable
class Registry:
def __init__(self): self._tools: dict[str, Tool] = {}
def register(self, tool): self._tools[tool.name] = tool
def run(self, tool_name, allow_writes=False, **kwargs):
tool = self._tools[tool_name]
if tool.risk == Risk.WRITE and not allow_writes:
return {"tool": tool_name, "allowed": False, "result": None}
return {"tool": tool_name, "allowed": True, "result": tool.run(**kwargs)}
reg = Registry()
reg.register(Tool("scale", Risk.WRITE, lambda dep, n: f"scaled {dep} to {n}"))
blocked = reg.run("scale", dep="web", n=3)
assert blocked["allowed"] is False
print(blocked)
print(reg.run("scale", allow_writes=True, dep="web", n=3))
✓ Checkpoint — ready for P4 when you can…
- Store a function in a dict and call it by name (tool dispatch).
- Explain what
tool.run(**block.input)does. - Split code across modules and import between them; fix a
ModuleNotFoundError. - Read a type-hinted signature, including
OptionalandLiteral. - Write a class with
__init__and a method that usesself. - Define a dataclass and an enum, and use the enum in a comparison.
Knowledge check check yourself
In this lesson, why does the agent's tool dispatch use tool.run(**block.input), and what does the ** operator actually do at the call site?
Show answer
** spreads a dict into named (keyword) arguments, turning something like {name:'web', replicas:3} into run(name='web', replicas=3). This lets the agent call any tool generically with whatever arguments the model chose, without knowing the tool's parameters in advance.The lesson calls the mutable-default trap a nasty bug for agents. What exactly goes wrong with def add_message(msg, history=[]), and what is the fix the course applies?
Show answer
[] is created once when the function is defined and shared across every call, so messages leak/accumulate across calls (and across agent instances). The fix is history=None plus if history is None: history = [] inside the function, which is why the course's agent sets self.messages = [] inside __init__ (per instance) rather than as a shared default.