Engineering · Part 1 · the operations pattern

One domain core, many thin transports

architectureapi

Written during auth v2. The architecture below is current and the prose now uses today’s nouns; the v3 operations rename (SPEC §24.8) changed the identifiers: PermissionSet → Policy, UserPermissionSet → Assignment, authorize(user, project, op, actor) → token.require(op), and Actor/ActorProject folded into Consent.

Reference documentation for a small, extractable Django pattern: one domain core, many thin transports. This is the foundation the other four documents build on.

You are building a multi-tenant blogging platform. A blog is a tenant; it contains posts; posts carry comments and tags; people join a blog through a membership, and a membership holds one or more policies — named subsets of the operations below. (Creating a blog mints two built-in policies, Owner and Collaborator; the owner can add more.) The same handful of operations — list posts, publish one, add a comment, tag it — must be reachable from at least five places:

• server-rendered HTML that a person clicks through;
• a JSON API, with OpenAPI docs generated for it;
• function-calling tools for an LLM agent;
• a remote MCP server external AI clients log into; and
• whatever you add next.

Implement those operations once per interface and you will, inevitably, end up with an authorization rule that lives in four places, three of which agree. This document describes the alternative: one stack, three layers, where every interface is a thin adapter over a single implementation.

api.py           the domain core — owns ALL authorization, speaks ORM objects
operations.py    the catalogue — one registry entry per capability + two entry points
registry.py      resolves the catalogue once + the shared transport machinery
transports       thin adapters: HTML views, JSON API, OpenAPI, agent tools, MCP

Two things to carry out of that picture. First, the return types diverge on purpose: the HTML lane gets real model instances and querysets back, because templates want models; every other lane gets schema-validated JSON, because machines want contracts. Second, the docs — OpenAPI, the LLM tool list, MCP's tools/list — are derived from the registry. They describe the doors; they are not extra doors.

One sentence governs every decision below: services authorize; transports attenuate. The service layer makes every authorization decision, because only it has the context. A transport may narrow what a request can attempt — a token scope, a per-client grant, a staff-only URL — but no transport check is ever the only thing standing between a request and your data.

Layer 1 — the domain core (api.py)

A service is a plain module-level function obeying a strict signature contract:

• user first, always — the acting principal, however the transport authenticated them.
• The scope second (here, the blog), as a resolved model instance. The transport resolves it — from the URL, the chat session, the OAuth grant — because scope is contextual: a person belongs to many blogs, and the request declares which one it is about. Account-level services omit it. Never derive the scope from the data being edited: that would let a member of blog A operate on blog B by quoting B's ids.
• Target rows arrive as ids plus validated keyword data — never as pre-fetched instances. The service fetches them itself, scoped, and that fetch is the authorization boundary.
• Returns are ORM objects — instances and querysets — so the HTML transport loses nothing.

Two primitives carry the invariants: the operation check and the scoped fetch. Notice what the check is not — there is no role rank. Each capability is a named operation, and a caller may perform it iff that operation is in one of the policies they hold on this blog:

# api.py   (the domain core — formerly services.py)
from django.core.exceptions import PermissionDenied

class NotFound(Exception):
    """A referenced id does not exist within this scope."""

def _member_operations(user, blog):
    """The operations this user's membership grants here: the union of the operations across the
    policies they hold on this blog. The built-in 'Owner' policy resolves to the WHOLE catalogue, so a
    new operation is always owner-able and the owner can never be locked out of one."""
    ops = set()
    for ups in Assignment.objects.filter(
            membership__user=user, membership__blog=blog, membership__accepted=True):
        ops |= ups.policy.resolved_operations()
    return ops

def authorize(user, blog, operation, actor=None):
    """Raise unless the caller may perform `operation` here. actor=None → the human acting on their
    own membership. An `actor` (an OAuth client's grant, §part 2) narrows to its BORROWED set ∩ the membership —
    so a machine can never out-reach the human it acts for, even if the set is later widened."""
    effective = (_member_operations(user, blog) if actor is None
                 else actor.operations_on(blog) & _member_operations(user, blog))
    if operation not in effective:
        raise PermissionDenied(f'not permitted to {operation} here')

def _get_in_blog(blog, model, pk):
    obj = model.objects.filter(pk=pk, blog=blog).first()
    if obj is None:
        raise NotFound(f'{model.__name__} {pk} not found in this blog')
    return obj

