API reference

Generated from source by mkdocstrings — never hand-written, so it cannot go stale.

Interactive OpenAPI documentation is served by the running application at /docs.

Endpoints

Method	Path	Purpose
GET	/healthz	Liveness
POST	/conversations	Start a run (202 with its id); the graph runs in the background. Pass follow_up_to (the latest run of a conversation) to ask a follow-up in the same conversation and patient
GET	/conversations/{id}	The durable record: status, answer, per-node cost
GET	/conversations/{id}/thread	Every turn of the conversation this run belongs to, oldest first
GET	/conversations/{id}/events	SSE: replays every stored step, then tails the run to completion
GET	/personas	The demonstration patients a visitor can ask as (a demo selector, not a login)
GET	/reviews	Drafts held for a clinician
POST	/reviews/{id}/decision	Approve, edit or decline; resumes the paused run from its checkpoint
POST	/conversations/stream	Phase 0 walking-skeleton stream, kept as the streaming proof

Every response carries the synthetic-data and not-a-medical-device statement. Endpoints are rate limited per client IP and per route: routes that start or resume a model run share a strict ceiling (10/minute by default), reads and the event stream a generous one (120/minute). A limited request gets 429. Request bodies are capped at the ASGI layer. POST /conversations only answers as one of the personas from GET /personas; any other patient_id gets a 404. Errors carry a human-readable detail (for example 404 for an unknown patient_id, 409 when a review was already decided), which the frontend shows as-is.

Settings

Settings

Bases: AgentSettings

Application configuration, sourced from the environment.

Source code in src/navigator/settings.py

class Settings(AgentSettings):
    """Application configuration, sourced from the environment."""

    model_config = SettingsConfigDict(env_file=".env", env_prefix="NAVIGATOR_", extra="ignore")

    langsmith_project: str = "clinical-care-navigator"

    record_db_path: str = "data/records.db"
    education_db_path: str = "data/education.db"
    policy_db_path: str = "data/policy.db"
    checkpoint_db_path: str = "data/checkpoints.db"
    # Reviews share the run-store database; the queue owns its own table.
    # A blank demo patient means "resolve the first patient in the record store"
    # so the patient-facing API has a sane default without hard-coding an id.
    demo_patient_id: str = ""
    # The API only answers as a demonstration persona (see GET /personas) rather than
    # any patient id a caller supplies. This is a demo guard, not authentication.
    demo_patients_only: bool = True

    # docs/PLAN.md §5.9 — moves the inform/recommend boundary only, never the
    # escalation boundary.
    autonomy_level: AutonomyLevel = "L2_balanced"

    # docs/PLAN.md §5.5 — a clinical answer is truncated conservatively, never silently.
    max_tool_calls: int = 12
    max_evidence_passes: int = 1
    max_run_seconds: float = 90.0
    citation_coverage_floor: float = 1.0

    # Tighter than the library default: a patient question is a sentence or two.
    max_request_body_bytes: int = 8_192

    # Split per-IP limits: routes that start or resume an LLM run spend money and
    # get the strict ceiling; reads and the SSE replay are cheap, and one page view
    # makes several of them, so a single shared ceiling locked the UI out.
    write_rate_limit_per_minute: int = 10
    read_rate_limit_per_minute: int = 120

    # Follow-up questions: how many turns one conversation may have, and how many of the
    # most recent ones are shown to the rewriter as context.
    max_conversation_turns: int = 8
    conversation_context_turns: int = 3

    frontend_origins: list[str] = ["http://localhost:3000"]

Streaming

stream_conversation_events(run_store, run_id, max_wait_seconds=None)

Replays every event already persisted for a conversation, then — while the run is still executing — tails newly-appended events until it reaches a terminal status. Works identically whether the patient connects the instant a run starts or reconnects long after it finished (docs/PLAN.md §5.8).

These events are the patient's own record and are deliberately not redacted; redaction guards the log/trace boundary, not the answer a patient reads (§5.7).

max_wait_seconds bounds the tail: a run that never reaches a terminal status ends the stream with its current status rather than holding the connection open forever.

Source code in src/navigator/api/streaming.py

