🐍 Python for AI Developers 🤖 — From 0 to Pro 🚀

One file, one path: from x = 1 to shipping an async, typed, tested AI service that survives production.

Every example is drawn from the code AI engineers actually write — agent loops, tool registries, token streams, Pydantic schemas, FastAPI endpoints, pytest suites. No foo/bar filler.

Companion reads: 🐹 Golang for AI Developers (the sibling to this guide), 📘 The Complete Guide to LLMs and AI Agents 🤖
to understand modern AI deeply, ⚠️ Common Issues 🪲 with LLMs & AI Agents — and How to Fix Them 🛠️, 🏗️ Building High-Quality AI Agents 🤖
for the agent architecture on top of this foundation, 🔄 The Agentic Loop Guide for the control loop itself, 🏢 Enterprise-Ready AI Agents, and 🛠️ The Senior Software Engineer Playbook 📖.

---

📖 How to read this guide

You are…	Start at	Skip
New to Python	Part 1 → read straight through	Parts 12–13 on first pass
Coming from Go/Java/TS	Part 1, then Part 4 and Part 8	Part 2 (skim the tables)
Writing agents already	Part 7, Part 8, Part 13	—
Reviewing code	Part 14 and Part 15	everything else

Convention in this guide: # ✅ = do this, # ❌ = don't. Snippets target Python 3.11+ unless a version is called out.

---

📋 Table of Contents

• 1. 🧠 The Python Mental Model
• 2. 🧱 Core Data Types & Syntax
• 3. 🔧 Functions
• 4. 🏷️ The Type System
• 5. 🧬 Objects, Classes & Dataclasses
• 6. 💥 Errors & Resource Management
• 7. 🌀 Iterators, Generators and Async
• 8. ⚡ Concurrency, the GIL, and Performance
• 9. 📦 The Standard Library & AI Toolkit
• 10. 🧪 Testing with pytest
• 11. 🗂️ Project Layout & Tooling
• 12. 🐞 Debugging & Profiling in VS Code
• 13. 🏛️ Patterns That Earn Their Keep
• 14. ⚖️ Good vs Bad, Side by Side
• 15. ⚠️ Anti-Patterns and Misconceptions
• 16. 🗺️ The 30-Day Path to Pro

---

1. 🧠 The Python Mental Model

Before syntax, internalize four facts. Almost every Python surprise traces back to one of them.

1.1 Python is interpreted — what that actually means

Python source is compiled to bytecode (.pyc files under __pycache__/), then executed by the CPython virtual machine, a loop that dispatches on bytecode instructions.

import dis
def add(a, b): return a + b
dis.dis(add)   # LOAD_FAST a; LOAD_FAST b; BINARY_OP +; RETURN_VALUE

There is no machine-code compile step, no linker, no binary. The consequence: errors surface when a line runs, not when the file loads. A typo in an except branch ships to production silently.

1.2 Python vs Go — the honest comparison

Dimension	Python (CPython)	Go
Execution	Bytecode → VM interpreter	Compiled to native machine code
Typing	Dynamic, gradual (hints optional, erased at runtime)	Static, enforced by compiler
Errors caught at	Runtime (unless you run mypy)	Compile time
Raw CPU speed	~10–100× slower on tight loops	Fast
Concurrency	GIL: 1 thread runs bytecode at a time; asyncio for I/O	Real parallel goroutines
Deploy	Interpreter + venv + wheels	Single static binary
Startup	30–300 ms (imports dominate)	~1 ms
Ecosystem	Owns ML/AI: torch, transformers, numpy, pandas	Owns infra/networking

Use Python when the heavy lifting happens inside C/CUDA libraries or another service, and your code is glue + I/O. Use Go when you need CPU-bound throughput, tiny deploys, or real thread parallelism. A very common production shape — and the one in this repo's CLAUDE.md — is Go as the API gateway calling a Python ML service.

1.3 Dynamic typing ≠ no typing

Python is strongly, dynamically typed. Strongly: "1" + 1 raises instead of guessing. Dynamically: types live on values, not variables.

x = 5        # x → int object
x = "five"   # perfectly legal; the name is just a label

Type hints are annotations, not enforcement. At runtime nothing checks them:

def embed(text: str) -> list[float]: ...
embed(42)     # runs fine; explodes later inside the function

They exist for mypy/pyright, your IDE, and the next human — and for libraries like Pydantic and FastAPI that do read them at runtime. Treat "typed Python" as "Python + a type checker in CI." Without the checker, hints are documentation.

1.4 Names, objects, and mutability

Every value is an object on the heap. Variables are names bound to references. Assignment rebinds a name; it never copies.

a = [1, 2]
b = a          # same object, two names
b.append(3)
print(a)       # [1, 2, 3]  ← surprised? this is the #1 beginner bug

Immutable (safe to share)	Mutable (aliasing hazard)
int, float, bool, str, bytes, tuple, frozenset, None, Enum members	list, dict, set, bytearray, most class instances

Rules of thumb that fall out of this:

• Only immutable objects can be dict keys / set members (they need a stable __hash__).
• Never use a mutable object as a default parameter (§3.2).
• "Pass by value or reference?" — neither. Python passes the reference by value; rebinding inside a function is local, mutating is visible outside.

1.5 Everything else follows

[your .py] → compile → [bytecode] → [CPython VM, holds the GIL]
                                        ↓ calls into
                          [C extensions: numpy, torch, orjson] → release the GIL

That diagram explains the GIL debate (§8), why numpy is fast, and why asyncio is the concurrency story for I/O.

🎯 Actionable rules

1. Add type hints from line one, and run mypy in CI — you are buying back what the compiler gives Go.
2. Assume any function you pass a list/dict to may mutate it; copy at boundaries you care about.
3. Don't fight Python on CPU speed — push hot loops into numpy/C or another service.

---

2. 🧱 Core Data Types & Syntax

2.1 Scalars

n: int      = 42            # arbitrary precision — no int64 overflow, ever
f: float    = 0.7           # IEEE 754 double
b: bool     = True          # bool is a subclass of int: True + True == 2
s: str      = "hello"       # immutable sequence of Unicode code points
raw: bytes  = b"\x00\x01"   # immutable sequence of 0–255 ints
nothing     = None          # the single NoneType instance

str vs bytes — the boundary that bites AI devs. Files, sockets, and HTTP bodies give you bytes; models and JSON want str.

data = "café".encode("utf-8")     # str → bytes: b'caf\xc3\xa9'  (5 bytes, 4 chars)
text = data.decode("utf-8")       # bytes → str
len("café"), len(data)            # (4, 5)

Never concatenate the two, and never guess an encoding — pass encoding= explicitly.

Constants. Python has none. Convention is UPPER_SNAKE_CASE at module level; typing.Final lets the checker enforce it.

from typing import Final
MAX_HISTORY_TURNS: Final[int] = 20
TOOL_TIMEOUT_SECONDS: Final = 30.0

Type conversion is explicit and constructor-shaped:

int("42"), int(3.9), int("ff", 16)     # 42, 3 (truncates), 255
float("0.7"), str(42), bool("")        # 0.7, '42', False
list("abc"), tuple([1, 2]), set([1, 1])  # ['a','b','c'], (1,2), {1}

2.2 Truthiness, and/or, is vs ==

Falsy: False, None, 0, 0.0, "", [], {}, (), set(). Everything else is truthy.

and/or return an operand, not a bool — that is why they work as defaults:

name = user_name or "anonymous"        # "" / None → "anonymous"
tools = cfg.tools and list(cfg.tools)  # None-safe: returns None or the list

⚠️ Trap: or fires on any falsy value, so timeout = user_timeout or 30 silently turns a deliberate 0 into 30. Use an explicit is None check when 0/""/False are valid inputs.

Operator	Asks	Use for
==	Same value (calls __eq__)	Almost everything
is	Same object (identity)	None, True/False, sentinels, enum members

if resp is None: ...              # ✅
if resp == None: ...              # ❌ works, but sloppy and slower
if isinstance(x, str): ...        # ✅ type check — accepts subclasses
if type(x) == str: ...            # ❌ brittle
isinstance(x, (int, float))       # tuple = "any of these"

2.3 Strings: f-strings and the methods you'll actually use

f-strings are the only interpolation style you need.

name, score = "calculator", 0.9421
f"{name} scored {score:.2f}"         # 'calculator scored 0.94'
f"{name!r}"                          # "'calculator'"  ← repr(): quotes + escapes
f"{score:>8.1%}"                     # '   94.2%'      ← align, width, percent
f"{name=}, {score=}"                 # "name='calculator', score=0.9421"  (debug, 3.8+)
f"{'\n'.join(lines)}"                # nested quotes/backslashes OK in 3.12+

!r vs !s — !r calls repr(), which shows quotes and escapes. Use it in logs and error messages so "" and " " are distinguishable:

raise ValueError(f"tool name must be non-empty, got {name!r}")
# → tool name must be non-empty, got '   '   ← the whitespace is visible

Multi-line and templating:

SYSTEM = f"""You are {agent_name}.
Available tools: {", ".join(tool_names)}
"""
"Hello {who}".format(who="world")     # runtime templates (user-supplied strings)

Never build a prompt with + in a loop, and never use f-strings for SQL — use parameterized queries.

String methods, ranked by how often you'll use them:

"  hi \n".strip()            # 'hi'      also lstrip/rstrip
"Calculate 2+2".lower()      # 'calculate 2+2'   (casefold() for Unicode-correct)
"a,b,c".split(",")           # ['a','b','c']     split() alone → splits on any whitespace
"calculate 10*5".split("calculate", 1)[-1].strip()   # '10*5'  ← maxsplit=1 keeps the tail
", ".join(["a", "b"])        # 'a, b'    ← join is a method ON the separator
"tool:web".startswith("tool:")   # True   endswith() likewise; both accept tuples
"result: ok".replace("ok", "done")
"api_key" in text            # substring test
"x".ljust(8), "5".zfill(3)   # 'x       ', '005'
"path/to/x".removeprefix("path/")    # 'to/x'  (3.9+, safer than lstrip)

⚠️ "abcx".lstrip("xa") strips characters, not a prefix — removeprefix is what you meant.

2.4 Collections at a glance

Type	Literal	Ordered	Mutable	Lookup	Use it for
list	[1, 2]	✅	✅	O(n)	Sequences you append to: messages, chunks
tuple	(1, 2)	✅	❌	O(n)	Fixed records, dict keys, *args, safe defaults
dict	{"a": 1}	✅ (insertion)	✅	O(1)	Everything keyed: JSON, registries, kwargs
set	{1, 2}	❌	✅	O(1)	Membership, dedupe, allow-lists
frozenset	frozenset({1})	❌	❌	O(1)	Hashable set: dict key, class constant

2.5 list

msgs = ["hi"]
msgs.append("there")            # add one → ['hi', 'there']
msgs.extend(["a", "b"])         # add many (append would nest the list!)
msgs.insert(0, "sys"); msgs.pop(); msgs.pop(0); msgs.remove("a")
msgs.sort(key=len, reverse=True)      # in place, returns None
top = sorted(msgs, key=len)           # new list  ← prefer this
list(reversed(msgs)); msgs[::-1]      # reversed view vs reversed copy
len(msgs); sum([1, 2, 3]); max(scores); min(scores)

⚠️ msgs = msgs.sort() sets msgs to None. Mutating methods return None by design.

Slicing — seq[start:stop:step], stop exclusive, all parts optional:

history[-10:]        # last 10 messages  ← the sliding-window idiom
history[:-1]         # everything but the last
tokens[::2]          # every other
text[::-1]           # reversed string
history[:] = []      # clear in place (keeps aliases in sync)

Slices never raise for out-of-range — history[-10:] on a 3-item list returns 3 items. That is a feature for context windows.

Comprehensions — the idiomatic map/filter. Read them left-to-right as "expression for item in iterable if cond".

names   = [t.name for t in tools]                        # map
enabled = [t for t in tools if t.enabled]                # filter
lengths = {t.name: len(t.schema) for t in tools}         # dict comp
uniq    = {m.role for m in history}                      # set comp
flat    = [tc for m in messages for tc in (m.tool_calls or [])]   # nested: outer loop first
lazy    = (t.name for t in tools)                        # generator — no list built

A real one from an agent test suite:

tool_calls = sorted(
    {
        tc["name"]
        for m in result["messages"]
        if isinstance(m, AIMessage)          # filter applies to the OUTER loop
        for tc in (m.tool_calls or [])       # then the inner loop
    }
)

Rule: if a comprehension needs a second if plus a ternary plus a nested loop, write a for loop.

2.6 dict

cfg = {"model": "claude-opus-5", "temp": 0.7}
cfg["model"]                    # KeyError if missing
cfg.get("temp")                 # None if missing
cfg.get("temp", 1.0)            # default if missing   ← use for optional config
cfg.setdefault("tools", []).append("calc")   # get-or-create in one step
cfg.update({"temp": 0.2}, top_p=0.9)         # merge in place
merged = {**defaults, **overrides, "stream": True}   # new dict; later wins
merged = defaults | overrides                        # same thing, 3.9+
cfg.pop("temp", None)           # remove, no raise
list(cfg.keys()); cfg.values(); cfg.items()
for k, v in cfg.items(): ...

.get() vs [] — decide by intent, not by fear:

Situation	Use	Why
Key is required; absence is a bug	cfg["model"]	KeyError names the key — fail loudly
Key is optional	cfg.get("temp", 0.7)	Explicit default
Need to know if it was absent	"k" in cfg / cfg.get("k")	None may be a legit value

⚠️ .get() everywhere turns a missing-key bug into an AttributeError: 'NoneType' fifty lines later. Loud beats silent.

Set-like operations on keys (dict - dict is not a thing; .keys() is):

missing = required.keys() - provided.keys()   # keys in A not in B
shared  = a.keys() & b.keys()
changed = {k: v for k, v in new.items() if old.get(k) != v}   # diff two dicts

2.7 set and frozenset

seen = set()                    # {} is an empty DICT — this is the trap
seen.add("doc-1"); seen.discard("x")     # discard = remove without KeyError
"doc-1" in seen                 # O(1) — the whole point
a | b, a & b, a - b, a ^ b      # union, intersection, difference, symmetric diff
ALLOWED = frozenset({"read", "grep"})    # hashable + immutable → safe class constant

Dedupe while preserving order: list(dict.fromkeys(items)).

2.8 tuple

Fixed-length, immutable, hashable — the right type for records and safe defaults.

point = (1.0, 2.0)
x, y = point                          # unpacking
first, *rest = [1, 2, 3]              # star-unpacking → 1, [2, 3]
allowed_tools: tuple[str, ...] = ()   # ← immutable default: safe as a class field
CACHE: dict[tuple[str, int], str] = {}  # composite key — a list can't do this

tuple[str, ...] = "any number of str". tuple[str, int] = exactly two, in that order.

2.9 enum — kill your magic strings

from enum import Enum, StrEnum, auto

class Role(StrEnum):        # 3.11+; members ARE str → JSON-serializable for free
    USER = "user"
    ASSISTANT = "assistant"
    SYSTEM = "system"

class Status(Enum):
    OK = auto(); RETRY = auto(); FAILED = auto()

