Build mizan-fastapi MVP — HTTP RPC + context bundling

The Blazr-critical surface for FastAPI. Forms, Channels, Shapes, SSR,
and MWT are out of scope (Ryth's call: defer until Blazr exercises
them; FastAPI projects use native equivalents anyway).

What ships:
- POST /api/mizan/call/      RPC dispatch with Pydantic input validation
- GET  /api/mizan/ctx/{name}/ bundled context fetch (all functions in
                              the named context, parallel-evaluated, single
                              JSON response)
- JSON-body invalidation transport (the 'invalidate' field on mutation
  responses, with auto-scoping when mutation arg names match context params)
- Auth check infrastructure expecting request.state.user populated by
  FastAPI middleware/deps (matches FastAPI idioms)
- Cache-Control: no-store on all responses

Built on existing mizan-core: registry (function lookup, context groups,
invalidation metadata), client.function (the @client decorator + ServerFunction
+ _FunctionWrapper). No code copied or duplicated from mizan-django — the
shared substrate is genuinely shared.

Package layout:
  backends/mizan-fastapi/
    pyproject.toml         distribution=mizan-fastapi, module=mizan_fastapi
    src/mizan_fastapi/
      executor.py          dispatch + auth + invalidation
      router.py            FastAPI APIRouter with the two endpoints
    tests/test_dispatch.py 11 e2e tests against TestClient

Test fixture establishes the registration pattern: explicit
register(fn_class, "name") after each @client. mizan-fastapi doesn't
ship discovery — apps register their functions explicitly. (mizan-django
keeps its DjangoAppVisitor discovery; FastAPI's lack of an app system
makes auto-discovery less natural.)

Makefile: install + test targets now include mizan-fastapi alongside
the other packages. New test-core / test-fastapi targets added for
symmetry.

Verified:
- mizan-core: 15/15
- mizan-django: 348 pass, 21 skip, 0 fail
- mizan-fastapi: 11/11
- mizan-ts edge-compat: 34/34

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Makefile

@@ -1,22 +1,31 @@
-.PHONY: install test test-django test-react test-integration docker-up docker-down clean
+.PHONY: install test test-core test-django test-fastapi test-react test-integration docker-up docker-down clean
 
+CORE = cores/mizan-python
 DJANGO = backends/mizan-django
+FASTAPI = backends/mizan-fastapi
 REACT = frontends/mizan-react
 
 # ─── Setup ───────────────────────────────────────────────────────────────────
 
 install:
-	cd cores/mizan-python && uv pip install -e .
+	cd $(CORE) && uv pip install -e .
 	cd $(DJANGO) && uv pip install -e ".[dev,channels]"
+	cd $(FASTAPI) && uv pip install -e ".[dev]"
 	cd $(REACT) && npm install
 
 # ─── Unit Tests ──────────────────────────────────────────────────────────────
 
-test: test-django test-react
+test: test-core test-django test-fastapi test-react
+
+test-core:
+	cd $(CORE) && uv run --extra dev pytest
 
 test-django:
 	cd $(DJANGO) && uv run pytest
 
+test-fastapi:
+	cd $(FASTAPI) && uv run pytest
+
 test-react:
 	cd $(REACT) && npm test
 

backends/mizan-fastapi/pyproject.toml

@@ -0,0 +1,32 @@
+[project]
+name = "mizan-fastapi"
+version = "0.1.0"
+description = "Mizan FastAPI backend adapter — HTTP RPC dispatch + context bundling, built on mizan-core."
+requires-python = ">=3.10"
+dependencies = [
+    "mizan-core",
+    "fastapi>=0.110",
+    "pydantic>=2.0",
+]
+
+[project.optional-dependencies]
+dev = [
+    "pytest>=8.0",
+    "httpx>=0.27",
+]
+
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"
+
+[tool.hatch.build.targets.wheel]
+packages = ["src/mizan_fastapi"]
+
+[tool.uv.sources]
+mizan-core = { path = "../../cores/mizan-python", editable = true }
+
+[tool.pytest.ini_options]
+pythonpath = ["src"]
+testpaths = ["tests"]
+python_classes = ["*Tests", "*Test", "Test*"]
+python_functions = ["test_*"]

