Part 7 · Module 1: Prerequisites

Lesson 1: The plain engineering underneath every agent

20 min read·13 Oct 2026

Module 1 · Lesson 1 · About 40 minutes · Needs: some Python; the course setup done

The situation

It is the week before the project starts. Priya wants an agent that works BrightCart's returns queue alone. Dev, who owns the order and returns systems, asks one thing first: "When my API is slow, or busy, or answers twice, what does your agent do?"

It is not a question about models. An agent is a loop of ordinary calls, and each can be slow, fail, or return the wrong shape. This lesson checks the engineering you need, on one request, R-1001: Aisha Rahman wants to return her robot vacuum, order 5521. Each topic is brief here and links to the AI Engineering course for depth.

A real case, first

On 11 February 2023 Will McGugan of Textualize published "The Heisenbug lurking in your async code" (textual.textualize.io). If you call asyncio.create_task and do not keep the task it returns, he wrote, "the task may disappear without warning when Python runs garbage collection". The code in your task "will stop running with no obvious indication why", and the bug is "challenging (read almost impossible) to reproduce".

The Python documentation says so, he notes, and a GitHub search suggested many projects carry the bug, perhaps because tasks feel like threads, which you can start and forget. His fix: keep a reference to each task while it must live, or use TaskGroups, which keep them for you.

The event loop holds only a weak reference to a task, so once your code drops its own reference, garbage collection can remove the task while it is still waiting (red: the failure).

Agents that start background work hit exactly this. Section 3 reproduces it.

What you will be able to do

Write typed tools that report expected failures and let bugs crash.
Run slow calls together under a time budget, without losing tasks.
Call an HTTP API with a token and retry only what is safe.
Validate model output with Pydantic; test with pytest and mocks; run it all in CI and Docker.

Blue is BrightCart's code, amber is the model: one request passes through reads, one model decision, validation and a retried write, all resting on tests, CI and a container.

The map for this lesson: blue is BrightCart's code, amber is the model. Little of the path is amber. The blue part is what makes an agent reliable, and each section covers one piece of it.

1. Python: typing, classes, error handling (Essential)

Every agent tool is a Python function, and its most important lines run when something goes wrong. Get those wrong and the agent reports a return that never happened. First, setup and shared data.

python

%pip install -q -U langchain langgraph langchain-groq langchain-google-genai langchain-ollama

python

import os
from langchain.chat_models import init_chat_model

PROVIDER = "groq"   # "groq" or "gemini" in Colab; "ollama" on your own computer
MODELS = {"groq": "groq:openai/gpt-oss-20b", "gemini": "google_genai:gemini-3.5-flash-lite", "ollama": "ollama:llama3.2:3b"}
KEY = {"groq": "GROQ_API_KEY", "gemini": "GOOGLE_API_KEY", "ollama": None}[PROVIDER]
if KEY and KEY not in os.environ:
    from google.colab import userdata   # keys live in Colab's Secrets panel
    os.environ[KEY] = userdata.get(KEY)
llm = init_chat_model(MODELS[PROVIDER], temperature=0, max_retries=6)   # waits and retries on rate limits

Code walkthrough

The standard setup cell: pick a provider, read its key from Colab's Secrets, create llm once (used in section 5).

python

from datetime import date

COURSE_DATE = date(2026, 10, 9)    # facts, model names and prices checked on this day
DATA_AS_OF = date(2026, 4, 20)     # the shared data is a snapshot from this morning

print(DATA_AS_OF.strftime("%A %d %B %Y"), "|", (COURSE_DATE - DATA_AS_OF).days, "days before the course date")

CUSTOMERS = {
    "cust-88": {"name": "Aisha Rahman", "email": "aisha@example.com"},
    "cust-12": {"name": "Tom Becker", "email": "tom@example.com"},
    "cust-31": {"name": "Mei Tanaka", "email": "mei@example.com"},
}