A policy is a per-blog row: a name plus a JSON list of operation names. It belongs to one blog, so editing it only ever affects that blog — editing is always safe. There are no global or rank roles to reason about across tenants; "what may this member do here" is a set membership test. _get_in_blog is the second load-bearing idiom: there is no separate "check permission, then fetch" to drift apart — an id from another blog simply does not exist here, and the caller cannot even learn the row exists (NotFound, not PermissionDenied). Now the services — each opens with one authorize line naming its own operation, then does its work:

def get_posts(user, blog, *, status=None, actor=None):
    authorize(user, blog, 'get_posts', actor=actor)
    qs = blog.posts.select_related('author').prefetch_related('tags')
    return qs.filter(status=status) if status else qs

def create_post(user, blog, *, title, body='', actor=None):
    authorize(user, blog, 'create_post', actor=actor)
    post = Post.objects.create(blog=blog, author=user, title=title, body=body, status='draft')
    post._effects = [_effect(post, 'create', blog.pk)]
    return post

def publish_post(user, blog, *, post_id, actor=None):
    authorize(user, blog, 'publish_post', actor=actor)  # in the 'Editor' set, not the 'Author' one
    post = _get_in_blog(blog, Post, post_id)
    before = _snapshot(post, ['status', 'published_at'])
    post.status = 'published'
    post.published_at = timezone.now()
    post.save(update_fields=['status', 'published_at'])
    post._effects = [_effect(post, 'publish', blog.pk)]
    post._history = _diff(before, _snapshot(post, ['status', 'published_at']))
    return post

def add_comment(user, blog, *, post_id, body, actor=None):
    authorize(user, blog, 'add_comment', actor=actor)
    post = _get_in_blog(blog, Post, post_id)
    comment = Comment.objects.create(post=post, author=user, body=body)
    comment._effects = [_effect(comment, 'create', blog.pk)]
    return comment

def delete_post(user, blog, *, post_id, actor=None):
    authorize(user, blog, 'delete_post', actor=actor)
    post = _get_in_blog(blog, Post, post_id)
    post.delete()
    return {'deleted': True, 'id': post_id,
            '_effects': [{'type': 'post', 'id': post_id, 'op': 'delete', 'blog': blog.pk}]}

def create_blog(user, *, name):                          # account-level: no scope, no actor
    blog = Blog.objects.create(name=name)
    m = Membership.objects.create(blog=blog, user=user, role='owner', accepted=True)
    owner_set, _collaborator = _mint_builtin_sets(blog)  # 'Owner' (whole catalogue) + 'Collaborator'
    Assignment.objects.create(membership=m, policy=owner_set)
    blog._effects = [_effect(blog, 'create', blog.pk)]
    return blog

The actor=None keyword on every scoped service is the whole of the machine-caller story at this layer: a human call omits it; a connector call fills it (part 2). One parameter, threaded by the entry point — no second code path, no contextvar.

Effects — every mutation names what it changed

You just saw _effects on every write. An effect is a small change pointer attached to the return value — a _effects attribute on a model, a _effects key on a delete's dict:

_EFFECT_TYPE = {Blog: 'blog', Post: 'post', Comment: 'comment', Tag: 'tag'}

def _effect(instance, op, blog_id):
    return {'type': _EFFECT_TYPE[type(instance)], 'id': instance.pk, 'op': op, 'blog': blog_id}

One line per mutator buys every transport, uniformly, the answer to "what did that just change?" — which is exactly what a live-updating UI needs (we consume them at the end). Because they are easy to forget, make forgetting impossible: a completeness test iterates every registered kind='write' operation and fails, by name, any whose exact effects are not asserted in your effects test file. The vocabulary is thus guaranteed complete without anyone remembering to keep it so.

The return is richer than the schema — and different transports capture different slices

Effects are the first example of a general idea worth naming outright. A service returns a rich ORM object; the to_json serialiser and its output_schema are a deliberately narrow projection of it for the wire. But the object can also carry transient side-band data that specific transports reach for. Effects are one. Before/after history is the second, and it exists for one caller in particular: the AI agent.