Role.USER.value             # 'user'
Role("user")                # lookup by value → Role.USER (raises ValueError if bad)
msg = {"role": Role.USER, "content": "hi"}   # StrEnum works directly in JSON
if status is Status.OK: ...  # identity compare — enum members are singletons
list(Role)                   # iterate all members

Why bother: typos become ValueError at the boundary instead of a silent no-match branch, and your IDE autocompletes the valid set.

2.10 Control flow

if score > 0.9:      ...
elif score > 0.5:    ...
else:                ...

label = "high" if score > 0.9 else "low"      # ternary

for i, msg in enumerate(history, start=1):    # index + item
    print(f"{i}. {msg.role}")

for name, score in zip(names, scores, strict=True):   # strict=True (3.10+) catches length mismatch
    ...

lookup = dict(zip(names, scores))             # two lists → dict

while retries < MAX_RETRIES:
    retries += 1
    if transient: continue                    # next iteration
    if fatal:     break                       # exit loop
else:
    raise RuntimeError("retries exhausted")   # runs only if NO break happened

for x in items: pass                          # `pass` = syntactic no-op placeholder

The for/while ... else clause is rare but perfect for search loops: else = "loop finished without finding anything."

match (3.10+) — structural pattern matching, not a C switch. It shines on the shape-dispatch that agent code is full of:

match event:
    case {"type": "tool_call", "name": str(name), "args": dict(args)}:
        run_tool(name, **args)
    case {"type": "text", "content": content} if content.strip():
        emit(content)
    case [first, *rest]:                       # sequence pattern
        ...
    case Status.FAILED:                        # enum / literal
        retry()
    case _:                                    # default
        log.warning("unhandled %r", event)

For a plain value-to-handler mapping, a dict is still better: HANDLERS[kind](payload).

Note on for vs async for: async for iterates an async generator (an LLM token stream, a paginated API). Covered in §7.

2.11 Builtins worth memorizing

all(t.enabled for t in tools)        # True if every item truthy (True on empty)
any(t.name == "bash" for t in tools) # True if at least one (False on empty; short-circuits)
len(x); sum(xs); min(xs); max(xs, key=len); abs(-1); round(0.746, 2)
sorted(items, key=lambda t: t.score, reverse=True)
sorted(range(len(points)), key=lambda i: scores[i], reverse=True)   # argsort: indices by score
enumerate(xs, 1); zip(a, b); reversed(xs); range(0, 10, 2)
isinstance(x, T); type(x).__name__; getattr(obj, "name", default); callable(fn)
repr(x); print(x, sep=" ", end="\n", flush=True)

print() is for scripts and demos. In services use logging (§9.5) — you get levels, structure, and timestamps.

🎯 Actionable rules

1. dict for keyed data, set for membership, tuple for fixed records, list for sequences you grow.
2. {} is an empty dict; set() is an empty set.
3. Use !r in every error message that quotes a value.
4. Replace magic strings with StrEnum the moment there are more than two of them.

---

3. 🔧 Functions

3.1 Anatomy

def summarize(text: str, *, max_words: int = 50) -> str:
    """Return a summary of `text`, capped at `max_words` words.

    Why: LLM context is finite; callers pass raw documents and need a
    bounded string back. Truncation is word-aligned, never mid-token.

    Args:
        text: Raw document. Whitespace is collapsed.
        max_words: Hard cap on output length. Must be > 0.

    Returns:
        The first `max_words` words, space-joined.

    Raises:
        ValueError: If `max_words` <= 0.
    """
    if max_words <= 0:
        raise ValueError(f"max_words must be positive, got {max_words!r}")
    return " ".join(text.split()[:max_words])

Docstrings — what and why. A """...""" as the first statement in a module/class/function becomes obj.__doc__. It powers help(), IDE hovers, and doc generators — and, increasingly, it is what an LLM reads when your function becomes a tool. Write the why, the contract, and the failure modes; the what is already in the signature. One line is fine for obvious helpers; skip nothing that surprises a reader.

3.2 Default arguments — the classic trap

Defaults are evaluated once, at def time, and stored on the function object. A mutable default is shared by every call.

def append_buggy(item: str, history: list = []) -> list:   # ❌ noqa: B006
    """Bug: `history` is created once at def-time and shared across calls."""
    history.append(item)
    return history

append_buggy("a")   # ['a']
append_buggy("b")   # ['a', 'b']  ← leaks across calls, across requests, across tests

def append_fixed(item: str, history: list | None = None) -> list:   # ✅
    """Fix: None sentinel, fresh list per call."""
    if history is None:
        history = []
    history.append(item)
    return history

The same applies to {}, set(), datetime.now(), and any object built at def time. Rule: default arguments must be immutable (None, 0, "", (), frozenset()). Ruff's B006 catches this — leave it on.

Where a class holds the default, use tuple[str, ...] = () or a dataclass field(default_factory=list) (§5.5).

3.3 Parameters: positional, keyword-only, *args, **kwargs

def call(name, /, *args, timeout: float = 30.0, **kwargs):
    #        ↑ positional-only     ↑ keyword-only (after *)
    ...

• *args packs extra positionals into a tuple.
• **kwargs packs extra keywords into a dict.
• A bare * in the signature makes everything after it keyword-only — the single highest-value readability trick in Python.

async def publish_ingest_request(
    client: redis.Redis,
    *,                      # everything below MUST be passed by name
    job_id: str,
    tenant: str,
    priority: int = 0,
) -> None: ...

await publish_ingest_request(r, job_id="j1", tenant="acme")   # ✅ self-documenting
await publish_ingest_request(r, "j1", "acme")                 # ❌ TypeError at the door

Use * for any function with 3+ arguments, booleans, or same-typed neighbours. It makes call sites readable and lets you reorder parameters without breaking callers.

Unpacking at the call site mirrors packing:

args = ("calculator",); kwargs = {"expression": "2+2"}
run_tool(*args, **kwargs)              # spread
run_tool(**{**base_kwargs, "timeout": 5})   # merge-then-spread
first, *middle, last = messages        # star-unpack a sequence
a, b = b, a                            # swap (tuple pack/unpack)

3.4 Framework-style defaults: Depends(...), Header(...)

FastAPI reads your annotations plus default values at import time to build the request pipeline. A default of Header(...) or Depends(fn) is not a value — it's a marker object the framework interprets.

from fastapi import Depends, Header, HTTPException
from typing import Annotated

async def get_ctx(x_tenant: Annotated[str, Header()]) -> dict:
    """Dependency: runs per request, result injected into any handler that asks."""
    if not x_tenant:
        raise HTTPException(401, "missing tenant")
    return {"tenant": x_tenant}

@app.post("/query")
async def query(
    body: QueryIn,                                   # parsed + validated from JSON body
    ctx: Annotated[dict, Depends(get_ctx)],          # injected
    trace_id: Annotated[str | None, Header()] = None # from the `trace-id` header
) -> QueryOut: ...

Dependencies are cached per request, can be nested, and are the clean place for auth, tenancy, DB sessions, and rate limits. Prefer the Annotated[...] form — it keeps the type and the metadata separate, and works with plain function calls in tests.

3.5 lambda, closures, and scope

sorted(tools, key=lambda t: t.score)       # ✅ tiny, inline, single expression
handler = lambda x: x + 1                  # ❌ just use def — you lose the name in tracebacks

A closure is a function that captures variables from its enclosing scope. It's the lightest possible way to carry configuration:

def make_retrier(attempts: int, backoff: float):
    """Factory → returns a configured function. `attempts` lives on in the closure."""
    def retry(fn):
        for i in range(attempts):
            try:
                return fn()
            except TransientError:
                time.sleep(backoff * 2 ** i)
        raise RuntimeError(f"failed after {attempts} attempts")
    return retry

retry_fast = make_retrier(attempts=3, backoff=0.1)

Scope resolution is LEGB: Local → Enclosing → Global → Builtins. Assignment makes a name local for the whole function, which is why this fails:

count = 0
def bump():
    count += 1        # ❌ UnboundLocalError: `count` is local because it's assigned

def bump_ok():
    global count      # module-level rebinding — legal, but a smell
    count += 1

def outer():
    n = 0
    def inner():
        nonlocal n    # rebind the ENCLOSING variable — the right tool for closures
        n += 1
    inner(); return n

⚠️ global mutable state is the enemy of testable, concurrent code. Prefer passing an object, a closure, or a dependency.

⚠️ Late-binding gotcha: closures capture the variable, not its value.

fns = [lambda: i for i in range(3)]      # ❌ all three return 2
fns = [lambda i=i: i for i in range(3)]  # ✅ bind now via default arg

🎯 Actionable rules

1. Mutable default → None sentinel. Always.
2. Put a bare * in any signature with more than two parameters.
3. Docstrings explain why and raises; the signature already says what.

---

4. 🏷️ The Type System

Hints are erased at runtime — but a checker turns them into Go-grade safety, and Pydantic/FastAPI turn them into validation. This is the highest-leverage chapter for anyone coming from a static language.

4.1 The basics

name: str
scores: list[float]                    # builtin generics (3.9+) — no typing.List needed
index: dict[str, list[int]]
pair: tuple[str, int]                  # exactly 2
names: tuple[str, ...]                 # N of the same
maybe: str | None = None               # 3.10+ ; same as Optional[str]
num: int | float                       # union

X | None is not optional-as-in-omittable — it means "this value may be None". A parameter is omittable when it has a default. Both often appear together: history: list | None = None.

4.2 Any vs object vs no annotation

Annotation	Checker behaviour	Use when
Any	Disables checking — every operation allowed	Untyped third-party boundary; escape hatch
object	Accepts anything, allows nothing until narrowed	You genuinely accept any value and will isinstance it
(missing)	Implicitly Any — silent hole	Never, in checked code

def dynamic_dispatch(obj: object, method: str, text: str) -> str:  # ✅ object, then narrow
    fn = getattr(obj, method, None)
    if not callable(fn):
        return f"no handler for '{method}'"
    return fn(text)

Any is contagious: one Any in a chain silences every downstream error. Quarantine it at the edge — parse into a real type immediately.

4.3 Literal, Final, NewType

from typing import Literal, Final, NewType

Mode = Literal["stream", "batch"]      # only these two strings type-check
def run(mode: Mode = "stream") -> None: ...
run("streaming")                       # ❌ mypy: not a valid Mode

MAX_TOKENS: Final = 4096               # rebinding is an error
TenantId = NewType("TenantId", str)    # distinct type at check time, plain str at runtime
def load(t: TenantId) -> None: ...
load("acme")                           # ❌ — forces you through TenantId("acme")

Literal is the cheapest way to model a small closed set inside a signature; Enum is better when the set is used in many places or needs behaviour.

4.4 Callable — typing functions

from collections.abc import Callable, Awaitable

ToolFn = Callable[[str, dict], str]              # (str, dict) -> str
AsyncToolFn = Callable[..., Awaitable[str]]      # ... = "any arguments"
Hook = Callable[[str], None]

REGISTRY: dict[str, ToolFn] = {}
def register(name: str) -> Callable[[ToolFn], ToolFn]:   # a decorator's type
    def deco(fn: ToolFn) -> ToolFn:
        REGISTRY[name] = fn
        return fn
    return deco

Import Callable, Iterable, Sequence, Mapping, Awaitable, AsyncIterator from collections.abc, not typing (the typing aliases are deprecated).

4.5 Generics — TypeVar and Generic

A generic preserves the relationship between input and output types.

# 3.12+ syntax — clean and preferred
def first[T](items: list[T]) -> T | None:
    return items[0] if items else None

class Cache[K, V]:
    def __init__(self) -> None: self._d: dict[K, V] = {}
    def get(self, k: K) -> V | None: return self._d.get(k)
    def put(self, k: K, v: V) -> None: self._d[k] = v

# Pre-3.12 equivalent
from typing import TypeVar, Generic
T = TypeVar("T")
def first_legacy(items: list[T]) -> T | None: ...
class CacheLegacy(Generic[K, V]): ...

cache: Cache[str, list[float]] = Cache()   # embeddings by doc id

Without generics you'd annotate -> Any and lose every downstream check. Bounded type vars constrain the family: def largest[T: (int, float)](xs: list[T]) -> T.

4.6 Protocol — duck typing the checker understands

Python's runtime does structural typing: "if it quacks, it's a duck." Protocol brings that to static checking — no base class, no registration, no import coupling.

from typing import Protocol, runtime_checkable

@runtime_checkable
class Tool(Protocol):
    name: str
    def run(self, **kwargs: object) -> str: ...

class Calculator:                       # does NOT inherit from Tool
    name = "calculator"
    def run(self, **kwargs: object) -> str:
        return str(eval_expr(str(kwargs["expression"])))

def execute(tool: Tool, **kw: object) -> str:   # accepts anything shaped right
    return tool.run(**kw)

execute(Calculator(), expression="2+2")         # ✅ type-checks, no inheritance
isinstance(Calculator(), Tool)                  # True — only with @runtime_checkable

	ABC / inheritance	Protocol
Coupling	Implementer imports the base	Zero — the interface can live in the consumer
Third-party classes	Must register()	Just work
Runtime isinstance	Always	Only with @runtime_checkable (checks method names only)

Use Protocol for interfaces you consume (an LLM client, a tool, a store) — it makes fakes in tests trivial. Use an ABC when you want shared implementation and enforced construction:

from abc import ABC, abstractmethod

class BaseTool(ABC):
    """ABC: a contract the subclass MUST fill, plus behaviour it inherits."""
    def __init__(self, name: str) -> None:
        self.name = name

    @abstractmethod
    def run(self, **kwargs: object) -> str: ...

    def describe(self) -> str:                # ← shared implementation; a Protocol can't give you this
        return f"{self.name}: {self.run.__doc__ or 'no docs'}"

class Calculator(BaseTool):
    def run(self, **kwargs: object) -> str:
        return str(safe_eval(str(kwargs["expression"])))

BaseTool("x")        # ❌ TypeError at instantiation: abstract method 'run' not implemented

The distinction in one line: an ABC is a base class you inherit; a Protocol is a shape you happen to match. ABCs enforce at instantiation, Protocols at type-check time.

Callback protocols type a function, including parameter names and defaults — which Callable[[...], T] cannot express. This is the right type for a keyword-driven tool registry:

class ToolFn(Protocol):
    def __call__(self, *, expression: str, precision: int = 2) -> str: ...

def register(name: str, fn: ToolFn) -> None: ...

def calc(*, expression: str, precision: int = 2) -> str: ...   # ✅ matches
def bad(expr: str) -> str: ...                                 # ❌ mypy: wrong parameter name

Async protocols are how you type an LLM client. Note the asymmetry: a coroutine method is declared async def, but a method returning an async generator is declared with a plain def returning AsyncIterator — because calling it hands you the iterator without awaiting:

from collections.abc import AsyncIterator

class LLMClient(Protocol):
    async def complete(self, prompt: str, *, max_tokens: int = 1024) -> str: ...
    def stream(self, prompt: str) -> AsyncIterator[str]: ...

class Anthropic:                                   # satisfies both, no inheritance
    async def complete(self, prompt: str, *, max_tokens: int = 1024) -> str: ...
    async def stream(self, prompt: str) -> AsyncIterator[str]:   # async gen fn → OK
        yield "token"

Protocols can be generic, which is what you want for stores and caches:

