Engineering · Part 4 · chat and tools engine

The chat agent: a backend-neutral transcript

Reference documentation for an extractable chat-and-tools engine: one transcript model, two LLM providers behind one adapter, and a plain synchronous loop that streams tokens, runs tools, and pushes effects to the page. It depends on nothing above it — you can lift it into another product unchanged.

The engine's one job: given a conversation and a set of tools, call a model, stream what it says, run any tools it asks for, and repeat until it stops. The design goal that shapes everything is provider neutrality — your transcript, tools and UI must not know or care whether the model is AWS Bedrock or an OpenAI-compatible endpoint.

Own your transcript

Do not store a vendor SDK's message objects. Store a neutral, lossless transcript of your own, because it is your durable record (audit, resumability, replay) and it must outlive any one provider. Three tables:

flowchart LR
    S["Session
user, purpose, status, system_prompt"]:::a
    M["Message
role, position, per-call usage, tools_offered"]:::b
    P["Part
type, text, tool_use_id, tool_input, tool_result, media_ref"]:::c
    S -->|ordered| M -->|ordered| P
    classDef a fill:#eff6ff,stroke:#1d4ed8;
    classDef b fill:#ecfdf5,stroke:#047857;
    classDef c fill:#fffbeb,stroke:#b45309;

A Session is identity and scope only — user, purpose, status, system_prompt. Crucially it carries no foreign key into your domain (no blog FK): that keeps the agent package dependency-free and extractable. The link from a domain object to its chat lives in the domain package, as its own small join model.
A Message has a role — there are three speakers: the human (user), the model (assistant), and the engine running a tool (tool). One model call produces exactly one assistant message, so assistant rows also carry the per-call record: backend, model id, stop reason, token usage, and tools_offered (a JSON list of the tool names offered on that call — the audit lives in the transcript itself).
A Part is a typed content block: text, reasoning, tool_use, tool_result, image, document. Everything is content — a tool call is just a tool_use part; its result is a tool_result part on a later message, paired by tool_use_id. Tool input/output are JSON (their shape is owned by each tool — the one legitimate use of a JSON column); everything else is a real column.

This shape is deliberately close to how one major provider (Bedrock Converse) already models a conversation, which makes the mapping to both providers cheap. Media is carried as a reference (an object-storage key), not a foreign key, so the agent stays decoupled from whatever owns your files.

One adapter, two providers

An adapter converts your neutral transcript to a provider's wire format, calls it, and yields a single normalised event stream back. The engine only ever sees these events:

{'kind': 'text',      'text': '…'}          # a token of visible output
{'kind': 'reasoning', 'text': '…'}          # a token of thinking (providers that expose it)
{'kind': 'tool_use',  'id': …, 'name': …, 'input': {…}}
{'kind': 'done',      'stop_reason': 'end_turn'|'tool_use'|…, 'usage': {…}}

class Backend:
    name = '...'
    def stream(self, *, system, messages, tools, model, inference):
        """Yield the normalised events above. Raise BackendError on any transport failure."""

Each provider adapter owns two translations — in (neutral messages → wire) and out (wire stream → neutral events) — and normalises two things that always differ: the stop reason (map each provider's finish vocabulary onto yours) and the usage shape (input/output/cache token counts). The interesting divergence to plan for is tool-result placement: one provider expects tool results folded into a user-role message, another expects a dedicated tool role. Your neutral tool message absorbs the difference — each adapter's "in" translation places it correctly. A third adapter, a deterministic fake that returns scripted text and tool calls with no network, is what drives the entire test suite (Part 3).

Because tool results must round-trip to a model as a top-level JSON object, normalise a list-shaped tool result to {"items": […]} once, at the tool boundary — so every backend sees the same shape and the adapters stay dumb. (This is the same object-wrap you met in the MCP server; it lives in one helper shared by both.)

The catalogue and the handlers, in memory

Tools have two halves: the handler (a Python callable — you cannot store a callable in a row) and the catalogue (name, description, schema — what the model is offered). Keep both in one in-memory registry, populated at startup. There is no tool table and no migration to sync it, so an empty catalogue — the failure mode where the model is handed no tools and hallucinates calls — is structurally impossible while the app is installed:

REGISTRY, CATALOG = {}, {}

def register(name, *, description='', input_schema=None, purposes=('chat',)):
    def deco(fn):
        REGISTRY[name] = fn
        CATALOG[name] = {'description': description,
                         'input_schema': input_schema or {'type': 'object', 'properties': {}},
                         'purposes': list(purposes)}
        return fn
    return deco

def catalog_for(purpose):                       # what THIS session may be offered
    return [{'name': n, **spec} for n, spec in sorted(CATALOG.items())
            if not spec['purposes'] or purpose in spec['purposes']]

The domain package registers its tools at startup, straight from the operation catalogue of Part 1 — the handler is perform_json plus context resolution:

# in the DOMAIN package's app startup — the agent never imports this direction
for name, op in exposed(OPERATIONS).items():
    register(name, description=op.description, input_schema=op.input_schema, purposes=('chat',))(
        lambda session, _n=name, **raw: _dispatch(session, _n, raw))

def _dispatch(session, name, raw):
    blog = ChatLink.objects.get(session=session).blog        # resolve scope from the session
    out, effects = perform_json(name, session.user, blog, raw)
    result = as_object(out)
    if effects:
        result['_effects'] = effects                         # rides back to the engine (below)
    return result

Purpose scoping keeps tool sets apart: a general chat session is offered purposes=('chat',) tools; a specialised session (say a proposition-authoring chat) is offered only its own. The model for one purpose never even sees another's tools.

The loop — synchronous, streamed, on gevent

flowchart TB
    Start([user message persisted]) --> Budget{over budget?}
    Budget -- yes --> Err[emit error event, stop]
    Budget -- no --> Build[build neutral messages from the DB]
    Build --> Call[adapter.stream: yield token / reasoning / tool_use / done]
    Call --> Persist[persist assistant Message + Parts + usage]
    Persist --> Tools{model asked for tools?}
    Tools -- no --> Done([stop])
    Tools -- yes --> Run[run each tool by name] --> Effects[stream any effects event] --> Budget

Write it as a plain synchronous generator of structured (event, payload) tuples — the engine itself knows no wire format; a thin UI wrapper encodes the tuples for its surface (ours: sse_response() in the django-toolchat-htmx-sse package renders them as server-sent events for the browser; a Slack surface renders the same tuples as messages). Do not reach for async: a straight-line loop is far easier to reason about, and the concurrency you need — many simultaneous streams — comes from running under a gevent worker, where each request is a cheap greenlet. One discipline matters under gevent, and it is the thing to get right: never hold a database transaction open across the stream. A stream can last minutes; wrapping the loop in a transaction would pin a connection and hold locks the whole time. Instead each persist is a discrete autocommit write, so between model bursts the connection sits idle with no transaction and no locks — harmless.

def run_agent(session):
    backend = get_backend()
    specs, offered_names = _offered(session)          # catalogue_for(purpose) → specs + names
    for _ in range(MAX_TOOL_CYCLES):                  # a hard ceiling on tool round-trips
        try:
            budget.check(session.user)                # DoW guard, before EVERY model call
        except BudgetExceeded as exc:
            yield ('error', {'message': str(exc)}); return

        messages = build_neutral_messages(session)    # a read burst from the transcript
        text, reasoning, tool_uses, usage, stop = [], [], [], {}, 'end_turn'
        try:
            for ev in backend.stream(system=session.system_prompt, messages=messages,
                                     tools=specs, model=model_id, inference=inference):
                if ev['kind'] == 'text':
                    text.append(ev['text']);       yield ('token', {'text': ev['text']})
                elif ev['kind'] == 'reasoning':
                    reasoning.append(ev['text']);  yield ('reasoning', {'text': ev['text']})
                elif ev['kind'] == 'tool_use':
                    tool_uses.append(ev);          yield ('tool_use', ev)
                elif ev['kind'] == 'done':
                    stop, usage = ev['stop_reason'], ev.get('usage') or {}
        except BackendError:
            logger.exception('backend error')            # raw error → the log
            yield ('error', {'message': GENERIC_MESSAGE}); return   # calm text → the user

        msg = persist_assistant(session, text, reasoning, tool_uses, usage, offered_names)
        if not tool_uses:
            return                                    # the model is done

        results = [(_run_tool(session, tu)) for tu in tool_uses]   # (result, status) each
        persist_tool_results(session, tool_uses, results)
        for result, _status in results:              # effects → the page (below)
            for e in (result.pop('_effects', []) if isinstance(result, dict) else []):
                yield ('effects', [e])
    yield ('error', {'message': TOO_MANY_CYCLES_MESSAGE})

Three robustness details worth copying verbatim: a hard MAX_TOOL_CYCLES ceiling so a confused model cannot loop forever; a strict split between the operator's error (the raw provider exception, to the log) and the user's error (a calm sentence, to the chat — never leak a botocore or httpx string into the UI); and a tool runner that turns any handler exception into a structured {'error': …} result the model can read and recover from, rather than crashing the stream.

Effects close the loop to the UI

Notice the _effects a tool result carries (Part 1): the engine strips that key from the model-facing result — the model must not see UI bookkeeping — and emits it as a separate effects server-sent event. The page's client subscribes and refreshes the fragments that care. So when the assistant publishes a post, the post list updates live, with no polling and no transport aware of any other. The same vocabulary, all the way from the mutator in Part 1 to the pixel.

Denial-of-wallet: a budget before every call

An agent that calls a paid model in a loop is a standing invitation to run up a bill — through abuse or a runaway loop. Guard it with a cheap check before every model call, across several windows:

def check(user):
    now = timezone.now()
    if per_min and tokens(usage(user, now - minutes(1)))  >= per_min:  raise BudgetExceeded('minute', …)
    if per_hour and tokens(usage(user, now - hours(1)))   >= per_hour: raise BudgetExceeded('hour', …)
    if monthly and cost_usd(usage(user, now - days(30)))  >= monthly:  raise BudgetExceeded('month', …)
    if global_cap and cost_usd(usage_all(now - days(30))) >= global_cap: raise BudgetExceeded('global', …)

Per-minute and per-hour caps are token counts (they throttle abuse and loops); the monthly cap is a dollar figure estimated from per-token prices (output dominates spend); and a global monthly dollar ceiling across all users is the account-wide backstop that trips before a compromised key can empty the account. The token counts and prices live only in this module; what escapes is a friendly per-bucket message. A UI counterpart turns the same buckets into fill fractions so a usage meter can stay hidden until the user is near a limit — the product should feel limitless right up until it must not.

A design choice to make deliberately: direct mutation

Should the agent apply changes directly, or propose them for a human to approve? Both are valid; choose consciously. Direct mutation — the agent calls the same mutators a person's UI calls, with a capped subset of the owner's own access (below) — makes for a fluid assistant and keeps exactly one write path (no parallel "proposed change" model to maintain). Its safety comes from three places already in this series. First, the agent does not run with the user's full account: it runs as a capped, client-less grant (an Actor when written; a Consent row since v3, SPEC §24.8) — a subset of the owner's own access that they lend the assistant in the same Clients screen the OAuth connectors use, intersected with their live seat on every call (Part 2). So it can never out-reach the human behind it, and narrowing or clearing a grant bites on the very next turn — with a separate owner-only gate deciding who may summon it at all. Second, each change is recorded in the immutable transcript with before/after history for audit and reversal. Third, the system prompt instructs the model to confirm before irreversible actions. If your domain is higher-stakes, keep the same engine and register propose_* tools that write a pending-change record instead — the loop does not change, only the handlers do.

The shape to copy

A neutral Session → Message → Part transcript, with no FK into your domain.
One Backend.stream interface; an adapter per provider normalising stop reason, usage and tool-result placement; a deterministic fake for tests.
An in-memory registry holding handler and catalogue, purpose-scoped; the domain registers perform_json-backed handlers at startup.
A synchronous streamed loop under gevent, no transaction across the stream, a cycle ceiling, operator-vs-user error split.
Effects streamed to the UI; a denial-of-wallet budget before every call.

Part 5 leaves the application entirely: how the box it all runs on boots itself — restoring the database before the app starts — without cloud-init.

← All posts