ontolith.interfaces

REST and GraphQL app factories — the two web interfaces with a genuine Python call-signature surface (CLI flags and MCP tool schemas are excluded from this pinned surface by ADR-0019 Decision #1).

ontolith.interfaces.rest

REST interface for Ontolith (SPEC §14.3, ADR-0021).

Exposes read, propose, direct write, retraction (governed, KI-057), proposal review (accept/reject/request_changes/resubmit/assign, KI-078), contradiction listing/flagging/resolution, namespace listing, and principal/token admin over HTTP. Full SPEC §14.3 resource parity (/query offset pagination, GraphQL) is out of scope (KI-022).

Authentication (ADR-0014, reused unchanged): every route requires an Authorization: Bearer <token> header, resolved server-side via the injected AuthProvider — never a caller-asserted principal ID. This includes read routes, a deliberate divergence from the shipped MCP server's unauthenticated read tools (KI-021 tracks closing that gap on the MCP side). Per ADR-0021 §2, authentication is the read-capability check — any successfully-authenticated principal can read any entity, assertion, provenance record, or proposal (including other principals' source/ rationale via GET /proposals) in this namespace; there is no per-principal or per-namespace read scoping in this slice.

FastAPI serves interactive docs (/docs, /redoc, /openapi.json) unauthenticated by default, exposing the API's shape (not its data) to anonymous callers. Pass docs_url=None, redoc_url=None, openapi_url=None to disable them for a given deployment — this factory forwards those three straight to FastAPI(...), defaulting to FastAPI's own (docs-enabled) behavior when omitted.

Usage

from ontolith.identity.token_auth import TokenAuthProvider
from ontolith.interfaces.rest import create_rest_app
app = create_rest_app(kb, TokenAuthProvider(kb.backend))
# uvicorn.run(app) to serve

create_rest_app

create_rest_app(
    kb: Ontology,
    auth_provider: AuthProvider,
    name: str = "ontolith",
    *,
    docs_url: str | None = "/docs",
    redoc_url: str | None = "/redoc",
    openapi_url: str | None = "/openapi.json",
) -> FastAPI

Build and return a FastAPI app bound to the given knowledge base.

Parameters:

Name	Type	Description	Default
kb	Ontology	Ontology instance (knowledge base) to expose	required
auth_provider	AuthProvider	Resolves caller-supplied bearer tokens to Principals (ADR-0014) — e.g. TokenAuthProvider(kb.backend)	required
name	str	API title advertised in the OpenAPI schema	'ontolith'
docs_url	str | None	Swagger UI path, or None to disable it. Unauthenticated like the rest of FastAPI's docs surface (module docstring).	'/docs'
redoc_url	str | None	ReDoc path, or None to disable it.	'/redoc'
openapi_url	str | None	OpenAPI schema JSON path, or None to disable it (also disables docs_url/redoc_url, which depend on it).	'/openapi.json'

Returns:

Type	Description
FastAPI	Configured FastAPI app with the read + propose route slice
FastAPI	(ADR-0021)

Source code in src/ontolith/interfaces/rest.py

