"""bricks.py — declare a use case once; get an HTTP route AND an MCP tool from it.
A brick has two front doors over the same use cases: the HTTP API for apps and people, and MCP for
agents such as Claude and ChatGPT. Written separately they drift — a field added to one, a
permission checked in one — and the promise that an agent can do what its person can do, and no
more, quietly stops being true. So neither door is written by hand. A use case is declared once:
QUOTE = Brick("quote")
@QUOTE.use_case(kind="command", route="POST /quotes/{quote_id}/accept",
permission="quote:quote:accept", idempotent=True)
def accept_quote(caller: dict, quote_id: str, version: int) -> dict:
'''Accept one sent version of a quote, superseding the others.'''
...
and `QUOTE.app()` serves it as `POST /quotes/{quote_id}/accept` AND as the MCP tool
`accept_quote` — the same permission, checked by the same rule, against an identity read by the
same parser, refused with the same reason, returning the same JSON.
Rules a use case follows:
* **The first parameter is `caller`** — the identity both doors hand it. Row visibility is decided
INSIDE the use case from `caller`, which is why an agent cannot see more than its person: there
is no second place for the check to be missing from.
* **Refuse with `Refused(status, reason)`**, never an HTTP exception. The use case does not know
which door it came through; each door renders the refusal in its own shape. A row the caller may
not see is `Refused(404, ...)`, never 403 — a 403 confirms the record exists.
* **A `query` is read-only and is GET; a `command` changes something and is not.** The kind sets
the hints an agent's client shows a person before it acts, and the `x-dodil-seam` extension on
the OpenAPI spec. One declaration, both surfaces.
* **A docstring is required.** It is the description an agent reads to decide whether to call the
tool; a use case without one is invisible to the agents this layer exists for.
"""
import inspect
import json
import logging
import os
import re
import typing
from contextlib import asynccontextmanager
from dataclasses import dataclass, field
from typing import Any, Callable, Optional
import anyio
import pydantic_core
from fastapi import APIRouter, Body, Depends, FastAPI, Header, Path, Query, Request
from fastapi.responses import JSONResponse
from mcp.server.mcpserver import Context, MCPServer
from mcp.server.mcpserver.exceptions import ToolError
from mcp.server.transport_security import TransportSecuritySettings
from mcp.types import ToolAnnotations
from pydantic import Field, create_model
from starlette.exceptions import HTTPException
from starlette.background import BackgroundTask
from starlette.routing import Route
from erp_platform import after, auth, idempotency, repository, requestlog, retry, types
KINDS = ("command", "query")
# 0.8: the use cases that show a brick's process as BPMN 2.0 — the standard one, and brick-hr's richer playbook
# view (0.4.0), which counts as the brick's process view as it stands. `erp-compose check` asks for one.
PROCESS_VIEW = "show_process_bpmn"
PROCESS_VIEWS = (PROCESS_VIEW, "show_playbook_bpmn")
_PATH_PARAM = re.compile(r"{(\w+)}")
_KW = inspect.Parameter.KEYWORD_ONLY
_EMPTY = inspect.Parameter.empty
class Answer:
"""0.9.8: a use case's result with its own success status, for the one answer `status=` cannot give alone.
A create that recognises a repeat by the caller's own reference (a batch_ref, an order number) answers 201
the first time and 200 for the repeat, as the signed specs say:
return Answer(batch.as_dict(), status=200) if existing else batch.as_dict()
Only the HTTP door reads `status`; the MCP tool returns `body` either way. 2xx with a body only (not 204).
A replay of an Idempotency-Key answers the route's declared status, as before."""
__slots__ = ("body", "status")
def __init__(self, body: Any, *, status: int = 200):
if not 200 <= status < 300 or status == 204:
raise ValueError("Answer(status=...) is a 2xx with a body (not 204); refuse with Refused")
self.body, self.status = body, status
class Refused(Exception):
"""A use case declining, carrying the status and the reason the person needs.
0.9.6: a refusal that a caller acts on by rule can name its GATE and the FIX, and both doors carry them:
raise Refused(409, f"{unit} has no e-invoice unit code yet", gate="einvoice_unit_code",
fix="give it (UN/ECE Rec 20 or 21) before a document uses it")
HTTP answers {"detail": "<reason>: <fix>", "gate": ..., "reason": ..., "fix": ...} with that status; MCP's
error reads "<status>: <reason>: <fix>". `fields` adds more keys to the HTTP body (never "detail").
`.reason` stays the whole sentence (the `detail`), as it always was."""
def __init__(self, status: int, reason: str, *, gate: Optional[str] = None, fix: Optional[str] = None,
fields: Optional[dict] = None):
detail = f"{reason}: {fix}" if fix else reason
super().__init__(detail)
self.status, self.reason = status, detail
self.gate, self.fix, self.why, self.fields = gate, fix, reason, dict(fields or {})
if "detail" in self.fields:
raise ValueError("Refused(fields=...) cannot set detail: it is the reason")
def body(self) -> dict:
"""The HTTP door's JSON body."""
out: dict = {"detail": self.reason}
if self.gate is not None or self.fix is not None:
out.update(gate=self.gate, reason=self.why, fix=self.fix)
out.update(self.fields)
return out
def encode(value: Any) -> Any:
"""The ONE encoder both doors use for what a use case returns.
FastAPI's default turns a Decimal into a float, while the MCP side would send a string — so
without this, the same quote totals 1250.5 through one door and "1250.50" through the other.
Money is exact; it leaves as the exact string, through both.
"""
return pydantic_core.to_jsonable_python(value)
def _is_id_key(key: Any, extra: frozenset) -> bool:
return isinstance(key, str) and (key in extra or key == "id" or key.endswith("_id") or key.endswith("_ids"))
def ids_as_text(value: Any, extra: frozenset = frozenset(), _key: Any = None) -> Any:
"""Every integer id in an (encoded) answer as a string: an int under a key ending `_id` (or `_ids`, or
named in `extra`), and each int in a list under such a key. A JavaScript client — the screens, and most
agents' tool runtimes — parses a JSON number into a double, and an id past 2**53 becomes ANOTHER
partner's id (the CRM audit, 2026-10-17: 45665172796086509 read back as …512). A brick opts in with
`Brick(..., ids_as_strings=True)`; its inputs take the digits as text or as a number (types.BigId)."""
if isinstance(value, dict):
return {k: ids_as_text(v, extra, k) for k, v in value.items()}
if isinstance(value, list):
return [ids_as_text(v, extra, _key) for v in value]
if isinstance(value, int) and not isinstance(value, bool) and _is_id_key(_key, extra):
return str(value)
return value
@dataclass(frozen=True)
class UseCase:
fn: Callable[..., Any]
name: str
kind: str
method: str
path: str
permission: Optional[str]
idempotent: bool
destructive: bool
description: str
inputs: tuple # (name, annotation, default) for every parameter after `caller`
path_params: frozenset
read_only: bool = False # a command whose route is POST by contract but changes nothing (a capability read)
if_match: Optional[str] = None # the input an `If-Match` header fills (an optimistic version)
etag: Optional[str] = None # a query: the answer's field sent back as the `ETag` header
tool: bool = True # False: a route only — a seam another brick calls, never an agent's tool
aliases: tuple = () # 0.9.7: ((python name, wire name), ...) — `date_from` served as `from`
status: int = 200 # 0.9.7: the HTTP success status (201 for a create); MCP has none
brick: Any = field(default=None, compare=False, repr=False)
def wire_name(self, name: str) -> str:
"""The name an input has on both doors: its alias (a Python keyword the spec uses, `from`), or itself."""
return dict(self.aliases).get(name, name)
def from_wire(self, args: dict) -> dict:
"""Arguments keyed by wire names, keyed by the use case's own parameter names."""
back = {w: n for n, w in self.aliases}
return {back.get(k, k): v for k, v in args.items()}
def to_wire(self, args: dict) -> dict:
"""Arguments keyed by the use case's parameter names, keyed as both doors take them."""
return {self.wire_name(k): v for k, v in args.items()}
@property
def takes_idempotency_key(self) -> bool:
"""A command that makes something new on every call takes an Idempotency-Key (erp_platform.idempotency)."""
return (self.kind == "command" and not self.idempotent and not self.read_only
and idempotency.ARGUMENT not in {n for n, _, _ in self.inputs})
@property
def annotations(self) -> ToolAnnotations:
"""What an agent's client may show a person before the call runs.
open_world_hint is False: a brick acts on the customer's own records, not the open web.
"""
if self.kind == "query" or self.read_only:
return ToolAnnotations(read_only_hint=True, destructive_hint=False,
idempotent_hint=True, open_world_hint=False)
return ToolAnnotations(read_only_hint=False, destructive_hint=self.destructive,
idempotent_hint=self.idempotent, open_world_hint=False)
@property
def seam(self) -> dict:
"""The `x-dodil-seam` OpenAPI extension — what OpenAPI cannot say on its own."""
return {"kind": self.kind, "permission": self.permission,
"idempotent": self.idempotent, "destructive": self.destructive}
# ── who signs a person in: the user pool, advertised to agents (RFC 9728) ───────────────────
#
# Claude.ai and ChatGPT connect to an MCP server AS A PERSON: they need to know where that person
# signs in. The MCP authorization spec answers with OAuth 2.0 Protected Resource Metadata
# (RFC 9728): an unauthenticated request gets HTTP 401 with
# WWW-Authenticate: Bearer resource_metadata="<base>/.well-known/oauth-protected-resource/mcp"
# and that document names the authorization server — here the brick's dodil-appid pool, whose
# issuer is {APPID_PUBLIC_URL}/{org}/{pool} (dodil-appid src/domain.rs, verified 2026-10-16).
# The client then runs authorization_code + PKCE against the pool and retries with a token, which
# the gateway verifies and turns into the identity headers.
APPID_PUBLIC_URL = os.environ.get("APPID_PUBLIC_URL", "https://appid.dodil.io").rstrip("/")
METADATA_PATH = "/.well-known/oauth-protected-resource"
def pool_issuer() -> Optional[str]:
"""The issuer of the pool people sign in to — `AUTH_ISSUER`, or built from DODIL_USER_POOL."""
explicit = os.environ.get("AUTH_ISSUER", "").strip().rstrip("/")
if explicit:
return explicit
pool = os.environ.get("DODIL_USER_POOL", "").strip().strip("/")
if pool.count("/") != 1:
return None
return f"{APPID_PUBLIC_URL}/{pool}"
def _public_base(scope: dict) -> str:
"""The URL a client used to reach us. PUBLIC_BASE_URL wins; otherwise the gateway's
forwarded headers, which it sets and which a client cannot inject past it."""
explicit = os.environ.get("PUBLIC_BASE_URL", "").strip().rstrip("/")
if explicit:
return explicit
h = {k.decode().lower(): v.decode() for k, v in scope.get("headers", [])}
proto = h.get("x-forwarded-proto", scope.get("scheme", "http")).split(",")[0].strip()
host = h.get("x-forwarded-host", h.get("host", "localhost")).split(",")[0].strip()
return f"{proto}://{host}"
_LOCAL_HOSTS = ["127.0.0.1:*", "localhost:*", "[::1]:*"]
def transport_security() -> TransportSecuritySettings:
"""Which Host headers the MCP endpoint answers — the SDK's DNS-rebinding protection, configured.
Left to the SDK (`host="127.0.0.1"`) the protection is on and accepts ONLY localhost, so behind
the gateway — where Host is the customer's domain — every agent call is refused 421 Misdirected
Request. Found 2026-10-16: the door tests had passed only because they connected to 127.0.0.1.
* Known public address (PUBLIC_BASE_URL and/or MCP_ALLOWED_HOSTS): protection on, those hosts
and their https origins (+ localhost for health checks).
* Local development (DEV_ALLOW_ANON): protection on, localhost only. This is the case DNS
rebinding exists to attack — a browser page reaching a server that trusts the network.
* Deployed with neither: off. Every call must carry identity headers that only the gateway
mints (it strips inbound copies), so a rebound browser request has nothing to act as.
"""
hosts = [h.strip() for h in os.environ.get("MCP_ALLOWED_HOSTS", "").split(",") if h.strip()]
base = os.environ.get("PUBLIC_BASE_URL", "").strip().rstrip("/")
if base:
hosts.append(base.split("://", 1)[-1])
if hosts:
origins = [f"https://{h}" for h in hosts] + [
o.strip() for o in os.environ.get("MCP_ALLOWED_ORIGINS", "").split(",") if o.strip()]
return TransportSecuritySettings(enable_dns_rebinding_protection=True,
allowed_hosts=hosts + _LOCAL_HOSTS,
allowed_origins=origins)
if auth.DEV_ALLOW_ANON:
return TransportSecuritySettings(enable_dns_rebinding_protection=True,
allowed_hosts=_LOCAL_HOSTS,
allowed_origins=["http://127.0.0.1:*", "http://localhost:*"])
return TransportSecuritySettings(enable_dns_rebinding_protection=False)
class _SignInGate:
"""In front of the MCP app: no person → HTTP 401 pointing at the sign-in metadata.
The tool-level 401 stays as a second line, but a client never sees it from here: an MCP client
starts its sign-in flow on an HTTP 401 with this header, and on nothing else — a refusal inside
a successful JSON-RPC reply would leave Claude.ai with a broken connector and no way to log in.
"""
def __init__(self, mcp_app: Any, mcp_path: str):
self.mcp_app, self.mcp_path = mcp_app, mcp_path
async def __call__(self, scope, receive, send):
if scope["type"] == "http":
requestlog.note(door="mcp")
headers = {k.decode(): v.decode() for k, v in scope.get("headers", [])}
try:
auth.identity_from_headers(headers)
except HTTPException as exc:
challenge = "Bearer"
if pool_issuer():
challenge += (f' resource_metadata="{_public_base(scope)}{METADATA_PATH}'
f'{self.mcp_path}"')
body = json.dumps({"detail": exc.detail}).encode()
await send({"type": "http.response.start", "status": exc.status_code,
"headers": [(b"content-type", b"application/json"),
(b"www-authenticate", challenge.encode()),
(b"content-length", str(len(body)).encode())]})
await send({"type": "http.response.body", "body": body})
return
scope = {**scope, "path": self.mcp_path} # /mcp/ and /mcp reach the same app
if self.mcp_app is None:
raise RuntimeError("the MCP app starts with the app's lifespan — serve it with a "
"server that runs lifespan (uvicorn does)")
await self.mcp_app(scope, receive, send)
MAX_BODY_BYTES = int(os.environ.get("ERP_MAX_BODY_BYTES", str(5 * 1024 * 1024)))
class BodyLimit:
"""ASGI middleware: a request body past `ERP_MAX_BODY_BYTES` (5 MiB) is 413 with a reason — through both
doors, before anything parses it. Without it a 12 MB body reached the engine and came back a 500 (the CRM
audit, 2026-10-17). Per field, the use case's own types bound what is stored (erp_platform.types)."""
def __init__(self, app, limit: Optional[int] = None):
self.app, self.limit = app, limit or MAX_BODY_BYTES
def _answer(self) -> tuple:
body = json.dumps({"detail": f"the request body is larger than the {self.limit // 1024} KiB this "
f"service accepts"}).encode()
return ({"type": "http.response.start", "status": 413,
"headers": [(b"content-type", b"application/json"), (b"content-length", str(len(body)).encode())]},
{"type": "http.response.body", "body": body})
async def __call__(self, scope, receive, send):
if scope["type"] != "http":
return await self.app(scope, receive, send)
length = next((v for k, v in scope.get("headers", []) if k.lower() == b"content-length"), b"")
if length.isdigit() and int(length) > self.limit:
for message in self._answer():
await send(message)
return
state = {"bytes": 0, "over": False, "sent": False}
async def counted():
if state["over"]:
return {"type": "http.disconnect"}
message = await receive()
if message["type"] == "http.request":
state["bytes"] += len(message.get("body", b""))
if state["bytes"] > self.limit:
# A chunked body with no length: stop reading, tell the app the client is gone, and answer
# 413 in place of whatever the app makes of that (below).
state["over"] = True
return {"type": "http.disconnect"}
return message
async def answered(message):
if not state["over"]:
return await send(message)
if not state["sent"]:
state["sent"] = True
for m in self._answer():
await send(m)
try:
await self.app(scope, counted, answered)
except Exception:
if not state["over"]:
raise
if not state["sent"]:
state["sent"] = True
for m in self._answer():
await send(m)
def _authorize(uc: UseCase, caller: dict) -> None:
if uc.permission and not auth.may_run(caller, uc.permission):
raise Refused(403, f"missing permission: {uc.permission}")
def _log_retry(uc: UseCase):
def log(attempt: int, kind: str, exc: BaseException) -> None:
# Runs on the use case's worker thread, which sees the request through a copied context — so
# the count reaches the request line, and this warning names the request it belongs to. The
# SQLSTATE is logged, never the exception's message: it can quote the statement's values.
requestlog.count_retry()
logging.getLogger("erp_platform.retry").warning(json.dumps({
"event": "retry", "request_id": requestlog.request_id(), "use_case": uc.name,
"attempt": attempt, "kind": kind, "sqlstate": retry.sqlstate(exc)}))
return log
def _check_text(inputs: dict) -> None:
"""Every string a door received, refused with its field's name if it holds a control character (tab,
newline and carriage return are text). A NUL cannot even be stored — it was a 500 in the CRM audit."""
where = types.control_character(inputs)
if where:
raise Refused(422, f"{where}: control characters are not allowed in text")
async def _execute(uc: UseCase, caller: dict, inputs: dict, idempotency_key: Optional[str] = None) -> Any:
return (await _execute_replayable(uc, caller, inputs, idempotency_key))[0]
def _unwrap(raw: Any) -> tuple:
"""(the result, the status it asks for or None)."""
return (raw.body, raw.status) if isinstance(raw, Answer) else (raw, None)
async def _execute_replayable(uc: UseCase, caller: dict, inputs: dict,
idempotency_key: Optional[str] = None, jobs: Optional[list] = None) -> tuple:
"""Run a use case for one door: (the encoded answer, whether it is a replay of an Idempotency-Key's).
`jobs` collects what the use case registered with erp_platform.after_response; the door runs the ready ones
after answering. Without it (the MCP door) they start on a thread once the answer is decided."""
own_jobs = jobs is None
jobs = [] if jobs is None else jobs
token = after.collecting(jobs)
try:
out = await _execute_collected(uc, caller, inputs, idempotency_key, jobs)
except BaseException:
jobs.clear() # refused, crashed or lost a claim: nothing runs after it
raise
finally:
after.reset(token)
if own_jobs:
after.start(jobs, uc.name)
return out
async def _execute_collected(uc: UseCase, caller: dict, inputs: dict, idempotency_key: Optional[str],
jobs: list) -> tuple:
requestlog.note(use_case=uc.name, sub=caller.get("sub"))
claim, store = None, None
try:
_authorize(uc, caller)
_check_text(inputs)
if idempotency_key is not None and uc.takes_idempotency_key:
store = idempotency.Store(getattr(uc.brick, "idempotency_connection", None))
scope = f"{getattr(uc.brick, 'name', '')}.{uc.name}"
try:
key = idempotency.valid_key(idempotency_key)
verdict, found = await anyio.to_thread.run_sync(lambda: idempotency.begin(
store, str(caller.get("sub") or ""), scope, key,
idempotency.request_hash(scope, encode(inputs))))
except idempotency.KeyRefused as e:
raise Refused(e.status, e.reason) from None
if verdict == "replay":
return found, True, None
claim = found
# Use cases are synchronous (the DataK3 engine is driven through sync SQLAlchemy), so they run
# on a worker thread rather than blocking the event loop both doors share. The WHOLE use case is
# retried on a transient database failure — see erp_platform.retry for which ones, and why a
# non-idempotent command is not retried when its commit may already have landed.
idempotent = uc.kind == "query" or uc.idempotent
attempts = [0]
def attempt():
# 0.9.13: each attempt reads through this request's identity map (erp_platform.repository): an aggregate is
# loaded once per key per caller, a retry starts from an empty map.
attempts[0] += 1
jobs.clear()
with repository.request_scope(caller, fresh=attempts[0] > 1):
return uc.fn(caller, **inputs)
raw, status = _unwrap(await anyio.to_thread.run_sync(
lambda: retry.run(attempt, idempotent=idempotent, on_retry=_log_retry(uc))))
out = encode(raw)
if getattr(uc.brick, "ids_as_strings", False):
out = ids_as_text(out, frozenset(getattr(uc.brick, "id_keys", ())))
if claim is not None:
done, claim = claim, None # the work is done: never released (re-run) from here on
try:
await anyio.to_thread.run_sync(lambda: store.finish(done, out))
except Exception as e: # noqa: BLE001 — the answer stands; the lease ends the claim
# The TYPE and SQLSTATE only, never the message: it quotes the stored answer (requestlog's rule).
logging.getLogger("erp_platform.idempotency").warning(json.dumps(
{"event": "finish_failed", "request_id": requestlog.request_id(), "use_case": uc.name,
"error": type(e).__name__, "sqlstate": retry.sqlstate(e)}))
return out, False, status
except Refused:
requestlog.mark("refused")
raise
except Exception as exc:
# An optimistic version lost (erp_platform.repository): someone saved after this caller loaded.
# Recognised here as well as in save(), because a race can be lost at COMMIT — after save() returned.
conflict = repository.as_conflict(exc)
if conflict is not None:
requestlog.mark("refused")
raise Refused(409, conflict.reason) from None
# Recorded here as well as by the middleware: through the MCP door the SDK turns a crash into
# an error RESULT on an HTTP 200, and the request line would otherwise say "ok".
requestlog.mark("error", exc)
raise
finally:
if claim is not None: # refused or crashed: the key may try again
c = claim
try:
await anyio.to_thread.run_sync(lambda: store.release(c["caller"], c["use_case"],
c["idempotency_key"], c["generation"]))
except Exception as e: # noqa: BLE001 — the lease frees it anyway; the answer stands
logging.getLogger("erp_platform.idempotency").warning(json.dumps(
{"event": "release_failed", "request_id": requestlog.request_id(), "use_case": uc.name,
"error": type(e).__name__, "sqlstate": retry.sqlstate(e)}))
class Brick:
"""A module that owns the tables it writes, served through both doors."""
def __init__(self, name: str, *, version: str = "0.1.0", instructions: str = "",
requires: tuple = (), ids_as_strings: bool = False, id_keys: tuple = (),
machine_permissions: tuple = (), namespace: Optional[str] = None,
ticks: Optional[tuple] = None):
self.name, self.version, self.instructions = name, version, instructions
# 0.9.11: the use cases a clock worker runs on a schedule (ar's post_pending…). None = let the platform
# find them (erp_platform.local.tick_use_cases); () = this brick has none. `serve --local` runs them.
self.ticks = None if ticks is None else tuple(ticks)
# 0.9.8: the permission namespace the module's spec declares (business-partner's is `bp`, contracts'
# is `contract`). Permissions the platform derives (a provided capability's door) use it, not the name.
self.namespace = namespace or name
# 0.9.2: permissions only a service client holds (a seam another brick calls, a scheduler's tick). No
# wildcard grant reaches them — `*:*:*` runs every person command and none of these (auth.has_permission).
self.machine_permissions = tuple(machine_permissions)
auth.declare_machine(self.machine_permissions, f"Brick({name!r}, machine_permissions=…)")
# 0.5: every integer `*_id` in an answer leaves as a string (see ids_as_text). Opt-in, so a brick's
# answers change only when it says so; `id_keys` names id fields not ending in _id ("parents").
self.ids_as_strings, self.id_keys = bool(ids_as_strings), tuple(id_keys)
# Where an Idempotency-Key's claims live (erp_platform.idempotency): a callable returning a connection
# context (engine.begin). None = erp_platform.db's engine; tests point it at SQLite.
self.idempotency_connection: Optional[Callable[[], Any]] = None
self.use_cases: list[UseCase] = []
# Capabilities (erp_platform.capabilities): what this brick asks others to DO, by contract name.
# A CapabilitySeam in the brick's module counts as well; erp-compose check reads both.
self.requires: tuple = tuple(requires)
# Where a provider brick keeps its capability claims and log: a callable returning a connection
# context (engine.begin). None = erp_platform.db's engine; tests point it at SQLite.
self.capability_connection: Optional[Callable[[], Any]] = None
# The public-link door (erp_platform.public): what a link of each grant and subject kind shows and does.
self.public_handlers: list = []
def public(self, grant: str, *, subject_kind: str):
"""Register the GET (render) of a public link of this grant and subject kind; `.act` on the result
registers its POST. Served only by the composition's public app. See erp_platform.public."""
from erp_platform import public
return public.handler(self, grant, subject_kind)
@property
def links(self):
"""Issue, revoke, reissue and read the status of this brick's public links (erp_platform.public)."""
from erp_platform import public
return public.Links(self.name)
@property
def provides(self) -> tuple:
"""The capabilities this brick provides — one per @BRICK.capability declaration."""
return tuple(c for c in (getattr(uc.fn, "__capability__", None) for uc in self.use_cases) if c)
def capability(self, name: str, *, secrets: tuple = (), lease_seconds: float = 120.0, attempts: int = 3):
"""Provide a capability from capabilities.yaml: generates POST /capabilities/<name> and the MCP tool
<name with underscores>, and implements the contract's claim, replay, dry run and action log around
the handler. See erp_platform.capabilities."""
from erp_platform import capabilities
return capabilities.provide(self, name, secrets=secrets, lease_seconds=lease_seconds, attempts=attempts)
@property
def process_view(self) -> Optional[str]:
"""The name of the use case that shows this brick's process as BPMN, or None (erp-compose check)."""
return next((u.name for u in self.use_cases if u.name in PROCESS_VIEWS), None)
def declare_process(self, source, *, permission: Optional[str] = None) -> None:
"""Declare the brick's process — its workflows, in the shape of a module spec's `workflows:` (and its
`actors:`, `id`, `version`) — and serve it as the standard query `show_process_bpmn(process_id?)` at
`GET /process/bpmn`, rendered by erp_platform.bpmn with the stable ids of modules/pattern.md.
`source` is a dict, or a YAML file: a relative path is read beside the module that calls this (keep it
in the package — `process.yaml` next to use_cases.py, listed in package-data). A route, not a tool: every
brick has one, so a composition offers agents ONE `show_process_bpmn(brick, process_id?)` instead
(compose.app). `permission` defaults to any signed-in caller: the process is not a secret among staff."""
import pathlib
from erp_platform import bpmn
if isinstance(source, (str, pathlib.Path)):
path = pathlib.Path(source)
if not path.is_absolute():
caller_file = inspect.stack()[1].filename
path = pathlib.Path(caller_file).parent / path
import yaml
data = yaml.safe_load(path.read_text())
name = str(path)
else:
data, name = dict(source), f"{self.name} process"
spec = bpmn.from_module({"id": self.name, **data}, name) # refused now, at import, if malformed
def show_process_bpmn(caller: dict, process_id: Optional[str] = None) -> dict:
"""This brick's process as BPMN 2.0 (one process per workflow; `process_id` picks one by its entity),
generated from the brick's declared workflows with stable ids (Task_<key>, Flow_<a>__<b>, …) —
the diagram a customer sees and signs (erp-compose signoff record). Read-only."""
from erp_platform.signoff import content_hash
one = spec
if process_id:
keep = [w for w in spec.workflows if w.entity == process_id]
if not keep:
raise Refused(404, f"no process {process_id} (there are: "
f"{', '.join(w.entity for w in spec.workflows)})")
one = bpmn.Spec(spec.process, spec.version, spec.name, keep, spec.actors, spec.source,
spec.external)
xml = bpmn.to_bpmn(one)
return {"brick": self.name, "process_id": process_id, "processes": [w.entity for w in one.workflows],
"bpmn": xml, "content_sha256": content_hash(xml)}
show_process_bpmn.__qualname__ = f"{self.name}.show_process_bpmn"
self.use_case(kind="query", route="GET /process/bpmn", permission=permission,
tool=False)(show_process_bpmn)
def use_case(self, *, kind: str, route: str, permission: Optional[str] = None,
idempotent: bool = False, destructive: bool = False, read_only: bool = False,
if_match: Optional[str] = None, etag: Optional[str] = None, tool: bool = True,
aliases: Optional[dict] = None, status: int = 200):
"""Declare a use case: a route and (unless `tool=False`) an MCP tool.
`status=201` (0.9.7): the route answers that success status (and OpenAPI says so); a refusal keeps its own,
and the MCP tool is unchanged. A 2xx with a body only — 204 has none to carry the answer.
`aliases={"date_from": "from", "date_to": "to"}` (0.9.7): the input `date_from` is `?from=` on the route (or
`"from"` in its body) and `from` in the MCP tool's schema — the name the spec uses, which Python cannot take
as a parameter. The use case still receives `date_from`. A path parameter keeps its name.
`if_match="version"`: an `If-Match: "3"` header on the route fills the input `version` when the body
leaves it out — the optimistic version a screen loaded (erp_platform.repository). `etag="version"` on
a query sends that field of the answer back as `ETag: "3"`. `tool=False` is a route another brick
calls through a seam and no agent should see — its permission is one no person holds."""
if kind not in KINDS:
raise ValueError(f"kind must be one of {KINDS}, not {kind!r}")
method, _, path = route.partition(" ")
method = method.upper()
if kind == "query" and method != "GET":
raise ValueError(f"a query reads and must be GET, not {method}: {route}")
if kind == "command" and method == "GET":
raise ValueError(f"a command changes something and must not be GET: {route}")
if kind == "query" and destructive:
raise ValueError("a query cannot be destructive")
auth.require_concrete(permission, f"{route}: permission")
if not isinstance(status, int) or not 200 <= status < 300 or status == 204:
raise ValueError(f"{route}: a success status is 2xx with a body (200, 201, 202), not {status!r}")
def register(fn: Callable[..., Any]) -> Callable[..., Any]:
params = list(inspect.signature(fn).parameters.values())
if not params or params[0].name != "caller":
raise TypeError(f"{fn.__name__}: the first parameter must be `caller`")
description = inspect.cleandoc(fn.__doc__ or "")
if not description:
raise TypeError(f"{fn.__name__}: a use case needs a docstring — it is what an "
"agent reads to decide whether to call it")
if any(u.name == fn.__name__ for u in self.use_cases):
raise ValueError(f"{fn.__name__}: a use case with this name is already declared")
# include_extras: an Annotated bound (erp_platform.types: a max length, a 64-bit id) is part of the
# input contract — without it both doors would publish and validate the bare type.
hints = typing.get_type_hints(fn, include_extras=True)
inputs = tuple((p.name, hints.get(p.name, Any), p.default) for p in params[1:])
path_params = frozenset(_PATH_PARAM.findall(path))
missing = path_params - {n for n, _, _ in inputs}
if missing:
raise TypeError(f"{fn.__name__}: the route names {sorted(missing)} but the "
"function takes no such parameter")
if if_match and (if_match not in {n for n, _, _ in inputs} or if_match in path_params):
raise TypeError(f"{fn.__name__}: if_match names {if_match!r}, which is not a body or query input")
if etag and kind != "query":
raise ValueError(f"{fn.__name__}: an ETag is sent by a query, not a command")
names = [n for n, _, _ in inputs]
for n, w in (aliases or {}).items():
if n not in names:
raise TypeError(f"{fn.__name__}: aliases names {n!r}, which is not an input")
if n in path_params:
raise TypeError(f"{fn.__name__}: {n!r} is a path parameter; the route names it")
if not isinstance(w, str) or not w or w.startswith("_") or w in names or w == idempotency.ARGUMENT:
raise ValueError(f"{fn.__name__}: {n!r} cannot be served as {w!r}")
if len(set((aliases or {}).values())) != len(aliases or {}):
raise ValueError(f"{fn.__name__}: two inputs share one alias")
self.use_cases.append(UseCase(
fn=fn, name=fn.__name__, kind=kind, method=method, path=path,
permission=permission, idempotent=idempotent, destructive=destructive,
description=description, inputs=inputs, path_params=path_params,
read_only=read_only, if_match=if_match, etag=etag, tool=tool,
aliases=tuple((aliases or {}).items()), status=status, brick=self))
return fn
return register
# ── door 1: HTTP ─────────────────────────────────────────────────────────────────────────
def router(self, hide: frozenset = frozenset()) -> APIRouter:
router = APIRouter()
for uc in self.use_cases:
if uc.name in hide:
continue
router.add_api_route(
uc.path, _endpoint(uc), methods=[uc.method], name=uc.name, status_code=uc.status,
summary=uc.description.splitlines()[0], description=uc.description,
openapi_extra={"x-dodil-seam": uc.seam})
return router
# ── door 2: MCP ──────────────────────────────────────────────────────────────────────────
def mcp(self) -> MCPServer:
server = MCPServer(name=self.name, version=self.version,
instructions=self.instructions or None)
for uc in self.use_cases:
if not uc.tool:
continue
server.add_tool(_tool(uc), name=uc.name, description=_tool_description(uc),
annotations=uc.annotations)
return server
def routes_app(self) -> FastAPI:
"""This brick's routes alone — what a neighbour reaches in process (tier 1) and what a composition
serves under `/<brick>/`. No MCP: a composition has ONE agent door for all its bricks."""
app = FastAPI(title=self.name, version=self.version)
app.include_router(self.router())
# Its own request lines, under the caller's id when a seam reaches it: at tier 1 a neighbour's
# routes run here, not behind the composition's middleware.
requestlog.configure()
app.add_middleware(requestlog.RequestLog)
return app
def app(self) -> FastAPI:
"""Both doors on one app: the routes at their paths, MCP at `/mcp`."""
return serve([self], name=self.name, version=self.version, instructions=self.instructions)
def serve(bricks: list, *, name: str, version: str, instructions: str = "",
prefix_routes: bool = False, hide: frozenset = frozenset()) -> FastAPI:
"""Any number of bricks as ONE app: their routes, ONE `/mcp` carrying every brick's tools, one sign-in
gate and one RFC 9728 document. One brick unprefixed is `Brick.app()`; a composition prefixes each
brick's routes with `/<brick>` so two bricks' paths can never meet.
MCP runs in STATELESS mode: Ignite scales an app to zero between requests, and a session held in one
replica's memory does not survive that. It is routed rather than mounted, so `/mcp` answers directly —
a mount would answer it with a 307 to `/mcp/`, and an MCP client is not obliged to follow a redirect
on a POST.
`hide` is a set of (brick name, use case name) left out of both doors — a composition hides the
capability of a provider it did not choose, whose tool name would otherwise collide with the chosen one.
"""
mcp_path = "/mcp"
gate = _SignInGate(None, mcp_path)
def build_mcp() -> MCPServer:
server = MCPServer(name=name, version=version, instructions=instructions or None)
for b in bricks:
for uc in b.use_cases:
if (b.name, uc.name) in hide or not uc.tool:
continue
server.add_tool(_tool(uc), name=uc.name, description=_tool_description(uc),
annotations=uc.annotations)
return server
@asynccontextmanager
async def lifespan(_app: FastAPI):
# Built here, once per start: the MCP session manager runs only once per instance, so an app
# whose MCP side was built up front could never be started a second time (a test suite, an
# in-process restart). And its lifespan has to be entered by hand — nothing runs it for an app
# served inside another one.
mcp_app = build_mcp().streamable_http_app(
streamable_http_path=mcp_path, stateless_http=True, json_response=True,
transport_security=transport_security())
gate.mcp_app = mcp_app
async with mcp_app.router.lifespan_context(mcp_app):
yield
app = FastAPI(title=name, version=version, lifespan=lifespan)
app.add_middleware(BodyLimit)
# Outside every route and the MCP door: each request gets an id, returns it, and ends in one line.
requestlog.configure()
app.add_middleware(requestlog.RequestLog)
@app.get("/healthz", include_in_schema=False)
def healthz():
"""Liveness for the platform's probe. Needs no identity and touches no database: a probe that
needed a sign-in would 401 forever, and one that needed DataK3 would restart the pod whenever the
data plane blinked. Is the process up and serving — nothing more."""
return {"ok": True, "name": name, "version": version}
for b in bricks:
app.include_router(b.router(frozenset(u for n, u in hide if n == b.name)),
prefix=f"/{b.name}" if prefix_routes else "")
for path in (mcp_path, mcp_path + "/"):
app.router.routes.append(Route(path, endpoint=gate, methods=["GET", "POST", "DELETE"]))
def metadata(request: Request):
issuer = pool_issuer()
if not issuer:
return JSONResponse({"detail": "no user pool configured (set DODIL_USER_POOL)"},
status_code=404)
base = _public_base(request.scope)
return JSONResponse({"resource": f"{base}{mcp_path}",
"authorization_servers": [issuer],
"bearer_methods_supported": ["header"],
"resource_name": name})
# RFC 9728 §3: the path-suffixed form is canonical; the bare form serves clients that look there first.
for path in (METADATA_PATH + mcp_path, METADATA_PATH):
app.add_api_route(path, metadata, methods=["GET"], include_in_schema=False)
return app
def _version_from_if_match(raw: str) -> int:
"""`If-Match: "3"` (or W/"3", or 3) → 3. Anything else is the caller's mistake."""
v = raw.strip()
if v.startswith("W/"):
v = v[2:]
v = v.strip().strip('"')
if not v.isdigit():
raise Refused(422, 'If-Match carries the version you loaded, e.g. If-Match: "3"')
return int(v)
def _endpoint(uc: UseCase) -> Callable[..., Any]:
aliased = dict(uc.aliases)
body_fields = {n: (typing.Annotated[a, Field(alias=aliased[n])] if n in aliased else a, ... if d is _EMPTY else d)
for n, a, d in uc.inputs if n not in uc.path_params}
body_model = (create_model(f"{uc.name}_input", **body_fields)
if uc.method != "GET" and body_fields else None)
async def endpoint(**kw: Any) -> Any:
caller = kw.pop("_caller")
body = kw.pop("_body", None)
if_match = kw.pop("_if_match", None)
key = kw.pop("_idempotency_key", None)
inputs = dict(kw)
if body is not None:
inputs.update({n: getattr(body, n) for n in body_fields})
elif body_model is not None: # a body of optional fields, left out: their defaults
inputs.update({n: f[1] for n, f in body_fields.items() if f[1] is not ...})
try:
if uc.if_match and if_match is not None:
v = _version_from_if_match(if_match)
if inputs.get(uc.if_match) is None:
inputs[uc.if_match] = v
elif int(inputs[uc.if_match]) != v:
raise Refused(422, f"If-Match says version {v} but the body says {inputs[uc.if_match]}")
jobs: list = []
out, replayed, status = await _execute_replayable(uc, caller, inputs, key, jobs)
headers = {}
if replayed:
headers["Idempotent-Replayed"] = "true"
if uc.etag and isinstance(out, dict) and out.get(uc.etag) is not None:
headers["ETag"] = f'"{out[uc.etag]}"'
todo = after.ready(jobs)
background = BackgroundTask(anyio.to_thread.run_sync, after.run, todo, uc.name) if todo else None
return JSONResponse(out, status_code=status or uc.status, headers=headers or None, background=background)
except Refused as refused:
return JSONResponse(refused.body(), status_code=refused.status)
params = [inspect.Parameter("_caller", _KW, default=Depends(auth.current_user))]
# Path()/Query() go INSIDE Annotated: as a parameter's default, FastAPI drops the annotation's own metadata —
# a 64-bit id's validator and string schema (erp_platform.types) would silently vanish from path and query.
for name, annotation, default in uc.inputs:
if name in uc.path_params:
params.append(inspect.Parameter(name, _KW, annotation=typing.Annotated[annotation, Path()]))
elif uc.method == "GET":
params.append(inspect.Parameter(
name, _KW, annotation=typing.Annotated[annotation, Query(alias=aliased.get(name))],
default=inspect.Parameter.empty if default is _EMPTY else default))
if body_model is not None:
# A body whose every field has a default may be left out altogether (a DELETE with an optional version,
# a POST /revise): FastAPI would otherwise answer 422 "body: Field required" to a request that is fine.
optional = all(f[1] is not ... for f in body_fields.values())
params.append(inspect.Parameter("_body", _KW, annotation=Optional[body_model] if optional else body_model,
default=Body(None) if optional else Body()))
if uc.if_match:
params.append(inspect.Parameter(
"_if_match", _KW, annotation=Optional[str],
default=Header(None, alias="If-Match",
description="the version you loaded (the GET's `version`), e.g. \"3\"; "
"a stale one is 409 \"changed since you opened it\"")))
if uc.takes_idempotency_key:
params.append(inspect.Parameter(
"_idempotency_key", _KW, annotation=Optional[str],
default=Header(None, alias=idempotency.HEADER,
description="your own unique string for this attempt: a repeat with the same key "
"returns the first answer instead of doing it twice")))
endpoint.__signature__ = inspect.Signature(params)
endpoint.__name__ = uc.name
return endpoint
def _tool(uc: UseCase) -> Callable[..., Any]:
takes_key = uc.takes_idempotency_key
async def tool(**kw: Any) -> Any:
ctx: Context = kw.pop("ctx")
key = kw.pop(idempotency.ARGUMENT, None) if takes_key else None
try:
caller = auth.identity_from_headers(ctx.headers or {})
return await _execute(uc, caller, uc.from_wire(kw), key)
except HTTPException as exc: # no identity: the shared parser's 401
requestlog.mark("refused")
raise ToolError(f"{exc.status_code}: {exc.detail}") from None
except Refused as refused:
# Same status and reason as the HTTP door's {"detail": ...}. The 2.x SDK presents it
# as "Error executing tool <name>: <status>: <reason>" — the framing is the SDK's and is
# kept, because it tells an agent which tool refused when it called several.
raise ToolError(f"{refused.status}: {refused.reason}") from None
except Exception:
# A crash. The HTTP door answers {"detail": "internal error", "request_id": …}; without this the
# agent read only "Error executing tool <name>" — no status, nothing to quote in a ticket. The
# exception's message stays out: it can quote the statement's values (requestlog's rule).
raise ToolError(f"500: internal error (request id {requestlog.request_id() or 'none'})") from None
# An aliased input keeps its Python name in the signature and carries the alias as a pydantic Field: the MCP
# SDK publishes the alias in the tool's schema, validates by it and hands the argument back under it.
aliased = dict(uc.aliases)
annotations = {n: typing.Annotated[a, Field(alias=aliased[n])] if n in aliased else a for n, a, _ in uc.inputs}
params = [inspect.Parameter(n, _KW, annotation=annotations[n], default=d) for n, _, d in uc.inputs]
if takes_key:
params.append(inspect.Parameter(idempotency.ARGUMENT, _KW, annotation=Optional[str], default=None))
annotations[idempotency.ARGUMENT] = Optional[str]
params.append(inspect.Parameter("ctx", _KW, annotation=Context))
tool.__signature__ = inspect.Signature(params)
tool.__annotations__ = annotations | {"ctx": Context}
tool.__name__ = uc.name
tool.__doc__ = _tool_description(uc)
return tool
def _tool_description(uc: UseCase) -> str:
if not uc.takes_idempotency_key:
return uc.description
return uc.description + ("\n\n`idempotency_key` (optional): your own unique string for this attempt. If the "
"reply is lost, call again with the SAME key and the same arguments: you get the "
"first answer back and nothing is done twice.")