ORDERS = {
    "5521": {"customer": "cust-88", "item": "RoboSweep X2 robot vacuum", "total": 129.00,
             "status": "delivered", "delivered_on": "2026-04-02"},
    "5730": {"customer": "cust-88", "item": "QuietBoil 1.7L kettle", "total": 39.00,
             "status": "shipped", "delivered_on": None},
    "7003": {"customer": "cust-12", "item": "CinemaView 55-inch TV", "total": 899.00,
             "status": "shipped", "delivered_on": None},
    "7120": {"customer": "cust-12", "item": "GlowLine smart bulb, 4-pack", "total": 45.00,
             "status": "delivered", "delivered_on": "2026-02-10"},
    "8204": {"customer": "cust-31", "item": "SoundPod wireless headphones", "total": 79.00,
             "status": "processing", "delivered_on": None},
}

for order_id, o in ORDERS.items():
    age = None if o["delivered_on"] is None else (DATA_AS_OF - date.fromisoformat(o["delivered_on"])).days
    print(order_id, o["customer"], f"${o['total']:.2f}", o["status"], "| days since delivery:", age)

Code walkthrough

The shared dates, customers and orders, copied unchanged. Order 5521 was delivered 18 days before the data date.

A class bundles state (orders, returns, audit) with the methods that act on it; systems is one instance with its own copy of the data.

A class bundles data with the functions that act on it. BrightCartSystems stands in for Dev's systems: its data is the orders, returns and audit trail; its methods are the actions. systems = BrightCartSystems() makes one instance, a fresh copy with its own state, and inside a method self is that instance.

python

import copy


class BrightCartSystems:
    """An in-memory stand-in for BrightCart's order, returns, payment and email systems."""

    def __init__(self, today=DATA_AS_OF):
        self.today = today
        self.orders = copy.deepcopy(ORDERS)
        self.returns, self.refunds = {}, {}          # idempotency key -> record
        self.approvals, self.escalations, self.emails, self.audit = [], [], [], []

    def _log(self, actor, action, **details):
        self.audit.append({"step": len(self.audit) + 1, "actor": actor, "action": action, **details})

    def get_order(self, order_id, actor="agent"):
        self._log(actor, "get_order", order_id=order_id)
        if order_id not in self.orders:
            raise KeyError(f"No order {order_id}")
        return {"order_id": order_id, **self.orders[order_id]}

    def days_since_delivery(self, order_id):
        delivered = self.orders[order_id]["delivered_on"]
        return None if delivered is None else (self.today - date.fromisoformat(delivered)).days

    def cancel_order(self, order_id, actor="agent"):
        order = self.orders[order_id]
        if order["status"] != "processing":
            raise ValueError(f"Order {order_id} is {order['status']} and can no longer be cancelled")
        order["status"] = "cancelled"
        self._log(actor, "cancel_order", order_id=order_id)
        return {"order_id": order_id, "status": "cancelled"}


systems = BrightCartSystems()
print(systems.get_order("5521")["item"], "| days since delivery:", systems.days_since_delivery("5521"))

Code walkthrough

__init__ runs when the instance is made and copies ORDERS, so runs never share state. get_order raises KeyError for an unknown order: raising is how a method refuses.

Type hints label what goes in and what comes out; editors, Pydantic and agent frameworks read them, but Python does not enforce them at run time.

A type hint says what a function takes and returns: order_id: str, -> ToolResult. Python does not check hints at run time. Editors, type checkers and Pydantic do, and agent frameworks read them to describe your tools to the model.

One tool, three orders: 5521 is Aisha's and is found (green: passed), 7003 is someone else's, 9999 does not exist; both refusals come back as results the agent can read.

Here is a lookup tool for R-1001's customer, tried on three orders. The order system fails in one expected way, a missing order, so the tool catches exactly that and returns a result the agent can read. Someone else's order is refused too.

python

from dataclasses import dataclass

R1001 = {"id": "R-1001", "customer": "cust-88", "order_id": "5521",
         "message": "I've changed my mind about the robot vacuum. It's unused and back in its box. Can I return it for a refund?"}

@dataclass
class ToolResult:
    ok: bool
    data: dict | None = None
    error: str | None = None          # "not_found" or "not_allowed"

def lookup(systems: BrightCartSystems, order_id: str, customer: str) -> ToolResult:
    try:
        order = systems.get_order(order_id)
    except KeyError:                  # the one failure we expect from this call
        return ToolResult(ok=False, error="not_found")
    if order["customer"] != customer:
        return ToolResult(ok=False, error="not_allowed")
    return ToolResult(ok=True, data=order)