def create_rest_app(
    kb: Ontology,
    auth_provider: AuthProvider,
    name: str = "ontolith",
    *,
    docs_url: str | None = "/docs",
    redoc_url: str | None = "/redoc",
    openapi_url: str | None = "/openapi.json",
) -> FastAPI:
    """Build and return a FastAPI app bound to the given knowledge base.

    Args:
        kb: Ontology instance (knowledge base) to expose
        auth_provider: Resolves caller-supplied bearer tokens to Principals
            (ADR-0014) — e.g. ``TokenAuthProvider(kb.backend)``
        name: API title advertised in the OpenAPI schema
        docs_url: Swagger UI path, or None to disable it. Unauthenticated
            like the rest of FastAPI's docs surface (module docstring).
        redoc_url: ReDoc path, or None to disable it.
        openapi_url: OpenAPI schema JSON path, or None to disable it (also
            disables docs_url/redoc_url, which depend on it).

    Returns:
        Configured FastAPI app with the read + propose route slice
        (ADR-0021)
    """
    app = FastAPI(title=name, docs_url=docs_url, redoc_url=redoc_url, openapi_url=openapi_url)

    def _resolve_principal(
        authorization: Annotated[str | None, Header()] = None,
    ) -> Principal:
        """Resolve the Authorization header to a Principal (ADR-0014).

        Raises:
            AuthError: header missing/malformed, or token invalid/revoked
        """
        if authorization is None or not authorization.startswith("Bearer "):
            raise AuthError("Missing or malformed Authorization header")
        token = authorization.removeprefix("Bearer ")
        return auth_provider.resolve(token)

    @app.exception_handler(OntolithError)
    def _handle_ontolith_error(_request: Request, exc: OntolithError) -> JSONResponse:
        """Map every OntolithError subtype to its HTTP status (SPEC §16).

        5xx errors (StorageError/PluginError) log the real message
        server-side but never return it — those messages interpolate raw
        internal exception text (e.g. sqlite3 constraint/transaction-state
        details) that a caller has no use for and shouldn't see.
        """
        status = _STATUS_BY_ERROR_TYPE.get(type(exc), 500)
        if status >= 500:
            # SPEC §18/ADR-0044: structured, correlated log via kb's sink —
            # replaces the ad hoc module logger this call site used to use.
            kb.observability.log(logging.ERROR, exc.message, code=exc.code)
            message = _GENERIC_SERVER_ERROR_MESSAGE
        else:
            message = exc.message
        body = ErrorOut(code=exc.code, message=message, detail=exc.detail)
        return JSONResponse(status_code=status, content=body.model_dump())

    @app.exception_handler(RequestValidationError)
    def _handle_request_validation_error(
        _request: Request, exc: RequestValidationError
    ) -> JSONResponse:
        """Map FastAPI's own request-body/query validation failures onto the
        same envelope as ValidationError (SPEC §16) — malformed input is a
        validation failure regardless of whether Pydantic or domain code
        caught it first."""
        body = ErrorOut(
            code=ValidationError.code,
            message="Request validation failed",
            detail={"errors": exc.errors()},
        )
        return JSONResponse(status_code=400, content=body.model_dump())

    # ------------------------------------------------------------------
    # GET /schema
    # ------------------------------------------------------------------

    @app.get("/schema")
    def get_schema(
        namespace: str = "default",
        _principal: Principal = Depends(_resolve_principal),
    ) -> SchemaOut:
        """Return the schema (concepts, properties, and relations) for a namespace."""
        ir = kb.backend.get_schema(namespace)
        if ir is None:
            return SchemaOut(concepts=[])
        concepts = [
            ConceptOut(
                name=concept_name,
                properties=[
                    PropertyOut(
                        name=prop_name,
                        type=prop_def.value_type,
                        cardinality=prop_def.cardinality,
                        temporality=prop_def.temporality,
                        required=prop_def.required,
                    )
                    for prop_name, prop_def in concept_def.properties.items()
                ],
                relations=[
                    RelationOut(
                        name=rel_name,
                        target_concept=rel_def.target_concept,
                        cardinality=rel_def.cardinality,
                        required=rel_def.required,
                        temporality=rel_def.temporality,
                        inverse=rel_def.inverse,
                    )
                    for rel_name, rel_def in concept_def.relations.items()
                ],
            )
            for concept_name, concept_def in ir.concepts.items()
        ]
        return SchemaOut(namespace=namespace, version=ir.version, concepts=concepts)

    # ------------------------------------------------------------------
    # POST /entities
    # ------------------------------------------------------------------

    @app.post("/entities", status_code=201)
    def create_entity_route(
        body: CreateEntityIn,
        principal: Principal = Depends(_resolve_principal),
    ) -> EntityOut:
        """Create a new entity (KI-082). Propose-tier — same capability
        gate as every other governed write (`Ontology.create_entity`
        rejects `read`-only principals)."""
        entity = kb.create_entity(
            concept=body.concept,
            author=principal.id,
            natural_key=body.natural_key,
        )
        return EntityOut(
            id=entity.id,
            concept=entity.concept,
            namespace=entity.namespace,
            natural_key=entity.natural_key,
            created_at=entity.created_at.isoformat(),
            created_by=entity.created_by,
        )

    # ------------------------------------------------------------------
    # GET /entities/{entity_id}
    # ------------------------------------------------------------------

    @app.get("/entities/{entity_id}")
    def get_entity_route(
        entity_id: str,
        _principal: Principal = Depends(_resolve_principal),
    ) -> EntityDetailOut:
        """Fetch an entity and its currently active assertions."""
        entity = kb.backend.get_entity(entity_id)
        if entity is None:
            raise NotFoundError(f"Entity {entity_id!r} not found")
        assertions = kb.backend.assertions(subject=entity_id, status="active")
        return EntityDetailOut(
            entity=EntityOut(
                id=entity.id,
                concept=entity.concept,
                namespace=entity.namespace,
                natural_key=entity.natural_key,
                created_at=entity.created_at.isoformat(),
                created_by=entity.created_by,
            ),
            assertions=[
                AssertionOut(
                    id=a.id,
                    predicate=a.predicate,
                    value=a.value,
                    value_type=a.value_type,
                    confidence=a.confidence,
                    author=a.author,
                    asserted_at=a.asserted_at.isoformat(),
                )
                for a in assertions
            ],
        )

    # ------------------------------------------------------------------
    # POST /query
    # ------------------------------------------------------------------

    @app.post("/query")
    def query_route(
        body: QueryIn,
        _principal: Principal = Depends(_resolve_principal),
    ) -> QueryOut:
        """Query entities of a concept, optionally filtered/ranked."""
        builder = kb.query(body.concept)
        if body.filters:
            builder = builder.where(**body.filters)
        if body.semantic is not None:
            builder = builder.semantic(body.semantic)
        if body.min_confidence is not None:
            builder = builder.min_confidence(body.min_confidence)
        if body.trust_at_least is not None:
            builder = builder.trust_at_least(body.trust_at_least)
        if body.include_flagged:
            builder = builder.include_flagged()
        if body.include_history:
            builder = builder.include_history()
        if body.limit is not None:
            builder = builder.limit(body.limit)
        entities = builder.all()
        return QueryOut(
            concept=body.concept,
            count=len(entities),
            entities=[
                EntitySummaryOut(
                    id=e.id,
                    concept=e.concept,
                    natural_key=e.natural_key,
                    created_at=e.created_at.isoformat(),
                )
                for e in entities
            ],
        )

    # ------------------------------------------------------------------
    # GET /provenance/{assertion_id}
    # ------------------------------------------------------------------

    @app.get("/provenance/{assertion_id}")
    def provenance_route(
        assertion_id: str,
        _principal: Principal = Depends(_resolve_principal),
    ) -> ProvenanceOut:
        """Return the full provenance record for a single assertion.

        ``superseded_ids`` carries the full predecessor set this assertion
        superseded — ``supersedes`` alone only records the first predecessor
        when one incoming assertion supersedes several concurrently-
        overlapping ones (KI-008).
        """
        prov = kb.provenance(assertion_id)  # raises NotFoundError -> 404 via the error handler
        a = prov.assertion
        return ProvenanceOut(
            id=a.id,
            subject=a.subject,
            predicate=a.predicate,
            value=a.value,
            value_type=a.value_type,
            status=a.status,
            author=a.author,
            confidence=a.confidence,
            source=a.source,
            rationale=a.rationale,
            model=a.model,
            asserted_at=a.asserted_at.isoformat(),
            valid_from=a.valid_from.isoformat() if a.valid_from else None,
            valid_to=a.valid_to.isoformat() if a.valid_to else None,
            proposal_id=a.proposal_id,
            supersedes=a.supersedes,
            superseded_ids=list(prov.superseded_ids),
            review_events=[
                ReviewEventOut(actor=e.actor, type=e.type, detail=e.detail, at=e.at.isoformat())
                for e in prov.review_events
            ],
        )

    # ------------------------------------------------------------------
    # POST /proposals
    # ------------------------------------------------------------------

    @app.post("/proposals", status_code=201)
    def create_proposal_route(
        body: ProposeIn,
        principal: Principal = Depends(_resolve_principal),
    ) -> ProposeOut:
        """Create a proposal to assert a fact or relation. Does NOT write directly.

        Conflict-routing temporality is resolved server-side from the
        active schema (SPEC §10.1). The acting principal is resolved from
        the bearer token (ADR-0014), never taken from the request body.
        """
        has_literal = body.value is not None and body.value_type is not None
        has_ref = body.target is not None
        if has_literal == has_ref:
            raise ValidationError("Provide exactly one of (value and value_type) or target")

        if has_ref:
            assert body.target is not None
            proposal, decision = kb.propose_ref(
                subject=body.subject,
                predicate=body.predicate,
                target=body.target,
                author=principal.id,
                confidence=body.confidence,
                source=body.source,
                rationale=body.rationale,
                acting_as=body.acting_as,
                model=body.model,
                supersedes=body.supersedes,
            )
        else:
            assert body.value is not None and body.value_type is not None
            proposal, decision = kb.propose(
                subject=body.subject,
                predicate=body.predicate,
                value=body.value,
                value_type=body.value_type,
                author=principal.id,
                confidence=body.confidence,
                source=body.source,
                rationale=body.rationale,
                acting_as=body.acting_as,
                model=body.model,
                supersedes=body.supersedes,
            )

        return ProposeOut(
            proposal=ProposalOut(
                id=proposal.id,
                namespace=proposal.namespace,
                author=proposal.author,
                acting_as=proposal.acting_as,
                state=proposal.state,
                created_at=proposal.created_at.isoformat(),
                decided_at=proposal.decided_at.isoformat() if proposal.decided_at else None,
                policy_reason=proposal.policy_reason,
                reviewers=proposal.reviewers,
            ),
            decision=type(decision).__name__,
        )

    # ------------------------------------------------------------------
    # GET /proposals
    # ------------------------------------------------------------------

    @app.get("/proposals")
    def list_proposals_route(
        state: str | None = "require_review",
        _principal: Principal = Depends(_resolve_principal),
    ) -> list[ProposalOut]:
        """List proposals, defaulting to those pending review.

        Pass ``state=pending`` to merge ``require_review`` and
        ``changes_requested`` — see ``Ontology.proposals`` (KI-027) for why
        that isn't the default: a caller iterating the default result and
        calling accept/reject on each entry would break the moment a
        ``changes_requested`` proposal showed up in it.

        Pass ``state=all`` to list proposals in every state — a plain
        empty query string value can't express "no filter" unambiguously,
        so ``"all"`` is used as an explicit sentinel instead (mirrors the
        CLI's ``proposal list --all`` flag, expressed as a query value
        here since REST has no separate boolean-flag convention).

        Raises:
            ValidationError: ``state`` is none of the values above — an
                unrecognized value (a typo, wrong case, or a plausible-
                sounding non-existent state) previously reached
                ``kb.proposals()``'s own ``WHERE state = ?`` unfiltered and
                silently matched zero rows, indistinguishable from "no
                proposals in that state" (KI-077).
        """
        # Structurally after auth, not just textually: `_principal` is a
        # FastAPI `Depends`, resolved before this body ever runs - an
        # unauthenticated caller gets 401 regardless of `state`, never a
        # free pre-auth probe of the accepted-value set (matches MCP's own
        # ontolith.list_contradictions, KI-076 review).
        if state not in (None, *_PROPOSAL_STATES, "pending", "all"):
            raise ValidationError(f"Invalid state: {state!r}")
        effective_state = None if state == "all" else state
        results = kb.proposals(state=effective_state)
        return [
            ProposalOut(
                id=p.id,
                namespace=p.namespace,
                author=p.author,
                acting_as=p.acting_as,
                state=p.state,
                created_at=p.created_at.isoformat(),
                decided_at=p.decided_at.isoformat() if p.decided_at else None,
                policy_reason=p.policy_reason,
                reviewers=p.reviewers,
            )
            for p in results
        ]

    # ------------------------------------------------------------------
    # POST /assertions
    # ------------------------------------------------------------------

    @app.post("/assertions", status_code=201)
    def write_assertion_route(
        body: WriteAssertionIn,
        principal: Principal = Depends(_resolve_principal),
    ) -> AssertionDetailOut:
        """Directly write an assertion, bypassing the proposal/policy pipeline.

        Requires write or admin capability and a non-AI principal
        (Ontology.assert_literal/assert_ref enforce both) — SPEC §10
        conflict routing still applies; this is a governed direct-write
        path, not a way around review for AI principals.
        """
        has_literal = body.value is not None and body.value_type is not None
        has_ref = body.target is not None
        if has_literal == has_ref:
            raise ValidationError("Provide exactly one of (value and value_type) or target")

        if has_ref:
            assert body.target is not None
            assertion = kb.assert_ref(
                subject=body.subject,
                predicate=body.predicate,
                target=body.target,
                author=principal.id,
                confidence=body.confidence,
                source=body.source,
                rationale=body.rationale,
                acting_as=body.acting_as,
                model=body.model,
                valid_from=body.valid_from,
                valid_to=body.valid_to,
                supersedes=body.supersedes,
            )
        else:
            assert body.value is not None and body.value_type is not None
            assertion = kb.assert_literal(
                subject=body.subject,
                predicate=body.predicate,
                value=body.value,
                value_type=body.value_type,
                author=principal.id,
                confidence=body.confidence,
                source=body.source,
                rationale=body.rationale,
                acting_as=body.acting_as,
                model=body.model,
                valid_from=body.valid_from,
                valid_to=body.valid_to,
                supersedes=body.supersedes,
            )

        return AssertionDetailOut(
            id=assertion.id,
            subject=assertion.subject,
            predicate=assertion.predicate,
            value=assertion.value,
            value_type=assertion.value_type,
            status=assertion.status,
            author=assertion.author,
            confidence=assertion.confidence,
            source=assertion.source,
            rationale=assertion.rationale,
            model=assertion.model,
            asserted_at=assertion.asserted_at.isoformat(),
            valid_from=assertion.valid_from.isoformat() if assertion.valid_from else None,
            valid_to=assertion.valid_to.isoformat() if assertion.valid_to else None,
            supersedes=assertion.supersedes,
        )

    # ------------------------------------------------------------------
    # POST /assertions/{assertion_id}/retract
    # ------------------------------------------------------------------

    @app.post("/assertions/{assertion_id}/retract", status_code=201)
    def retract_assertion_route(
        assertion_id: str,
        acting_as: str | None = None,
        principal: Principal = Depends(_resolve_principal),
    ) -> ProposeOut:
        """Propose retraction of an assertion through the governed
        proposal/policy pipeline (SPEC §9). Does NOT delete or write
        directly — same propose/policy/conflict-routing pipeline as POST
        /proposals. When acting_as is set the retraction is made on
        behalf of another principal (delegation, ADR-0003)."""
        proposal, decision = kb.retract(assertion_id, author=principal.id, acting_as=acting_as)
        return ProposeOut(
            proposal=ProposalOut(
                id=proposal.id,
                namespace=proposal.namespace,
                author=proposal.author,
                acting_as=proposal.acting_as,
                state=proposal.state,
                created_at=proposal.created_at.isoformat(),
                decided_at=proposal.decided_at.isoformat() if proposal.decided_at else None,
                policy_reason=proposal.policy_reason,
                reviewers=proposal.reviewers,
            ),
            decision=type(decision).__name__,
        )

    # ------------------------------------------------------------------
    # POST /proposals/{proposal_id}/accept
    # ------------------------------------------------------------------

    @app.post("/proposals/{proposal_id}/accept")
    def accept_proposal_route(
        proposal_id: str,
        principal: Principal = Depends(_resolve_principal),
    ) -> ProposalOut:
        """Accept a pending proposal, replaying its operations. Requires
        review or admin capability; the reviewer must not be the
        proposal's own author or delegate (self-review is blocked)."""
        proposal = kb.accept_proposal(proposal_id, principal.id)
        return ProposalOut(
            id=proposal.id,
            namespace=proposal.namespace,
            author=proposal.author,
            acting_as=proposal.acting_as,
            state=proposal.state,
            created_at=proposal.created_at.isoformat(),
            decided_at=proposal.decided_at.isoformat() if proposal.decided_at else None,
            policy_reason=proposal.policy_reason,
            reviewers=proposal.reviewers,
        )

    # ------------------------------------------------------------------
    # POST /proposals/{proposal_id}/reject
    # ------------------------------------------------------------------

    @app.post("/proposals/{proposal_id}/reject")
    def reject_proposal_route(
        proposal_id: str,
        body: RejectIn,
        principal: Principal = Depends(_resolve_principal),
    ) -> ProposalOut:
        """Reject a pending proposal. Requires review or admin capability;
        same self-review guard as accept."""
        proposal = kb.reject_proposal(proposal_id, principal.id, reason=body.reason)
        return ProposalOut(
            id=proposal.id,
            namespace=proposal.namespace,
            author=proposal.author,
            acting_as=proposal.acting_as,
            state=proposal.state,
            created_at=proposal.created_at.isoformat(),
            decided_at=proposal.decided_at.isoformat() if proposal.decided_at else None,
            policy_reason=proposal.policy_reason,
            reviewers=proposal.reviewers,
        )

    # ------------------------------------------------------------------
    # POST /proposals/{proposal_id}/review
    # ------------------------------------------------------------------

    @app.post("/proposals/{proposal_id}/review")
    def review_proposal_route(
        proposal_id: str,
        body: ReviewIn,
        principal: Principal = Depends(_resolve_principal),
    ) -> ProposalOut:
        """Request changes on a pending proposal (SPEC §9.1/§9.4's third
        under_review outcome, alongside accept/reject). Requires review or
        admin capability; same self-review guard as accept/reject."""
        proposal = kb.request_changes(proposal_id, principal.id, reason=body.reason)
        return ProposalOut(
            id=proposal.id,
            namespace=proposal.namespace,
            author=proposal.author,
            acting_as=proposal.acting_as,
            state=proposal.state,
            created_at=proposal.created_at.isoformat(),
            decided_at=proposal.decided_at.isoformat() if proposal.decided_at else None,
            policy_reason=proposal.policy_reason,
            reviewers=proposal.reviewers,
        )

    # ------------------------------------------------------------------
    # POST /proposals/{proposal_id}/assign
    # ------------------------------------------------------------------

    @app.post("/proposals/{proposal_id}/assign")
    def assign_reviewers_route(
        proposal_id: str,
        body: AssignIn,
        principal: Principal = Depends(_resolve_principal),
    ) -> ProposalOut:
        """Reassign a pending proposal's reviewers (SPEC §9.4's `assign`
        action, KI-078). Requires review or admin capability; same
        self-review guard as accept/reject/review — replaces the reviewer
        list wholesale, including clearing it via an empty list."""
        proposal = kb.assign_reviewers(proposal_id, body.reviewers, principal.id)
        return ProposalOut(
            id=proposal.id,
            namespace=proposal.namespace,
            author=proposal.author,
            acting_as=proposal.acting_as,
            state=proposal.state,
            created_at=proposal.created_at.isoformat(),
            decided_at=proposal.decided_at.isoformat() if proposal.decided_at else None,
            policy_reason=proposal.policy_reason,
            reviewers=proposal.reviewers,
        )

    # ------------------------------------------------------------------
    # POST /proposals/{proposal_id}/resubmit
    # ------------------------------------------------------------------

    @app.post("/proposals/{proposal_id}/resubmit")
    def resubmit_proposal_route(
        proposal_id: str,
        principal: Principal = Depends(_resolve_principal),
    ) -> ProposeOut:
        """Resubmit a proposal after changes were requested (KI-027),
        re-running policy evaluation against the unedited payload. Only the
        proposal's own author or delegate may call this."""
        proposal, decision = kb.resubmit(proposal_id, principal.id)
        return ProposeOut(
            proposal=ProposalOut(
                id=proposal.id,
                namespace=proposal.namespace,
                author=proposal.author,
                acting_as=proposal.acting_as,
                state=proposal.state,
                created_at=proposal.created_at.isoformat(),
                decided_at=proposal.decided_at.isoformat() if proposal.decided_at else None,
                policy_reason=proposal.policy_reason,
                reviewers=proposal.reviewers,
            ),
            decision=type(decision).__name__,
        )

    # ------------------------------------------------------------------
    # GET /contradictions
    # ------------------------------------------------------------------

    @app.get("/contradictions")
    def list_contradictions_route(
        state: str | None = "open",
        _principal: Principal = Depends(_resolve_principal),
    ) -> list[ContradictionOut]:
        """List contradictions, defaulting to open ones.

        Pass ``state=all`` to list contradictions in every state (mirrors
        GET /proposals's ``all`` sentinel — see that route for why).

        Raises:
            ValidationError: ``state`` is none of "open"/"resolved"/"all" —
                an unrecognized value previously reached
                ``kb.contradictions()``'s own ``WHERE state = ?`` unfiltered
                and silently matched zero rows, indistinguishable from "no
                contradictions in that state" (KI-077; same class of bug
                KI-076 fixed for MCP's ``ontolith.list_contradictions``).
        """
        # Structurally after auth, not just textually — see
        # list_proposals_route's identical comment above.
        if state not in (None, *_CONTRADICTION_STATES, "all"):
            raise ValidationError(f"Invalid state: {state!r}")
        effective_state = None if state == "all" else state
        results = kb.contradictions(state=effective_state)
        return [
            ContradictionOut(
                id=c.id,
                namespace=c.namespace,
                subject=c.subject,
                predicate=c.predicate,
                state=c.state,
                member_ids=c.member_ids,
                created_at=c.created_at.isoformat(),
                raised_by=c.raised_by,
                resolved_by=c.resolved_by,
                resolved_at=c.resolved_at.isoformat() if c.resolved_at else None,
                metadata=c.metadata,
            )
            for c in results
        ]

    # ------------------------------------------------------------------
    # POST /contradictions/flag
    # ------------------------------------------------------------------

    @app.post("/contradictions/flag", status_code=201)
    def flag_contradiction_route(
        body: FlagContradictionIn,
        principal: Principal = Depends(_resolve_principal),
    ) -> FlagContradictionOut:
        """Flag two assertions as contradictory, opening or extending a
        contradiction. Requires propose capability or higher (ADR-0008)."""
        contradiction, action = kb.flag_contradiction(
            body.assertion_id_a,
            body.assertion_id_b,
            principal.id,
            rationale=body.rationale,
        )
        return FlagContradictionOut(
            contradiction=ContradictionOut(
                id=contradiction.id,
                namespace=contradiction.namespace,
                subject=contradiction.subject,
                predicate=contradiction.predicate,
                state=contradiction.state,
                member_ids=contradiction.member_ids,
                created_at=contradiction.created_at.isoformat(),
                raised_by=contradiction.raised_by,
                resolved_by=contradiction.resolved_by,
                resolved_at=(
                    contradiction.resolved_at.isoformat() if contradiction.resolved_at else None
                ),
                metadata=contradiction.metadata,
            ),
            action=action,
        )

    # ------------------------------------------------------------------
    # POST /contradictions/{contradiction_id}/resolve
    # ------------------------------------------------------------------

    @app.post("/contradictions/{contradiction_id}/resolve")
    def resolve_contradiction_route(
        contradiction_id: str,
        body: ResolveContradictionIn,
        principal: Principal = Depends(_resolve_principal),
    ) -> ContradictionOut:
        """Resolve an open contradiction by selecting a winning assertion.
        Requires review or admin capability."""
        contradiction = kb.resolve_contradiction(
            contradiction_id, body.winner_assertion_id, principal.id
        )
        return ContradictionOut(
            id=contradiction.id,
            namespace=contradiction.namespace,
            subject=contradiction.subject,
            predicate=contradiction.predicate,
            state=contradiction.state,
            member_ids=contradiction.member_ids,
            created_at=contradiction.created_at.isoformat(),
            raised_by=contradiction.raised_by,
            resolved_by=contradiction.resolved_by,
            resolved_at=(
                contradiction.resolved_at.isoformat() if contradiction.resolved_at else None
            ),
            metadata=contradiction.metadata,
        )

    # ------------------------------------------------------------------
    # POST /principals
    # ------------------------------------------------------------------

    @app.post("/principals", status_code=201)
    def create_principal_route(
        body: CreatePrincipalIn,
        principal: Principal = Depends(_resolve_principal),
    ) -> PrincipalOut:
        """Create a new principal. Requires admin capability.

        Unlike every other write route here, Ontology.create_principal()
        has no built-in capability gate of its own — this route calls
        Ontology.require_admin() explicitly first, the same gate
        issue_token/revoke_token/list_tokens already use internally.
        """
        kb.require_admin(principal.id)
        created = kb.create_principal(
            body.principal_id,
            kind=body.kind,
            auth_method=body.auth_method,
            owner=body.owner,
            default_capability=body.default_capability,
            trust_level=body.trust_level,
            metadata=body.metadata,
            author=principal.id,
        )
        return PrincipalOut(
            id=created.id,
            kind=created.kind,
            owner=created.owner,
            auth_method=created.auth_method,
            default_capability=created.default_capability,
            trust_level=created.trust_level,
            created_at=created.created_at.isoformat(),
        )

    # ------------------------------------------------------------------
    # GET /principals
    # ------------------------------------------------------------------

    @app.get("/principals")
    def list_principals_route(
        principal: Principal = Depends(_resolve_principal),
    ) -> list[PrincipalOut]:
        """List all principals. Requires admin capability (KI-022)."""
        principals = kb.list_principals(author=principal.id)
        return [
            PrincipalOut(
                id=p.id,
                kind=p.kind,
                owner=p.owner,
                auth_method=p.auth_method,
                default_capability=p.default_capability,
                trust_level=p.trust_level,
                created_at=p.created_at.isoformat(),
            )
            for p in principals
        ]

    # ------------------------------------------------------------------
    # POST /principals/{principal_id}/tokens
    # ------------------------------------------------------------------

    @app.post("/principals/{principal_id}/tokens", status_code=201)
    def issue_token_route(
        principal_id: str,
        principal: Principal = Depends(_resolve_principal),
    ) -> TokenIssuedOut:
        """Issue a new bearer token for a principal. Requires admin capability.

        The raw token is returned exactly once, here, and cannot be
        recovered afterward — store it immediately.
        """
        token, credential_id = kb.issue_token(principal_id, author=principal.id)
        return TokenIssuedOut(token=token, credential_id=credential_id)

    # ------------------------------------------------------------------
    # GET /principals/{principal_id}/tokens
    # ------------------------------------------------------------------

    @app.get("/principals/{principal_id}/tokens")
    def list_tokens_route(
        principal_id: str,
        principal: Principal = Depends(_resolve_principal),
    ) -> list[CredentialOut]:
        """List credentials issued to a principal. Requires admin
        capability. Never returns the raw token or its hash."""
        credentials = kb.list_tokens(principal_id, author=principal.id)
        return [
            CredentialOut(
                id=c.id,
                principal_id=c.principal_id,
                created_at=c.created_at.isoformat(),
                revoked_at=c.revoked_at.isoformat() if c.revoked_at else None,
                issued_by=c.issued_by,
                revoked_by=c.revoked_by,
            )
            for c in credentials
        ]

    # ------------------------------------------------------------------
    # DELETE /principals/{principal_id}/tokens/{credential_id}
    # ------------------------------------------------------------------

    @app.delete("/principals/{principal_id}/tokens/{credential_id}", status_code=204)
    def revoke_token_route(
        principal_id: str,
        credential_id: str,
        principal: Principal = Depends(_resolve_principal),
    ) -> None:
        """Revoke a previously issued token by its credential ID. Requires
        admin capability.

        Unlike Ontology.revoke_token() itself (which identifies the
        credential solely by credential_id), this route verifies the
        credential actually belongs to ``principal_id`` before revoking —
        otherwise a caller could revoke a different principal's token
        while believing, from the URL alone, that they'd scoped the
        action to ``principal_id``.
        """
        kb.require_admin(principal.id)
        credential = kb.backend.get_credential(credential_id)
        if credential is None or credential.principal_id != principal_id:
            raise NotFoundError(f"Credential {credential_id!r} not found for {principal_id!r}")
        kb.revoke_token(credential_id, author=principal.id)

    # ------------------------------------------------------------------
    # GET /admin-events
    # ------------------------------------------------------------------

    @app.get("/admin-events")
    def list_admin_events_route(
        actor: str | None = None,
        target: str | None = None,
        principal: Principal = Depends(_resolve_principal),
    ) -> list[AdminEventOut]:
        """List recorded admin-action events (create_principal/apply_schema/
        register_plugin), optionally filtered by actor or target (KI-072,
        ADR-0042). Requires admin capability — closes the read half of
        KI-060's audit trail: `issued_by`/`revoked_by` on GET
        /principals/{id}/tokens covers token issuance/revocation, this
        route covers the other three admin actions.
        """
        events = kb.get_admin_events(author=principal.id, actor=actor, target=target)
        return [
            AdminEventOut(
                id=e.id,
                actor=e.actor,
                action=e.action,
                target=e.target,
                at=e.at.isoformat(),
                detail=e.detail,
            )
            for e in events
        ]

    # ------------------------------------------------------------------
    # GET /namespaces
    # ------------------------------------------------------------------

    @app.get("/namespaces")
    def list_namespaces_route(
        _principal: Principal = Depends(_resolve_principal),
    ) -> list[NamespaceOut]:
        """List all registered namespaces (SPEC §12.2, KI-022).

        Ungated beyond authentication, same as GET /proposals — namespace
        metadata carries nothing as sensitive as GET /principals's
        owner/trust_level fields.
        """
        return [
            NamespaceOut(id=n.id, created_at=n.created_at.isoformat(), metadata=n.metadata)
            for n in kb.list_namespaces()
        ]

    return app

