"""A use case declared once must behave identically through both doors — the HTTP API and MCP.
One small quote brick, served by a real uvicorn server, called over HTTP and through an actual MCP
client session. Every case renders both replies into one shape and compares them, because the claim
under test is "an agent can do exactly what its person can do through the API, and no more".
"""
import base64
import json
import socket
import threading
import time
from decimal import Decimal
import anyio
import httpx
import pytest
import uvicorn
from mcp import ClientSession
from mcp.client.streamable_http import create_mcp_http_client, streamable_http_client
from erp_platform import auth
from erp_platform.bricks import Answer, Brick, Refused
# ── the brick under test ────────────────────────────────────────────────────────────────────
QUOTES: dict = {}
DEPOSITS: set = set()
def reset() -> None:
QUOTES.clear()
QUOTES.update({
"Q-1": {"owner": "alice@shop.example",
"totals": {1: Decimal("1250.50"), 2: Decimal("1100.00")}, "accepted": None},
"Q-2": {"owner": "bob@shop.example", "totals": {1: Decimal("980.00")}, "accepted": None},
})
def _visible(caller: dict, quote_id: str) -> dict:
q, scope = QUOTES.get(quote_id), auth.owner_scope(caller)
if q is None or (scope is not None and q["owner"] != scope):
raise Refused(404, f"no quote {quote_id}") # 404, never 403: a 403 confirms it exists
return q
QUOTE = Brick("quote", instructions="Quotes a company has sent its customers.")
@QUOTE.use_case(kind="query", route="GET /quotes")
def list_quotes(caller: dict, mine: bool = False) -> dict:
"""List the quotes the caller may see. Managers see every quote; sales see their own."""
scope = auth.owner_scope(caller, mine)
return {"quotes": sorted(q for q, v in QUOTES.items() if scope is None or v["owner"] == scope)}
@QUOTE.use_case(kind="query", route="GET /quotes/{quote_id}")
def get_quote(caller: dict, quote_id: str) -> dict:
"""One quote, if the caller may see it."""
q = _visible(caller, quote_id)
return {"quote_id": quote_id, "owner": q["owner"], "accepted": q["accepted"]}
@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. Accepting the same version again changes nothing."""
q = _visible(caller, quote_id)
if version not in q["totals"]:
raise Refused(422, f"quote {quote_id} has no version {version}")
if q["accepted"] not in (None, version):
raise Refused(409, f"version {q['accepted']} of {quote_id} is already accepted")
q["accepted"] = version
return {"quote_id": quote_id, "accepted_version": version, "total": q["totals"][version]}
@QUOTE.use_case(kind="query", route="GET /period", aliases={"date_from": "from", "date_to": "to"})
def quotes_in_period(caller: dict, date_from: str, date_to: str = "9999-12-31") -> dict:
"""The period the caller asked about, as the use case received it (0.9.7: `from`/`to` on both doors)."""
return {"date_from": date_from, "date_to": date_to}
@QUOTE.use_case(kind="command", route="POST /period-report", idempotent=True,
aliases={"date_from": "from", "date_to": "to"})
def report_period(caller: dict, date_from: str, date_to: str) -> dict:
"""A command taking the period in its body as `from`/`to` (0.9.7)."""
return {"date_from": date_from, "date_to": date_to}
@QUOTE.use_case(kind="command", route="POST /drafts", status=201)
def create_draft(caller: dict, customer: str) -> dict:
"""Create a draft quote (0.9.7: answered 201 Created on HTTP)."""
return {"draft_for": customer}
@QUOTE.use_case(kind="command", route="POST /deposits", status=201)
def expect_deposit(caller: dict, batch_ref: str) -> dict:
"""201 the first time, 200 for the same batch_ref (0.9.8: Answer)."""
seen = batch_ref in DEPOSITS
DEPOSITS.add(batch_ref)
body = {"batch_id": batch_ref}
return Answer(body, status=200) if seen else body
# ── who is calling ──────────────────────────────────────────────────────────────────────────
def as_user(email: str, roles: list, permissions: list) -> dict:
"""The two headers the gateway mints after verifying a token."""
payload = base64.urlsafe_b64encode(json.dumps({"permissions": permissions}).encode()).decode()
return {"X-Dodil-User": json.dumps({"sub": email, "email": email, "app_roles": roles}),
"X-Dodil-User-JWT": f"h.{payload.rstrip('=')}.s"}
ALICE = as_user("alice@shop.example", ["sales"], ["quote:quote:accept"])
BOB = as_user("bob@shop.example", ["sales"], ["quote:quote:accept"])
MANAGER = as_user("mgr@shop.example", ["manager"], ["quote:quote:accept"])
VIEWER = as_user("vic@shop.example", ["manager"], []) # sees all, may accept nothing
# ── the server, and the two doors rendered into one shape ───────────────────────────────────
@pytest.fixture(scope="module")
def base():
with socket.socket() as s:
s.bind(("127.0.0.1", 0))
port = s.getsockname()[1]
server = uvicorn.Server(uvicorn.Config(QUOTE.app(), host="127.0.0.1", port=port,
log_level="critical"))
thread = threading.Thread(target=server.run, daemon=True)
thread.start()
deadline = time.time() + 10
while not server.started and time.time() < deadline:
time.sleep(0.02)
assert server.started
yield f"http://127.0.0.1:{port}"
server.should_exit = True
thread.join(5)
@pytest.fixture(autouse=True)
def _fresh():
reset()
def via_http(base: str, headers: dict, method: str, path: str, body=None, params=None) -> tuple:
r = httpx.request(method, f"{base}{path}", headers=headers, json=body, params=params)
if r.status_code < 300:
return ("ok", r.json())
return ("refused", f"{r.status_code}: {r.json()['detail']}")
async def _mcp(base: str, headers: dict, tool: str, args: dict) -> tuple:
async with create_mcp_http_client(headers=headers) as http:
async with streamable_http_client(f"{base}/mcp", http_client=http) as (read, write):
async with ClientSession(read, write) as session:
await session.initialize()
result = await session.call_tool(tool, args)
text = result.content[0].text if result.content else ""
if result.is_error:
# The MCP 2.x SDK frames every tool error as "Error executing tool <name>: <message>"
# (1.x passed ToolError through bare). The message — status and reason — is ours and must
# match the HTTP door exactly. The framing is the SDK's and is kept: it tells an agent
# WHICH tool refused. Asserted precisely, so a framing change fails here, loudly.
frame = f"Error executing tool {tool}: "
assert text.startswith(frame), f"MCP error framing changed: {text!r}"
return ("refused", text[len(frame):])
return ("ok", json.loads(text))
def via_mcp(base: str, headers: dict, tool: str, args: dict | None = None) -> tuple:
return anyio.run(_mcp, base, headers, tool, args or {})
def both(base, headers, *, tool, method, path, args=None, body=None, params=None) -> tuple:
"""Call one use case through both doors from the same starting state."""
reset()
http = via_http(base, headers, method, path, body=body, params=params)
reset()
mcp = via_mcp(base, headers, tool, args)
return http, mcp
# ── parity ──────────────────────────────────────────────────────────────────────────────────
def test_row_visibility_is_the_same_through_both_doors(base):
for who, expect in ((ALICE, ["Q-1"]), (BOB, ["Q-2"]), (MANAGER, ["Q-1", "Q-2"])):
http, mcp = both(base, who, tool="list_quotes", method="GET", path="/quotes")
assert http == mcp == ("ok", {"quotes": expect})
def test_a_row_out_of_scope_is_404_not_403_through_both(base):
http, mcp = both(base, ALICE, tool="get_quote", method="GET", path="/quotes/Q-2",
args={"quote_id": "Q-2"})
assert http == mcp == ("refused", "404: no quote Q-2")
def test_the_permission_is_enforced_through_both(base):
http, mcp = both(base, VIEWER, tool="accept_quote", method="POST", path="/quotes/Q-1/accept",
args={"quote_id": "Q-1", "version": 1}, body={"version": 1})
assert http == mcp == ("refused", "403: missing permission: quote:quote:accept")
def test_money_leaves_exact_through_both(base):
"""FastAPI's default would send 1250.5 (a float); MCP would send "1250.50"."""
http, mcp = both(base, ALICE, tool="accept_quote", method="POST", path="/quotes/Q-1/accept",
args={"quote_id": "Q-1", "version": 1}, body={"version": 1})
assert http == mcp == ("ok", {"quote_id": "Q-1", "accepted_version": 1, "total": "1250.50"})
def test_a_business_refusal_carries_its_reason_through_both(base):
def accept_other_version(door):
QUOTES["Q-1"]["accepted"] = 2
return door()
reset()
http = accept_other_version(lambda: via_http(base, ALICE, "POST", "/quotes/Q-1/accept",
body={"version": 1}))
reset()
mcp = accept_other_version(lambda: via_mcp(base, ALICE, "accept_quote",
{"quote_id": "Q-1", "version": 1}))
assert http == mcp == ("refused", "409: version 2 of Q-1 is already accepted")
def test_no_identity_is_refused_through_both(base):
"""The API refuses with 401. The agent door refuses at the HTTP level too — an MCP client starts
its sign-in flow on an HTTP 401, and would see a refusal inside a 200 as a broken server."""
http = via_http(base, {}, "GET", "/quotes")
assert http[0] == "refused" and http[1].startswith("401: end-user login required")
r = httpx.post(f"{base}/mcp", json={"jsonrpc": "2.0", "id": 1, "method": "tools/list"},
headers={"Accept": "application/json, text/event-stream"})
assert r.status_code == 401
assert r.json()["detail"].startswith("end-user login required")
# ── an agent signing a person in: RFC 9728, as the MCP authorization spec uses it ───────────
def test_without_a_pool_the_challenge_is_bare_and_no_metadata_is_served(base, monkeypatch):
monkeypatch.delenv("DODIL_USER_POOL", raising=False)
monkeypatch.delenv("AUTH_ISSUER", raising=False)
r = httpx.post(f"{base}/mcp", json={})
assert r.headers["www-authenticate"] == "Bearer"
assert httpx.get(f"{base}/.well-known/oauth-protected-resource/mcp").status_code == 404
def test_the_401_points_an_agent_at_the_pool_it_signs_in_to(base, monkeypatch):
monkeypatch.setenv("DODIL_USER_POOL", "acme/staff")
monkeypatch.delenv("AUTH_ISSUER", raising=False)
monkeypatch.delenv("PUBLIC_BASE_URL", raising=False)
gw = {"X-Forwarded-Proto": "https", "X-Forwarded-Host": "crm.acme.example"}
r = httpx.post(f"{base}/mcp", json={}, headers=gw)
assert r.status_code == 401
assert r.headers["www-authenticate"] == ('Bearer resource_metadata="https://crm.acme.example'
'/.well-known/oauth-protected-resource/mcp"')
for path in ("/.well-known/oauth-protected-resource/mcp", "/.well-known/oauth-protected-resource"):
meta = httpx.get(f"{base}{path}", headers=gw).json()
assert meta == {"resource": "https://crm.acme.example/mcp",
"authorization_servers": ["https://appid.dodil.io/acme/staff"],
"bearer_methods_supported": ["header"], "resource_name": "quote"}
def test_the_metadata_is_not_in_the_api_spec(base):
assert not any("well-known" in p for p in httpx.get(f"{base}/openapi.json").json()["paths"])
def test_mcp_answers_with_and_without_the_trailing_slash(base):
"""No 307: an MCP client is not obliged to follow a redirect on a POST."""
async def names(url):
async with create_mcp_http_client(headers=ALICE) as http:
async with streamable_http_client(url, http_client=http) as (read, write):
async with ClientSession(read, write) as session:
await session.initialize()
return sorted(t.name for t in (await session.list_tools()).tools)
assert anyio.run(names, f"{base}/mcp") == anyio.run(names, f"{base}/mcp/")
assert httpx.post(f"{base}/mcp/", json={}, headers=ALICE).status_code != 307
# ── an agent never sees another caller's rows ───────────────────────────────────────────────
def test_concurrent_agents_each_see_only_their_own_rows(base):
"""Many interleaved MCP calls from two people. Identity is read per request from that
request's headers — if it ever leaked across calls, one of these would see the other's rows."""
seen: list = []
async def one(headers: dict, expect: list) -> None:
verdict = await _mcp(base, headers, "list_quotes", {})
seen.append(verdict == ("ok", {"quotes": expect}))
async def storm() -> None:
async with anyio.create_task_group() as tg:
for i in range(12):
tg.start_soon(one, ALICE if i % 2 else BOB, ["Q-1"] if i % 2 else ["Q-2"])
anyio.run(storm)
assert len(seen) == 12 and all(seen)
# ── what each surface declares ──────────────────────────────────────────────────────────────
def test_every_use_case_is_both_a_route_and_a_tool(base):
async def tools() -> list:
async with create_mcp_http_client(headers=ALICE) as http:
async with streamable_http_client(f"{base}/mcp", http_client=http) as (read, write):
async with ClientSession(read, write) as session:
await session.initialize()
return (await session.list_tools()).tools
listed = {t.name: t for t in anyio.run(tools)}
paths = httpx.get(f"{base}/openapi.json").json()["paths"]
# A use case counts as routed only if its path+method exists AND carries the seam declaration.
routed = {u.name for u in QUOTE.use_cases
if "x-dodil-seam" in paths.get(u.path, {}).get(u.method.lower(), {})}
assert set(listed) == {u.name for u in QUOTE.use_cases} == routed
assert listed["list_quotes"].annotations.read_only_hint is True
assert listed["accept_quote"].annotations.read_only_hint is False
assert listed["accept_quote"].annotations.idempotent_hint is True
assert listed["accept_quote"].annotations.destructive_hint is False
assert "Accept one sent version" in listed["accept_quote"].description
def test_an_aliased_input_is_from_and_to_through_both_doors(base):
"""0.9.7: a spec's `from`/`to` (Python keywords) served from the use case's `date_from`/`date_to`."""
want = ("ok", {"date_from": "2026-01-01", "date_to": "2026-03-31"})
http, mcp = both(base, ALICE, tool="quotes_in_period", method="GET", path="/period",
args={"from": "2026-01-01", "to": "2026-03-31"}, params={"from": "2026-01-01", "to": "2026-03-31"})
assert http == mcp == want
http, mcp = both(base, ALICE, tool="report_period", method="POST", path="/period-report",
args={"from": "2026-01-01", "to": "2026-03-31"}, body={"from": "2026-01-01", "to": "2026-03-31"})
assert http == mcp == want
# The Python name is not a second spelling: the route's `from` is required, and so is the tool's.
r = httpx.get(f"{base}/period", headers=ALICE, params={"date_from": "2026-01-01"})
assert r.status_code == 422
assert via_mcp(base, ALICE, "quotes_in_period", {"date_from": "2026-01-01"})[0] == "refused"
async def tools() -> list:
async with create_mcp_http_client(headers=ALICE) as http_:
async with streamable_http_client(f"{base}/mcp", http_client=http_) as (read, write):
async with ClientSession(read, write) as session:
await session.initialize()
return (await session.list_tools()).tools
schema = {t.name: t.input_schema for t in anyio.run(tools)}["quotes_in_period"]
assert set(schema["properties"]) == {"from", "to"} and schema["required"] == ["from"]
spec = httpx.get(f"{base}/openapi.json").json()["paths"]
assert {p["name"] for p in spec["/period"]["get"]["parameters"]} == {"from", "to"}
body_ref = spec["/period-report"]["post"]["requestBody"]["content"]["application/json"]["schema"]["$ref"]
body = httpx.get(f"{base}/openapi.json").json()["components"]["schemas"][body_ref.rsplit("/", 1)[1]]
assert set(body["properties"]) == {"from", "to"}
def test_a_use_case_declares_its_success_status_and_mcp_is_unchanged(base):
r = httpx.post(f"{base}/drafts", headers=ALICE, json={"customer": "Acme"}, params=None)
assert (r.status_code, r.json()) == (201, {"draft_for": "Acme"})
assert via_mcp(base, ALICE, "create_draft", {"customer": "Acme"}) == ("ok", {"draft_for": "Acme"})
op = httpx.get(f"{base}/openapi.json").json()["paths"]["/drafts"]["post"]
assert "201" in op["responses"] and "200" not in op["responses"]
assert httpx.post(f"{base}/drafts", headers=ALICE, json={}).status_code == 422 # a refusal keeps its status
for bad in (204, 302, 404):
with pytest.raises(ValueError, match="success status"):
Brick("x").use_case(kind="command", route="POST /x", status=bad)(create_draft)
def test_a_use_case_answers_200_for_a_repeat_it_recognises(base):
first = httpx.post(f"{base}/deposits", headers=ALICE, json={"batch_ref": "dep:7"})
again = httpx.post(f"{base}/deposits", headers=ALICE, json={"batch_ref": "dep:7"})
assert (first.status_code, again.status_code) == (201, 200)
assert first.json() == again.json() == {"batch_id": "dep:7"}
assert via_mcp(base, ALICE, "expect_deposit", {"batch_ref": "dep:7"}) == ("ok", {"batch_id": "dep:7"})
for bad in (204, 302, 404):
with pytest.raises(ValueError, match="2xx with a body"):
Answer({}, status=bad)
@pytest.mark.parametrize("aliases, error", [
({"nope": "from"}, "not an input"), ({"y": "from"}, "path parameter"), ({"a": "b"}, "cannot be served"),
({"a": "from", "b": "from"}, "share one alias")])
def test_bad_aliases_are_refused(aliases, error):
def f(caller, y: str, a: str = "", b: str = ""):
"""d"""
with pytest.raises((TypeError, ValueError), match=error):
Brick("x").use_case(kind="query", route="GET /x/{y}", aliases=aliases)(f)
def test_openapi_carries_the_seam_declaration(base):
op = httpx.get(f"{base}/openapi.json").json()["paths"]["/quotes/{quote_id}/accept"]["post"]
assert op["x-dodil-seam"] == {"kind": "command", "permission": "quote:quote:accept",
"idempotent": True, "destructive": False}
# ── mistakes caught at declaration, not in production ───────────────────────────────────────
@pytest.mark.parametrize("kwargs, fn_src, error", [
({"kind": "query", "route": "POST /x"}, "def f(caller): 'd'", "must be GET"),
({"kind": "command", "route": "GET /x"}, "def f(caller): 'd'", "must not be GET"),
({"kind": "query", "route": "GET /x", "destructive": True}, "def f(caller): 'd'",
"cannot be destructive"),
({"kind": "query", "route": "GET /x"}, "def f(who): 'd'", "first parameter must be `caller`"),
({"kind": "query", "route": "GET /x"}, "def f(caller): pass", "needs a docstring"),
({"kind": "query", "route": "GET /x/{y}"}, "def f(caller): 'd'", "names ['y']"),
])
def test_bad_declarations_are_refused(kwargs, fn_src, error):
scope: dict = {}
exec(fn_src, scope)
with pytest.raises((TypeError, ValueError), match=error.replace("[", r"\[").replace("]", r"\]")):
Brick("t").use_case(**kwargs)(scope["f"])
def test_the_same_app_can_be_started_more_than_once():
"""The MCP session manager runs once per instance; the app must build a fresh one per start."""
app = QUOTE.app()
async def start_and_list():
async with app.router.lifespan_context(app):
import httpx2
async with httpx2.AsyncClient(transport=httpx2.ASGITransport(app=app), headers=ALICE,
base_url="http://brick") as http:
async with streamable_http_client("http://brick/mcp", http_client=http) as (r, w):
async with ClientSession(r, w) as session:
await session.initialize()
return len((await session.list_tools()).tools)
assert anyio.run(start_and_list) == anyio.run(start_and_list) == len(QUOTE.use_cases)
# ── reachable behind the gateway, where Host is the customer's domain ──────────────────────
def _initialize(base, headers):
return httpx.post(f"{base}/mcp", headers={**headers, "Accept": "application/json, text/event-stream"},
json={"jsonrpc": "2.0", "id": 1, "method": "initialize",
"params": {"protocolVersion": "2025-06-18", "capabilities": {},
"clientInfo": {"name": "t", "version": "1"}}})
def test_a_customer_domain_is_not_refused_421(base, monkeypatch):
"""The SDK default accepts only localhost; behind the gateway that refused every agent call."""
r = _initialize(base, {**ALICE, "Host": "crm.acme.example"})
assert r.status_code == 200, r.text
def test_with_a_public_address_other_hosts_are_refused():
import os
from erp_platform.bricks import transport_security
os.environ["PUBLIC_BASE_URL"] = "https://crm.acme.example"
try:
s = transport_security()
finally:
del os.environ["PUBLIC_BASE_URL"]
assert s.enable_dns_rebinding_protection
assert "crm.acme.example" in s.allowed_hosts and "evil.example" not in s.allowed_hosts
assert s.allowed_origins[0] == "https://crm.acme.example"
def test_healthz_answers_without_identity(base):
r = httpx.get(f"{base}/healthz")
assert r.status_code == 200 and r.json()["ok"] is True
def test_a_use_case_is_retried_through_the_door_when_the_database_says_not_now(base, monkeypatch):
"""Through a real route: the use case fails once with 57P03, then answers — the caller sees 200."""
import sys
from sqlalchemy.exc import OperationalError
class Pg(Exception):
sqlstate = "57P03"
fails = {"left": 1}
module = sys.modules[__name__]
real = module._visible
def flaky_visible(caller, quote_id):
if fails["left"]:
fails["left"] -= 1
raise OperationalError("SELECT", {}, Pg("cannot connect now"))
return real(caller, quote_id)
monkeypatch.setattr(module, "_visible", flaky_visible)
status, body = via_http(base, ALICE, "GET", "/quotes/Q-1")
assert status == "ok" and body["quote_id"] == "Q-1" and fails["left"] == 0