backends/mizan-fastapi/src/mizan_fastapi/__init__.py

@@ -0,0 +1,30 @@
+"""
+mizan-fastapi — FastAPI backend adapter for the Mizan protocol.
+
+Provides HTTP RPC dispatch and context bundling on top of mizan-core's
+function registry. Channels, Forms, Shapes, SSR are out of scope for
+the FastAPI adapter — FastAPI projects use native equivalents (WebSocket,
+Pydantic models, ORM-of-choice, server-side rendering frameworks).
+
+Usage:
+    from fastapi import FastAPI
+    from mizan_fastapi import router as mizan_router
+
+    app = FastAPI()
+    app.include_router(mizan_router, prefix="/api/mizan")
+
+    # Register your @client-decorated functions
+    from mizan_core.client.function import client
+    from .my_functions import *  # noqa
+"""
+
+from .router import router
+from .executor import execute_function, ErrorCode, FunctionError, FunctionResult
+
+__all__ = [
+    "router",
+    "execute_function",
+    "ErrorCode",
+    "FunctionError",
+    "FunctionResult",
+]

backends/mizan-fastapi/src/mizan_fastapi/executor.py

@@ -0,0 +1,196 @@
+"""
+RPC dispatch — looks up registered functions by name, validates input
+against the function's Pydantic Input model, executes, and returns the
+result wrapped in a normalized FunctionResult / FunctionError.
+
+Backend-agnostic where possible. The only FastAPI-specific bits are the
+Request type hint (kept loose as Any) and the auth-check mechanism, which
+expects FastAPI to populate request.state.user via dependency injection
+or middleware before dispatch.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from enum import Enum
+from typing import Any
+
+from pydantic import BaseModel, ValidationError
+
+from mizan_core.registry import get_function
+
+
+# ─── Error / Result types ───────────────────────────────────────────────────
+
+
+class ErrorCode(str, Enum):
+    NOT_FOUND = "NOT_FOUND"
+    BAD_REQUEST = "BAD_REQUEST"
+    VALIDATION_ERROR = "VALIDATION_ERROR"
+    UNAUTHORIZED = "UNAUTHORIZED"
+    FORBIDDEN = "FORBIDDEN"
+    NOT_IMPLEMENTED = "NOT_IMPLEMENTED"
+    INTERNAL_ERROR = "INTERNAL_ERROR"
+
+
+@dataclass
+class FunctionError:
+    code: ErrorCode
+    message: str
+    details: dict[str, Any] | None = None
+
+
+@dataclass
+class FunctionResult:
+    data: Any  # serialized return value (dict, list, primitive, or Pydantic model_dump)
+
+
+# ─── Auth ───────────────────────────────────────────────────────────────────
+
+
+def _check_auth(request: Any, auth_requirement: Any) -> FunctionError | None:
+    """
+    Verify the request meets the function's auth requirement.
+
+    The auth value comes from @client(auth=...). FastAPI projects are expected
+    to populate `request.state.user` (or compatible) via middleware. If
+    request has no `.state` or `.state.user`, treats the user as anonymous.
+    """
+    if auth_requirement is None:
+        return None
+
+    user = getattr(getattr(request, "state", None), "user", None)
+    is_authenticated = bool(user) and getattr(user, "is_authenticated", True)
+
+    if auth_requirement is True:
+        if not is_authenticated:
+            return FunctionError(ErrorCode.UNAUTHORIZED, "Authentication required")
+        return None
+
+    if auth_requirement == "staff":
+        if not is_authenticated:
+            return FunctionError(ErrorCode.UNAUTHORIZED, "Authentication required")
+        if not getattr(user, "is_staff", False):
+            return FunctionError(ErrorCode.FORBIDDEN, "Staff access required")
+        return None
+
+    if auth_requirement == "superuser":
+        if not is_authenticated:
+            return FunctionError(ErrorCode.UNAUTHORIZED, "Authentication required")
+        if not getattr(user, "is_superuser", False):
+            return FunctionError(ErrorCode.FORBIDDEN, "Superuser access required")
+        return None
+
+    if callable(auth_requirement):
+        if not auth_requirement(request):
+            return FunctionError(ErrorCode.FORBIDDEN, "Permission denied")
+        return None
+
+    return FunctionError(
+        ErrorCode.INTERNAL_ERROR,
+        f"Unknown auth requirement: {auth_requirement!r}",
+    )
+
+
+# ─── Dispatch ───────────────────────────────────────────────────────────────
+
+
+def execute_function(
+    request: Any,
+    fn_name: str,
+    input_data: dict[str, Any] | None = None,
+) -> FunctionResult | FunctionError:
+    """
+    Look up a registered function by name, validate input, execute, return result.
+    """
+    view_class = get_function(fn_name)
+    if view_class is None:
+        return FunctionError(ErrorCode.NOT_FOUND, "Function not found")
+
+    meta = getattr(view_class, "_meta", {})
+
+    if meta.get("private"):
+        return FunctionError(ErrorCode.FORBIDDEN, "Function is not client-callable")
+
+    auth_error = _check_auth(request, meta.get("auth"))
+    if auth_error is not None:
+        return auth_error
+
+    view = view_class(request)
+    input_cls = view.Input
+
+    # Pydantic input validation
+    has_fields = bool(getattr(input_cls, "model_fields", None)) if input_cls else False
+
+    if input_data is not None and has_fields:
+        if not isinstance(input_data, dict):
+            return FunctionError(
+                ErrorCode.BAD_REQUEST,
+                f"Input must be an object, got {type(input_data).__name__}",
+            )
+        try:
+            validated_input = input_cls(**input_data)
+        except ValidationError as e:
+            return FunctionError(
+                ErrorCode.VALIDATION_ERROR,
+                "Input validation failed",
+                details={"errors": e.errors()},
+            )
+    elif has_fields:
+        # Function expects input but none provided
+        required = input_cls.model_json_schema().get("required", [])
+        if required:
+            return FunctionError(
+                ErrorCode.VALIDATION_ERROR,
+                "Input validation failed",
+                details={"fields": {field: ["Field required"] for field in required}},
+            )
+        validated_input = input_cls()
+    else:
+        validated_input = None
+
+    # Execute. The wrapper's call(input) always takes the input arg,
+    # passing None when the function has no fields.
+    try:
+        result = view.call(validated_input)
+    except NotImplementedError as e:
+        return FunctionError(ErrorCode.NOT_IMPLEMENTED, str(e) or "Not implemented")
+    except Exception as e:
+        return FunctionError(ErrorCode.INTERNAL_ERROR, str(e))
+
+    # Serialize Pydantic models to plain dicts
+    if isinstance(result, BaseModel):
+        return FunctionResult(data=result.model_dump(mode="json"))
+    return FunctionResult(data=result)
+
+
+# ─── Invalidation ───────────────────────────────────────────────────────────
+
+
+def compute_invalidation(view_class: Any, input_data: dict[str, Any] | None) -> list[Any]:
+    """
+    Build the invalidate list for a mutation response from the function's
+    @client(affects=...) metadata. Auto-scopes to params when the mutation's
+    arg names overlap with a context's params.
+    """
+    meta = getattr(view_class, "_meta", {})
+    affects = meta.get("affects")
+    if not affects:
+        return []
+
+    out: list[Any] = []
+    for target in affects:
+        target_type = target.get("type")
+        target_name = target.get("name")
+        if target_type == "context":
+            scope_params = target.get("params") or {}
+            if scope_params and input_data:
+                # Auto-scope: include matching param values
+                matched = {k: input_data[k] for k in scope_params if k in input_data}
+                if matched:
+                    out.append({"context": target_name, "params": matched})
+                    continue
+            out.append(target_name)
+        elif target_type == "function":
+            out.append({"function": target_name})
+    return out

backends/mizan-fastapi/src/mizan_fastapi/router.py

@@ -0,0 +1,118 @@
+"""
+FastAPI router exposing Mizan's HTTP endpoints:
+
+    POST /call/                — RPC dispatch
+    GET  /ctx/{context_name}/  — bundled context fetch
+
+Mount under /api/mizan in your FastAPI app:
+
+    from fastapi import FastAPI
+    from mizan_fastapi import router as mizan_router
+
+    app = FastAPI()
+    app.include_router(mizan_router, prefix="/api/mizan")
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from fastapi import APIRouter, Request
+from fastapi.responses import JSONResponse
+
+from mizan_core.registry import get_context_groups, get_function
+
+from .executor import (
+    ErrorCode,
+    FunctionError,
+    FunctionResult,
+    compute_invalidation,
+    execute_function,
+)
+
+
+router = APIRouter()
+
+
+_HTTP_STATUS_FOR_ERROR = {
+    ErrorCode.NOT_FOUND: 404,
+    ErrorCode.BAD_REQUEST: 400,
+    ErrorCode.VALIDATION_ERROR: 422,
+    ErrorCode.UNAUTHORIZED: 401,
+    ErrorCode.FORBIDDEN: 403,
+    ErrorCode.NOT_IMPLEMENTED: 501,
+    ErrorCode.INTERNAL_ERROR: 500,
+}
+
+
+def _error_response(err: FunctionError) -> JSONResponse:
+    body: dict[str, Any] = {
+        "error": {
+            "code": err.code.value,
+            "message": err.message,
+        }
+    }
+    if err.details:
+        body["error"]["details"] = err.details
+    return JSONResponse(
+        body,
+        status_code=_HTTP_STATUS_FOR_ERROR.get(err.code, 500),
+        headers={"Cache-Control": "no-store"},
+    )
+
+
+@router.post("/call/")
+async def function_call(request: Request) -> JSONResponse:
+    """RPC dispatch — `{"fn": "name", "args": {...}}` → `{"result": ..., "invalidate": [...]}`."""
+    try:
+        body = await request.json()
+    except Exception:
+        return _error_response(
+            FunctionError(ErrorCode.BAD_REQUEST, "Invalid JSON body")
+        )
+
+    if not isinstance(body, dict):
+        return _error_response(
+            FunctionError(ErrorCode.BAD_REQUEST, "Body must be an object")
+        )
+
+    fn_name = body.get("fn")
+    if not fn_name or not isinstance(fn_name, str):
+        return _error_response(
+            FunctionError(ErrorCode.BAD_REQUEST, "Missing or invalid 'fn' field")
+        )
+
+    args = body.get("args", {})
+
+    outcome = execute_function(request, fn_name, args)
+    if isinstance(outcome, FunctionError):
+        return _error_response(outcome)
+
+    view_class = get_function(fn_name)
+    invalidate = compute_invalidation(view_class, args)
+
+    return JSONResponse(
+        {"result": outcome.data, "invalidate": invalidate},
+        headers={"Cache-Control": "no-store"},
+    )
+
+
+@router.get("/ctx/{context_name}/")
+async def context_fetch(context_name: str, request: Request) -> JSONResponse:
+    """Bundled context fetch — returns `{function_name: result, ...}` for every function in the named context."""
+    groups = get_context_groups()
+    fn_names = groups.get(context_name)
+    if not fn_names:
+        return _error_response(
+            FunctionError(ErrorCode.NOT_FOUND, f"Context '{context_name}' not found")
+        )
+
+    params = dict(request.query_params)
+    result: dict[str, Any] = {}
+    for fn_name in fn_names:
+        outcome = execute_function(request, fn_name, params)
+        if isinstance(outcome, FunctionError):
+            return _error_response(outcome)
+        result[fn_name] = outcome.data
+
+    return JSONResponse(result, headers={"Cache-Control": "no-store"})