ontolith.interfaces.graphql

GraphQL interface for Ontolith (SPEC §14.3, ADR-0037).

Exposes the Entity, Assertion, Proposal, Contradiction, and Principal types SPEC §14.3 names, with query, propose, createEntity, and review operations "mirroring the SDK" (SPEC's own phrasing) — narrower than the REST interface (interfaces/rest.py, ADR-0021/0022), which also exposes a direct-write route and principal/token admin as REST-specific extensions beyond SPEC parity (KI-022). This module deliberately does not add GraphQL equivalents of those: no direct-write mutation, no principal creation or token issuance/revocation. principals is exposed as a read-only query only, since SPEC explicitly names Principal as one of the five exposed types. createEntity (KI-082) is propose-tier, matching every other governed write here — added because, without it, a GraphQL-only caller could never introduce a genuinely new entity into the KB at all, not merely reach a narrower surface than REST's.

Authentication (ADR-0014, reused unchanged, same posture as REST/ADR-0021): every query and mutation requires an Authorization: Bearer <token> header, resolved server-side via the injected AuthProvider — never a caller-asserted principal ID. Unlike REST's Depends-based dependency (which raises before the route body runs), GraphQL has no single request-level auth gate: auth resolution happens in context_getter (storing either the resolved Principal or the AuthError in context, never raising there) and each field resolver calls _require_principal(info) first, which raises the stored AuthError if resolution failed — the exception then flows through GraphQL's normal per-field error path into the response body, same as any other domain error (see below). This deferred-to-resolver-time design exists so introspection queries (__schema) can stay reachable without a token when a deployment opts into that (introspection=True — see create_graphql_app's own docstring; False by default since KI-056, so this isn't the out-of-the-box behavior).

Error handling: GraphQL has no HTTP-status channel for domain errors the way REST does (SPEC §16's code/message/detail envelope doesn't map onto HTTP status codes here — by GraphQL convention, every response is HTTP 200 and errors surface in the response body's errors[] array). _OntolithSchema.process_errors is the single centralized hook (mirrors REST's one @app.exception_handler(OntolithError) instead of per-route try/except): for every error whose original_error is an OntolithError, it sets extensions = {"code": ..., "detail": ...}; StorageError/PluginError (the same two REST maps to 5xx) additionally get their message redacted to a generic string and the real message logged server-side only — the raw text can carry internal exception detail (e.g. sqlite3 constraint/transaction-state messages) a caller has no use for. Any other resolver exception (not an OntolithError at all — a genuine bug) is redacted the same way, with extensions.code = "INTERNAL_ERROR": REST's equivalent path gets FastAPI's generic, code-less 500, never the raw message, and this must fail closed to match rather than leak by omission.

Concurrency (KI-052): every resolver is async def. _require_principal/ _kb are cheap dict lookups and run inline, but everything that touches kb/kb.backend — the actual blocking SQLite/DuckDB I/O — is factored into a plain sync helper function and run via starlette.concurrency.run_in_threadpool. Unlike REST, whose plain def routes Starlette dispatches to a thread pool automatically, strawberry's GraphQLRouter executes resolvers inline on the ASGI event loop by default; without this, one slow resolver would block every concurrent request (not just database-bound ones) for its full duration — measured directly during review, closed here.

Usage

from ontolith.identity.token_auth import TokenAuthProvider
from ontolith.interfaces.graphql import create_graphql_app
app = create_graphql_app(kb, TokenAuthProvider(kb.backend))
# uvicorn.run(app) to serve; GraphiQL served at /graphql by default, but
# introspection is off by default (KI-056) so it can't load a schema —
# pass introspection=True for a working interactive dev experience:
# app = create_graphql_app(kb, TokenAuthProvider(kb.backend), introspection=True)

create_graphql_app

create_graphql_app(
    kb: Ontology,
    auth_provider: AuthProvider,
    name: str = "ontolith",
    *,
    graphql_ide: Literal[
        "graphiql", "apollo-sandbox", "pathfinder"
    ]
    | None = "graphiql",
    introspection: bool = False,
    docs_url: str | None = "/docs",
    redoc_url: str | None = "/redoc",
    openapi_url: str | None = "/openapi.json",
) -> FastAPI

Build and return a FastAPI app serving GraphQL at /graphql.

Parameters:

Name	Type	Description	Default
kb	Ontology	Ontology instance (knowledge base) to expose.	required
auth_provider	AuthProvider	Resolves caller-supplied bearer tokens to Principals (ADR-0014) — e.g. TokenAuthProvider(kb.backend).	required
name	str	API title.	'ontolith'
graphql_ide	Literal['graphiql', 'apollo-sandbox', 'pathfinder'] | None	Which IDE to serve at GET /graphql ("graphiql", "apollo-sandbox", "pathfinder"), or None to disable it. Unauthenticated, like REST's docs_url — it exposes the API's shape via the IDE's own introspection call, not its data.	'graphiql'
introspection	bool	Whether __schema/__type introspection queries are answered at all. False by default (KI-056): introspection queries are self-referentially recursive over the schema's own type graph, and depth/alias limiting can't bound them (QueryDepthLimiter hardcodes an introspection carve-out in strawberry's own depth-limiting validator, not graphql-core's — see the module-level comment above _MAX_QUERY_DEPTH) — so unlike REST's docs_url (a fixed, bounded JSON document with no recursive amplification potential), leaving introspection on by default would leave an unauthenticated recursive-response amplification vector open with no other mitigation in this factory. Set this True for a deployment that accepts that tradeoff (local development, trusted-network staging, or any deployment that wants self-documenting tooling and doesn't face untrusted traffic). Leaving graphql_ide at its default while this is False still serves the IDE page itself (GET /graphql returns 200), but the IDE won't be able to load a schema through it either — pass introspection=True alongside the default graphql_ide for a working interactive dev experience (see Usage above).	False
docs_url	str | None	FastAPI Swagger UI path, or None to disable it. Kept for parity with create_rest_app — this app has no REST routes of its own, but mounting under a shared FastAPI app is a documented usage pattern, and the FastAPI docs surface exists regardless.	'/docs'
redoc_url	str | None	ReDoc path, or None to disable it.	'/redoc'
openapi_url	str | None	OpenAPI schema JSON path, or None to disable it (also disables docs_url/redoc_url, which depend on it).	'/openapi.json'

