Published about 16 hours ago 16 min read

🐍 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

Add type hints from line one, and run mypy in CI — you are buying back what the compiler gives Go.

Assume any function you pass a list/dict to may mutate it; copy at boundaries you care about.

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

dict for keyed data, set for membership, tuple for fixed records, list for sequences you grow.

{} is an empty dict; set() is an empty set.

Use !r in every error message that quotes a value.

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

Mutable default → None sentinel. Always.

Put a bare * in any signature with more than two parameters.

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

Annotate every public signature; let inference handle locals.

Protocol for interfaces you depend on; Enum/Literal instead of bare strings.

Any only at the untyped boundary, and parse it into a real type immediately.

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

Reach for @dataclass before writing __init__ by hand.

field(default_factory=...) for every mutable field.

Prefer frozen=True until you have a reason to mutate.

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

Catch the narrowest exception that can actually occur, as close to the cause as possible.

Always raise ... from exc when translating an error.

Every acquire has a with. If a library doesn't provide one, wrap it in @contextmanager.

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

Generators for anything large or streaming; never build a list you'll consume once.

TaskGroup > gather for anything with failure semantics.

Every await on an external call gets a timeout, and every fan-out gets a semaphore.

If it blocks and you can't fix it, to_thread it.

(...to be continued...) Read full version here https://dev.to/truongpx396/python-for-ai-developers-from-0-to-pro-5600

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! 😃