When the agent publishes a post through the tool-calling route, we want the transcript to record what it changed — so a later turn can audit the change, explain it, or reverse it. So a mutator snapshots the fields it touches and attaches a {before, after, changed} block, exactly as it attaches effects:

def _snapshot(obj, fields):
    return {f: getattr(obj, f) for f in fields}

def _diff(before, after):
    changed = [f for f in before if before[f] != after[f]]
    return {'before': {f: before[f] for f in changed},
            'after':  {f: after[f]  for f in changed},
            'changed': changed}

The two kinds of extra are captured by different paths, and the difference is deliberate — it turns on whether the model should see the data:

• Effects are hidden from the model. perform_json pops _effects off the result before serialising (you saw it above), returning them as the second tuple element. They are UI bookkeeping; the model has no use for them, and streaming transports route them to the page.
• History is shown to the model. It rides into the JSON output, folded in by a serialiser wrapper, so it becomes part of the tool result the model reads and the transcript stores. A write operation opts in by wrapping its serialiser:

def _with_history(serialise):
    def to_json(obj):
        out = serialise(obj)
        history = getattr(obj, '_history', None)
        if history:
            out['_history'] = _jsonsafe(history)      # dates → iso, Decimals → float, …
        return out
    return to_json

# in the registry, write ops opt in:
Operation('publish_post', api.publish_post, kind='write',
          to_json=_with_history(post_to_json), ...)

Now watch the same perform/perform_json return feed three transports, each capturing what its consumer needs and nothing more:

Concretely, the agent tool handler is the one path that captures everything:

def tool_handler(session, name, **args):
    blog = ChatLink.objects.get(session=session).blog       # resolve scope from the session
    out, effects = perform_json(name, session.user, blog, args)
    stream_effects(effects)          # → server-sent event → the page refreshes a fragment
    return out                       # → stored as the tool_result Part → audit + reversibility
    # `out` carries {before, after, changed}: a later turn can call the inverse operation.

The lesson generalises past these two examples: keep the service return rich and let each transport project what it needs, rather than pushing every consumer's requirements down into a single wire shape. When a new consumer wants more — a diff, a provenance id, a computed total — it is a side-band attribute and a serialiser wrapper, not a schema migration that ripples through every other caller.

Layer 2 — the operation catalogue

Every public service gets exactly one entry in a registry. The entry is that capability's entire public contract, and the constructor enforces a rule with teeth: no private APIs.

SURFACES = frozenset({'tool', 'api', 'mcp'})

class Operation:
    def __init__(self, name, service, *, kind, description='', to_json=None,
                 input_schema=None, output_schema=None, veto=None, gated=True,
                 account=False, from_json=None):
        assert veto or (to_json and input_schema is not None and output_schema is not None), \
            f'{name}: needs schemas + a serialiser, or a written veto reason'
        self.name, self.service, self.kind = name, service, kind
        self.description = description
        self.to_json, self.input_schema, self.output_schema = to_json, input_schema, output_schema
        self.veto = veto        # a written reason this capability is human-only
        self.gated = gated      # False = a "fix-it flow" that must run even when the gates fail
        self.account = account  # True = account-level; the service takes no scope
        self.from_json = from_json  # optional inverse of to_json (raw JSON → service kwargs)

An operation either carries full schemas and a serialiser (and is therefore served on every machine surface), or it carries a written veto — a stated reason it must stay human-only. There is no third state, so "I haven't written the schema yet" is an AssertionError at import time, not a silently hidden endpoint. Vetoes are rare and greppable:

Operation('accept_terms', api.accept_terms, kind='write', account=True, gated=False,
          veto='agreeing to legal terms must be a human act')

That one earns it: an AI accepting a terms-of-service on the user's behalf would hollow out the very consent record the operation exists to create. The registry itself:

# JSON Schema helpers: _obj(props, required) → an object schema; ID / STR primitives.
OPERATIONS = {op.name: op for op in [
    Operation('get_posts', api.get_posts, kind='read',
              to_json=lambda qs: [post_to_json(p) for p in qs],
              input_schema=_obj({'status': {'enum': ['draft', 'published']}}),
              output_schema={'type': 'array', 'items': _obj(POST_OUT)},
              description="List the blog's posts (optionally filtered by status)."),
    Operation('create_post', api.create_post, kind='write', to_json=post_to_json,
              input_schema=_obj({'title': STR, 'body': STR}, required=['title']),
              output_schema=_result(POST_OUT),
              description='Create a draft post.'),
    Operation('publish_post', api.publish_post, kind='write', to_json=post_to_json,
              input_schema=_obj({'post_id': ID}, required=['post_id']),
              output_schema=_result(POST_OUT),
              description='Publish a draft post (editor only).'),
    Operation('delete_post', api.delete_post, kind='write', to_json=lambda d: d,
              input_schema=_obj({'post_id': ID}, required=['post_id']),
              output_schema=_obj({'deleted': {'type': 'boolean'}, 'id': ID}, ['deleted', 'id']),
              description='Delete a post.'),
    Operation('create_blog', api.create_blog, kind='write', account=True,
              to_json=blog_to_json,
              input_schema=_obj({'name': STR}, required=['name']),
              output_schema=_result(BLOG_OUT),
              description='Create a blog (you become its owner).'),
]}

def exposed(operations):
    """The machine-served slice: everything without a veto."""
    return {name: op for name, op in operations.items() if not op.veto}

The schemas do double duty. They validate input and output shape on every machine call, and they advertise the capability — the same input_schema becomes the LLM tool signature, the OpenAPI request body, and the MCP tool schema. Write the contract once and five surfaces cannot disagree about what create_post takes. A serialiser is small:

def post_to_json(p):
    return {'id': p.id, 'title': p.title, 'body': p.body, 'status': p.status,
            'author_id': p.author_id, 'tags': [t.name for t in p.tags.all()]}

The two entry points

Nothing outside the domain layer calls a service directly. Everything comes through one of two doors, built once by a small factory. This is the whole of it:

import jsonschema

def make_entrypoints(operations, *, gate=None):

    def perform(name, user, scope=None, /, **kwargs):
        """Product gates (unless gated=False), then registry dispatch. Returns ORM objects."""
        op = operations[name]
        if gate is not None and op.gated:
            gate(user, None if op.account else scope)
        if op.account:
            return op.service(user, **kwargs)
        return op.service(user, scope, **kwargs)

    def perform_json(name, user, scope, raw, /):
        """The JSON skin: validate in → decode → perform → serialise → validate out."""
        op = operations[name]
        raw = raw or {}
        jsonschema.validate(raw, op.input_schema)
        kwargs = op.from_json(raw) if op.from_json else raw
        result = perform(name, user, scope, **kwargs)
        # effects ride on the return; extract them so they never reach to_json / the output schema.
        if isinstance(result, dict):
            effects = result.pop('_effects', [])
        else:
            effects = getattr(result, '_effects', [])
        out = op.to_json(result)
        jsonschema.validate(out, op.output_schema)
        return out, effects

    return perform, perform_json

(The / makes the leading parameters positional-only — necessary, because operations legitimately have their own name keyword arguments.) perform is the domain door: gates, dispatch, ORM objects out. perform_json is the machine door — validate, decode, perform, serialise, validate again — returning (out, effects). Note the output is validated too: a serialiser bug fails loudly at the boundary instead of silently inside a client.

Gates, and the deliberate bypass

Product gates are the checks that are not authorization but still guard everything — "has this user accepted the current terms?", "does this blog have a live subscription?". Centralise the predicates and enforce them inside perform:

# gates.py
def gate_message(user, blog=None):
    if user.is_staff:                                     # support bypass
        return None
    if not terms_accepted(user):
        return 'you must accept the current terms — visit /legal/accept'
    if blog is not None and billing_enabled() and not subscription_live(blog):
        return f'the subscription for "{blog.name}" is not active — renew under Billing'
    return None

def require(user, blog=None):
    message = gate_message(user, blog)
    if message:
        raise PermissionDenied(message)

# operations.py
perform, perform_json = make_entrypoints(OPERATIONS, gate=gates.require)

Two subtleties make this robust in practice:

• The bypass is a per-operation flag, not a URL list. Some operations must run precisely when the gate fails — you cannot re-accept the terms if accepting terms is behind the terms gate. Mark those gated=False on their entry. The exemption is explicit, greppable, and reviewed alongside the operation.
• Middleware survives only as UX. A browser user deserves a friendly redirect ("add a card to start your trial") rather than a raw 403, and middleware is the right place to produce it. But middleware protects URLs, and your next transport will not share your URL structure — so middleware is presentation and defence in depth. Enforcement lives in the door every caller must pass.