class Store[T](Protocol):                          # 3.12+ syntax
    def get(self, key: str) -> T | None: ...
    def put(self, key: str, value: T) -> None: ...

⚠️ @runtime_checkable is weaker than it looks. isinstance checks that the member names exist (hasattr) — never signatures, never types:

class Broken:
    name = "broken"
    def run(self) -> None:              # wrong parameters, wrong return type
        print("nope")

isinstance(Broken(), Tool)              # ⚠️ True — names matched, nothing else was checked
issubclass(Broken, Tool)                # ❌ TypeError: protocols with non-method members
                                        #    don't support issubclass()

So use it as a cheap plugin filter, not as validation. If you want the checker to verify a class at its definition site instead of at every call site, inherit from the Protocol explicitly — that's allowed, and you also pick up any default method bodies it defines:

class Calculator(Tool):      # explicit: mypy reports a mismatch HERE, not 40 files away
    ...

4.7 TypedDict and Annotated

from typing import TypedDict, NotRequired, Annotated

class ToolCall(TypedDict):
    name: str
    args: dict[str, object]
    id: NotRequired[str]                # optional key (3.11+)

tc: ToolCall = {"name": "calc", "args": {"expression": "1+1"}}
tc["nmae"]                              # ❌ mypy catches the typo

TypedDict types JSON-ish dicts you can't or won't turn into classes (LangChain state, API payloads). For anything you validate, prefer a Pydantic model (§9.1).

Annotated[T, ...] attaches metadata to a type without changing it — the mechanism behind FastAPI and Pydantic constraints:

Temp = Annotated[float, Field(ge=0.0, le=2.0)]
ctx: Annotated[dict, Depends(get_ctx)]

4.8 Self and forward references

from typing import Self

class Builder:
    def with_tool(self, name: str) -> Self:   # 3.11+ — correct for subclasses
        self._tools.append(name); return self
    def build(self) -> "Agent":               # string = forward ref to a later class
        ...

from __future__ import annotations at the top of a file makes all annotations lazy strings — no more quoting forward refs, and cheaper imports. Caveat: libraries that read annotations at runtime (older Pydantic setups) may need model_rebuild().

4.9 Narrowing — how the checker follows your logic

def describe(x: str | int | None) -> str:
    if x is None:            return "empty"       # x narrowed out of the union
    if isinstance(x, int):   return f"n={x}"      # x is int here
    return x.upper()                              # x is str here — .upper() is safe

parsed: object = json.loads(raw)
keys = list(parsed.keys()) if isinstance(parsed, dict) else []   # narrow before use

assert x is not None, isinstance, is None, and truthiness checks all narrow. cast(T, x) lies to the checker — use it only when you've proven the invariant elsewhere.

4.10 Running the checker

# pyproject.toml
[tool.mypy]
python_version = "3.12"
strict = true                    # turn everything on, then relax
warn_unreachable = true
plugins = ["pydantic.mypy"]

[[tool.mypy.overrides]]
module = ["untyped_lib.*"]
ignore_missing_imports = true

uv run mypy src/          # or: pyright src/

Start with strict = true on a new project. On an old one, enable per-module and ratchet. A type error found in CI costs seconds; the same error found at 3 a.m. in an agent loop costs hours.

🎯 Actionable rules

1. Annotate every public signature; let inference handle locals.
2. Protocol for interfaces you depend on; Enum/Literal instead of bare strings.
3. Any only at the untyped boundary, and parse it into a real type immediately.
4. Hints without a checker in CI are just comments.

---

5. 🧬 Objects, Classes & Dataclasses

5.1 A class, annotated

class Agent:
    """One conversational agent instance."""

    MAX_STEPS: int = 10                    # class attribute — shared by all instances

    def __init__(self, config: AgentConfig) -> None:
        self.config = config               # instance attributes live on `self`
        self._history: list[Message] = []  # leading _ = "internal, don't touch"

    def add_message(self, role: Role, content: str) -> None:
        self._history.append(Message(role=role, content=content))

    @property
    def history(self) -> tuple[Message, ...]:
        """Read-only view — callers can't mutate our list."""
        return tuple(self._history)

    @classmethod
    def from_env(cls) -> "Agent":
        """Alternative constructor. `cls` = the actual class, so subclasses work."""
        return cls(AgentConfig(model=os.environ["MODEL"]))

    @staticmethod
    def supported_models() -> list[str]:
        """No self/cls needed — namespaced utility."""
        return ["claude-opus-5", "claude-sonnet-5"]

self is explicit because Python resolves attributes at runtime; the first parameter is the instance. Nothing magic — Agent.add_message(a, ...) and a.add_message(...) are the same call.

Decorator	First arg	Use for
(none)	self	Normal behaviour
@classmethod	cls	Alternative constructors, factories, registry hooks
@staticmethod	—	Pure helpers that belong to the namespace
@property	self	Computed/read-only attribute access

Python has no private. _name is a convention; __name triggers name-mangling (_Class__name) which prevents accidental subclass collisions, not access.

5.2 Dunder methods — the protocol layer

"Dunder" = double underscore. These hook your class into language syntax.

class Message:
    def __init__(self, role: Role, content: str) -> None:
        if not content.strip():
            raise ValueError(f"content must be non-blank, got {content!r}")
        self.role, self.content = role, content

    def __repr__(self) -> str:              # what devs/logs see — make it unambiguous
        return f"Message(role={self.role.value!r}, content={self.content[:20]!r})"

    def __str__(self) -> str:               # what users see; falls back to __repr__
        return f"{self.role.value}: {self.content}"

    def __eq__(self, other: object) -> bool:
        if not isinstance(other, Message): return NotImplemented
        return (self.role, self.content) == (other.role, other.content)

    def __hash__(self) -> int:              # define WITH __eq__ or the class becomes unhashable
        return hash((self.role, self.content))

    def __len__(self) -> int:  return len(self.content)
    def __bool__(self) -> bool: return bool(self.content.strip())

Dunder	Enables
__init__ / __new__	Construction
__repr__ / __str__	repr(x) / str(x), f-strings, logs
__eq__ + __hash__	==, dict keys, set members
__lt__ __le__ __gt__ __ge__	<, sorted(), min/max
__len__ __bool__	len(), truthiness
__iter__ / __next__	for loops
__aiter__ / __anext__	async for
__enter__ / __exit__	with
__aenter__ / __aexit__	async with
__call__	Instance becomes callable
__getattr__	Fallback for missing attributes (proxies, lazy loading)

Ordering without writing all four comparisons:

from functools import total_ordering

@total_ordering
class Score:
    def __init__(self, v: float) -> None: self.v = v
    def __eq__(self, o: object) -> bool: return isinstance(o, Score) and self.v == o.v
    def __lt__(self, o: "Score") -> bool: return self.v < o.v
    # __le__, __gt__, __ge__ are generated

Two dunders you'll read constantly in library code:

type(exc).__name__      # 'ValueError' — the class name of an exception, for logs
__name__                # module's own name: "__main__" when run directly (see §11.4)

5.3 Dynamic attribute access — getattr and friends

class Handlers:
    def summarize(self, text: str) -> str: return f"summary({text})"
    def classify(self, text: str)  -> str: return f"class({text})"

def dynamic_dispatch(obj: object, method: str, text: str) -> str:
    """
    Look up a method by string name — core pattern in plugin/tool registries.
    getattr() resolves the attribute; callable() guards against non-methods.
    """
    fn = getattr(obj, method, None)      # 3rd arg = default instead of AttributeError
    if not callable(fn):
        return f"no handler for '{method}'"
    return fn(text)

Why it matters: an LLM returns a tool name as a string. getattr is how a string becomes a call. Companions: hasattr, setattr, vars(obj), dir(obj).

Why to be careful: it defeats static checking and autocompletion, and unguarded getattr(obj, user_input) is an arbitrary-attribute-access vulnerability. Always validate the name against an explicit allow-list (a dict registry is usually better than getattr on self).

5.4 Copying: assignment vs shallow vs deep

import copy
orig = {"tools": ["calc"], "cfg": {"temp": 0.7}}

alias   = orig                       # same object
shallow = copy.copy(orig)            # new dict, SAME inner objects  (also dict(orig), orig[:])
deep    = copy.deepcopy(orig)        # new dict, new inner objects, recursively

shallow["tools"].append("bash")      # ⚠️ mutates orig["tools"] too
deep["cfg"]["temp"] = 0.1            # orig untouched

Use shallow copies for flat structures (cheap), deepcopy for nested state you must isolate (agent state snapshots, test fixtures). deepcopy is slow and chokes on sockets, locks, and open files — for those, define __deepcopy__ or restructure. Best of all: use immutable data so the question disappears.

5.5 Dataclasses

@dataclass generates __init__, __repr__, and __eq__ from annotated fields. It's the default choice for internal value objects.

from dataclasses import dataclass, field, asdict, replace

@dataclass(slots=True)                       # slots=True: less memory, faster attrs (3.10+)
class AgentConfig:
    name: str
    model: str = "claude-opus-5"
    temperature: float = 0.7
    tools: list[str] = field(default_factory=list)   # ✅ fresh list per instance
    # tools: list[str] = []                          # ❌ ValueError at class creation

cfg = AgentConfig(name="researcher", tools=["web"])
asdict(cfg)                                   # → dict, recursively
replace(cfg, temperature=0.1)                 # → new instance with one field changed

Frozen = immutable (and hashable), which makes instances safe as dict keys, safe to share across threads/tasks, and safe as defaults:

@dataclass(frozen=True, slots=True)
class ToolResult:
    tool_name: str
    output: str
    ok: bool = True

r = ToolResult("calculator", "42")
r.ok = False                    # ❌ FrozenInstanceError
r2 = replace(r, ok=False)       # ✅ make a new one

Other useful knobs: order=True (generates comparisons), kw_only=True (all fields keyword-only), field(compare=False) (exclude from __eq__), field(repr=False) (keep secrets out of logs).

__post_init__ runs after the generated __init__ — the place for validation and derived fields:

@dataclass
class Window:
    max_turns: int
    def __post_init__(self) -> None:
        if self.max_turns < 1:
            raise ValueError(f"max_turns must be >= 1, got {self.max_turns!r}")

5.6 Which container type should I use?

Need	Choose
Internal value object, no validation	@dataclass(slots=True)
Immutable key / shared constant	@dataclass(frozen=True) or NamedTuple
Data crossing a trust boundary (API, LLM output, config file)	Pydantic BaseModel (§9.1)
Loose JSON shape you only read	TypedDict
Behaviour + state + inheritance	plain class

🎯 Actionable rules

1. Reach for @dataclass before writing __init__ by hand.
2. field(default_factory=...) for every mutable field.
3. Prefer frozen=True until you have a reason to mutate.
4. Define __repr__ on anything that will appear in a log line.

---

6. 💥 Errors & Resource Management

6.1 try / except / else / finally

try:
    result = await call_model(prompt)
except (httpx.TimeoutException, httpx.ConnectError) as exc:      # catch related errors together
    log.warning("transient failure: %s: %s", type(exc).__name__, exc)
    raise RetryableError("model unreachable") from exc            # ← chain, don't swallow
except httpx.HTTPStatusError as exc:
    if exc.response.status_code == 429:
        raise RateLimitError(retry_after(exc.response)) from exc
    raise                                                         # bare raise = re-raise as-is
else:
    log.info("ok in %d tokens", result.usage.output_tokens)       # runs only if NO exception
finally:
    await client.aclose()                                         # always runs

• else keeps the happy path out of the try block, so you don't accidentally catch exceptions from your own success handling.
• finally always runs — including on return and on break. Never return from finally; it discards the in-flight exception.

raise ... from exc preserves the cause. Without it you lose the original traceback and debugging becomes archaeology:

except concurrent.futures.TimeoutError as exc:
    raise TimeoutError(
        f"Tool call exceeded the {TOOL_TIMEOUT_SECONDS}s timeout."
    ) from exc

Use from None deliberately when the inner error is noise you must hide (e.g. leaking a secret in the message).

6.2 The built-in errors you'll meet

Exception	Raised when	Typical agent-code cause
ValueError	Right type, wrong value	int("abc"), invalid temperature, blank content
TypeError	Wrong type / bad arguments	"a" + 1, missing required kwarg
KeyError	Missing dict key	payload["tool_calls"] on a text-only response
IndexError	Out-of-range index	parts[1] after a split that found nothing
AttributeError	Missing attribute	None.content — an unhandled .get()
RuntimeError	Invalid state	Loop already running, generator reused, retries exhausted
TimeoutError	Deadline exceeded	asyncio.wait_for, tool timeouts
StopIteration / StopAsyncIteration	Iterator exhausted	Raised by next() / __anext__
asyncio.CancelledError	Task cancelled	Client disconnected — must not be swallowed
NotImplementedError	Abstract method	Unfinished subclass hook

Custom exceptions, in a small hierarchy so callers can catch broadly or narrowly:

class AgentError(Exception):
    """Base for everything this package raises."""

class ToolError(AgentError):
    def __init__(self, tool: str, msg: str) -> None:
        super().__init__(f"{tool}: {msg}")
        self.tool = tool                     # structured fields → structured logs

class RetryableError(AgentError): ...

6.3 Catch narrow, catch late

try:
    data = json.loads(raw)
except Exception:        # ❌ swallows KeyboardInterrupt path, typos, CancelledError logic
    data = {}

try:
    data = json.loads(raw)
except json.JSONDecodeError as exc:          # ✅ exactly the failure you predicted
    log.warning("model returned non-JSON: %s", exc)
    data = {}

except Exception is acceptable in exactly one place: the outermost loop of a long-running worker, where you log with log.exception(...) and continue. Never except: (bare) — it catches SystemExit and KeyboardInterrupt too.

⚠️ In async code, asyncio.CancelledError inherits from BaseException (3.8+), so except Exception won't eat it — but except BaseException will, and that breaks graceful shutdown.

ExceptionGroup / except* (3.11+) — for concurrent failures, where several tasks can fail at once:

try:
    async with asyncio.TaskGroup() as tg:
        for t in tools:
            tg.create_task(t.run())
except* ToolError as eg:                      # eg.exceptions = every ToolError raised
    log.error("%d tools failed", len(eg.exceptions))

6.4 with — deterministic cleanup

with guarantees teardown even on exception or early return. Anything that opens, locks, connects, or times should be a context manager.

with open("prompt.txt", encoding="utf-8") as f:      # closed automatically
    prompt = f.read()

with open("a") as fa, open("b") as fb:               # multiple
    ...

async with httpx.AsyncClient(timeout=30) as client:  # async version
    r = await client.post(url, json=payload)

async with asyncio.timeout(10):                      # 3.11+ deadline for a whole block
    await agent.run(user_input)

The protocol is two dunders:

class Span:
    """Manual context manager: __enter__ returns the `as` value; __exit__ cleans up."""
    def __enter__(self) -> "Span":
        self.t0 = time.perf_counter()
        return self
    def __exit__(self, exc_type, exc, tb) -> bool:   # return True to SUPPRESS the exception
        log.info("span %s took %.1fms (err=%s)",
                 self.name, (time.perf_counter() - self.t0) * 1000,
                 exc_type.__name__ if exc_type else None)
        return False                                  # ← False: let exceptions propagate