systems = BrightCartSystems()
for order_id in ["5521", "7003", "9999"]:
    print(order_id, lookup(systems, order_id, R1001["customer"]))

Code walkthrough

@dataclass writes the constructor and printed form of ToolResult. dict | None means "a dictionary or nothing".
The try holds one line, the call that can raise KeyError. Output: 5521 found, 7003 not_allowed (Tom Becker's TV), 9999 not_found.

Break it: a try block that is too wide

Keep the try narrow: wrapped around too much, it catches a typo in your own code and reports a real order as not found.

Move the ownership check inside the try, with a typo. The typo also raises KeyError, and the wide except reports it as a missing order. The agent would tell Aisha her order does not exist.

python

def lookup_wide(systems: BrightCartSystems, order_id: str, customer: str) -> ToolResult:
    try:
        order = systems.get_order(order_id)
        if order["custmer"] != customer:          # a typo: a bug in our own code
            return ToolResult(ok=False, error="not_allowed")
        return ToolResult(ok=True, data=order)
    except KeyError:                              # now it catches the typo as well
        return ToolResult(ok=False, error="not_found")

print("narrow try:", lookup(systems, "5521", "cust-88").ok)
print("wide try:  ", lookup_wide(systems, "5521", "cust-88"))

Code walkthrough

order["custmer"] is a bug that raises the same error as a missing order. The wide version prints not_found for a real order. Keep try blocks to the one call that can fail, and let bugs crash. Depth: AI Engineering, Python core.

2. async / await and the event loop (Essential)

An agent spends most of its time waiting for models and APIs. Async code lets one program wait for several things at once, so three reads take the time of one.

The event loop starts each call (dark blue), parks it at its await, and moves on, so all three waits overlap instead of queueing.

An async def function returns a coroutine: an object that does nothing until something awaits it. The event loop runs coroutines. When one reaches await on something slow, the loop parks it and runs another. Nothing runs in parallel; the waits overlap.

python

import asyncio
import time

def run_async(coro):
    """Run a coroutine from a plain script or from a notebook cell."""
    try:
        asyncio.get_running_loop()
    except RuntimeError:
        return asyncio.run(coro)                 # plain Python: start a loop, run, stop it
    import nest_asyncio; nest_asyncio.apply()    # notebook: a loop is already running
    return asyncio.run(coro)

Code walkthrough

run_async works in a script, where no loop is running, and in a notebook, where one is; nest_asyncio (in Colab) lets the notebook's loop be reused. In your own scripts, write asyncio.run(main()) once.

python

async def call(name, seconds, answer):
    await asyncio.sleep(seconds)                 # stands in for a network call
    return name, answer

def reads_for(request):                          # the three reads R-1001 needs, 0.3 s each
    oid = request["order_id"]
    return [call("order", 0.3, systems.get_order(oid)["status"]),
            call("days", 0.3, systems.days_since_delivery(oid)),
            call("help", 0.3, "HC-01 How to return an item")]

async def one_by_one(request):
    return [await c for c in reads_for(request)]

async def together(request):
    return await asyncio.gather(*reads_for(request))

for fn in (one_by_one, together):
    t0 = time.perf_counter()
    answers = run_async(fn(R1001))
    print(f"{fn.__name__:10} {time.perf_counter() - t0:.2f}s", answers)

Code walkthrough

call stands in for a network call: await asyncio.sleep(0.3) hands control back to the loop.
one_by_one awaits each call before starting the next; asyncio.gather starts all three and waits for them together.

Measured: R-1001's three reads take 0.90 seconds one by one and 0.30 seconds with gather.

Same answers in a third of the time: 0.90 seconds one by one, 0.30 together. Depth: AI Engineering, Async Python.

3. Concurrency, cancellation, timeouts (Essential)

Starting calls together is easy; stopping them is hard. A call that never answers holds the request forever, and each request has a 60-second limit.

A TaskGroup keeps a reference to every task it starts and the block ends only when all are done; if one fails, the rest are cancelled.

A TaskGroup (Python 3.11 and later) runs tasks together safely. It holds a reference to every task it starts, waits for all of them before its async with block ends, and if one fails, cancels the rest. This is McGugan's recommended fix.

At the 0.5 s budget the search is cancelled at its await, tidies up and re-raises, and the agent gets a timeout result; the whole step took 0.50 s.

A timeout sets a budget. When it runs out, Python cancels the task, raising CancelledError at its await. The task may tidy up but must let the error through: the Python docs warn that TaskGroup and timeouts "might misbehave if a coroutine swallows asyncio.CancelledError".

python

log = []

async def slow_help_search(query):
    try:
        await asyncio.sleep(5)                   # the help centre is having a bad day
        return "HC-01"
    except asyncio.CancelledError:
        log.append("help search cancelled, connection closed")
        raise                                    # let the cancellation carry on

async def within(seconds, coro):
    try:
        async with asyncio.timeout(seconds):
            return await coro
    except TimeoutError:
        return {"error": "timeout", "message": f"no answer within {seconds} s"}

async def reads_with_budget(request):
    async with asyncio.TaskGroup() as tg:        # holds every task until all are done
        order = tg.create_task(call("order", 0.3, request["order_id"]))
        help_ = tg.create_task(within(0.5, slow_help_search("return a vacuum")))
    return order.result(), help_.result()

t0 = time.perf_counter()
print(run_async(reads_with_budget(R1001)), f"{time.perf_counter() - t0:.2f}s")
print(log)

Code walkthrough

slow_help_search takes 5 seconds; on cancellation it logs a tidy-up and re-raises.
within wraps a call in asyncio.timeout and turns TimeoutError into a result the agent can act on. Output: the order read succeeds, the search times out, the step takes 0.50 seconds, and the tidy-up ran.

Break it: the disappearing task

Run for real: after garbage collection the unreferenced task is gone, with only a log line to show it, while the task kept in a set is still waiting.

Now McGugan's bug. A background task waits for the warehouse to scan R-1001's parcel. The event loop keeps only a weak reference to it, one that does not keep it alive. Drop your own reference and garbage collection can remove the task mid-wait.

python

import gc
import weakref

async def wait_for_scan(request_id):
    scan = asyncio.get_running_loop().create_future()   # the warehouse scan; only this task holds it
    await scan
    return request_id

async def start_background(keep):
    background = set()
    task = asyncio.create_task(wait_for_scan("R-1001"))
    probe = weakref.ref(task)            # watches the task without keeping it alive
    if keep:
        background.add(task)             # a strong reference, as the Python docs advise
    del task
    await asyncio.sleep(0)               # the task starts and parks on its await
    gc.collect()                         # Python may do this at any moment
    return "still there" if probe() else "gone"

print("no reference kept:", run_async(start_background(keep=False)))
print("kept in a set:    ", run_async(start_background(keep=True)))

Code walkthrough

Only the task holds the scan future. weakref.ref watches the task without keeping it alive; gc.collect() forces what Python may do any time.
Unreferenced, the task is gone, with only a "Task was destroyed but it is pending!" log line. Kept in a set, as the docs advise, it is still there.

4. HTTP, REST, auth, retries with backoff (Essential)

Every BrightCart write is an HTTP request to Dev's services. A careless retry makes a second return label, or, for a refund, pays twice.

One write as HTTP: the method and address say what, the bearer token says who, the idempotency key names the action, and the server answers 201 with RET-001.

In REST, the address names a thing (/returns) and the method says what to do: GET reads, POST creates. Auth is usually a bearer token in the Authorization header, read from the environment, never from code. The idempotency key names this one action, R-1001:return, so the server can spot a repeat.

python

import os
import httpx

os.environ.setdefault("RETURNS_API_TOKEN", "demo-token")   # a real token lives in Secrets, never in code

class FakeReturnsAPI:
    """BrightCart's returns endpoint, simulated. Answers from a script; can honour Idempotency-Key."""

    def __init__(self, script, honour_keys=True):
        self.script, self.honour_keys, self.records = list(script), honour_keys, []

    def __call__(self, request: httpx.Request) -> httpx.Response:
        if request.headers.get("Authorization") != "Bearer " + os.environ["RETURNS_API_TOKEN"]:
            return httpx.Response(401, json={"error": "bad token"})
        step = self.script.pop(0)
        if step in (429, 503):
            return httpx.Response(step)                       # busy: nothing was done
        key = request.headers["Idempotency-Key"]
        same = [r for r in self.records if self.honour_keys and r["key"] == key]
        if not same:
            same = [{"key": key, "return_id": f"RET-{len(self.records) + 1:03d}"}]
            self.records.append(same[0])
        if step == "lost":
            raise httpx.ReadTimeout("the return was created, but the reply never arrived")
        return httpx.Response(201, json=same[0])

def make_client(api, token=None):
    token = token or os.environ["RETURNS_API_TOKEN"]
    return httpx.Client(base_url="https://returns.brightcart.example", transport=httpx.MockTransport(api),
                        headers={"Authorization": f"Bearer {token}"}, timeout=5.0)

Code walkthrough

FakeReturnsAPI plays Dev's returns service: it checks the token, answers from a script of statuses, and keeps one record per key unless told not to. "lost" creates the return, then drops the reply.
make_client builds a real httpx.Client with the fake as its transport, the token and a 5-second timeout.

The status code decides the next move: success is used, a 4xx is reported because resending changes nothing, and only 429, 5xx and timeouts are retried.

The status code decides the next move. 2xx: done. 400, 401, 403, 404: the request itself is wrong, and resending gets the same answer. 429 (slow down), 5xx and timeouts are often temporary; only those are worth retrying.

Measured run: 429, wait 0.13 s, 503, wait 0.23 s, then 201; each wait doubles, scaled by jitter, and only one return exists at the end.

A retry with backoff waits longer after each failure: 0.2 seconds, then 0.4, then 0.8. Jitter scales each wait by a random factor, so many agents that failed together do not retry in the same instant.

python

import random

RETRY_ON = {429, 500, 502, 503, 504}
jitter = random.Random(7)                                 # fixed seed: the same waits every run

def post_with_retries(client, path, body, key, attempts=4, base=0.2, sleep=time.sleep):
    waits = []
    for attempt in range(1, attempts + 1):
        try:
            r = client.post(path, json=body, headers={"Idempotency-Key": key})
            if r.status_code not in RETRY_ON:
                r.raise_for_status()                      # 4xx: our mistake, so stop at once
                return r.json(), waits
        except httpx.TransportError:                      # timeouts and dropped connections
            pass
        if attempt < attempts:
            wait = base * 2 ** (attempt - 1) * jitter.uniform(0.5, 1.0)   # backoff with jitter
            waits.append(round(wait, 2))
            sleep(wait)
    raise RuntimeError(f"gave up after {attempts} attempts")

api = FakeReturnsAPI([429, 503, 201])
record, waits = post_with_retries(make_client(api), "/returns", {"order_id": "5521"}, "R-1001:return")
print(record, "| waits:", waits, "| returns on the server:", len(api.records))

Code walkthrough

raise_for_status() stops at once on a 4xx; httpx.TransportError covers timeouts and dropped connections. sleep is a parameter so tests can replace it.
Output: waits of 0.13 and 0.23 seconds, then RET-001, one return on the server.

Break it: retry without an idempotency key

The reply is lost after the return was made: without a key the retry creates RET-002, a second label; with the key it gets RET-001 back.

A retry is safe only if a repeat does nothing new. Here the reply is lost after the return was created, and the client retries.

python

for honour in (False, True):
    api = FakeReturnsAPI(["lost", 201], honour_keys=honour)
    record, waits = post_with_retries(make_client(api), "/returns", {"order_id": "5521"}, "R-1001:return")
    print("keys honoured:", honour, "| returns created:", [r["return_id"] for r in api.records])

Code walkthrough

Keys ignored: the retry creates RET-002, two labels for one vacuum. Keys honoured: the retry gets RET-001 back.

SDKs choose these defaults for you. The OpenAI Python SDK retries "2 times by default, with a short exponential backoff" and times out "after 10 minutes" (README, checked October 2026). Check every client you use. Depth: AI Engineering, Working with APIs.

5. Pydantic and schema validation (Essential)

A model's decision is text until something checks it. Before R-1001's decision creates anything, code must know the order number, action and amount are valid.

The schema for the model's decision: five fields, each with a type and a rule that Pydantic enforces.

A schema describes a valid shape. Pydantic lets you write one as a class: each field has a type and, where needed, a rule. Literal lists the allowed values; Field(ge=0) means zero or more. The same class becomes a JSON schema you can send to a model.

python

from typing import Literal
from pydantic import BaseModel, Field, ValidationError

class ReturnDecision(BaseModel):
    """What the agent decides for one return request, in a shape code can check."""
    order_id: str = Field(pattern=r"^\d{4}$", description="The four-digit order number")
    action: Literal["create_return", "decline", "not_yet_delivered", "escalate"]
    refund_usd: float = Field(ge=0, description="Refund in US dollars, 0 if none")
    needs_finance: bool = Field(description="True when the refund is above 100 dollars")
    reason: str

good = {"order_id": "5521", "action": "create_return", "refund_usd": "129.00",
        "needs_finance": True, "reason": "18 days since delivery, unused"}
bad = {"order_id": "order 5521", "action": "refund_everything", "refund_usd": -129, "reason": "asked"}
print(ReturnDecision.model_validate(good))
try:
    ReturnDecision.model_validate(bad)
except ValidationError as e:
    print(e.error_count(), "errors:", [(err["loc"][0], err["type"]) for err in e.errors()])

Code walkthrough

pattern=r"^\d{4}$" allows exactly four digits. model_validate(good) returns a typed object; "129.00" became 129.0, because Pydantic converts values when the meaning is clear.
The bad dictionary fails with 4 errors, each naming its field.

Validation gives one of two things: a typed object (the text "129.00" became the number 129.0) or a list of named errors, here 4.

That is schema validation: an object you can trust, or a list of exactly what is wrong, which the agent sends back to the model or a person instead of guessing.

With structured output the model's reply (amber: uncertain) is always checked: the agent gets a ReturnDecision or an error, never loose text to act on.

Now with the model. with_structured_output sends the schema with the prompt and validates the reply into a ReturnDecision. Use method="json_schema", which works on both Groq and Gemini (checked October 2026).

python

facts = systems.get_order(R1001["order_id"])
prompt = (f"You handle BrightCart return requests. Rules: returns within 30 days of delivery, "
          f"unused, in the box; refunds above $100 need finance approval.\n"
          f"Order: {facts}. Days since delivery: {systems.days_since_delivery(R1001['order_id'])}.\n"
          f"Customer message: {R1001['message']}")

decider = llm.with_structured_output(ReturnDecision, method="json_schema")
try:
    decision = decider.invoke(prompt)
except Exception:                                    # a provider may reject a malformed reply: try once more
    try:
        decision = decider.invoke(prompt)
    except Exception as err:
        decision = None
        print(f"no answer (provider refused: {type(err).__name__})")
if decision:
    print(type(decision).__name__, decision.model_dump())

Code walkthrough

The facts come from the systems, not the customer's message. invoke returns a ReturnDecision, not a string: on our run, create_return, $129.00, needs_finance true. The reason wording varies; the shape does not.
Some providers reject a malformed reply with an error; the try asks once more, then prints "no answer" instead of crashing.

6. pytest and mocking (Essential)

You will change the retry code, schema and prompt many times. pytest reruns your checks in under a second, so a broken change never reaches Aisha.

In the tests the slow, costly or random parts are swapped for stand-ins (grey: replaced), so all 3 tests pass in under a second with no network or key.

A pytest test is a function named test_... that uses plain assert. A mock is a stand-in that records how it was called and returns what you set. Here one replaces time.sleep, so tests do not wait, and one replaces the model, so tests need no key.

python

import pathlib

pathlib.Path("tests").mkdir(exist_ok=True)
pathlib.Path("tests/test_returns.py").write_text('''
from unittest.mock import Mock
import httpx, pytest
from notebook_code import FakeReturnsAPI, make_client, post_with_retries, ReturnDecision, ValidationError

def test_busy_api_is_retried_and_one_return_is_created():
    api, fake_sleep = FakeReturnsAPI([429, 503, 201]), Mock()
    record, _ = post_with_retries(make_client(api), "/returns", {"order_id": "5521"}, "R-1001:return", sleep=fake_sleep)
    assert record["return_id"] == "RET-001" and len(api.records) == 1
    assert fake_sleep.call_count == 2                     # two waits, none of them real

def test_bad_token_is_not_retried():
    api, fake_sleep = FakeReturnsAPI([201]), Mock()
    with pytest.raises(httpx.HTTPStatusError):
        post_with_retries(make_client(api, token="wrong"), "/returns", {}, "R-1001:return", sleep=fake_sleep)
    fake_sleep.assert_not_called()

def test_model_output_outside_the_schema_is_rejected():
    model = Mock()                                        # stands in for the LLM: no key, no cost
    model.invoke.return_value = {"order_id": "5521", "action": "refund_everything"}
    with pytest.raises(ValidationError):
        ReturnDecision.model_validate(model.invoke("R-1001"))
''')

Code walkthrough

Three tests: a busy API is retried into one return (call_count proves two waits); a bad token fails with no sleep; a mocked model's bad output is rejected. pytest.raises passes only if the error is raised.

python

import pytest
import sys
import types

notebook = types.ModuleType("notebook_code")              # lesson-only; the lab's tests import returns_desk.triage
notebook.__dict__.update(globals())
sys.modules["notebook_code"] = notebook
sys.modules.pop("test_returns", None)                   # forget the copy from an earlier run

exit_code = pytest.main(["-q", "-p", "no:cacheprovider", "-W", "ignore::pytest.PytestAssertRewriteWarning",
                        "tests/test_returns.py"])
print("exit code:", exit_code)

Code walkthrough

The types.ModuleType lines let the test file import this notebook's functions (in the repository it imports from returns_desk). pytest.main runs pytest as pytest -q would (-W hides a harmless notebook warning): 3 passed, exit code: 0. Depth: AI Engineering, Testing.

7. Git and CI basics (Essential)

Prompts and tools change weekly in an agent project, and a bad change can move money. Git records each change so you can see and undo it; CI (continuous integration) runs the tests on every change before it is merged.

Every push runs the same tests on a clean machine: green lets the change merge, red stops it before it reaches the agent.

The CI workflow is a file in the repository. On every push or pull request, a GitHub runner checks out the code, installs Python 3.12 and the requirements, and runs pytest. A failing test turns the run red and blocks the merge.

python

import yaml

workflow = """
name: tests
on: [push, pull_request]
jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v7
      - uses: actions/setup-python@v7
        with:
          python-version: "3.12"
      - run: pip install -r requirements.txt
      - run: pytest -q
"""
pathlib.Path("requirements.txt").write_text("httpx\npydantic\npytest\n")
path = pathlib.Path(".github/workflows/tests.yml")
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(workflow.lstrip())
steps = yaml.safe_load(path.read_text())["jobs"]["test"]["steps"]
print(len(steps), "steps, last one:", steps[-1]["run"])

Code walkthrough

on lists the events, steps the commands; checkout@v7 and setup-python@v7 are current releases (checked 12 October 2026). yaml.safe_load confirms the file parses. Not run on GitHub here; verified against the workflow syntax docs.

bash

# Not run here: run in a terminal, in your returns_desk repository
git switch -c retrying-client
git add requirements.txt tests/ .github/workflows/tests.yml
git commit -m "Retry busy returns API; tests for retries and decisions"
git push -u origin retrying-client

Code walkthrough

switch -c starts a branch apart from main; add and commit record the files with a message; push sends the branch to GitHub, where a pull request triggers CI. Depth: AI Engineering, Version control and CI/CD.

8. Docker fundamentals (Essential)

Your agent will run unattended on a server, not in a notebook. Docker packs code, Python and libraries into one image; a running copy of an image is a container, the same on your laptop, in CI and in production.

Each Dockerfile line adds a layer to the image, least-changing first, and the same image runs as a container on a laptop, in CI and on a server.

A Dockerfile is the recipe, and each line adds a layer. Docker reuses unchanged layers, so put what changes least first: the base, then the requirements, then your code.

text

FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY returns_desk/ returns_desk/
COPY tests/ tests/
CMD ["pytest", "-q"]

Code walkthrough

FROM picks the base image. Installing before copying code keeps the slow step cached. CMD runs the tests for now; Module 16 runs the agent.

bash

# Not run here, verified against https://docs.docker.com/reference/cli/docker/ (checked October 2026)
docker build -t returns-desk:0.1 .
docker run --rm -e RETURNS_API_TOKEN returns-desk:0.1

Code walkthrough

build -t names and tags the image. run --rm starts a container and removes it afterwards; -e passes the token in from your environment, never baked into the image. Depth: AI Engineering, Containers.

In production

Agent teams lean on these habits most. Every outbound call has a timeout. Retries live in one place, with backoff, only for errors that may pass. Every write carries an idempotency key, because agents retry, crash and resume. Every model output is validated before code acts on it. Background work lives in a TaskGroup, or in a job queue (Module 15) when it must survive a restart. And CI checks every change: a prompt edit can break an agent as surely as code, as the returns desk finds in week 11.

What you added to the returns desk

A typed lookup that reports not_found and not_allowed and lets bugs crash.
Concurrent reads under a time budget, in a TaskGroup.
A retrying HTTP client with backoff, jitter and an idempotency key.
ReturnDecision, three pytest tests, a CI workflow and a Dockerfile.

Check yourself

Why should a tool's try hold only the call that can fail?
What can happen to a task from create_task if nothing keeps a reference to it?
Which of 401, 429 and 503 should be retried, and why does the retry need an idempotency key?

A wide try also catches bugs and reports them as facts, such as "no such order".
The loop holds only a weak reference, so garbage collection can remove it mid-run, silently.
429 and 503, which are usually temporary; 401 will fail the same way again. The first attempt may have worked and lost only its reply; the key makes the server return that record instead of a second one.

Exercise

Add a fourth test: a returns API that answers 503 four times must make post_with_retries raise RuntimeError after exactly three waits. Use a Mock for sleep. Write it inside the string in section 6's first cell, rerun that cell, then run this check:

python

import pathlib
import sys
import pytest

sys.modules.pop("test_returns", None)
sys.modules["notebook_code"].__dict__.update(globals())   # in case you edited code above

code = pytest.main(["-q", "-p", "no:cacheprovider", "-W", "ignore::pytest.PytestAssertRewriteWarning",
                   "tests/test_returns.py"])
tests = pathlib.Path("tests/test_returns.py").read_text().count("\ndef test_")
if code == 0 and tests >= 4:
    print(f"ok: {tests} tests, all passed")
else:
    print(f"not yet: {tests} tests, pytest exit code {code}")

Code walkthrough

It reruns the test file and prints ok only when there are at least four tests and all pass; until then, not yet.

Key terms

Term	Meaning
Coroutine	What an `async def` function returns; runs only when awaited
Event loop	Runs coroutines, switching between them while they wait
TaskGroup	Holds the tasks it starts and waits for all of them
Bearer token	A secret in the `Authorization` header that proves who is calling
Backoff with jitter	Waiting longer after each failure, by a slightly random amount
Idempotency key	A name for one intended action, so a repeat creates nothing new
Schema validation	Checking data against a declared shape before using it
Mock	A stand-in that records calls and returns set answers
Container	A running copy of an image with code, Python and libraries

Sources

Will McGugan (Textualize), "The Heisenbug lurking in your async code", 11 February 2023. https://textual.textualize.io/blog/2023/02/11/the-heisenbug-lurking-in-your-async-code/
Python documentation, "Coroutines and Tasks", checked October 2026. https://docs.python.org/3/library/asyncio-task.html
OpenAI, openai-python README, "Retries" and "Timeouts", checked October 2026. https://github.com/openai/openai-python/blob/main/README.md
GitHub, "Workflow syntax for GitHub Actions", checked October 2026. https://docs.github.com/en/actions/reference/workflows-and-actions/workflow-syntax
Docker, "Docker CLI reference", checked October 2026. https://docs.docker.com/reference/cli/docker/