backends/mizan-fastapi/tests/test_dispatch.py

@@ -0,0 +1,151 @@
+"""End-to-end dispatch tests against a real FastAPI app + TestClient."""
+
+from __future__ import annotations
+
+import pytest
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+from pydantic import BaseModel
+
+from mizan_core.client.function import client
+from mizan_core.registry import clear_registry, register
+from mizan_fastapi import router as mizan_router
+
+
+# ─── Fixtures ───────────────────────────────────────────────────────────────
+
+
+class EchoOutput(BaseModel):
+    message: str
+
+
+class SumOutput(BaseModel):
+    total: int
+
+
+class UserOutput(BaseModel):
+    email: str
+    authenticated: bool
+
+
+@pytest.fixture
+def app():
+    """Build a fresh FastAPI app + Mizan router with a few @client functions."""
+    clear_registry()
+
+    @client
+    def echo(request, text: str) -> EchoOutput:
+        return EchoOutput(message=f"echo: {text}")
+
+    @client
+    def add(request, a: int, b: int) -> SumOutput:
+        return SumOutput(total=a + b)
+
+    @client(context="user")
+    def current_user(request) -> UserOutput:
+        return UserOutput(email="anon@example.com", authenticated=False)
+
+    @client(context="user")
+    def user_count(request) -> SumOutput:
+        return SumOutput(total=42)
+
+    @client(affects="user")
+    def update_email(request, email: str) -> EchoOutput:
+        return EchoOutput(message=f"updated: {email}")
+
+    register(echo, "echo")
+    register(add, "add")
+    register(current_user, "current_user")
+    register(user_count, "user_count")
+    register(update_email, "update_email")
+
+    fastapi_app = FastAPI()
+    fastapi_app.include_router(mizan_router, prefix="/api/mizan")
+
+    yield fastapi_app
+
+    clear_registry()
+
+
+@pytest.fixture
+def http(app):
+    return TestClient(app)
+
+
+# ─── RPC dispatch ───────────────────────────────────────────────────────────
+
+
+class FunctionCallTests:
+    def test_simple_call_returns_result(self, http):
+        r = http.post("/api/mizan/call/", json={"fn": "echo", "args": {"text": "hi"}})
+        assert r.status_code == 200
+        body = r.json()
+        assert body["result"]["message"] == "echo: hi"
+        assert body["invalidate"] == []
+
+    def test_call_with_typed_input(self, http):
+        r = http.post("/api/mizan/call/", json={"fn": "add", "args": {"a": 2, "b": 3}})
+        assert r.status_code == 200
+        assert r.json()["result"]["total"] == 5
+
+    def test_unknown_function_returns_not_found(self, http):
+        r = http.post("/api/mizan/call/", json={"fn": "ghost"})
+        assert r.status_code == 404
+        assert r.json()["error"]["code"] == "NOT_FOUND"
+
+    def test_validation_error_returns_422(self, http):
+        r = http.post("/api/mizan/call/", json={"fn": "add", "args": {"a": "not-int", "b": 3}})
+        assert r.status_code == 422
+        assert r.json()["error"]["code"] == "VALIDATION_ERROR"
+
+    def test_missing_required_input_returns_validation_error(self, http):
+        r = http.post("/api/mizan/call/", json={"fn": "add", "args": {}})
+        assert r.status_code == 422
+        assert r.json()["error"]["code"] == "VALIDATION_ERROR"
+
+    def test_missing_fn_field_returns_400(self, http):
+        r = http.post("/api/mizan/call/", json={})
+        assert r.status_code == 400
+        assert r.json()["error"]["code"] == "BAD_REQUEST"
+
+    def test_invalid_json_returns_400(self, http):
+        r = http.post("/api/mizan/call/", content=b"not json", headers={"content-type": "application/json"})
+        assert r.status_code == 400
+
+    def test_response_carries_no_store(self, http):
+        r = http.post("/api/mizan/call/", json={"fn": "echo", "args": {"text": "x"}})
+        assert r.headers.get("cache-control") == "no-store"
+
+
+# ─── Context bundling ───────────────────────────────────────────────────────
+
+
+class ContextFetchTests:
+    def test_context_returns_bundled_results(self, http):
+        r = http.get("/api/mizan/ctx/user/")
+        assert r.status_code == 200
+        body = r.json()
+        assert "current_user" in body
+        assert "user_count" in body
+        assert body["current_user"]["email"] == "anon@example.com"
+        assert body["user_count"]["total"] == 42
+
+    def test_unknown_context_returns_not_found(self, http):
+        r = http.get("/api/mizan/ctx/ghost/")
+        assert r.status_code == 404
+        assert r.json()["error"]["code"] == "NOT_FOUND"
+
+
+# ─── Invalidation ───────────────────────────────────────────────────────────
+
+
+class InvalidationTests:
+    def test_mutation_emits_invalidate_list(self, http):
+        r = http.post(
+            "/api/mizan/call/",
+            json={"fn": "update_email", "args": {"email": "new@example.com"}},
+        )
+        assert r.status_code == 200
+        body = r.json()
+        # affects='user' is a context-name string → invalidate list contains 'user'
+        assert "user" in body["invalidate"]