Async version: __aenter__ / __aexit__, used with async with.

6.5 contextlib — the shortcut

@contextmanager turns a generator into a context manager: everything before yield is setup, the yield is the body, everything after is teardown.

from contextlib import contextmanager, asynccontextmanager, suppress, ExitStack

@contextmanager
def timed(label: str):
    """Wrap any block to measure elapsed time. `yield` is the body of the with-block."""
    t0 = time.perf_counter()
    try:
        yield
    finally:                                    # finally ⇒ teardown runs even on error
        print(f"[{label}] {(time.perf_counter()-t0)*1000:.1f}ms")

with timed("retrieval"):
    docs = search(query)

@asynccontextmanager
async def db_session():
    session = await pool.acquire()
    try:
        yield session
    finally:
        await pool.release(session)

with suppress(FileNotFoundError):               # ✅ intentional, scoped ignore
    Path("cache.json").unlink()

with ExitStack() as stack:                      # N context managers known at runtime
    files = [stack.enter_context(open(p)) for p in paths]

@asynccontextmanager is also how FastAPI expresses app startup/shutdown (lifespan=).

🎯 Actionable rules

1. Catch the narrowest exception that can actually occur, as close to the cause as possible.
2. Always raise ... from exc when translating an error.
3. Every acquire has a with. If a library doesn't provide one, wrap it in @contextmanager.
4. Log with log.exception() inside except — it captures the traceback for free.

---

7. 🌀 Iterators, Generators and Async

This is where AI code lives: token streams, paginated retrievals, parallel tool calls.

7.1 Iterables vs iterators

An iterable can produce an iterator (__iter__). An iterator produces values one at a time (__next__) and is exhausted after one pass.

xs = [1, 2, 3]          # iterable
it = iter(xs)           # iterator
next(it), next(it)      # 1, 2
next(it, "done")        # 3 ; a 4th call returns "done" instead of raising StopIteration

for x in xs: is sugar for "call iter(), then next() until StopIteration."

⚠️ Iterators are single-use. list(gen) twice gives you the data then an empty list. If you need two passes, materialize once: items = list(gen).

7.2 Generators — lazy sequences with yield

A function containing yield returns a generator. Execution pauses at each yield and resumes on the next next(). Memory stays O(1) regardless of length.

def token_stream(text: str):
    """Yield tokens one at a time — nothing is buffered."""
    for word in text.split():
        yield word + " "

for tok in token_stream("hello there friend"):
    print(tok, end="")

import types
assert isinstance(token_stream("hi"), types.GeneratorType)   # calling it does NOT run the body

That last line matters: calling a generator function executes nothing. The body runs only when you iterate. A generator that never gets consumed never does its work — a classic silent bug.

def read_chunks(path: str, size: int = 8192):
    """Stream a huge file without loading it into RAM."""
    with open(path, "rb") as f:
        while chunk := f.read(size):     # walrus := assigns and tests in one expression
            yield chunk

def batched(items, n):
    """yield from delegates to another iterable/generator."""
    it = iter(items)
    while batch := list(itertools.islice(it, n)):
        yield batch

Generator expressions are comprehensions with parentheses — use them when feeding an aggregate:

total = sum(len(m.content) for m in history)      # no intermediate list
first_hit = next((d for d in docs if d.score > 0.9), None)   # short-circuits

7.3 The async model in one picture

       ┌──────────────── Event loop (ONE thread) ────────────────┐
       │  ready queue: [coro A, coro B, coro C]                  │
       │    ↓ run A until it `await`s something not-ready        │
       │    ↓ park A, run B …                                    │
       │  epoll/kqueue watches sockets → wakes coros when ready  │
       └─────────────────────────────────────────────────────────┘

Async gives you concurrency, not parallelism. One thread interleaves thousands of waiting operations. It makes I/O-bound work (model calls, HTTP, DB, Redis) fast, and does nothing for CPU-bound work.

async def fetch(url: str) -> str:            # coroutine function
    async with httpx.AsyncClient() as c:
        r = await c.get(url)                 # await = "park me until this resolves"
        return r.text

asyncio.run(fetch("https://..."))            # entry point: creates the loop, runs, closes

Rules:

• await is only legal inside async def.
• Calling fetch(url) without await creates a coroutine object and runs nothing (you'll get a RuntimeWarning: coroutine was never awaited).
• One blocking call (time.sleep, requests.get, a big for loop) freezes every task on the loop.

7.4 Running things concurrently

# Sequential — 3 × latency
a, b, c = await fetch(u1), await fetch(u2), await fetch(u3)

# Concurrent — 1 × latency
a, b, c = await asyncio.gather(fetch(u1), fetch(u2), fetch(u3))

results = await asyncio.gather(*coros, return_exceptions=True)   # failures come back as values
oks = [r for r in results if not isinstance(r, Exception)]

# Structured concurrency (3.11+) — preferred: cancels siblings on failure, no orphans
async with asyncio.TaskGroup() as tg:
    tasks = [tg.create_task(t.run()) for t in tools]
outputs = [t.result() for t in tasks]

# Deadlines
out = await asyncio.wait_for(agent.run(q), timeout=30)   # raises TimeoutError
async with asyncio.timeout(30):                          # 3.11+, block-scoped
    out = await agent.run(q)

# Bounded fan-out — don't open 10 000 sockets
sem = asyncio.Semaphore(8)
async def guarded(u: str):
    async with sem:
        return await fetch(u)

await asyncio.sleep(0)      # yield control without waiting (rarely needed)

⚠️ Keep a reference to fire-and-forget tasks. asyncio.create_task(f()) without storing the result can be garbage-collected mid-flight. Store it in a set and discard on completion, or use a TaskGroup.

7.5 Async generators and async for

An async generator is async def + yield. It's the natural type for an LLM token stream.

from collections.abc import AsyncGenerator

async def stream_tokens(self, text: str) -> AsyncGenerator[str, None]:
    for word in text.split():
        await asyncio.sleep(0.01)          # simulates network latency
        yield word + " "

async for tok in agent.stream_tokens("hello there"):
    print(tok, end="", flush=True)

AsyncGenerator[Y, S]: Y = yielded type, S = type accepted by .asend() (usually None). AsyncIterator[str] is the simpler annotation when you only yield.

Per-chunk timeouts — you often need "no single chunk may stall more than N seconds", which wait_for around the whole stream can't express. Drive the protocol manually:

aiter = stream.__aiter__()
while True:
    try:
        chunk = await asyncio.wait_for(aiter.__anext__(), timeout=5.0)
    except StopAsyncIteration:
        break                                   # stream finished normally
    except TimeoutError:
        raise RuntimeError("stream stalled >5s") from None
    handle(chunk)

__aiter__() returns the async iterator; __anext__() returns an awaitable for the next item and raises StopAsyncIteration at the end. async for does exactly this, minus the deadline.

Always close async generators you abandon early — aclose(), or let async with contextlib.aclosing(gen) do it.

7.6 Escaping the loop: to_thread and executors

Blocking call inside async code? Push it to a thread so the loop keeps spinning.

# Blocking library (sync SDK, file I/O, subprocess wait)
text = await asyncio.to_thread(pdf_extract, path)              # 3.9+, one-liner

# Same thing with an explicit pool (reusable, size-controlled)
loop = asyncio.get_running_loop()
with concurrent.futures.ThreadPoolExecutor(max_workers=4) as pool:
    text = await loop.run_in_executor(pool, pdf_extract, path)

# CPU-bound work → processes, not threads (see §8)
with concurrent.futures.ProcessPoolExecutor() as pool:
    vecs = await loop.run_in_executor(pool, embed_batch, docs)

7.7 Bridging sync ↔ async

Sometimes a sync codebase (a CLI, a Django view, a test) must call async code. Three cases:

# 1. No loop running yet — just run it
result = asyncio.run(agent.run("hi"))

# 2. Inside a running loop, calling sync code — see §7.6 (to_thread)

# 3. Sync code that must reach a loop living in another thread:
import threading, asyncio

_loop: asyncio.AbstractEventLoop | None = None
ready = threading.Event()

def _run_loop() -> None:
    global _loop
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)          # bind this loop to THIS thread
    _loop = loop
    ready.set()                           # signal: the loop exists and is usable
    loop.run_forever()                    # blocks this thread, servicing callbacks

threading.Thread(
    target=_run_loop, daemon=True, name="agent-loop"
).start()
ready.wait()                              # don't race — wait until the loop is up

def call_from_sync(coro, timeout: float = 30.0):
    """Submit a coroutine to the background loop and block for the result."""
    fut = asyncio.run_coroutine_threadsafe(coro, _loop)   # thread-safe handoff
    return fut.result(timeout=timeout)                    # concurrent.futures.Future

daemon=True means the thread won't block interpreter exit. ready.set() / ready.wait() is the standard "wait for initialization" handshake — without it, _loop may still be None when the first call lands.

Use this only at a real boundary (a plugin host, a notebook, a legacy service). Two event loops in one process is a debugging tax.

7.8 Async mistakes checklist

Symptom	Cause	Fix
RuntimeWarning: coroutine ... never awaited	Missing await	Add await or create_task
Everything is slow despite async	Blocking call on the loop	asyncio.to_thread, or an async library
RuntimeError: This event loop is already running	asyncio.run inside a loop	Use await / nest_asyncio only in notebooks
Tasks vanish silently	GC'd fire-and-forget task	Keep refs or use TaskGroup
Shutdown hangs	Swallowed CancelledError	Re-raise it; clean up in finally
requests in async code	Sync HTTP client	Use httpx.AsyncClient / aiohttp

🎯 Actionable rules

1. Generators for anything large or streaming; never build a list you'll consume once.
2. TaskGroup > gather for anything with failure semantics.
3. Every await on an external call gets a timeout, and every fan-out gets a semaphore.
4. If it blocks and you can't fix it, to_thread it.

---

8. ⚡ Concurrency, the GIL, and Performance

8.1 The one question that decides everything: I/O-bound or CPU-bound?

	I/O-bound	CPU-bound
Time goes to	Waiting: network, disk, DB, model API	Computing: parsing, math, tokenizing, image ops
Examples in AI code	LLM calls, vector DB queries, S3, Redis	Chunking 10k docs, cosine similarity in pure Python, PDF rendering
Right tool	asyncio (thousands of tasks, one thread)	multiprocessing or a C/numpy library
Threads help?	Yes (they release the GIL while waiting)	No — the GIL serializes them

Measure before choosing. time.perf_counter() around the suspicious block answers it in two minutes.

8.2 The GIL, precisely

The Global Interpreter Lock is one mutex per interpreter that must be held to execute Python bytecode. Consequences:

• Only one thread executes Python bytecode at a time, even on 32 cores.
• Threads do run concurrently when they're not executing bytecode — i.e. while blocked on I/O, or inside a C extension that released the GIL.
• Python-level operations on built-in types are individually atomic, but multi-step logic is not — you still need locks for if k not in d: d[k] = ....

# Reading a socket → GIL released while waiting → threads genuinely overlap
# Multiplying a million ints in a Python loop → GIL held → threads take turns

How C extensions escape it: numpy, torch, polars, orjson, lxml drop the GIL around their heavy C/CUDA work.

# 4 numpy matmuls in 4 threads DO run in parallel — the GIL is released inside BLAS
with ThreadPoolExecutor(4) as ex:
    list(ex.map(lambda _: A @ B, range(4)))

That is the whole reason Python is viable for ML: your Python code orchestrates; the compute happens under released locks.

Free-threaded Python. CPython 3.13 shipped an experimental no-GIL build (PEP 703); 3.14 makes it an officially supported build. It is not the default interpreter, and the C ecosystem is still catching up. Plan today's architecture as if the GIL exists; revisit when your dependency tree is verified free-threading-ready.

8.3 Choosing a concurrency primitive

Primitive	Parallel CPU?	Cost per unit	Shared memory	Use for
asyncio	❌	~KB	Yes (single thread)	Thousands of concurrent I/O ops
threading / ThreadPoolExecutor	❌ (except in C ext)	~MB stack	Yes → needs locks	Blocking libraries, modest fan-out
multiprocessing / ProcessPoolExecutor	✅	~10s MB + startup	No → pickling	Real CPU work in pure Python
Another service (Go/Rust worker)	✅	Deploy unit	No	Sustained CPU-heavy workloads

from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor, as_completed

with ThreadPoolExecutor(max_workers=8) as ex:              # I/O with a sync client
    futures = {ex.submit(fetch_doc, u): u for u in urls}
    for fut in as_completed(futures):                       # results as they land
        try:
            docs.append(fut.result(timeout=30))
        except Exception:
            log.exception("fetch failed for %s", futures[fut])

with ProcessPoolExecutor() as ex:                           # CPU: defaults to os.cpu_count()
    vectors = list(ex.map(embed_chunk, chunks, chunksize=16))

Process-pool rules: arguments and return values must be picklable (no lambdas, no open sockets), every task pays a serialization cost, and on macOS/Windows the spawn start method re-imports your module — so guard entry points with if __name__ == "__main__": (§11.4).

Thread safety when you do share state:

lock = threading.Lock()
with lock:
    cache[key] = value          # compound operations need the lock
q = queue.Queue()               # already thread-safe — prefer message passing over locks

8.4 WSGI vs ASGI, and how FastAPI actually runs your code

	WSGI (Flask, Django sync, gunicorn)	ASGI (FastAPI, Starlette, uvicorn)
Model	One request occupies one worker thread/process, start to finish	Event loop multiplexes thousands of in-flight requests
Concurrency ceiling	≈ number of workers × threads	≈ open sockets
Streaming / WebSockets / SSE	Awkward or impossible	Native
Best for	CPU-ish, short, sync handlers	LLM calls, streaming, long waits

For an AI service — where a request may sit for 30 seconds waiting on a model — ASGI is the difference between 20 concurrent users and 2000.

The rule that trips everyone up:

@app.get("/a")
async def a():           # runs ON the event loop → must never block
    return await client.get(url)          # ✅
    # time.sleep(5)                       # ❌ freezes EVERY concurrent request

@app.get("/b")
def b():                 # plain def → FastAPI runs it in a threadpool automatically
    return blocking_sdk_call()            # ✅ safe; the loop keeps serving

So: async def for awaitable work, plain def for blocking libraries you can't avoid. The worst combination is async def containing a blocking call — it looks fast and serializes your whole server. The threadpool has a fixed size (~40 by default), so def handlers cap out earlier than async def ones; use them for genuinely blocking code, not as the default.

Deployment shape:

# One process per core; each process runs its own event loop.
uvicorn src.main:app --workers 4 --host 0.0.0.0 --port 8000
# In containers, let the orchestrator scale replicas instead: --workers 1 per pod.

• Workers are processes → they sidestep the GIL, but share nothing. Keep handlers stateless; put shared state in Redis/Postgres (CLAUDE.md says exactly this).
• Start at workers = cpu_cores for mixed load; I/O-bound services often do fine with fewer, because the loop absorbs the waiting.
• --reload is dev-only; it costs a file watcher and forks.

[client] → [uvicorn worker ×N (processes)]
                 ↓
          [event loop] ──async def──→ coroutines (concurrent, 1 thread)
                 └──── def ────────→ threadpool (blocking, ~40 slots)
                 └──── to_thread ──→ threadpool
                 └──── ProcessPool ─→ separate processes (true parallelism)

