Appendix · Python for AI Agents · Part 6

Real-World Python Engineering

The final part: the Python that separates a script from a system. Memory-safe processing at scale, correct resource & connection handling, production resilience (rate-limit + circuit-breaker), config/logging done right, a real pytest toolkit, packaging your agent as an installable CLI, and the quality toolchain pros run in CI — capped by a real-world mini-project that ties it all together.

⏱️ ~2.5 hours🎯 Expert · production🏭 real engineeringrunnable

⚙️ To run this for realThe code here is complete and correct as written. To actually execute it you'll need:

an Anthropic API key (ANTHROPIC_API_KEY) + pip install anthropic

Reading and learning works without any of this — run when you're ready. IDs/ARNs in examples are placeholders; swap in your own.

Learning objectives

Process large data without exploding memory (generators, streaming, __slots__).
Manage resources & connections correctly (pools, cleanup, reentrancy).
Implement production resilience: token-bucket rate limiting + a circuit breaker.
Do configuration, logging, and error reporting the way production teams do.
Write a real pytest suite: fixtures, parametrize, async, coverage, markers.
Package your agent as an installable command-line tool and run a quality toolchain.

What this is (and isn't) optional, max-level

This page is maximal on purpose — it's the "learn as much as possible" material for someone who wants to build agents like a professional software engineer, not just make them work. None of it is required for the course or projects. Skim the headers; go deep on whatever matches what you're building. Everything is runnable and grounded in real agent scenarios.

1 · Memory & performance at scale expert

When you process thousands of documents or a huge log stream, loading everything into a list will exhaust memory. Generators keep memory flat; __slots__ shrinks objects; profiling finds the real bottleneck.

Example code for learning — review, test, and adapt it before running against real or production systems. Commands can create, change, or delete resources. See the Terms & Disclaimer.

Try it

Illustrative fragment — defines demo values / files are needed before this runs standalone.

python# ✗ loads the whole file into memory
lines = open("huge.log").readlines()
errors = [l for l in lines if "ERROR" in l]

# ✓ streams line by line — constant memory, even for a 10GB file
def error_lines(path):
    with open(path) as f:
        for line in f:              # file objects are lazy iterators
            if "ERROR" in line:
                yield line             # produce one at a time

# generator pipelines compose without intermediate lists
def parse(lines):
    for l in lines:
        yield l.split("|", 2)
first_10 = list(islice(parse(error_lines("huge.log")), 10))   # only reads what it needs

# __slots__: drop the per-instance __dict__ -> big memory + speed win for
# millions of small objects (e.g. chunk records in a RAG index)
class Chunk:
    __slots__ = ("id", "text", "vec")      # no arbitrary attributes, less RAM
    def __init__(self, id, text, vec):
        self.id, self.text, self.vec = id, text, vec

▶ How this works

The big idea here is don't load a giant file into memory all at once. The first two lines read an entire log into a list — fine for a small file, fatal for a 10 GB one. The rest of the block shows the memory-safe way using generators (functions that hand back values one at a time).

open(path) inside a with block gives you a file object you can loop over line by line. Python reads only one line at a time from disk, so memory stays flat no matter how huge the file is.
yield line is what makes error_lines a generator. Instead of building a whole list and returning it, yield produces one value, pauses, and resumes on the next request. Nothing is computed until someone asks for it.
parse is a second generator that takes the lines from the first and splits each one. Chaining generators like this is a pipeline — data flows through without ever creating a big intermediate list.
islice(..., 10) pulls just the first 10 results, so the pipeline reads only as much of the file as it needs and then stops.
__slots__ = ("id", "text", "vec") tells Python this object will only ever have those three attributes. That lets Python skip the per-object dictionary it normally keeps, saving a lot of RAM when you have millions of these little objects.

What the output means: Nothing prints on its own — these are building blocks. The payoff is that error_lines and parse use roughly the same tiny amount of memory whether the log is 1 MB or 10 GB, because they never hold the whole thing at once.

Try this: Picture a 10 GB log. The first approach (readlines()) would try to allocate 10 GB of RAM and crash; the generator version reads a line, checks it, and forgets it. That difference is why generators matter at scale.

Measure, don't guessBefore optimizing, profile: python -m cProfile -s cumtime script.py, or py-spy top --pid <pid> on a running service. The bottleneck is almost never where you think — and in agent apps it's usually the API latency, not your Python, so concurrency (P5 §5) beats micro-optimization.

🔗 Course relevanceGenerators are how you'd process the doc-intel project's invoice folder or the DevOps agent's log streams without loading everything. __slots__ matters when the Ch 3 RAG index holds hundreds of thousands of chunks.

2 · Resource & connection management expert

DB connections, HTTP clients, and file handles are limited resources. Leaking them crashes a service under load. The rules: reuse clients, pool connections, and always clean up.

Try it

Uses the client / objects set up earlier in this lesson, and needs ANTHROPIC_API_KEY set. Run the earlier blocks first.

python# ✗ creating a client per request leaks connections & is slow
def bad_handler(q):
    client = Anthropic()          # new connection pool every call!
    return client.messages.create(...)

# ✓ create ONE client at module load, reuse it (the SDK pools connections)
client = Anthropic()              # module-level singleton
def good_handler(q):
    return client.messages.create(...)

# a reusable connection with guaranteed cleanup via a context manager
from contextlib import contextmanager
import sqlite3

@contextmanager
def readonly_db(path):
    con = sqlite3.connect(f"file:{path}?mode=ro", uri=True)
    try:
        yield con
    finally:
        con.close()               # ALWAYS closes, even on exception

with readonly_db("shop.db") as con:
    rows = con.execute("SELECT ...").fetchall()

# ExitStack: manage a dynamic number of resources cleanly
from contextlib import ExitStack
with ExitStack() as stack:
    files = [stack.enter_context(open(p)) for p in paths]
    ...                           # all closed when the block exits

▶ How this works

This block is about not leaking limited resources — network connections, database handles, open files. Each has a hard cap, and forgetting to close them will eventually take a service down under load. The two rules shown: reuse one client instead of making new ones, and always clean up with a context manager.

bad_handler creates a new Anthropic() client on every call. Each client opens its own pool of network connections, so this slowly leaks connections and is slow. The fix right below it creates one client at module load and reuses it everywhere.
The @contextmanager decorator turns readonly_db into something you can use with with. The code before yield is setup (open the connection); the code after is teardown (close it).
The try / finally is the whole point: finally runs no matter what — even if your query raises an error — so con.close() always happens and the connection is never leaked.
with readonly_db("shop.db") as con: is how you use it. When the with block ends (normally or via an exception), Python automatically runs the teardown for you.
ExitStack handles the case where you don't know how many resources you'll open. Every file added with stack.enter_context(...) is closed together when the block exits — clean cleanup for a dynamic list of resources.

What the output means: Again, structure rather than printed output. The lesson is behavioural: the with versions guarantee the connection or files are closed even when something throws, which is what keeps a long-running service healthy.

Try this: Imagine the query on the SELECT line crashes. Trace what happens: the exception jumps to finally, con.close() still runs, then the error propagates. That guarantee is the reason to prefer with over manually calling .close().

🔗 Course relevanceThe data-analyst project's readonly_db is exactly this pattern — and reusing one Anthropic() client (not one per request) is the fix for the connection leak that kills a naive support-bot under load (Ch 6 scaling).

3 · Resilience patterns — rate limiting & circuit breaker expert · production

P4 showed retry+backoff. Production adds two more: rate limiting (don't exceed your quota) and a circuit breaker (stop hammering a failing dependency). Both are short, real, and reusable.

Try it — token-bucket rate limiter

ratelimit.pyimport time, threading

class RateLimiter:
    """Allow `rate` calls per second (token bucket). Thread-safe."""
    def __init__(self, rate: float, burst: int):
        self.rate, self.capacity = rate, burst
        self.tokens = burst
        self.updated = time.monotonic()
        self.lock = threading.Lock()

    def acquire(self):
        with self.lock:
            now = time.monotonic()
            self.tokens = min(self.capacity,
                              self.tokens + (now - self.updated) * self.rate)
            self.updated = now
            if self.tokens < 1:
                sleep = (1 - self.tokens) / self.rate
                time.sleep(sleep)         # wait for a token
                self.tokens = 0
            else:
                self.tokens -= 1

limiter = RateLimiter(rate=10, burst=20)   # 10/s, bursts to 20
def call(...): limiter.acquire(); return client.messages.create(...)

▶ How this works

This is a token-bucket rate limiter — the standard way to stay under an API's calls-per-second quota. Picture a bucket that refills with tokens over time; every call must spend one token, and if the bucket is empty the call waits. rate is how fast tokens refill; burst is the bucket's max size (how many you can spend at once).

__init__ stores the settings and starts with a full bucket (self.tokens = burst). time.monotonic() is a clock that only moves forward — perfect for measuring elapsed time (unlike wall-clock time, it can't jump backward).
self.lock = threading.Lock() makes the limiter thread-safe: if several threads call at once, the with self.lock: block lets only one adjust the token count at a time, so two threads can't both think a token is free.
Every acquire() first refills: (now - self.updated) * self.rate is how many tokens accumulated since last time. min(self.capacity, ...) caps it so the bucket never overflows past burst.
If there's less than one token, we time.sleep(...) for exactly long enough for one token to refill, then proceed. Otherwise we simply spend a token (self.tokens -= 1) and return immediately.

What the output means: A caller doing limiter.acquire() before each request will be allowed to burst up to 20 quick calls, then automatically throttle to about 10 per second — sleeping just enough to stay under the limit.

Try this: Set rate=1, burst=1 and imagine calling acquire() in a tight loop: the first returns instantly, then each one sleeps ~1 second. That's the bucket forcing you down to 1 call/sec.

Try it — circuit breaker

breaker.pyimport time

class CircuitBreaker:
    """Trip open after N consecutive failures; recover after a cooldown."""
    def __init__(self, threshold=5, cooldown=30):
        self.threshold, self.cooldown = threshold, cooldown
        self.failures, self.opened_at = 0, None

    def call(self, fn, *a, **k):
        if self.opened_at and time.monotonic() - self.opened_at < self.cooldown:
            raise RuntimeError("circuit open — failing fast")   # don't even try
        try:
            result = fn(*a, **k)
        except Exception:
            self.failures += 1
            if self.failures >= self.threshold:
                self.opened_at = time.monotonic()       # trip open
            raise
        self.failures, self.opened_at = 0, None       # success resets
        return result

▶ How this works

A circuit breaker protects you from a dependency that has gone down. Like an electrical breaker, it "trips" after too many failures and then fails fast for a while instead of letting every call hang and pile up. It has two settings: threshold (how many failures trip it) and cooldown (how long it stays tripped).

State lives in two fields: self.failures counts consecutive failures, and self.opened_at records when the breaker tripped (or None if it's healthy).
call(self, fn, *a, **k) wraps any function. First it checks: if the breaker is open and we're still inside the cooldown window, it raises immediately with "circuit open — failing fast" — it doesn't even attempt the call.
It then tries to run fn. On an exception it bumps self.failures; once that reaches threshold, it sets self.opened_at to now (trips open) and re-raises the error.
On success the last line resets self.failures to 0 and clears opened_at — so a single good call closes the breaker again and normal service resumes.

What the output means: With threshold=5: the first 5 failures pass the error through while counting up; the 6th and later calls fail instantly for 30 seconds without touching the broken dependency, giving it time to recover.

Try this: Compare the cost: without a breaker, a dead backend means every call waits for a timeout (seconds each) and stacks up. With one, calls fail in microseconds after it trips — saving latency, threads, and money.

Why a breaker matters for agentsIf a tool's backend (a database, the GitHub API, a monitoring endpoint) goes down, retrying every call just piles up latency and cost. A breaker fails fast after N failures, gives the dependency time to recover, then retries. Combined with retry+backoff (P4) and the rate limiter above, you have the full resilience stack the Ch 6 "circuit breaker / graceful degradation" section describes.

🔗 Course relevanceThe rate limiter is the "token-bucket limiter" named in Ch 6 scaling and the DevOps/analyst projects (respect API + DB limits). The breaker protects the agent when a tool's backend is down — degrade gracefully instead of hanging.

4 · Configuration & logging, done right production

Try it — typed settings

config.pyfrom pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_file=".env", env_prefix="AGENT_")
    anthropic_api_key: str                     # required — fails fast if missing
    model: str = "claude-opus-4-8"
    max_tokens: int = 1024
    confidence_threshold: float = 0.6
    rung: str = "observe"

settings = Settings()      # validated once at startup; typed everywhere after

▶ How this works

This replaces scattered os.environ.get("...") calls with one typed, validated settings object. It uses pydantic-settings, which reads values from environment variables (and a .env file), converts them to the right type, and fails at startup if something required is missing.

The class inherits from BaseSettings. The model_config line says: load from a file named .env, and every env var is prefixed with AGENT_ (so anthropic_api_key comes from AGENT_ANTHROPIC_API_KEY).
Each attribute is a setting with a type. anthropic_api_key: str has no default, so it's required — if it's missing, creating the object errors immediately instead of failing confusingly mid-request later.
The others have defaults (model, max_tokens = 1024, confidence_threshold = 0.6). Pydantic also coerces types: an env var is always text, but max_tokens: int means the string "1024" is turned into the number 1024 for you.
settings = Settings() runs the validation once, at import time. After that, settings.model and friends are typed and safe to use everywhere in the codebase.

What the output means: If the API key isn't set, the program stops at launch with a clear validation error naming the missing field — far better than a cryptic failure halfway through handling a user request.

Try this: Think about the alternative: with os.environ.get(), a missing key returns None and blows up later with a confusing message. "Fail fast at startup" means you find out the moment you run, not in production.

Try it — structured logging

logging_setup.pyimport logging, json, sys

class JsonFormatter(logging.Formatter):
    def format(self, record):
        return json.dumps({
            "ts": self.formatTime(record),
            "level": record.levelname,
            "msg": record.getMessage(),
            **getattr(record, "extra_fields", {}),   # attach trace_id, tokens...
        })

handler = logging.StreamHandler(sys.stdout)
handler.setFormatter(JsonFormatter())
logging.basicConfig(level=logging.INFO, handlers=[handler])
log = logging.getLogger("agent")

# one structured line per request -> queryable in any log tool
log.info("llm_call", extra={"extra_fields":
    {"trace_id": "abc", "model": "opus", "out_tokens": 120}})

▶ How this works

This sets up structured logging: instead of printing plain sentences, each log line is a JSON object. Machines can filter and search JSON logs ("show me all calls where out_tokens > 100"), which plain text makes painful. This is the format real log dashboards expect.

JsonFormatter subclasses Python's built-in logging.Formatter and overrides format. Whatever this method returns is the text of one log line — here, a JSON string built with json.dumps(...).
The dict includes a timestamp, the level (INFO/WARNING/...), and the message. The **getattr(record, "extra_fields", {}) line merges in any custom fields you attached to that log call, like a trace_id or token counts.
The handler decides where logs go (here, sys.stdout) and uses our JSON formatter. basicConfig(level=logging.INFO, ...) wires it up and sets the minimum level to record.
The final log.info("llm_call", extra={"extra_fields": {...}}) shows the payoff: you log an event plus structured data (trace_id, model, out_tokens) that flows straight into the JSON line.

What the output means: Each call to log.info(...) prints one line like {"ts": "...", "level": "INFO", "msg": "llm_call", "trace_id": "abc", "out_tokens": 120} — a record a log tool can index and query.

Try this: Compare to print(f"call for {trace_id}"). The JSON version lets you later ask a dashboard "average out_tokens per model today?" — impossible to do reliably against free-form text.

🔗 Course relevanceTyped Settings upgrades the scattered os.environ.get calls into one validated object — and fails fast at startup if the API key is missing (better than a confusing error mid-request). The JSON formatter is the production form of the Ch 6 observability logging, ready to ship to a dashboard.

5 · Real pytest — fixtures, parametrize, async, coverage essential for shipping

The projects' tests are deliberately simple. Here's the full pytest toolkit you'd actually use — building on the "mock the model" technique from P5 §8.

Try it

Requires: pip install pytest

conftest.py + test_agent.pyimport pytest, types
from unittest.mock import MagicMock

# a fixture: reusable setup, injected by name into any test
@pytest.fixture
def mock_client():
    c = MagicMock()
    c.messages.create.return_value = types.SimpleNamespace(
        content=[types.SimpleNamespace(type="text", text="ok")],
        stop_reason="end_turn")
    return c

# parametrize: run the SAME test over many inputs (a mini eval-in-tests)
@pytest.mark.parametrize("query,expected_kb", [
    ("reset password", "kb-1"),
    ("refund window", "kb-2"),
    ("enable 2fa", "kb-3"),
])
def test_retrieval(query, expected_kb):
    from agent import kb
    _, ids = kb.context_for(query)
    assert expected_kb in ids     # 3 tests from one function

# testing exceptions
def test_bad_sql_is_blocked():
    from agent import db
    with pytest.raises(ValueError, match="blocked"):
        db.run_query("DROP TABLE orders")

# async tests (needs pytest-asyncio)
@pytest.mark.asyncio
async def test_async_path():
    result = await some_async_agent("q")
    assert result

# markers to separate fast unit tests from slow eval tests
@pytest.mark.eval          # run with: pytest -m eval   (needs an API key)
def test_real_quality(): ...

terminalpytest -v                      # run all
pytest -m "not eval"            # fast, key-free tests only (CI on every commit)
pytest --cov=agent             # coverage report (needs pytest-cov)
pytest -x -q                   # stop at first failure, quiet

▶ How this works

This is the real pytest toolkit. pytest is Python's testing framework: you write functions named test_*, use assert to state what should be true, and pytest runs them and reports pass/fail. This block shows the four features you'll actually use: fixtures, parametrize, exception tests, and markers.

A fixture (@pytest.fixture on mock_client) is reusable setup. Any test that takes an argument named mock_client automatically receives what the fixture returns. Here it builds a MagicMock — a fake Anthropic client that returns a canned reply, so tests never hit the real API.
parametrize runs the same test over a table of inputs. The three (query, expected_kb) rows become three separate tests from one function — a compact way to check many cases (a mini eval).
with pytest.raises(ValueError, match="blocked"): asserts that the code inside must throw a ValueError whose message contains "blocked". This is how you test that bad input is correctly rejected.
Markers label tests. @pytest.mark.asyncio lets an async test run; @pytest.mark.eval tags the slow, API-key-requiring tests so you can run or skip them as a group. The terminal block below shows the commands: pytest -m "not eval" runs only the fast, free tests.

What the output means: Running pytest -v lists each test and its result. The parametrized test_retrieval shows up as three lines (one per row), and test_bad_sql_is_blocked passes only if the query really was blocked with that error.

Try this: Look at the terminal commands: pytest -m "not eval" is what CI runs on every commit (fast, no API key), while pytest -m eval runs the model-calling tests on demand. Splitting them this way keeps your test suite fast and free by default.

The pro test strategy for agentsMark key-free logic tests to run on every commit (fast, free), and mark model-calling evals as @pytest.mark.eval to run on PRs / nightly (with a key). parametrize turns your golden set into a clean table of test cases. This is exactly how the course's projects split "tests" (no key) from "evals" (key) — now with the real pytest machinery.

🔗 Course relevanceEvery project ships a tests/ dir; this is how you'd grow them properly. parametrize over a golden set is the bridge between the Ch 5 eval concept and a real test suite.

6 · Packaging your agent as an installable CLI production

Turn your agent from "a folder of scripts" into a tool your team can pip install and run as a command.

Try it

pyproject.toml[project]
name = "devops-agent"
version = "0.1.0"
requires-python = ">=3.11"
dependencies = ["anthropic>=0.40", "pydantic>=2", "python-dotenv"]

[project.scripts]
devops-agent = "devops_agent.cli:main"    # `devops-agent` on the command line

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

terminalpip install -e .          # editable install
devops-agent "checkout-api is crashing"   # now a real command

# modern, fast alternative to pip/venv:
uv venv && uv pip install -e .            # uv — much faster

▶ How this works

This is a pyproject.toml — not Python, but the config file that turns a folder of scripts into an installable package. It's the modern standard: one file describing what your project is, what it needs, and how to build it. It's organized into sections written in [brackets].

[project] declares the package's identity: its name (what you'd pip install), version, the Python it needs (requires-python), and its dependencies — the other libraries pip should install alongside it.
[project.scripts] is the magic that creates a command. The line devops-agent = "devops_agent.cli:main" means: when someone types devops-agent in the terminal, run the main function in the devops_agent/cli.py file.
[build-system] tells pip which tool to use to actually build the package (here hatchling). You rarely touch this — it's boilerplate that makes the install work.
The terminal block shows the payoff: pip install -e . installs your project in editable mode (code changes take effect without reinstalling), and afterwards devops-agent "..." is a real command anyone on your team can run.

What the output means: After pip install -e ., typing devops-agent "checkout-api is crashing" anywhere runs your agent — it behaves like any installed CLI tool, not a script you have to cd into.

Try this: Note -e (editable): without it you'd reinstall after every code edit. With it, the install points at your source folder, so you edit and re-run instantly — the normal way to develop a package.

A clean CLI with argparseThe projects already have cli.py files using argparse. The [project.scripts] entry point turns one into an installed command. For richer CLIs (subcommands, colors, help), typer or click are worth learning — but argparse ships with Python and is enough for most agents.

🔗 Course relevanceThe DevOps and analyst projects have cli.py entry points — this is how you'd ship them as installable tools a team runs, rather than files they cd into.

7 · The code-quality toolchain production

What professional Python teams run in CI. All fast, all worth adopting once your agent is more than a script.

Tool	Does	Command
ruff	Lint + format (replaces flake8/black/isort) — extremely fast	`ruff check . && ruff format .`
mypy / pyright	Static type checking — enforces the P5 §1 types	`mypy agent/`
pytest + pytest-cov	Tests + coverage	`pytest --cov`
pre-commit	Run all the above automatically before each git commit	`pre-commit run --all`
uv	Fast dependency & venv manager (Rust-based)	`uv pip install -r requirements.txt`

Minimum viable quality gateFor any agent you'll maintain: ruff (format + lint), mypy (types), pytest -m "not eval" (fast tests) on every commit via pre-commit, and the eval suite on PRs. That's the same discipline the course's CI-gate sections (Ch 5/6) describe, with concrete tools.

Capstone mini-project · a production-shaped agent runner putting it ALL together

One file that composes the real-world patterns: typed settings, a reused client, rate limiting, a circuit breaker, retry, structured logging, and request-scoped trace ids. This is what a production agent's core actually looks like.

Capstone example

runner.pyimport time, random, uuid, logging, contextvars
from anthropic import Anthropic, RateLimitError, APIStatusError

log = logging.getLogger("agent")
trace_id = contextvars.ContextVar("trace_id", default="-")
client = Anthropic()                       # reused (§2)

class AgentError(Exception): ...           # custom exception (P4)

class Runner:
    def __init__(self, rate=8, breaker_threshold=5):
        self._tokens, self._rate, self._updated = rate, rate, time.monotonic()
        self._fails, self._opened = 0, None
        self._threshold = breaker_threshold

    def _rate_limit(self):                  # token bucket (§3)
        now = time.monotonic()
        self._tokens = min(self._rate, self._tokens + (now-self._updated)*self._rate)
        self._updated = now
        if self._tokens < 1:
            time.sleep((1-self._tokens)/self._rate); self._tokens = 0
        else: self._tokens -= 1

    def ask(self, prompt, *, max_retries=4):
        if self._opened and time.monotonic()-self._opened < 30:
            raise AgentError("circuit open")   # fail fast (§3)
        for attempt in range(max_retries):
            self._rate_limit()
            try:
                t0 = time.monotonic()
                r = client.messages.create(model="claude-opus-4-8",
                        max_tokens=1024, messages=[{"role":"user","content":prompt}])
                self._fails, self._opened = 0, None              # reset breaker
                log.info("ok", extra={"extra_fields":{"trace_id":trace_id.get(),
                    "ms":round((time.monotonic()-t0)*1000),
                    "out":r.usage.output_tokens}})
                return next((b.text for b in r.content if b.type=="text"), "")
            except (RateLimitError, APIStatusError) as e:
                self._fails += 1
                if self._fails >= self._threshold: self._opened = time.monotonic()
                delay = min(30, 2**attempt + random.uniform(0,1))   # backoff (P4)
                log.warning("retry", extra={"extra_fields":{"trace_id":trace_id.get(),
                    "attempt":attempt, "delay":round(delay,1)}})
                time.sleep(delay)
        raise AgentError("exhausted retries")

def handle(prompt, tenant):
    trace_id.set(str(uuid.uuid4()))        # request-scoped (§P5 contextvars)
    return Runner().ask(prompt)

▶ How this works

This is the capstone: one file that combines every production pattern from the lesson into a realistic agent runner. Read it as a checklist — each piece you learned separately appears here, working together. Don't worry about memorizing it; understand which pattern each part represents.

Setup (top): client = Anthropic() is created once and reused (§2). trace_id is a contextvars.ContextVar — a per-request id so every log line from one request can be tied together. AgentError is a custom exception type.
_rate_limit: the same token-bucket logic from §3, inlined — refill tokens based on elapsed time, and either sleep for a token or spend one. This throttles the agent to stay under quota.
ask (the circuit breaker + retry loop): first it fails fast if the breaker is open (§3). Then it loops up to max_retries times: rate-limit, call the model, and on success reset the breaker and log a structured "ok" line with timing and token counts.
On failure: it counts the failure (tripping the breaker at the threshold), computes a 2**attempt + random jittered backoff delay (P4), logs a "retry" line, and sleeps before trying again. If all retries are exhausted it raises AgentError.
handle: the entry point sets a fresh trace_id per request (so logs are traceable) and delegates to Runner().ask(...). Swap the body of ask for a real agent loop and this is a shippable core.

What the output means: A successful call returns the model's text and emits one JSON "ok" log line (with trace_id, latency in ms, and output tokens). Failures emit "retry" lines and back off; a truly dead dependency trips the breaker so later calls fail instantly.

Try this: Go through the code and name the pattern beside each part: reused client, token bucket, circuit breaker, retry with backoff, custom exception, structured logging, request-scoped trace id. If you can point to all seven, you've got the production skeleton.

Read it as a checklistThis one class demonstrates: reused client (§2), token-bucket rate limiting (§3), a circuit breaker (§3), retry with jittered backoff (P4), custom exceptions (P4), structured logging with trace ids (§4 + P5), and request-scoped context (P5). That's the production skeleton behind any of the six projects — swap ask()'s body for the project's agent loop.

The complete Python journey expert

Part	Level	You can now…
P1	Beginner	read & write any lab
P2	Beginner→Int.	handle messages, chunks, JSON, logs
P3	Intermediate	structure the agent: modules, OOP, dataclasses, enums
P4	Int.→Advanced	Pydantic, exceptions, streaming, async, secrets
P5	Expert	typing, functools/itertools, concurrency, testing the model
P6 (here)	Production	scale, resilience, config, real pytest, packaging, quality

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

Implement a token-bucket rate limiter

"Limit to N calls/sec." Refill tokens over time; allow a call only if a token is available.

pythonimport time
class RateLimiter:
    def __init__(self, rate, per=1.0):
        self.rate, self.per = rate, per
        self.tokens = rate
        self.updated = time.monotonic()
    def allow(self):
        now = time.monotonic()
        self.tokens = min(self.rate,
            self.tokens + (now - self.updated) * self.rate / self.per)
        self.updated = now
        if self.tokens >= 1:
            self.tokens -= 1
            return True
        return False

▶ How this works

The interview version of the rate limiter. The task: "limit to N calls/sec." The key insight interviewers want is the token bucket — you don't track a list of timestamps, you track a refilling count of tokens and check if one is available.

__init__ stores rate (tokens per per seconds), starts the bucket full (self.tokens = rate), and records the current time with time.monotonic().
In allow(), the refill formula (now - self.updated) * self.rate / self.per adds tokens for the time that has passed, and min(self.rate, ...) caps the bucket so it can't overfill.
If at least one token is available, it spends one (self.tokens -= 1) and returns True (call allowed); otherwise it returns False (call denied).

What the output means: allow() returns True when a call is permitted and False when you'd exceed the rate — the caller decides whether to drop, queue, or wait. (The §3 version sleeps instead of returning False; same math, different policy.)

Try this: Be ready to explain why token-bucket beats "store the last N timestamps": it's O(1) memory and time, handles bursts naturally via the bucket size, and needs no cleanup of old entries.

Circuit breaker — stop hammering a dead dependency

Open after N failures, fail fast during a cooldown, then half-open to test recovery.

pythonclass CircuitBreaker:
    def __init__(self, threshold=5):
        self.fails = 0; self.threshold = threshold; self.open = False
    def call(self, fn):
        if self.open:
            raise RuntimeError("circuit open")
        try:
            r = fn()
        except Exception:
            self.fails += 1
            if self.fails >= self.threshold: self.open = True
            raise
        self.fails = 0
        return r

▶ How this works

The interview version of the circuit breaker. Explain it as a state machine: closed (calls flow), open (fail fast after too many errors), and half-open (after a cooldown, let one trial call test whether the dependency recovered). This minimal version shows closed and open.

__init__ tracks a failure count (self.fails), the trip threshold, and a boolean self.open that starts False (closed / healthy).
call(self, fn) first checks: if self.open is true, raise "circuit open" immediately — the fail-fast behaviour that stops hammering a dead dependency.
It runs fn(); on an exception it increments self.fails, and once that hits threshold it flips self.open = True (trips) and re-raises.
On success it resets self.fails = 0 and returns the result. In a full version you'd add a cooldown timer and the half-open trial call (mentioned in the prompt) to auto-recover — the §3 breaker shows the cooldown piece.

What the output means: While closed, calls pass through and errors count up. After threshold failures the breaker opens and every subsequent call() raises instantly until it's reset.

Try this: In the interview, sketch the three states and the transitions between them. Mention that real breakers add a cooldown and a half-open probe so they recover automatically instead of staying open forever.

🪜 Practice ladder beginner → industry

Six graded exercises, easy to real-world. Try each before opening its solution.

Exercise 1 · Stream a big file with a generatorBeginner

Context: Log files and datasets routinely dwarf available RAM, so readlines() is a memory bomb waiting to happen. A generator that yields matching lines one at a time keeps memory flat regardless of file size — the foundational streaming pattern for any log or data pipeline.

Your task: Write a generator error_lines(source) that yields only the lines containing ERROR, keeping memory constant, and demonstrate it on an in-memory list of lines (a file object would iterate the same way).

Requirements:

The function is a generator that yields matching lines lazily
Only lines containing ERROR are emitted; others are skipped
It never accumulates all lines into a list internally
The demo runs offline against a list, noting a real file iterates identically
Confirm the count of error lines matches expectation

💡 Hint: Iterating a file object yields lines just like iterating a list, so the same generator body works on both — filter inside the loop and yield the survivors.

Show solution

Generators keep memory constant regardless of file size. Runnable (adapted to a list so it runs offline):

def error_lines(lines):
    for line in lines:          # file objects iterate the same way
        if "ERROR" in line:
            yield line

sample = ["INFO ok\n", "ERROR auth failed\n", "ERROR 503\n", "INFO served\n"]
errs = list(error_lines(sample))
print(len(errs), "error lines")   # 2 error lines

Exercise 2 · Shrink objects with __slots__Intermediate

Context: When you hold millions of small records — log entries, events, embeddings metadata — the per-instance __dict__ is pure overhead. __slots__ removes it, shrinking each object and, as a side effect, rejecting stray attributes that would otherwise slip through as silent bugs.

Your task: Define a small record class that declares __slots__ for its fields, and show that assigning an attribute not in the slots raises — the observable signal that slots are active.

Requirements:

Declare __slots__ as a tuple of the allowed field names
Set those fields in __init__ and read them back normally
Assigning an undeclared attribute raises AttributeError
Catch that error and report it rather than crashing
No per-instance __dict__ exists once slots are declared

💡 Hint: The rejection is the proof — wrap the illegal assignment in try/except AttributeError and print the message to show slots are enforcing the field set.

Show solution

__slots__ drops the per-instance __dict__, saving memory at scale. Runnable:

class LogRecord:
    __slots__ = ("svc", "level", "msg")
    def __init__(self, svc, level, msg):
        self.svc, self.level, self.msg = svc, level, msg

r = LogRecord("checkout-api", "ERROR", "auth failed")
print(r.svc, r.level)
try:
    r.extra = "nope"           # not in __slots__
except AttributeError as e:
    print("blocked:", e)

Exercise 3 · Manage a resource with a context managerAdvanced

Context: Any code that opens a resource — a connection, a file, a lock — must guarantee it gets closed, even when the body raises. A class-based context manager encodes that setup/teardown contract so cleanup is automatic and exceptions still propagate correctly.

Your task: Write a class-based context manager that "opens" a resource in __enter__ and guarantees "close" in __exit__ even when the with body raises an exception.

Requirements:

__enter__ marks the resource open and returns it for as
__exit__ always runs the close, whether or not an error occurred
__exit__ returns falsy so it does not swallow the exception
Demonstrate a body that raises and confirm the resource is still closed
The raised error is caught outside the with, not hidden by it

💡 Hint: Returning False (or nothing) from __exit__ lets the exception re-raise after cleanup; put the close in __exit__'s body so it fires on every exit path.

Show solution

Deterministic cleanup, the with-statement contract. Runnable:

class Connection:
    def __init__(self, name): self.name = name; self.open = False
    def __enter__(self):
        self.open = True
        print(f"opened {self.name}")
        return self
    def __exit__(self, exc_type, exc, tb):
        self.open = False
        print(f"closed {self.name}")
        return False              # don't swallow exceptions

try:
    with Connection("db") as c:
        print("using", c.name)
        raise RuntimeError("boom")
except RuntimeError:
    print("error handled; connection still closed")

Exercise 4 · A token-bucket rate limiterExpert

Context: Calling a rate-limited model API in a burst gets you throttled; a token-bucket limiter smooths that out by refilling tokens at a steady rate and allowing short bursts up to a cap. It is the canonical client-side throttle, and it must be thread-safe because agents often fan out across threads.

Your task: Implement a thread-safe token-bucket RateLimiter(rate, burst) whose acquire consumes a token when one is available, then fire several rapid acquisitions and count how many were granted immediately.

Requirements:

Refill tokens based on elapsed time × rate, capped at burst
Use a monotonic clock so time going backwards can't corrupt the bucket
Guard the token state with a threading.Lock
Grant a call only when at least one token is available, decrementing on success
A rapid run of acquisitions grants roughly burst of them immediately

💡 Hint: On each attempt, first add (now − last) × rate tokens clamped to the capacity, then spend one if you can — hold the lock across that whole read-modify-write.

Show solution

The production rate-limiter, made deterministic for the demo (no real sleeping). Runnable:

import time, threading

class RateLimiter:
    def __init__(self, rate, burst):
        self.rate, self.capacity = rate, burst
        self.tokens = burst
        self.updated = time.monotonic()
        self.lock = threading.Lock()
    def try_acquire(self):
        with self.lock:
            now = time.monotonic()
            self.tokens = min(self.capacity,
                              self.tokens + (now - self.updated) * self.rate)
            self.updated = now
            if self.tokens >= 1:
                self.tokens -= 1
                return True
            return False

lim = RateLimiter(rate=10, burst=3)
granted = sum(lim.try_acquire() for _ in range(10))
print("immediately granted:", granted)   # ~3 (the burst)

Exercise 5 · A circuit breakerProfessional

Context: When a downstream dependency is failing, retrying it just piles on load and slows everything upstream. A circuit breaker trips after N consecutive failures and refuses calls while open, giving the dependency room to recover — a core resilience primitive in any distributed agent system.

Your task: Build a circuit breaker that opens after a threshold of consecutive failures and refuses further calls while open, then drive it with a function that always fails and show it trips.

Requirements:

Count consecutive failures and open the circuit once the threshold is hit
While open, a call short-circuits with a dedicated CircuitOpen exception instead of invoking the function
A success resets the failure count back to zero
The wrapped function's own exceptions still propagate on the way to tripping
Driving it with an always-failing function trips it after exactly the threshold count

💡 Hint: Track failures on the instance; in call, check the open flag first, then run the function under try/except, incrementing on failure and flipping open when the count reaches the threshold.

Show solution

Stops hammering a failing dependency. Runnable (stdlib):

class CircuitOpen(Exception): pass

class CircuitBreaker:
    def __init__(self, threshold=3):
        self.threshold = threshold
        self.failures = 0
        self.open = False
    def call(self, fn, *args, **kwargs):
        if self.open:
            raise CircuitOpen("circuit is open; refusing call")
        try:
            result = fn(*args, **kwargs)
        except Exception:
            self.failures += 1
            if self.failures >= self.threshold:
                self.open = True
            raise
        self.failures = 0
        return result

def always_fails(): raise ValueError("down")

cb = CircuitBreaker(threshold=3)
for i in range(5):
    try:
        cb.call(always_fails)
    except CircuitOpen as e:
        print(i, "->", e); break
    except ValueError:
        print(i, "-> failed")

Exercise 6 · A production-shaped agent runnerIndustry scenario

Context: A production agent runner is the sum of this whole page: it reads config, throttles with a rate limiter, guards each task with a circuit breaker, logs structurally, and returns a summary the caller can act on. Wiring these together resiliently — so one bad task doesn't take down the batch — is the real deliverable.

Your task: Combine config, a rate limiter, a circuit breaker, and structured logging into a run(tasks) that processes tasks resiliently and returns a summary of what succeeded and failed, then prove its behaviour with an assertion.

Requirements:

Each task is executed through the circuit breaker from the previous rung
A tripped breaker stops the run early with a logged warning rather than continuing
Task-level failures are counted and logged, not allowed to abort the whole batch
Use the logging module for structured output, not bare prints
Return a summary dict (e.g. completed results and a failure count)
Assert the summary equals the expected value for a known input

💡 Hint: Reuse the breaker you built, loop the tasks catching CircuitOpen (break) separately from a task's own error (count and continue), and accumulate results into the summary you return.

Show solution

The capstone runner shape — each task guarded by the breaker, throttled by the limiter, all logged. Runnable (stdlib), reusing the components above:

import logging

logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")
log = logging.getLogger("runner")

class CircuitOpen(Exception): pass

class Breaker:
    def __init__(self, threshold=3): self.t=threshold; self.f=0; self.open=False
    def call(self, fn, x):
        if self.open: raise CircuitOpen()
        try:
            r = fn(x); self.f = 0; return r
        except Exception:
            self.f += 1
            if self.f >= self.t: self.open = True
            raise

def process(x):
    if x < 0: raise ValueError("bad task")
    return x * 2

def run(tasks):
    br = Breaker(threshold=2)
    done, failed = [], 0
    for t in tasks:
        try:
            done.append(br.call(process, t))
        except CircuitOpen:
            log.warning("circuit open, stopping"); break
        except ValueError:
            failed += 1; log.error("task %s failed", t)
    return {"done": done, "failed": failed}

summary = run([1, 2, 3])
assert summary == {"done": [2, 4, 6], "failed": 0}
print(summary)

✓ Checkpoint — you can engineer agents professionally when you can…

Process large data with generators & know when __slots__ helps.
Reuse clients & manage resources with context managers / ExitStack.
Implement a rate limiter and a circuit breaker, and say when each fires.
Load typed settings and emit structured logs with trace ids.
Write a real pytest suite (fixtures, parametrize, markers) splitting fast tests from evals.
Package your agent as an installable CLI and run ruff/mypy/pytest in CI.
Read the capstone runner and name every production pattern in it.

🎓 You've completed the full Python track (P1–P6)From your first print() to a production-shaped, tested, packaged agent runner. Combined with the course chapters and the six projects, you now have everything to build and ship real agentic systems. Go build one.

Knowledge check check yourself

✓ Knowledge check

Contrast the circuit breaker with the token-bucket rate limiter as described in this lesson: what failure does each address, and when does each 'fire'?

Show answer

The rate limiter (token bucket) keeps you under your calls-per-second quota — it refills tokens over time and sleeps/denies when the bucket is empty, allowing bursts up to burst then throttling to rate. The circuit breaker protects against a failing dependency — it trips open after N consecutive failures and fails fast during a cooldown so you stop hammering a dead backend, resetting on a success.

✓ Knowledge check

Why does the lesson insist on creating one module-level Anthropic() client instead of one per request, and how do @contextmanager/ExitStack support correct resource handling?

Show answer

Creating a client per request opens a new connection pool each time, which leaks connections and is slow under load; a single reused module-level client pools connections. A @contextmanager with try/finally guarantees cleanup (e.g. con.close()) even if the body raises, and ExitStack cleanly manages a dynamic, unknown number of resources, closing them all when the block exits.

Educational content, provided as-is and without warranty. Code samples are examples — review, test, and adapt them before using in production. See the Terms of Use & Disclaimer. Use at your own risk.

Learning objectives

What this is (and isn't) optional, max-level

1 · Memory & performance at scale expert

2 · Resource & connection management expert

3 · Resilience patterns — rate limiting & circuit breaker expert · production

4 · Configuration & logging, done right production

5 · Real pytest — fixtures, parametrize, async, coverage essential for shipping

6 · Packaging your agent as an installable CLI production

7 · The code-quality toolchain production

Capstone mini-project · a production-shaped agent runner putting it ALL together

The complete Python journey expert

🎯 Interview practice interview

🪜 Practice ladder beginner → industry

✓ Checkpoint — you can engineer agents professionally when you can…

Knowledge check check yourself

More in Python