def stream_conversation_events(
    run_store: RunStore, run_id: str, max_wait_seconds: float | None = None
) -> AsyncIterator[str]:
    """Replays every event already persisted for a conversation, then — while the
    run is still executing — tails newly-appended events until it reaches a
    terminal status. Works identically whether the patient connects the instant a
    run starts or reconnects long after it finished (docs/PLAN.md §5.8).

    These events are the patient's own record and are deliberately not redacted;
    redaction guards the log/trace boundary, not the answer a patient reads (§5.7).

    ``max_wait_seconds`` bounds the tail: a run that never reaches a terminal
    status ends the stream with its current status rather than holding the
    connection open forever.
    """
    deadline = None if max_wait_seconds is None else time.monotonic() + max_wait_seconds

    def poll_end() -> dict[str, Any] | None:
        run = run_store.get_run(run_id)
        still_running = run is not None and run.status == "running"
        timed_out = deadline is not None and time.monotonic() >= deadline
        if still_running and not timed_out:
            return None
        status = run.status if run is not None else "not_found"
        answer = json.loads(run.answer_json) if run is not None and run.answer_json else None
        return {"status": status, "answer": answer}

    return replay_and_tail(partial(run_store.get_events, run_id), poll_end)

Graph

The only file that wires nodes and edges (docs/PLAN.md §9.3).

build_skeleton_graph is the Phase 0 walking skeleton, kept so the streaming path stays proven. build_navigator_graph wires the real graph (§5.1): intake loads the patient header; the pre-flight gate (screen_rules ∥ classify_intent ∥ detect_language → resolve_policy) routes to a templated branch or into the investigate loop; the investigate loop drives the scoped executor; draft_answer produces the cited PatientAnswer; post-flight (extract_claims -> post_flight) runs the three safety checks and routes to publish, a templated escalation, the review queue, or back into the loop for a missing citation (§5.3).

Node filename == registered node name == trace span name (§9.3 rule 3).

build_navigator_graph(settings, *, intent_chain=None, answer_writer_chain=None, explainer=None, claim_extractor_chain=None, scope_judge_chain=None, contextualizer_chain=None, checkpointer=None)

The only function that wires the navigator graph's nodes and edges.

The LLM-backed pieces are injectable so the graph can be assembled and tested offline with stubs; in production they are built from the configured models.

Source code in src/navigator/graph/builder.py