8.5 Making Python fast

Order matters — do these top to bottom:

1. Measure. Guessing is how you optimize the 3% path.

   python -m cProfile -s cumtime -m src.main | head -30
   py-spy top --pid 1234          # sampling profiler, works on a live prod process
   python -X importtime -c "import src.main"   # slow startup? usually imports

1. Fix the algorithm. set membership instead of list scans; one batched query instead of N+1; cache what repeats.

   from functools import lru_cache, cache
   @cache                                # unbounded memoization (3.9+)
   def tokenizer_for(model: str): ...
   @lru_cache(maxsize=1024)
   def embed_cached(text: str) -> tuple[float, ...]: ...   # args must be hashable

1. Vectorize. Push loops into numpy — one C call beats a million interpreter steps.

   sims = (M @ q) / (np.linalg.norm(M, axis=1) * np.linalg.norm(q))   # ~100× a Python loop

1. Use faster libraries. orjson over json, polars over pandas, uvloop as the asyncio loop, msgspec for hot serialization.
2. Concurrency. asyncio for I/O; processes for CPU (§8.3).
3. Compile the hot spot. | Tool | What it is | Good for | |---|---|---| | Cython | Python-ish → C extension | Annotated hot loops in an existing codebase | | mypyc | Compiles typed Python | Whole modules already annotated | | PyPy | JIT interpreter | Long-running pure-Python CPU work; not for C-extension-heavy ML stacks | | Rust + PyO3 | Native extension | New hot paths you're willing to rewrite | | Numba | @njit JIT for numeric loops | Array math that doesn't vectorize cleanly |
4. Move the workload. If it's sustained CPU, the honest answer may be a Go/Rust service — the architecture this repo already uses.

Micro-tips that are free: __slots__ / slots=True on hot classes, local variable lookups in tight loops, join() instead of += on strings, generators instead of intermediate lists, and importing heavy modules lazily inside functions to cut startup time.

🎯 Actionable rules

1. Classify the workload before choosing a primitive; almost all AI serving is I/O-bound.
2. Never block the event loop — async def + blocking call is the #1 production stall.
3. Profile, then optimize; py-spy on a live process finds in minutes what reading finds in days.
4. Uvicorn workers are processes: keep handlers stateless.

---

9. 📦 The Standard Library & AI Toolkit

The 20 modules that cover ~95% of AI-service code. For each: what it is, why you care, the snippet you'll copy.

9.1 pydantic — your data contract

What: runtime validation and parsing driven by type hints. Why: every byte entering your system (HTTP body, LLM JSON, YAML config) is untrusted; Pydantic turns it into a typed object or a precise error, at the boundary. It's the backbone of FastAPI and of structured LLM output.

from pydantic import BaseModel, Field, field_validator, ConfigDict

class AgentConfig(BaseModel):
    model_config = ConfigDict(extra="forbid", frozen=True)   # reject unknown keys

    name: str = Field(min_length=1, max_length=64)
    model: str = "claude-opus-5"
    temperature: float = Field(default=0.7, ge=0.0, le=2.0)  # bounds enforced at runtime
    tools: list[str] = Field(default_factory=list)
    api_key: str | None = Field(default=None, repr=False)     # kept out of logs

    @field_validator("tools")
    @classmethod
    def no_duplicates(cls, v: list[str]) -> list[str]:
        if len(set(v)) != len(v):
            raise ValueError("duplicate tool names")
        return v

cfg = AgentConfig.model_validate(raw_dict)     # dict → typed object, or ValidationError
cfg = AgentConfig.model_validate_json(body)    # bytes/str → object (fast path)
cfg.model_dump(exclude_none=True)              # → dict
cfg.model_dump_json(indent=2)                  # → JSON str
AgentConfig.model_json_schema()                # → JSON Schema, i.e. your LLM tool schema

That last line is the killer feature: one model gives you validation, serialization, OpenAPI docs, and the tool schema you hand the model. Pair with pydantic-settings for typed env config.

extra="forbid" is a deliberate choice: silently dropping unknown keys hides client bugs and typo'd config.

9.2 json

import json
data = json.loads(raw)                                  # str/bytes → Python
raw  = json.dumps(obj, ensure_ascii=False, indent=2)    # Python → str
raw  = json.dumps(obj, default=str)                     # fallback for datetime/UUID/etc.
with open(p, encoding="utf-8") as f: cfg = json.load(f) # load/dump = file variants

⚠️ Always wrap model output in try/except json.JSONDecodeError — LLMs emit prose, markdown fences, and truncated objects. Prefer orjson (2–5× faster, bytes in/out) on hot paths.

9.3 re — regular expressions