Returns:

Type	Description
FastAPI	Configured FastAPI app with the query/propose/review surface
FastAPI	(ADR-0037).

Source code in src/ontolith/interfaces/graphql.py

def create_graphql_app(
    kb: Ontology,
    auth_provider: AuthProvider,
    name: str = "ontolith",
    *,
    graphql_ide: Literal["graphiql", "apollo-sandbox", "pathfinder"] | None = "graphiql",
    introspection: bool = False,
    docs_url: str | None = "/docs",
    redoc_url: str | None = "/redoc",
    openapi_url: str | None = "/openapi.json",
) -> FastAPI:
    """Build and return a FastAPI app serving GraphQL at ``/graphql``.

    Args:
        kb: Ontology instance (knowledge base) to expose.
        auth_provider: Resolves caller-supplied bearer tokens to Principals
            (ADR-0014) — e.g. ``TokenAuthProvider(kb.backend)``.
        name: API title.
        graphql_ide: Which IDE to serve at GET /graphql ("graphiql",
            "apollo-sandbox", "pathfinder"), or None to disable it.
            Unauthenticated, like REST's docs_url — it exposes the API's
            shape via the IDE's own introspection call, not its data.
        introspection: Whether ``__schema``/``__type`` introspection
            queries are answered at all. False by default (KI-056):
            introspection queries are self-referentially recursive over the
            schema's own type graph, and depth/alias limiting can't bound
            them (`QueryDepthLimiter` hardcodes an introspection carve-out
            in *strawberry's own* depth-limiting validator, not
            graphql-core's — see the module-level comment above
            `_MAX_QUERY_DEPTH`) — so unlike REST's `docs_url` (a fixed,
            bounded JSON document with no recursive amplification
            potential), leaving introspection on by default would leave an
            *unauthenticated* recursive-response amplification vector open
            with no other mitigation in this factory. Set this True for a
            deployment that accepts that tradeoff (local development,
            trusted-network staging, or any deployment that wants
            self-documenting tooling and doesn't face untrusted traffic).
            Leaving ``graphql_ide`` at its default while this is False still
            serves the IDE page itself (GET /graphql returns 200), but the
            IDE won't be able to load a schema through it either — pass
            ``introspection=True`` alongside the default ``graphql_ide`` for
            a working interactive dev experience (see ``Usage`` above).
        docs_url: FastAPI Swagger UI path, or None to disable it. Kept for
            parity with ``create_rest_app`` — this app has no REST routes
            of its own, but mounting under a shared FastAPI app is a
            documented usage pattern, and the FastAPI docs surface exists
            regardless.
        redoc_url: ReDoc path, or None to disable it.
        openapi_url: OpenAPI schema JSON path, or None to disable it (also
            disables docs_url/redoc_url, which depend on it).

    Returns:
        Configured FastAPI app with the query/propose/review surface
        (ADR-0037).
    """
    app = FastAPI(title=name, docs_url=docs_url, redoc_url=redoc_url, openapi_url=openapi_url)

    # Factory callables, not instances: passing an already-constructed
    # extension instance to `extensions=[...]` is deprecated in strawberry
    # (a fresh instance must be built per request).
    extensions: list[Callable[[], SchemaExtension]] = [
        lambda: QueryDepthLimiter(max_depth=_MAX_QUERY_DEPTH),
        lambda: MaxAliasesLimiter(max_alias_count=_MAX_ALIAS_COUNT),
    ]
    if not introspection:
        extensions.append(DisableIntrospection)
    schema = _OntolithSchema(query=Query, mutation=Mutation, extensions=extensions, kb=kb)

    async def _get_context(
        authorization: Annotated[str | None, Header()] = None,
    ) -> dict[str, object]:
        """Resolve the bearer token into context, never raising here (module docstring).

        Catches OntolithError broadly, not just AuthError: auth_provider.resolve()
        is a StorageBackend-backed lookup and can raise StorageError too — that
        must flow through the same deferred-to-resolver-time path (and so
        through process_errors' redaction) rather than escaping this
        dependency as a raw, unmapped 500 with no SPEC §16 envelope at all.
        """
        context: dict[str, object] = {"kb": kb}
        if authorization is None or not authorization.startswith("Bearer "):
            context["auth_error"] = AuthError("Missing or malformed Authorization header")
            return context
        token = authorization.removeprefix("Bearer ")
        try:
            context["principal"] = await run_in_threadpool(auth_provider.resolve, token)
        except OntolithError as exc:
            context["auth_error"] = exc
        return context

    graphql_router = GraphQLRouter[dict[str, object], None](
        schema, context_getter=_get_context, graphql_ide=graphql_ide
    )
    app.include_router(graphql_router, prefix="/graphql")
    return app