Layer 3 — the transports

HTML views: forms in, models out

@blog_member_required                    # resolves request.blog from the URL, or 404
def post_create(request, pk):
    form = PostForm(request.POST)
    if not form.is_valid():
        return render(request, 'posts/_form.html', {'form': form}, status=422)
    post = perform('create_post', request.user, request.blog, **form.cleaned_data)
    return render(request, 'posts/_row.html', {'post': post})

Everything medium-specific stayed in the view. The Django Form is the HTML transport's shape gate — field errors re-rendered in place, widgets, coercion — exactly as JSON Schema is the machine transports'. Because perform returns the real model, the template renders properties and relations directly; and because reads return querysets, views keep composing:

posts = perform('get_posts', request.user, request.blog)   # a queryset
recent = posts.filter(status='published').order_by('-published_at')[:10]

The JSON API: about twenty lines, zero business rules

def api_call(request, pk, name):                 # /blogs/<pk>/api/<name>
    ops = exposed(OPERATIONS)
    if name not in ops:
        return JsonResponse({'error': f'unknown operation: {name}'}, status=404)
    op = ops[name]
    want = 'GET' if op.kind == 'read' else 'POST'          # reads safe/cacheable; writes CSRF
    if request.method != want:
        return JsonResponse({'error': f'use {want} for {name}'}, status=405)
    raw = (coerce_query(op, request.GET) if want == 'GET'
           else json.loads(request.body or '{}'))
    try:
        out, _effects = perform_json(name, request.user, request.blog, raw)
    except ValidationError as e:
        return JsonResponse({'error': 'invalid input', 'detail': e.message}, status=400)
    except PermissionDenied as e:
        return JsonResponse({'error': 'forbidden', 'detail': str(e)}, status=403)
    except NotFound as e:
        return JsonResponse({'error': 'not found', 'detail': str(e)}, status=404)
    return JsonResponse(out, safe=False)

The verb convention: kind='read' ⇒ GET with query parameters (safe, cacheable, no CSRF concern), kind='write' ⇒ POST with a JSON body (session-cookie callers stay CSRF-protected by the framework). Account-level operations get a scope-less mount, /api/<name>, whose handler passes scope=None.

OpenAPI: generated, and it attenuates

def openapi_spec(user=None):
    paths = {}
    for app in visible_apps(user):               # per-user: staff-only apps only for staff
        for op in app.operations.values():
            if op.veto:
                continue                         # human-only operations are not advertised
            path = (app.account_path_template if op.account else app.path_template) \
                       .format(pk='{pk}', name=op.name)
            if op.kind == 'read':
                params = query_params_from(op.input_schema)       # properties → ?query params
                paths[path] = {'get': {'parameters': params, 'responses': _resp(op)}}
            else:
                body = {'content': {'application/json': {'schema': op.input_schema}}}
                paths[path] = {'post': {'requestBody': body, 'responses': _resp(op)}}
    return {'openapi': '3.1.0', 'paths': paths}

This is attenuation in action, and it is important to be precise about what it is not. Filtering the document (dropping vetoed operations, hiding staff-only apps from non-staff) is least-surprise and courtesy — a caller who guesses a hidden operation's name still reaches the service layer's authorization and is refused there. Hiding is UX; the services are security. The same exposed()/veto filter drives every machine surface, so "what can a machine see?" has exactly one answer.

Agent tools and MCP: the same door again

An LLM tool handler is perform_json plus context resolution — the chat session knows the user and the blog; the tool call supplies name and arguments. A remote MCP server is the same thing over OAuth: the bearer token yields the acting user and an actor (a grant carrying, per blog, the policy the client may borrow), and it is perform_json all the way down — the actor threaded through to authorize. The same bearer token also works on the JSON API: /api/ and /mcp are two transports over one resource server, not two APIs. That server — the protocol, the OAuth, and the client model — is the subject of Part 2.

The attenuation stack

Once transports authenticate differently — a session cookie here, an OAuth bearer token there — it is tempting to scatter permission checks wherever they are convenient. Resist it by giving the checks a strict geometry: they form a stack, and every layer above the bottom may only narrow.