def build_navigator_graph(
    settings: Settings,
    *,
    intent_chain: IntentClassifierChain | None = None,
    answer_writer_chain: AnswerWriterChain | None = None,
    explainer: Explainer | None = None,
    claim_extractor_chain: ClaimExtractorChain | None = None,
    scope_judge_chain: ScopeJudgeChain | None = None,
    contextualizer_chain: ContextualizerChain | None = None,
    checkpointer: Checkpointer = None,
) -> NavigatorGraph:
    """The only function that wires the navigator graph's nodes and edges.

    The LLM-backed pieces are injectable so the graph can be assembled and tested
    offline with stubs; in production they are built from the configured models.
    """
    record_store = RecordStore(Path(settings.record_db_path))
    education_store = EducationStore(Path(settings.education_db_path))
    policy_store = PolicyStore(Path(settings.policy_db_path))
    registry = build_registry(record_store, education_store)
    executor = ScopedToolExecutor(registry)
    rule_engine = RuleEngine(policy_store.enabled_rules())

    if (
        intent_chain is None
        or answer_writer_chain is None
        or explainer is None
        or claim_extractor_chain is None
        or scope_judge_chain is None
    ):
        fast_model = build_fast_model(settings)
        reasoning_model = build_reasoning_model(settings)
        if intent_chain is None:
            intent_chain = build_intent_classifier_chain(fast_model)
        if answer_writer_chain is None:
            answer_writer_chain = build_answer_writer_chain(reasoning_model)
        if explainer is None:
            # The explainer is bound to the full registry; the scoped executor
            # enforces the per-run ToolScope on every call, so the model can
            # propose anything and only in-scope calls execute (§3.3, §3.4).
            explainer = build_explainer(fast_model, list(registry.tools.values()))
        if claim_extractor_chain is None:
            # Claim extraction is a mechanical decomposition -> fast model.
            claim_extractor_chain = build_claim_extractor_chain(fast_model)
        if scope_judge_chain is None:
            # The scope judge is the one graded call -> reasoning model.
            scope_judge_chain = build_scope_judge_chain(reasoning_model)
        if contextualizer_chain is None:
            # Rewriting a follow-up is mechanical -> fast model. Only ever called when a
            # conversation has earlier turns.
            contextualizer_chain = build_contextualizer_chain(fast_model)

    workflow = StateGraph(NavigatorState)
    # mypy cannot resolve add_node's overloads against a factory-returned
    # Callable (vs. a plain top-level function) — confirmed upstream limitation,
    # not a real type error; each node is unit-tested directly in tests/graph/.
    workflow.add_node("intake", build_intake_node(record_store, settings))  # type: ignore[arg-type]
    # With no chain (the all-stub offline graphs) a follow-up is simply not rewritten.
    workflow.add_node("contextualize", build_contextualize_node(contextualizer_chain))  # type: ignore[arg-type]
    workflow.add_node("screen_rules", build_screen_rules_node(rule_engine))  # type: ignore[arg-type]
    workflow.add_node("classify_intent", build_classify_intent_node(intent_chain))  # type: ignore[arg-type]
    workflow.add_node("detect_language", detect_language_node)
    workflow.add_node(
        "resolve_policy",
        build_resolve_policy_node(policy_store, registry, DEFAULT_ROW_CAP),  # type: ignore[arg-type]
    )
    workflow.add_node(
        "investigate",
        build_investigate_node(  # type: ignore[arg-type]
            explainer, executor, load_prompt("explainer_v1")
        ),
    )
    workflow.add_node("draft_answer", build_draft_answer_node(answer_writer_chain))  # type: ignore[arg-type]
    workflow.add_node(
        "extract_claims",
        build_extract_claims_node(claim_extractor_chain),  # type: ignore[arg-type]
    )
    workflow.add_node(
        "post_flight",
        build_post_flight_node(  # type: ignore[arg-type]
            record_store.reference_range,
            scope_judge_chain,
            floor=settings.citation_coverage_floor,
            max_evidence_passes=settings.max_evidence_passes,
        ),
    )
    workflow.add_node("publish", build_publish_node())  # type: ignore[arg-type]
    workflow.add_node("enqueue_review", build_enqueue_review_node())  # type: ignore[arg-type]
    workflow.add_node("template_response", build_template_response_node(policy_store))  # type: ignore[arg-type]
    workflow.add_node("budget_exceeded", budget_exceeded_node)

    workflow.set_entry_point("intake")
    # screen_rules, classify_intent and detect_language run in parallel after
    # contextualize (which follows intake, §5.1); resolve_policy is their join.
    workflow.add_edge("intake", "contextualize")
    workflow.add_edge("contextualize", "screen_rules")
    workflow.add_edge("contextualize", "classify_intent")
    workflow.add_edge("contextualize", "detect_language")
    workflow.add_edge("screen_rules", "resolve_policy")
    workflow.add_edge("classify_intent", "resolve_policy")
    workflow.add_edge("detect_language", "resolve_policy")
    workflow.add_conditional_edges(
        "resolve_policy",
        route_after_resolve_policy,
        {
            "emergency": "template_response",
            "crisis": "template_response",
            "out_of_scope": "template_response",
            "clinician_review": "template_response",
            "allow": "investigate",
        },
    )
    workflow.add_conditional_edges(
        "investigate",
        route_after_investigate,
        {
            "investigate": "investigate",
            "draft_answer": "draft_answer",
            "budget_exceeded": "budget_exceeded",
        },
    )
    # Post-flight: the draft is decomposed, then the three checks run and route.
    workflow.add_edge("draft_answer", "extract_claims")
    workflow.add_edge("extract_claims", "post_flight")
    workflow.add_conditional_edges(
        "post_flight",
        route_after_post_flight,
        {
            "publish": "publish",
            "escalate": "template_response",
            "review": "enqueue_review",
            "investigate": "investigate",
        },
    )
    workflow.add_edge("publish", END)
    workflow.add_edge("enqueue_review", END)
    workflow.add_edge("template_response", END)
    workflow.add_edge("budget_exceeded", END)
    return workflow.compile(checkpointer=checkpointer)

build_skeleton_graph()

Phase 0 walking skeleton: echo -> done.

Source code in src/navigator/graph/builder.py

def build_skeleton_graph() -> SkeletonGraph:
    """Phase 0 walking skeleton: echo -> done."""
    workflow = StateGraph(SkeletonState)
    workflow.add_node("echo", echo)
    workflow.add_node("done", done)
    workflow.set_entry_point("echo")
    workflow.add_edge("echo", "done")
    workflow.add_edge("done", END)
    return workflow.compile()