import re
FENCE = re.compile(r"```

(?:json)?\s*(.*?)

```", re.DOTALL)   # compile once at module level
m = FENCE.search(text)
payload = m.group(1) if m else text

re.findall(r"\{\{(\w+)\}\}", template)          # ['name', 'tools'] — template vars
re.sub(r"sk-[A-Za-z0-9]{20,}", "[REDACTED]", log_line)      # scrub secrets
re.split(r"\n{2,}", doc)                        # paragraph chunking

Use raw strings (r"..."), compile patterns you reuse, and prefer str methods when they suffice — text.startswith("tool:") beats a regex in both speed and clarity.

9.4 collections & itertools

from collections import defaultdict, Counter, deque, namedtuple
from collections.abc import Sequence, Mapping, Iterable, Callable   # for type hints

by_tool = defaultdict(list)
by_tool["calc"].append(result)                # no KeyError, auto-creates the list
Counter(words).most_common(3)                 # word frequency in one line
window = deque(maxlen=20)                     # O(1) both ends; auto-evicts → context window
window.append(msg)                            # oldest drops off automatically

import itertools
itertools.chain(sys_msgs, history)            # concatenate iterables lazily
itertools.islice(stream, 100)                 # take first N of anything
itertools.groupby(sorted(rs, key=k), key=k)   # group (⚠️ sort first!)
itertools.product(models, temps)              # grid search

deque(maxlen=N) is the cleanest sliding-window implementation in the language.

9.5 logging

What: leveled, structured, configurable output. Why: print has no severity, no timestamp, no module, no way to silence in prod, and writes to stdout unbuffered from every thread.

import logging
log = logging.getLogger(__name__)              # ✅ module-scoped, hierarchical

logging.basicConfig(                            # configure ONCE, in main() only
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(name)s %(message)s",
)

log.debug("prompt=%r", prompt)                  # ✅ %-style: formatting is lazy
log.info("tool %s ok in %.0fms", name, ms)
log.warning("retrying after %s", exc)
log.exception("tool failed")                    # inside except: adds the traceback
log.error("failed", extra={"tenant": tid, "job": jid})   # structured fields

Never log secrets, prompts containing PII, or full request bodies by default. Libraries should never call basicConfig — only applications configure handlers.

9.6 functools

from functools import wraps, partial, lru_cache, cache, reduce

def timed(fn):
    @wraps(fn)                       # ← copies __name__, __doc__, __wrapped__
    def inner(*args, **kwargs):      # without it, every decorated fn is named "inner"
        t0 = time.perf_counter()
        try:
            return fn(*args, **kwargs)
        finally:
            log.info("%s took %.1fms", fn.__name__, (time.perf_counter()-t0)*1000)
    return inner

search_docs = partial(search, index="docs", top_k=5)   # freeze arguments → new callable
callbacks.append(partial(on_done, job_id=jid))          # avoids the late-binding lambda bug

@wraps is mandatory on every decorator — without it you break introspection, docs, and pytest.

9.7 pathlib, os, sys

from pathlib import Path
p = Path(__file__).parent / "prompts" / "system.md"     # / operator, cross-platform
p.exists(); p.is_file(); p.stem; p.suffix; p.name
text = p.read_text(encoding="utf-8")                    # no open() needed
p.write_text(rendered, encoding="utf-8")
p.parent.mkdir(parents=True, exist_ok=True)
list(Path("data").rglob("*.md"))                        # recursive glob

import os, sys
os.environ["MODEL"]                    # KeyError if unset → fail fast at startup ✅
os.getenv("PORT", "8000")              # optional with default
sys.argv; sys.exit(1); sys.stderr.write("...")

pathlib over os.path — always. String path concatenation is a bug waiting for Windows.

9.8 datetime & time

from datetime import datetime, timezone, timedelta, UTC
now = datetime.now(UTC)                        # ✅ ALWAYS timezone-aware (3.11+ shorthand)
datetime.utcnow()                              # ❌ deprecated, returns naive — don't
now.isoformat()                                # '2026-08-24T09:15:00+00:00'
datetime.fromisoformat(s)
deadline = now + timedelta(minutes=5)

import time
t0 = time.perf_counter()                       # ✅ monotonic — for measuring durations
elapsed_ms = (time.perf_counter() - t0) * 1000
time.time()                                    # wall clock — for timestamps, can jump
time.sleep(0.1)                                # ❌ never inside async — use asyncio.sleep

Store UTC, convert at the edge. Naive datetimes in a distributed system are a bug you find on a Sunday.

9.9 httpx — HTTP for AI services

import httpx

async with httpx.AsyncClient(
    timeout=httpx.Timeout(30.0, connect=5.0),                 # ALWAYS set timeouts
    limits=httpx.Limits(max_connections=100),
) as client:
    r = await client.post(url, json=payload, headers={"x-api-key": key})
    r.raise_for_status()                                       # → HTTPStatusError on 4xx/5xx
    data = r.json()

    async with client.stream("POST", url, json=payload) as resp:   # SSE / token streaming
        async for line in resp.aiter_lines():
            if line.startswith("data: "):
                handle(json.loads(line[6:]))

Reuse one client for the app lifetime (connection pooling); creating one per request destroys throughput. requests is sync-only — never use it in async code.

9.10 numpy

import numpy as np
v = np.array(embedding, dtype=np.float32)          # float32 halves memory vs float64
M = np.stack(vectors)                              # (n_docs, dim)
sims = M @ v / (np.linalg.norm(M, axis=1) * np.linalg.norm(v))   # cosine, vectorized
top_k = np.argsort(-sims)[:5]                      # indices of the 5 best
M.shape, M.dtype, M[:, :10], M[sims > 0.8]         # slicing & boolean masking

Rule: if you're writing a for loop over floats, there's a numpy one-liner that's 100× faster and releases the GIL while it runs.

9.11 The rest, in one breath

Module	Use it for
asyncio	Event loop, tasks, queues, locks (§7)
threading	Lock, Event, background threads (§7.7)
concurrent.futures	Thread/process pools with a uniform Future API (§8.3)
multiprocessing	Process-level parallelism, Queue, shared memory
contextlib	@contextmanager, suppress, ExitStack (§6.5)
dataclasses	Value objects (§5.5)
enum	Closed sets (§2.9)
typing	Protocol, TypedDict, Annotated, Literal (§4)
io	BytesIO/StringIO — in-memory files for uploads, PDFs, images without touching disk
subprocess	subprocess.run([...], capture_output=True, timeout=30, check=True) — list args, never shell=True with user input
uuid	uuid.uuid4().hex for job/trace ids
hashlib	hashlib.sha256(text.encode()).hexdigest()[:16] — cache keys, dedupe
secrets	secrets.token_urlsafe(32) for API keys (never random for security)
random	Sampling, jitter: delay * (1 + random.random())
argparse	CLIs (§11.4)
os / sys / pathlib	Env, exit codes, paths
textwrap	textwrap.dedent(prompt) — keep prompts indented in source, flat at runtime

from io import BytesIO
img = Image.open(BytesIO(await file.read()))       # upload → PIL, no temp file

🎯 Actionable rules

1. Validate at the boundary with Pydantic; inside your service, trust your types.
2. logging with %-style args and getLogger(__name__), never print, in anything long-lived.
3. Every network call gets an explicit timeout.
4. Loop over floats → reach for numpy instead.

---

10. 🧪 Testing with pytest

10.1 Why pytest wins

Plain assert, plain functions, no boilerplate class hierarchy — and on failure it rewrites the assertion to show you both sides.

# tests/unit/test_message.py
def test_blank_content_raises():
    with pytest.raises(ValueError, match="non-blank"):     # ✅ assert the message too
        Message(role=Role.USER, content="   ")

def test_history_is_windowed():
    agent = Agent(AgentConfig(name="t"))
    for i in range(MAX_HISTORY_TURNS + 5):
        agent.add_message(Role.USER, f"msg {i}")
    assert len(agent.history) == MAX_HISTORY_TURNS         # shows both numbers on failure

pytest.raises(Exception) alone is weak — it passes on a typo'd NameError. Name the class and match the message.

10.2 The three properties every test needs

Isolated — no shared mutable state, no order dependence. Deterministic — no real clock, no real network, no randomness without a seed. Fast — milliseconds, so you run them on save.

# ❌ leaks state between tests, and hits the network
REGISTRY = {}
def test_register(): REGISTRY["calc"] = Calculator()
def test_run():      assert REGISTRY["calc"].run(...)   # passes only if the other ran first

# ✅ each test builds what it needs
@pytest.fixture
def registry() -> dict[str, Tool]:
    return {"calc": Calculator()}
def test_run(registry): assert registry["calc"].run(expression="1+1") == "2"

Isolation is also what makes pytest -n auto (xdist) safe — parallel tests that share a temp file or a global registry fail randomly, which is worse than failing always.

10.3 Fixtures and conftest.py

Fixtures are dependency injection for tests: request one by parameter name and pytest builds it.

# tests/conftest.py — auto-discovered by every test in this directory and below
import pytest

@pytest.fixture
def agent() -> Agent:                       # function-scoped: fresh per test (default)
    return Agent(AgentConfig(name="test-agent", tools=["calculator"]))

@pytest.fixture(scope="session")            # built once for the whole run
def embeddings() -> np.ndarray:
    return np.load("tests/fixtures/vectors.npy")

@pytest.fixture
def temp_index(tmp_path: Path) -> Path:     # tmp_path: built-in, auto-cleaned
    p = tmp_path / "index.json"
    p.write_text("{}")
    yield p                                 # everything after yield is teardown
    p.unlink(missing_ok=True)

Built-ins worth knowing: tmp_path, capsys (captured stdout), caplog (captured log records), monkeypatch.

10.4 parametrize — one test, many cases

@pytest.mark.parametrize(
    ("expr", "expected"),
    [("1+1", "2"), ("10 * 5", "50"), ("2 ** 8", "256")],
    ids=["add", "mul", "pow"],
)
def test_calculator(expr: str, expected: str):
    assert Calculator().run(expression=expr) == expected

@pytest.mark.parametrize("bad", ["", "   ", None])
def test_rejects_blank(bad):
    with pytest.raises((ValueError, TypeError)):
        Message(role=Role.USER, content=bad)

Each case is a separate test with its own name — you see exactly which input broke.

10.5 monkeypatch — surgical, auto-reverting patches

What: temporarily replaces attributes, dict items, and env vars, then restores them when the test ends. Why: it removes the network, the clock, and the filesystem from your unit tests without a mock framework.

def test_uses_configured_model(monkeypatch):
    monkeypatch.setenv("MODEL", "claude-sonnet-5")            # env var
    assert AgentConfig.from_env().model == "claude-sonnet-5"

def test_retries_on_timeout(monkeypatch):
    calls = {"n": 0}
    async def fake_post(*a, **kw):                             # a fake, not a mock
        calls["n"] += 1
        if calls["n"] < 3:
            raise httpx.TimeoutException("boom")
        return httpx.Response(200, json={"content": "ok"})
    monkeypatch.setattr(httpx.AsyncClient, "post", fake_post)  # attribute
    assert asyncio.run(call_model("hi")).content == "ok"
    assert calls["n"] == 3

def test_feature_flag(monkeypatch):
    monkeypatch.setitem(SETTINGS, "streaming", False)          # dict entry

Patch where the name is looked up, not where it's defined: if agent.py does from httpx import AsyncClient, patch agent.AsyncClient. This is the single most common patching mistake.

Prefer injecting a fake over patching at all — a Protocol-shaped FakeLLM needs no patching and survives refactors:

class FakeLLM:                                  # satisfies the LLMClient Protocol
    def __init__(self, replies: list[str]): self._r = iter(replies)
    async def complete(self, prompt: str) -> str: return next(self._r)

agent = Agent(cfg, llm=FakeLLM(["Result: 42"]))   # deterministic, no network, no patching

10.6 Async tests

# Option A: pytest-asyncio (add asyncio_mode = "auto" to pyproject → no marker needed)
@pytest.mark.asyncio
async def test_streams_tokens():
    out = [t async for t in Agent().stream_tokens("hello there")]
    assert "".join(out).strip() == "hello there"

# Option B: no plugin — wrap with asyncio.run
def test_calculator_path():
    async def _run():
        resp = await Agent().run("calculate 10 * 5")
        assert resp.status == Status.OK
        assert any(t.tool_name == "calculator" for t in resp.tool_results)
    asyncio.run(_run())

Option A is cleaner for a suite; Option B is handy for a one-off inside an otherwise sync file.

10.7 Organizing tests

class TestDynamicDispatch:            # grouping class — no base class, no __init__
    h = Handlers()
    def test_known(self):   assert "summary" in dynamic_dispatch(self.h, "summarize", "t")
    def test_unknown(self): assert "no handler" in dynamic_dispatch(self.h, "embed", "t")

def test_is_generator_type():         # assert on structure, not just values
    assert isinstance(token_stream("hi"), types.GeneratorType)

pytest                       # everything
pytest tests/unit -q         # fast subset
pytest -k "calculator"       # by name substring
pytest -m "not integration"  # by marker
pytest -x --lf               # stop at first failure; rerun last failures
pytest -n auto               # parallel (pytest-xdist)
pytest --cov=src --cov-report=term-missing

[tool.pytest.ini_options]
testpaths = ["tests"]
asyncio_mode = "auto"
addopts = "-q --strict-markers"
markers = ["integration: needs live services", "slow: >1s"]

Mirror this repo's layering (CLAUDE.md): unit tests in tests/unit/, service-dependent tests in tests/integration/ behind a marker, fast tests only in pre-commit, everything in CI.

10.8 Testing LLM-powered code

You can't assert on generated prose. Assert on the machinery instead:

• Contract: output parses into your Pydantic model; required fields exist.
• Routing: the right tool was called with the right args (any(t.tool_name == "calculator" for t in resp.tool_results)).
• Bounds: history windowed, token budget respected, timeouts enforced.
• Failure modes: malformed JSON → fallback path; 429 → backoff; timeout → TimeoutError.
• Golden tests against recorded responses for the few end-to-end flows that matter.
• Live-model tests exist, but they're @pytest.mark.integration, nightly, and never gate a PR.

🎯 Actionable rules

1. One behaviour per test; the name states the behaviour.
2. monkeypatch for env and third-party attributes; injected fakes for your own interfaces.
3. Deterministic by construction — no live model, clock, or RNG in unit tests.
4. Test the plumbing around the LLM, not the LLM's prose.

---

11. 🗂️ Project Layout & Tooling

11.1 Virtual environments — non-negotiable

Python installs packages per environment. Without a venv, every project shares one global site-packages and your torch version becomes a company-wide decision.

# Classic
python -m venv .venv && source .venv/bin/activate   # Windows: .venv\Scripts\activate
pip install -e ".[dev]"
deactivate

# Modern (uv — 10–100× faster resolver, manages Python versions too)
uv venv                    # creates .venv
uv add fastapi pydantic    # installs + records in pyproject.toml + updates uv.lock
uv add --dev pytest ruff mypy
uv run pytest              # runs inside the env, no activation needed
uv sync                    # reproduce the locked env exactly (CI, Docker)

Commit uv.lock (or requirements.txt from pip freeze); never commit .venv/.

File	Role
pyproject.toml	Source of truth: metadata, deps, and every tool's config
uv.lock / requirements.txt	Exact pinned versions for reproducible installs
requirements.txt (legacy)	Still fine for simple Docker images: pip install -r requirements.txt

11.2 pyproject.toml, annotated

[project]
name = "agent-service"
version = "0.1.0"
requires-python = ">=3.12"
dependencies = [
  "fastapi>=0.115",
  "pydantic>=2.9",
  "httpx>=0.27",
]

[project.optional-dependencies]
dev = ["pytest>=8", "pytest-asyncio", "pytest-cov", "ruff", "mypy"]

[project.scripts]
agent = "src.cli:main"            # installs an `agent` command on the PATH

[tool.ruff]
line-length = 100
target-version = "py312"

[tool.ruff.lint]
select = ["E", "F", "I", "B", "UP", "ASYNC", "SIM", "RUF"]
# E/F pycodestyle+pyflakes · I import sort · B bugbear (catches the mutable-default bug)
# UP pyupgrade · ASYNC async footguns · SIM simplifications · RUF ruff-specific
ignore = ["E501"]                 # the formatter owns line length

[tool.mypy]
python_version = "3.12"
strict = true
plugins = ["pydantic.mypy"]

[tool.pytest.ini_options]
testpaths = ["tests"]
asyncio_mode = "auto"

One tool, one config file. Ruff replaces black + isort + flake8 + a dozen plugins and runs in milliseconds:

uv run ruff check --fix .     # lint + autofix
uv run ruff format .          # format (black-compatible)
uv run mypy src/
uv run pytest

Wire these into pre-commit so they run before the code exists in history — the same "fast checks local, full suite in CI" split this repo uses.

11.3 Layout that scales

agent-service/
├── pyproject.toml
├── uv.lock
├── .env.example              # committed; .env is NOT
├── src/
│   ├── __init__.py
│   ├── main.py               # FastAPI app factory
│   ├── __main__.py           # python -m src
│   ├── routers/              # HTTP surface (thin)
│   ├── services/             # business logic (no HTTP, no DB driver)
│   ├── schemas/              # Pydantic request/response models
│   ├── clients/              # LLM / vector DB / Redis wrappers
│   └── deps.py               # DI providers for FastAPI
└── tests/
    ├── conftest.py
    ├── unit/
    └── integration/

The src/ layout forces you to install your package, so tests exercise the installed code — not a lucky sys.path accident.

Import discipline: routers → services → clients, one direction only. Absolute imports (from src.services.agent import Agent) over relative ones beyond a single dot. Circular imports mean your layering is wrong; fix the design, not the import.

11.4 __init__.py, __main__, and entry points

• __init__.py marks a directory as a package and runs on import. Keep it near-empty — re-export the public API at most. Heavy work here slows every import and creates cycles.
• __pycache__/ holds compiled bytecode; .pytest_cache/, .ruff_cache/, .mypy_cache/ are tool caches. All are generated — .gitignore them all.
• if __name__ == "__main__": — __name__ is "__main__" only when the file is run directly, and the module's dotted name when imported. The guard keeps import-time side effects out, and is required for multiprocessing on macOS/Windows.

# src/__main__.py   →  python -m src --model claude-opus-5 --verbose
import argparse, logging, sys

def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(prog="agent", description="Run the agent CLI.")
    parser.add_argument("prompt", help="user prompt")
    parser.add_argument("--model", default="claude-opus-5")
    parser.add_argument("--max-steps", type=int, default=10)
    parser.add_argument("--verbose", action="store_true")
    args = parser.parse_args(argv)

    logging.basicConfig(level=logging.DEBUG if args.verbose else logging.INFO)
    resp = asyncio.run(Agent(AgentConfig(name="cli", model=args.model)).run(args.prompt))
    print(resp.messages[-1].content)
    return 0                       # exit code: 0 = success

if __name__ == "__main__":
    sys.exit(main())

Returning the exit code from main() (instead of calling sys.exit inside) makes the function testable: assert main(["hi", "--model", "x"]) == 0.

11.5 Config and secrets

from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_file=".env", extra="ignore")
    database_url: str                       # required → app refuses to start if unset
    redis_url: str = "redis://localhost:6379/0"
    anthropic_api_key: str
    python_port: int = 8000

settings = Settings()                       # validated once, at import

Fail fast at startup on missing config — an agent that dies on request #4000 because REDIS_URL was blank is far worse than one that never starts. Never commit .env; commit .env.example.

11.6 A production-grade Dockerfile

Python ships as source plus an interpreter plus a dependency tree, so "it works on my machine" is the default failure mode. A good image is small, reproducible, cached, non-root, and shuts down cleanly. Here is the whole thing, then the reasoning line by line.

# syntax=docker/dockerfile:1.9

# ────────────────────────── Stage 1: builder ──────────────────────────
# Compilers, headers, and the package manager live here — and stay here.
FROM python:3.12-slim-bookworm AS builder

# uv as a static binary; pin an exact tag in CI (e.g. ghcr.io/astral-sh/uv:0.9.2)
COPY --from=ghcr.io/astral-sh/uv:latest /uv /uvx /bin/

ENV UV_COMPILE_BYTECODE=1 \
    UV_LINK_MODE=copy \
    UV_PYTHON_DOWNLOADS=never

WORKDIR /app

# Only if a dep needs to compile (asyncpg, psycopg[c], some ML wheels):
# RUN apt-get update && apt-get install -y --no-install-recommends \
#       build-essential gcc && rm -rf /var/lib/apt/lists/*

# 1️⃣  Dependencies FIRST, from the lockfile only.
#     This layer is reused on every build until uv.lock changes.
RUN --mount=type=cache,target=/root/.cache/uv \
    --mount=type=bind,source=uv.lock,target=uv.lock \
    --mount=type=bind,source=pyproject.toml,target=pyproject.toml \
    uv sync --locked --no-install-project --no-dev

# 2️⃣  THEN the source. Editing a handler no longer reinstalls torch.
COPY src/ ./src/
RUN --mount=type=cache,target=/root/.cache/uv \
    uv sync --locked --no-dev

# ────────────────────────── Stage 2: runtime ──────────────────────────
# No compilers, no uv, no build cache, no dev dependencies.
FROM python:3.12-slim-bookworm AS runtime

RUN groupadd --system --gid 1001 app \
 && useradd  --system --uid 1001 --gid app --no-create-home app

# Runtime-only OS packages. curl is for HEALTHCHECK; libgomp1 is needed by
# numpy/scikit-learn/torch wheels. Add nothing you cannot justify.
RUN apt-get update && apt-get install -y --no-install-recommends \
      curl libgomp1 \
 && rm -rf /var/lib/apt/lists/*

ENV PATH="/app/.venv/bin:$PATH" \
    PYTHONUNBUFFERED=1 \
    PYTHONDONTWRITEBYTECODE=1 \
    PYTHONFAULTHANDLER=1 \
    PYTHONHASHSEED=random

WORKDIR /app
COPY --from=builder --chown=app:app /app/.venv /app/.venv
COPY --from=builder --chown=app:app /app/src   /app/src

USER app
EXPOSE 8000

HEALTHCHECK --interval=30s --timeout=3s --start-period=20s --retries=3 \
  CMD curl -fsS http://localhost:8000/healthz || exit 1

# Exec form (JSON array): uvicorn becomes PID 1 and receives SIGTERM directly.
CMD ["uvicorn", "src.main:app", \
     "--host", "0.0.0.0", "--port", "8000", \
     "--workers", "1", \
     "--timeout-graceful-shutdown", "30"]

Why each decision:

Decision	Reason
python:3.12-slim-bookworm, not alpine	Alpine uses musl, so PyPI's manylinux wheels don't match — pip compiles numpy/pandas/torch from source. Builds go from 40 s to 20 min and images often end up larger. Slim is the right default for Python.
Not :latest, and pin by digest in CI	FROM python:3.12-slim-bookworm@sha256:… makes rebuilds byte-identical and blocks a surprise base-image change.
Two stages	Compilers, headers, and uv never reach production. Smaller image, smaller CVE surface, nothing an attacker can build with.
Deps before source	Docker caches per layer. Dependencies change monthly, source changes hourly — install them in that order and a code edit rebuilds in seconds. Getting this backwards is the single most common Python Dockerfile mistake.
--mount=type=cache	BuildKit keeps the wheel/uv cache between builds without baking it into a layer.
--mount=type=bind for the lockfile	The file is visible during that one RUN and leaves no layer behind.
uv sync --locked	Fails if uv.lock is stale instead of silently resolving new versions. Reproducibility is the whole point. (pip install -r requirements.txt with fully pinned, hashed deps is the equivalent.)
--no-dev	pytest, ruff, and mypy are build-time tools. Shipping them adds weight and attack surface.
UV_COMPILE_BYTECODE=1	Precompiles .pyc at build time → faster cold starts, and the container filesystem can stay read-only.
Non-root app user	A container escape starts as an unprivileged user. Also required by most hardened Kubernetes policies (runAsNonRoot: true).
PYTHONUNBUFFERED=1	Without it stdout is block-buffered when not a TTY, so your last log lines are lost exactly when the process crashes.
PYTHONFAULTHANDLER=1	Dumps a Python traceback on segfault — the only clue you'll get when a native extension dies.
Copy the whole .venv + PATH	No activation scripts, no pip in the final image, one self-contained directory.
HEALTHCHECK hitting /healthz	The orchestrator needs to know ready vs alive. Keep the endpoint dependency-free — a health check that queries Postgres takes your service down when Postgres blips.
Exec-form CMD	Shell form (CMD uvicorn …) makes /bin/sh PID 1, which does not forward SIGTERM. Your pods then take the full 30 s termination grace period and drop in-flight requests on every deploy.
--workers 1	One process per container; scale with replicas so the orchestrator can schedule, autoscale, and restart at the right granularity (§8.4). Use --workers N only when you deliberately run one big container per node.
--timeout-graceful-shutdown	Long LLM streams need time to finish; set it below your orchestrator's grace period.

The .dockerignore matters as much as the Dockerfile — without it, COPY ships your .venv, .git, and every model checkpoint into the build context:

.venv/
.git/
__pycache__/
*.pyc
.pytest_cache/
.mypy_cache/
.ruff_cache/
.env
tests/
notebooks/
data/
*.ipynb
Dockerfile

Secrets never go in the image. Layers are permanent and readable — an ARG or a deleted file is still in the history. Use runtime env vars (parsed by Settings in §11.5) or a build-time secret mount that leaves nothing behind:

RUN --mount=type=secret,id=pip_token \
    UV_INDEX_URL="https://$(cat /run/secrets/pip_token)@pypi.internal/simple" uv sync --locked

Building and shipping it:

DOCKER_BUILDKIT=1 docker build --platform linux/amd64 -t agent-service:1.4.2 .
docker run --rm -p 8000:8000 --env-file .env \
  --read-only --tmpfs /tmp --cap-drop=ALL \
  agent-service:1.4.2
docker scout cves agent-service:1.4.2      # or trivy image agent-service:1.4.2

--platform linux/amd64 is not optional on an Apple-silicon laptop deploying to x86 nodes — otherwise you build an arm64 image that dies with exec format error in production.

Pre-ship checklist: image under ~300 MB for a plain API (multi-GB is normal once torch/CUDA is involved) · non-root confirmed with docker run … whoami · SIGTERM stops it in under a second (docker stop) · no secrets in docker history · vulnerability scan clean · a code-only edit rebuilds in seconds, not minutes.

For local development, don't use this image: bind-mount the source and run uvicorn --reload in a compose service (make dev in this repo's CLAUDE.md). Production images are for production.

🎯 Actionable rules

1. One venv per project; lockfile committed; .venv/ ignored.
2. pyproject.toml is the only config file you need — ruff, mypy, pytest all live there.
3. src/ layout, one-way imports, empty __init__.py.
4. Validate config at startup with pydantic-settings.
5. Multi-stage, slim base, deps-before-source, non-root, exec-form CMD — every image, every time.

---

12. 🐞 Debugging & Profiling in VS Code

Print-debugging an agent loop that runs 40 steps and calls 6 tools is a losing game. Learn the debugger once; it pays back weekly.

12.1 .vscode/launch.json

{
  "version": "0.2.0",
  "configurations": [
    {
      "name": "Module: python -m src",
      "type": "debugpy",
      "request": "launch",
      "module": "src",
      "args": ["what is 2+2", "--verbose"],
      "console": "integratedTerminal",
      "justMyCode": false,           // step INTO libraries — essential for pydantic/httpx bugs
      "env": { "LOG_LEVEL": "DEBUG", "PYTHONASYNCIODEBUG": "1" }
    },
    {
      "name": "FastAPI: uvicorn --reload",
      "type": "debugpy",
      "request": "launch",
      "module": "uvicorn",
      "args": ["src.main:app", "--reload", "--port", "8000"],
      "jinja": true,
      "envFile": "${workspaceFolder}/.env"
    },
    {
      "name": "Pytest: current file",
      "type": "debugpy",
      "request": "launch",
      "module": "pytest",
      "args": ["${file}", "-vv", "-s", "--no-cov"],   // -s keeps stdout; disable cov for speed
      "console": "integratedTerminal"
    },
    {
      "name": "Attach: running container",
      "type": "debugpy",
      "request": "attach",
      "connect": { "host": "localhost", "port": 5678 },
      "pathMappings": [
        { "localRoot": "${workspaceFolder}", "remoteRoot": "/app" }
      ]
    }
  ]
}

For the attach config, the process must be listening:

# top of src/main.py, dev only
if os.getenv("DEBUGPY"):
    import debugpy; debugpy.listen(("0.0.0.0", 5678))
    if os.getenv("DEBUGPY_WAIT"): debugpy.wait_for_client()

# docker compose: expose 5678 and run with DEBUGPY=1
uvicorn src.main:app --reload --workers 1     # ⚠️ debug with ONE worker

12.2 .vscode/settings.json

{
  "python.defaultInterpreterPath": "${workspaceFolder}/.venv/bin/python",
  "python.testing.pytestEnabled": true,
  "python.testing.pytestArgs": ["tests"],
  "python.analysis.typeCheckingMode": "strict",
  "editor.formatOnSave": true,
  "editor.defaultFormatter": "charliermarsh.ruff",
  "editor.codeActionsOnSave": { "source.organizeImports.ruff": "explicit" },
  "files.exclude": { "**/__pycache__": true, "**/.pytest_cache": true }
}

12.3 Breakpoints beyond the red dot

Right-click any breakpoint → Edit Breakpoint:

Kind	Example	Use when
Conditional	msg.role == Role.TOOL and step > 5	Loop runs 200 times; you want iteration 201's cause
Hit count	>= 50	Failure appears only after N retries
Logpoint	step={step} tool={tool.name} tokens={usage.output_tokens}	You want a trace without editing code or restarting — logpoints don't pause
Exception	Check Raised for ValueError	Something swallows an exception and you need the raise site
Function	run_tool	Break wherever a function is called, without opening the file

Logpoints are the underrated one: they give you print-style tracing that vanishes when you close the session, so no debug prints ship.

12.4 Inspecting and changing runtime state

• Variables pane: locals, globals, self. Right-click → Copy Value to grab a whole prompt.
• Watch: pin real expressions — len(agent._history), sum(m.tokens for m in msgs), [t.name for t in tools if t.enabled].
• Call Stack: click any frame to inspect its locals — the fastest way to find which caller passed the bad argument. For async, each task has its own stack.
• Debug Console: a live REPL in the paused frame. This is where the leverage is — you can mutate state to force a branch you can't otherwise reach:

>>> resp.status = Status.FAILED       # force the error path without a real failure
>>> cfg.temperature = 2.0             # test a bound
>>> tool_result.output = "x" * 100_000  # simulate an oversized tool response
>>> import json; json.loads(raw)      # reproduce the parse error inline

Then continue execution and watch the branch you just forced.

12.5 Async and multi-process debugging

• "PYTHONASYNCIODEBUG": "1" warns about slow callbacks and never-awaited coroutines.
• Debug with --workers 1 and no --reload when a breakpoint won't bind.
• Subprocesses: debugpy follows child processes by default; for ProcessPoolExecutor it's usually faster to test the worker function directly in a unit test.
• Set "justMyCode": false when the bug is in how you call a library — that's where most "library bugs" actually live.

12.6 When the debugger isn't the right tool

python -m cProfile -s cumtime -m src.main | head -30   # where does the time go?
py-spy top --pid $(pgrep -f uvicorn)                   # live prod process, no restart
py-spy dump --pid 1234                                 # stack of every thread — find the hang
python -m tracemalloc ...                              # memory growth
python -X importtime -c "import src.main"              # slow startup

py-spy dump on a hung production service — showing every thread's Python stack without stopping it — has ended more incidents than any breakpoint.

🎯 Actionable rules

1. Commit launch.json; a shared debug config is team infrastructure.
2. Conditional breakpoints and logpoints instead of print + restart.
3. Mutate state in the Debug Console to reach error paths cheaply.
4. Hang in prod → py-spy dump. Slow in prod → py-spy top.

---

13. 🏛️ Patterns That Earn Their Keep

Not a catalogue — the eight patterns that actually appear in production AI code, each with what and why.

13.1 Registry + decorator — pluggable tools

What: a dict from name → implementation, populated by a decorator at import time. Why: an LLM hands you a tool name as a string; you need string → callable without an if/elif chain that grows forever.

TOOLS: dict[str, Callable[..., str]] = {}

def tool(name: str) -> Callable[[Callable[..., str]], Callable[..., str]]:
    """Register a function under `name`. Returns the function unchanged."""
    def deco(fn: Callable[..., str]) -> Callable[..., str]:
        if name in TOOLS:
            raise ValueError(f"duplicate tool {name!r}")     # fail at import, not at runtime
        TOOLS[name] = fn
        return fn
    return deco

@tool("calculator")
def calculator(expression: str) -> str:
    """Evaluate a simple arithmetic expression."""
    return str(safe_eval(expression))

def dispatch(name: str, **kwargs) -> str:
    fn = TOOLS.get(name)
    if fn is None:
        raise ToolError(name, f"unknown tool; available: {sorted(TOOLS)}")
    return fn(**kwargs)

Adding a tool = adding a decorated function. No central file to edit, no merge conflicts. (Registry beats getattr(self, name) — the allow-list is explicit.)

13.2 Decorator with @wraps — cross-cutting behaviour

What: a function that wraps a function. Why: retries, timing, tracing, and rate limits belong around your logic, not inside it.

def with_retry(attempts: int = 3, base_delay: float = 0.5):
    def deco(fn):
        @wraps(fn)                                     # preserve name/doc/signature
        async def inner(*args, **kwargs):
            for i in range(attempts):
                try:
                    return await fn(*args, **kwargs)
                except RetryableError:
                    if i == attempts - 1:
                        raise
                    await asyncio.sleep(base_delay * 2 ** i * (1 + random.random()))
        return inner
    return deco

@with_retry(attempts=4)
async def call_model(prompt: str) -> str: ...

Three nesting levels because a parameterized decorator is a factory returning a decorator returning a wrapper. If you don't need parameters, drop the outer layer.

13.3 Protocol + injection — swappable dependencies

What: depend on a structural interface, receive the implementation via the constructor. Why: tests get a fake for free, and swapping Anthropic → a local model touches one line.

class LLMClient(Protocol):
    async def complete(self, prompt: str, *, max_tokens: int = 1024) -> str: ...

class Agent:
    def __init__(self, cfg: AgentConfig, llm: LLMClient) -> None:   # injected, not imported
        self.cfg, self.llm = cfg, llm

agent = Agent(cfg, llm=AnthropicClient())      # prod
agent = Agent(cfg, llm=FakeLLM(["Result: 42"])) # test — no patching, no network

13.4 Factory / closure — configured behaviour

What: a function that returns a configured function or object. Why: cheaper than a class when the only state is configuration.

def make_chunker(size: int, overlap: int) -> Callable[[str], list[str]]:
    step = size - overlap
    def chunk(text: str) -> list[str]:
        return [text[i:i + size] for i in range(0, len(text), step)]
    return chunk

chunk_docs = make_chunker(size=1000, overlap=200)

functools.partial(search, index="docs", top_k=5) is the same idea in one line.

13.5 Context manager — scoped resources and scoped state

What: setup/teardown bound to a block (§6.5). Why: cleanup that can't be forgotten, plus a natural home for spans, tenancy, and budgets.

@asynccontextmanager
async def token_budget(limit: int):
    """Scoped budget: raises if the block exceeds `limit` tokens."""
    used = Counter()
    try:
        yield used
    finally:
        if used["total"] > limit:
            log.warning("budget exceeded: %d/%d", used["total"], limit)

13.6 Result object — return outcomes, don't throw control flow

What: a frozen dataclass/model carrying status + payload. Why: an agent step has expected failures (tool errored, budget hit, needs approval); exceptions are for the unexpected.

@dataclass(frozen=True, slots=True)
class ToolResult:
    tool_name: str
    output: str
    status: Status = Status.OK
    error: str | None = None

    @property
    def ok(self) -> bool: return self.status is Status.OK

The agent loop stays readable, and every outcome is inspectable and loggable.

13.7 Generator pipeline — streaming without buffers

What: chained generators, each doing one transformation. Why: constant memory, early results, composable stages.

def read_docs(paths): 
    for p in paths: yield p.read_text(encoding="utf-8")
def chunk(docs, size=1000):
    for d in docs: yield from (d[i:i+size] for i in range(0, len(d), size))
def embed(chunks, batch=32):
    for b in batched(chunks, batch): yield from model.encode(b)

vectors = embed(chunk(read_docs(paths)))     # nothing runs until you iterate

13.8 Sentinel — distinguishing "absent" from None

What: a unique object meaning "not provided". Why: when None is a legitimate value, you need a third state.

_MISSING = object()

def update(cfg: dict, temperature: float | None | object = _MISSING) -> dict:
    if temperature is not _MISSING:        # `None` here means "explicitly clear it"
        cfg["temperature"] = temperature
    return cfg

13.9 Putting it together

class Agent:
    async def run(self, user_input: str) -> AgentResponse:
        self.add_message(Role.USER, user_input)

        if "calculate" in user_input.lower():
            expr = user_input.lower().split("calculate", 1)[-1].strip()
            r = self.run_tool("calculator", expression=expr)
            self._tool_results.append(r)
            reply = f"Result: {r.output}"
        elif "count" in user_input.lower():
            r = self.run_tool("word_count", text=user_input)
            self._tool_results.append(r)
            top3 = [f"{k}={v}" for k, v in list(r.output.items())[:3]]   # comprehension
            reply = "Top words: " + ", ".join(top3)
        else:
            reply = f"Echo [{self.config.name}]: {user_input}"

        self.add_message(Role.ASSISTANT, reply)
        return AgentResponse(
            messages=self.history,
            tool_results=self._tool_results,
            total_steps=1,
            status=Status.OK,
        )

Count the ideas in 20 lines: enum roles (§2.9), split(sep, 1)[-1].strip() (§2.3), keyword args, a comprehension over dict.items() (§2.5), join (§2.3), a frozen result object (§13.6), and an async signature that stays awaitable even on the pure-Python path. That's the whole language, working together.

🎯 Actionable rules

1. String → behaviour: use a registry dict, never if/elif chains or unguarded getattr.
2. Cross-cutting concerns go in decorators; resources go in context managers.
3. Inject dependencies as Protocols; construct them at the edge (main, deps.py).
4. Expected failures → result objects. Unexpected failures → exceptions.

---

14. ⚖️ Good vs Bad, Side by Side

Twenty-four rewrites you can apply in your next code review.

1. Guard clauses beat nesting

# ❌ arrow code
def handle(req):
    if req is not None:
        if req.tools:
            if req.budget > 0:
                return run(req)
            else: return error("no budget")
        else: return error("no tools")
    else: return error("no request")

# ✅ fail fast, one indent level
def handle(req: Request | None) -> Response:
    if req is None:      return error("no request")
    if not req.tools:    return error("no tools")
    if req.budget <= 0:  return error("no budget")
    return run(req)

2. Loop over the thing, not over its indices

# ❌
for i in range(len(messages)):
    print(i, messages[i].role)

# ✅
for i, msg in enumerate(messages):
    print(i, msg.role)

3. Build strings with join

# ❌ O(n²): every += copies the whole string
prompt = ""
for m in history:
    prompt += f"{m.role}: {m.content}\n"

# ✅ O(n), and reads better
prompt = "\n".join(f"{m.role}: {m.content}" for m in history)

4. Membership tests belong on sets

# ❌ O(n) per check, inside a loop over 10k docs
if doc_id in seen_list: continue

# ✅ O(1)
seen: set[str] = set()
if doc_id in seen: continue
seen.add(doc_id)

5. Await concurrently when calls are independent

# ❌ 3 × latency
docs = await search(q)
summary = await summarize(q)
intent = await classify(q)

# ✅ 1 × latency
docs, summary, intent = await asyncio.gather(search(q), summarize(q), classify(q))

6. Don't block the event loop

# ❌ freezes every concurrent request on this worker
@app.post("/parse")
async def parse(f: UploadFile):
    return heavy_pdf_parse(await f.read())

# ✅ offload; the loop keeps serving
@app.post("/parse")
async def parse(f: UploadFile):
    return await asyncio.to_thread(heavy_pdf_parse, await f.read())

7. Catch what you predicted

# ❌ hides typos, cancellation, and every future bug
try:    data = json.loads(raw)
except Exception: data = {}

# ✅ narrow, logged, intentional
try:
    data = json.loads(raw)
except json.JSONDecodeError as exc:
    log.warning("non-JSON model output: %s", exc)
    data = {}

8. Validate at the boundary

# ❌ every access is a guess; failures surface deep in the call stack
def handle(payload: dict):
    temp = payload.get("temperature", 0.7)
    if temp > 2: ...     # TypeError if the client sent a string

# ✅ one parse, then trust your types
class QueryIn(BaseModel):
    query: str = Field(min_length=1)
    temperature: float = Field(default=0.7, ge=0.0, le=2.0)

def handle(payload: QueryIn):    # FastAPI returns a 422 with field-level detail
    ...

9. Names carry types

# ❌
def proc(d, l, f=False): ...

# ✅
def rerank_documents(docs: list[Document], limit: int, *, dedupe: bool = False) -> list[Document]: ...

10. Resources get a with

# ❌ leaks on exception; new connection pool per call
f = open(path); data = f.read(); f.close()
r = httpx.AsyncClient().post(url)

# ✅
data = Path(path).read_text(encoding="utf-8")
async with self._client_pool as client:      # one client, app-lifetime
    r = await client.post(url, timeout=30)

11. Comprehensions have a complexity budget

# ❌ one line, zero readability
r = [f(x) if p(x) else g(x) for xs in data for x in xs if q(x) and len(x) > 3]

# ✅ a loop is not a failure
r = []
for xs in data:
    for x in xs:
        if not q(x) or len(x) <= 3:
            continue
        r.append(f(x) if p(x) else g(x))

12. Module-level mutable state is a bug generator

# ❌ shared across requests, threads, and tests
CACHE = {}
def get(k): return CACHE.setdefault(k, expensive(k))

# ✅ owned, injectable, testable
class Cache:
    def __init__(self, maxsize: int = 1024) -> None:
        self._d: OrderedDict[str, str] = OrderedDict(); self._max = maxsize
    def get(self, k: str) -> str: ...

13. print → logging

# ❌ no level, no timestamp, no way to silence, leaks the prompt
print("calling model", prompt)

# ✅
log.info("calling model", extra={"model": cfg.model, "prompt_tokens": n})

14. Never eval model output

# ❌ remote code execution with extra steps
result = eval(model_expression)

# ✅ parse with a restricted grammar, or a sandboxed evaluator
import ast, operator
_OPS = {ast.Add: operator.add, ast.Mult: operator.mul, ast.Sub: operator.sub}
def safe_eval(expr: str) -> float:
    def ev(n):
        match n:
            case ast.Constant(value=float() | int() as v): return v
            case ast.BinOp() if type(n.op) in _OPS:
                return _OPS[type(n.op)](ev(n.left), ev(n.right))
            case _:
                raise ValueError(f"unsupported expression: {expr!r}")
    return ev(ast.parse(expr, mode="eval").body)

15. Batch the call, don't N+1 it

# ❌ 1000 round-trips: 1000 × latency, and you hit the rate limit at request ~200
vectors = [await embed(chunk) for chunk in chunks]

# ✅ same tokens, ~30× fewer round-trips
vectors: list[list[float]] = []
for batch in itertools.batched(chunks, 32):       # 3.12+; see §9.4 for the manual version
    vectors.extend(await embed_many(batch))

16. Bound the fan-out

# ❌ 10 000 concurrent sockets → instant 429s, exhausted file descriptors, one failure kills all
results = await asyncio.gather(*(fetch(u) for u in urls))

# ✅ at most 8 in flight, each with a deadline, failures isolated
sem = asyncio.Semaphore(8)
async def bounded(u: str) -> Doc | None:
    async with sem:
        try:
            return await asyncio.wait_for(fetch(u), timeout=20)
        except (TimeoutError, httpx.HTTPError) as exc:
            log.warning("skipping %s: %s", u, exc)
            return None
docs = [d for d in await asyncio.gather(*map(bounded, urls)) if d is not None]

17. Back off with jitter; never retry in a tight loop

# ❌ hammers a rate-limited API forever, and every client retries in lockstep
while True:
    try:    return await call_model(prompt)
    except RateLimitError: await asyncio.sleep(1)

# ✅ bounded attempts, exponential backoff, jitter to de-synchronize clients
for attempt in range(MAX_ATTEMPTS):
    try:
        return await call_model(prompt)
    except RateLimitError as exc:
        if attempt == MAX_ATTEMPTS - 1:
            raise
        delay = exc.retry_after or 0.5 * 2 ** attempt      # honour the server's hint first
        await asyncio.sleep(delay * (0.5 + random.random()))

18. Let length mismatches fail loudly

# ❌ if one embedding was dropped, zip truncates silently and EVERY id shifts by one
records = [{"id": i, "vec": v} for i, v in zip(doc_ids, vectors)]

# ✅ strict=True (3.10+) raises ValueError — a corrupted index is worse than a crash
records = [{"id": i, "vec": v} for i, v in zip(doc_ids, vectors, strict=True)]

19. assert is not input validation

# ❌ `python -O` strips every assert — your validation vanishes in production
def run(cfg):
    assert 0.0 <= cfg.temperature <= 2.0, "bad temperature"

# ✅ raise for untrusted input; keep `assert` for internal invariants you control
def run(cfg: AgentConfig) -> None:
    if not 0.0 <= cfg.temperature <= 2.0:
        raise ValueError(f"temperature out of range: {cfg.temperature!r}")

20. Own your state — copy at the boundary

# ❌ the caller's list IS the agent's internal state (see §1.4)
class Agent:
    def __init__(self, tools: list[Tool]) -> None:
        self._tools = tools

tools = [calculator]
agent = Agent(tools)
tools.clear()                      # agent._tools is now empty too

# ✅ snapshot on the way in, read-only view on the way out
class Agent:
    def __init__(self, tools: Sequence[Tool]) -> None:
        self._tools = list(tools)
    @property
    def tools(self) -> tuple[Tool, ...]:
        return tuple(self._tools)

21. Stream instead of buffering

# ❌ a 2 GB file becomes 2 GB of RSS; the user stares at a spinner for 30s
data = Path(big_file).read_text(encoding="utf-8")
answer = await client.complete(prompt)

# ✅ constant memory, and first token on screen in ~300ms
for chunk in read_chunks(big_file):            # generator, see §7.2
    index(chunk)

async for token in client.stream(prompt):      # async generator, see §7.5
    await websocket.send_text(token)

22. Top-k without sorting everything

# ❌ O(n log n) over a million candidates to keep five
top5 = sorted(docs, key=lambda d: d.score, reverse=True)[:5]

# ✅ O(n log k) in pure Python — or O(n) once the scores are an array
top5 = heapq.nlargest(5, docs, key=lambda d: d.score)
top5_idx = np.argpartition(-scores, 5)[:5]     # unordered; sort just these five if needed

23. Don't .get() your way into a distant None

# ❌ the missing key surfaces 20 frames later as 'NoneType' has no attribute 'lower'
name = payload.get("tool_call", {}).get("name")
args = payload.get("tool_call", {}).get("args", {})
run_tool(name.lower(), **args)

# ✅ required keys fail where they are missing, and the message names the key
try:
    call = payload["tool_call"]
    name, args = call["name"], call.get("args", {})
except KeyError as exc:
    raise ToolError("router", f"malformed tool call, missing {exc.args[0]!r}") from exc

24. Mutable class attributes are shared by every instance

# ❌ the §3.2 bug, class-shaped: one list for the whole process
class Agent:
    history: list[Message] = []
    def add(self, m: Message) -> None: self.history.append(m)

a, b = Agent(), Agent()
a.add(msg)
len(b.history)                     # 1 — b sees a's conversation

# ✅ per-instance state via default_factory (or plain assignment in __init__)
@dataclass(slots=True)
class Agent:
    history: list[Message] = field(default_factory=list)

---

15. ⚠️ Anti-Patterns and Misconceptions

15.1 Misconceptions that cost real hours

Belief	Reality
"Type hints make Python safe"	They do nothing at runtime. Safety comes from mypy in CI + Pydantic at boundaries.
"async makes code faster"	It makes waiting concurrent. CPU-bound async is the same speed, plus overhead.
"The GIL means Python can't do parallelism"	Processes are parallel; C extensions release the GIL. Only pure-Python threads are serialized.
"Threads speed up my numpy loop"	Only because numpy releases the GIL. Pure-Python threads won't help.
"is is a faster =="	It compares identity. x is 300 is False on most builds — small-int caching is an implementation detail.
"Copying a list with b = a protects the original"	It creates a second name for one object (§1.4).
"Private means private"	_x is a convention; __x is name-mangling. Neither enforces access.
"except Exception is defensive"	It's a way to convert a loud bug into a silent one.
"Adding @lru_cache is free performance"	It's an unbounded memory leak on high-cardinality keys, and wrong on anything non-pure.
"More uvicorn workers = more throughput"	Each is a full process. Past cpu_count() you're paying memory to context-switch.
"pandas/pytorch are Python-fast"	They're C/CUDA-fast. Your Python loop around them is the bottleneck.
"requirements.txt pins my env"	Only if pinned and transitively locked. Use a lockfile.

15.2 Anti-patterns, with the fix

1. God module. utils.py grows to 2000 lines and imports everything. → Split by domain (text.py, retry.py, tokens.py). If two modules need each other, extract the shared piece.

2. Import-time side effects. Opening a DB connection or reading env in __init__.py makes imports slow, tests fragile, and failures cryptic. → Do work in functions; construct at startup in main() / a lifespan handler.

3. Stringly-typed everything. status == "ok", role == "user", tool == "calculator". One typo = a silently dead branch. → StrEnum / Literal.

4. Exceptions as control flow. Raising StopProcessing to exit two levels of loop. → Return a result object; reserve exceptions for the unexpected.

5. Swallowing CancelledError. except Exception: pass inside a task breaks graceful shutdown and leaks connections. → Let it propagate; clean up in finally.

6. Mutable default arguments. Still the most common Python bug in production. → None sentinel (§3.2).

7. Unbounded everything. No timeout on the model call, no cap on history, no limit on retries, no semaphore on fan-out. Agents amplify all four into runaway cost. → Bound every loop, every wait, every list.

8. Re-creating clients per request. A new httpx.AsyncClient (or DB pool) per call destroys throughput and exhausts sockets. → One client for the app lifetime.

9. Logging the whole prompt. Blows up log costs and leaks PII/secrets. → Log token counts, ids, hashes, and truncated previews.

10. Testing the model instead of the code. Assertions on generated prose are flaky by construction. → Test parsing, routing, and bounds (§10.8).

11. Premature abstraction. A BaseAbstractToolHandlerFactory before the second tool exists. → Write it twice, abstract on the third.

12. sys.path hacking. sys.path.append("../..") at the top of a file. → src/ layout + pip install -e ..

13. Catching, logging, and re-raising at every level. The same error appears five times in the logs. → Log once, at the boundary that handles it.

14. Comparing floats with ==. 0.1 + 0.2 != 0.3. → math.isclose(a, b, rel_tol=1e-9); Decimal for money.

15. Mutating a list while iterating it. Silently skips elements. → Iterate a copy (for x in items[:]) or build a new list.

---

16. 🗺️ The 30-Day Path to Pro

Days	Focus	Ship this
1–3	§1–§2: types, collections, control flow	A CLI that chunks a text file and prints word stats
4–6	§3–§4: functions, typing, mypy	Add full hints; get mypy --strict to pass
7–9	§5–§6: dataclasses, errors, context managers	A Tool protocol + two tools + a timing context manager
10–13	§7: generators and asyncio	Stream tokens from a real model API with a per-chunk timeout
14–17	§9: Pydantic, httpx, logging	A FastAPI endpoint with validated I/O and structured logs
18–21	§10: pytest, fixtures, fakes	80% coverage with zero network calls in unit tests
22–24	§8: concurrency and profiling	Profile it; make one path 10× faster; write down why
25–27	§11–§12: tooling and debugging	Ruff + mypy + pytest in CI; a committed launch.json
28–30	§13–§15: patterns and review	Refactor with a registry + injected Protocol; review against §14

The one-page cheat sheet

# Data
{}                     # empty DICT (set() for an empty set)
d.get(k, default)      # optional      | d[k] — required, fails loud
{**a, **b}  /  a | b   # merge dicts (right wins)
required.keys() - got.keys()          # key diff
list(dict.fromkeys(xs))               # dedupe, order preserved
deque(maxlen=20)                      # sliding window
seq[-10:]  seq[::-1]  seq[::2]        # last N | reversed | stride

# Strings
f"{x!r}"  f"{x:.2f}"  f"{x=}"         # repr | format | debug
s.split(sep, 1)[-1].strip()           # take the tail
", ".join(parts)                      # never += in a loop

# Functions
def f(a, *, b=None, **kw)             # keyword-only after *
def f(x, hist=None): hist = hist or []# never a mutable default

# Types
x: str | None = None
Protocol / Literal / TypedDict / Annotated / Self
Callable[[str], Awaitable[str]]

# Objects
@dataclass(frozen=True, slots=True)
field(default_factory=list)
type(exc).__name__

# Async
await asyncio.gather(*coros)
async with asyncio.TaskGroup() as tg: tg.create_task(...)
async with asyncio.timeout(30): ...
await asyncio.to_thread(blocking_fn, arg)
async for chunk in stream: ...

# Errors
raise ToolError(name, msg) from exc
with suppress(FileNotFoundError): ...
log.exception("failed")               # inside except

# Test
pytest.raises(ValueError, match="...")
@pytest.mark.parametrize(...)
monkeypatch.setattr / setitem / setenv

# Run
uv sync && uv run pytest -n auto
uv run ruff check --fix . && uv run mypy src/
uvicorn src.main:app --workers 4
py-spy dump --pid $(pgrep -f uvicorn)

The ten habits that separate pro from proficient

1. Types at the boundary, checker in CI. Pydantic in, mypy over everything.
2. Bound every loop, wait, and list. Timeouts, retry caps, history windows, semaphores.
3. Classify the workload before choosing concurrency. I/O → asyncio. CPU → processes or C.
4. Never block the event loop. It's the #1 cause of "our AI service is slow."
5. Inject dependencies; construct at the edge. Testability is an architecture property.
6. Fail loudly at startup, gracefully at runtime. Missing config kills the process; a failed tool returns a result.
7. Log structured events, never secrets. Ids, counts, durations — not prompts.
8. Measure before optimizing. py-spy beats intuition every time.
9. Immutable by default. frozen=True, tuples, None sentinels — most concurrency bugs never appear.
10. Delete code. The fastest, safest, most readable line is the one you didn't write.

---

Where to go next: 🐹 Golang for AI Developers for the other half of the stack, 📘 The Complete Guide to LLMs and AI Agents 🤖
to understand modern AI deeply, ⚠️ Common Issues 🪲 with LLMs & AI Agents — and How to Fix Them 🛠️, 🏗️ Building High-Quality AI Agents for the agent architecture on top of this foundation, 🔄 The Agentic Loop Guide for the control loop itself, and 🏢 Enterprise-Ready AI Agents for multi-tenancy, security, and scale, and 🛠️ The Senior Software Engineer Playbook 📖.

Python is a small language wearing a large ecosystem. Learn the twelve concepts in Parts 1–7 properly and the rest is API documentation.

---

If you found this helpful, let me know by leaving a 👍 or a comment!, or if you think this post could help someone, feel free to share it! Thank you very much! 😃