• Token scopes — retired in v3 (SPEC §24.4): wire tokens are scope-less. The machine cap is the next bullet — the consent's per-blog policy — intersected with the human's own live membership; a coarse read/write scope added nothing the policy didn't already say more precisely.
• The actor (remote connectors + OAuth clients): the grant carries, per blog, the policy the client may borrow — and a request naming a blog the client wasn't granted stops here. Below, authorize intersects that borrowed set with the user's own live membership, so the machine is capped twice: to the granted set and to what the human could do anyway.
• Product gates: inside perform, for every caller, with the gated=False escape.
• Service authorization (authorize): the policy membership test (∩ the actor's borrowed set) plus the scoped fetch — the only layer that can say yes.

The geometry makes the system reason-about-able. Any upper layer can be deleted and you lose convenience or defence in depth — never correctness, because the bottom layer runs for every caller unconditionally. The practical review test: if removing a transport-level check would open a hole, that check was load-bearing, and something is in the wrong layer.

Closing the loop: effects drive the UI

Return to the change pointers every mutator emits. When a streaming transport applies a mutation, the effects are split off the model-facing result and emitted as their own server-sent event:

flowchart LR
    M["mutator runs
attaches _effects"]:::a --> PJ["perform_json
returns (out, effects)"]:::b
    PJ --> T["transport emits
an SSE 'effects' event"]:::c
    T --> C["client matches types
htmx re-fetches the fragment"]:::d
    C -. "the user sees it update; the next mutation goes round again" .-> M
    classDef a fill:#ecfdf5,stroke:#047857;
    classDef b fill:#f3f4f6,stroke:#374151;
    classDef c fill:#fffbeb,stroke:#b45309;
    classDef d fill:#eff6ff,stroke:#1d4ed8;

event: effects
data: [{"type": "post", "id": 42, "op": "publish", "blog": 7}]

source.addEventListener('effects', (e) => {
  const effects = JSON.parse(e.data);
  if (effects.some(x => ['post', 'tag'].includes(x.type))) {
    htmx.trigger('#post-list', 'refresh');   // an hx-get re-fetches just that fragment
  }
});

So when an AI assistant publishes a post mid-conversation, the list updates without a reload — and no transport had to know how any other transport renders. Start minimal (one event, one fragment); the point is that the write path already describes its own consequences, uniformly, so richer live-update never needs a retrofit through every mutator.

Making the architecture un-driftable

Rules that live in prose decay. Encode them as tests — Python's ast module makes each one a few lines:

TRANSPORT_MODULES = ['app/views.py', 'app/transport.py', 'app/tools.py', 'app/mcp.py']

def test_transports_never_call_the_domain_directly(self):
    for rel in TRANSPORT_MODULES:
        for node in ast.walk(ast.parse(Path(rel).read_text())):
            if (isinstance(node, ast.Call)
                    and isinstance(node.func, ast.Attribute)
                    and isinstance(node.func.value, ast.Name)
                    and node.func.value.id == 'api'):          # the domain module (was 'services')
                self.fail(f'{rel}:{node.lineno} — api.{node.func.attr}() called '
                          f'directly; go through perform()')

The precision matters: calling api.create_post(...) from a transport fails with a file and line; referencing api.create_post as a registry entry is fine — the AST distinguishes them, which a text-based lint rule cannot. Companion contracts worth the same treatment: the registry-derived effects completeness test; a dependency-direction check (the agent package imports nothing from the domain package, so it stays extractable); and a no-signals rule (object-lifecycle signals hide behaviour from the code path that triggers them and fight fixture loading — prefer explicit creation at explicit seams).

The recipe

1. Domain function: user first, one authorize line (policy check ∩ actor), scoped fetch, return the object with its effects.
2. Registry entry: schemas, serialiser, kind, description — or a written veto.
3. Tests: the service's behaviour, and its exact effects (the completeness contract will demand this by name).

You touched no view plumbing, no API code, no docs, no tool wiring — yet the capability is now callable from the browser, the JSON API, the generated docs, the AI assistant, and any MCP client: identically gated, identically authorized, self-describing. That is the entire trade — a strict shape for the domain layer, in exchange for transports that never grow business logic and interfaces that cannot drift apart.

Part 2 turns this catalogue into a real MCP server, and the OAuth that guards it.

← All posts