routes

routes

Route handlers for the OpenAI-compatible API server.

Classes

Functions:

chat_completions async

chat_completions(
    request_body: ChatCompletionRequest, request: Request
)

Handle chat completion requests (streaming and non-streaming).

Source code in src/diapason/server/routes.py

@router.post("/v1/chat/completions")
async def chat_completions(request_body: ChatCompletionRequest, request: Request):
    """Handle chat completion requests (streaming and non-streaming)."""
    engine = request.app.state.engine
    agent = getattr(request.app.state, "agent", None)
    model = request_body.model
    config = getattr(request.app.state, "config", None)
    latency = ChatLatency()

    # Le cliché du bureau se rafraîchit en parallèle de la requête (~100 ms
    # d'osascript) ; l'ancre du prompt ne lit que le cache — même mécanique
    # que la voix (Atlas, 24 août 2026).
    try:
        from diapason.desktop.etat_bureau import etat_du_bureau

        asyncio.get_running_loop().run_in_executor(None, etat_du_bureau)
    except Exception:  # noqa: BLE001 - la perception est un bonus
        pass

    # Trusted desktop fast path.  It runs BEFORE memory retrieval, complexity
    # scoring and inference, turning explicit low-risk commands into one local
    # OS call.  action_mode defaults to off and tools disable the path, so the
    # OpenAI-compatible API cannot unexpectedly control the host.
    query_text_for_action = ""
    for _message in reversed(request_body.messages):
        if _message.role == "user" and _message.content:
            query_text_for_action = _message.content
            break
    if (
        request_body.action_mode == "auto"
        and not request_body.tools
        and query_text_for_action
        and _host_actions_allowed(request, config)
    ):
        service = getattr(request.app.state, "lightning_actions", None)
        if service is not None:

            def generate_action_text(instruction: str) -> str:
                messages = _to_messages(request_body.messages)
                messages.insert(
                    0,
                    Message(
                        role=Role.SYSTEM,
                        content=(
                            "Create only the polished content that should be "
                            "inserted into the requested application. Follow the "
                            "user's language and instruction. Do not add commentary, "
                            "quotes, or a preface."
                        ),
                    ),
                )
                result = engine.generate(
                    messages,
                    model=model,
                    temperature=min(request_body.temperature, 0.7),
                    max_tokens=min(max(request_body.max_tokens, 128), 2048),
                )
                return str(result.get("content") or "")

            outcome = await asyncio.to_thread(
                service.handle,
                query_text_for_action,
                text_generator=generate_action_text,
            )
            if outcome.handled:
                if request_body.stream:
                    return measure_response(
                        _handle_lightning_stream(model, outcome), latency
                    )
                return ChatCompletionResponse(
                    model=model,
                    choices=[
                        Choice(
                            message=ChoiceMessage(
                                role="assistant",
                                content=outcome.message,
                            ),
                            finish_reason="stop",
                        )
                    ],
                    usage=UsageInfo(),
                    lightning=outcome.public_metadata(),
                )

    # Relevé AVANT toute injection : après, on ne peut plus distinguer le
    # cadrage du client de celui que le serveur vient d'ajouter.
    client_system = any(m.role == "system" for m in request_body.messages)

    # Inject memory context into messages before dispatching
    memory_backend = getattr(request.app.state, "memory_backend", None)
    if (
        config is not None
        and memory_backend is not None
        and config.agent.context_from_memory
        and request_body.messages
    ):
        try:
            from diapason.tools.storage.context import ContextConfig, inject_context

            # Extract query from the last user message
            query_text = ""
            for m in reversed(request_body.messages):
                if m.role == "user" and m.content:
                    query_text = m.content
                    break

            if query_text:
                messages = _to_messages(request_body.messages)
                ctx_cfg = ContextConfig(
                    top_k=config.memory.context_top_k,
                    min_score=config.memory.context_min_score,
                    max_context_tokens=config.memory.context_max_tokens,
                )
                # 19/09/2026 : la recherche SQLite tournait sur la boucle
                # du serveur ; un disque occupé retenait aussi les flux ouverts.
                with latency.phase("memoryMs"):
                    enriched = await asyncio.to_thread(
                        inject_context,
                        query_text,
                        messages,
                        memory_backend,
                        config=ctx_cfg,
                    )
                if len(enriched) > len(messages):
                    # Seuls les ajouts du serveur bougent ; les messages système
                    # fournis par le client et les échanges d'outils restent intacts.
                    from diapason.server.models import ChatMessage

                    new_msgs = []
                    for msg in enriched[: len(enriched) - len(messages)]:
                        new_msgs.append(
                            ChatMessage(
                                role=msg.role.value,
                                content=msg.content,
                                name=msg.name,
                                tool_call_id=getattr(msg, "tool_call_id", None),
                            )
                        )
                    request_body.messages = inserer_au_tour_courant(
                        request_body.messages, new_msgs
                    )
        except Exception:
            logging.getLogger("diapason.server").debug(
                "Memory context injection failed",
                exc_info=True,
            )

    # 20/09/2026 : « Que veut dire "Self Aware" ? » payait le 27b choisi dans
    # le sélecteur (89 s). Un tour léger part sur le modèle léger configuré ;
    # le modèle demandé reste pour tout le reste. Voir server/tour_leger.py.
    routage = await asyncio.to_thread(
        choisir_le_modele,
        model,
        _to_messages(request_body.messages),
        config,
        trousse_du_client=bool(request_body.tools),
        engine=engine,
    )
    if routage.substitue:
        model = routage.modele
        latency.routage = routage.public()

    # Run complexity analysis on the last user message
    complexity_info = None
    query_text_for_complexity = ""
    for m in reversed(request_body.messages):
        if m.role == "user" and m.content:
            query_text_for_complexity = m.content
            break
    if query_text_for_complexity:
        try:
            from diapason.learning.routing.complexity import (
                adjust_tokens_for_model,
                score_complexity,
            )

            cr = score_complexity(query_text_for_complexity)
            suggested = adjust_tokens_for_model(
                cr.suggested_max_tokens,
                model,
            )
            suggested = max(suggested, budget_quantite(query_text_for_complexity))
            complexity_info = ComplexityInfo(
                score=cr.score,
                tier=cr.tier,
                suggested_max_tokens=suggested,
            )
            # Bump max_tokens when complexity suggests more than what
            # the client requested — never reduce below the request value.
            if suggested > request_body.max_tokens:
                request_body.max_tokens = suggested
        except Exception:
            logging.getLogger("diapason.server").debug(
                "Complexity analysis failed",
                exc_info=True,
            )

    if request_body.stream:
        # When the client passes `tools`, stream the model's raw
        # OpenAI-compat function-calling decision directly from the engine
        # (bypassing the agent) — the streaming mirror of the non-streaming
        # #454 fix.  Routing tools through the agent stream bridge ignored
        # `request_body.tools`, ran the agent's own tool loop, and
        # word-split generic filler content into fake token deltas, so the
        # caller's tool_calls were dropped entirely (the streaming analog of
        # #414).  For plain chat (no tools), stream token-by-token directly
        # from the engine for true real-time output.
        if request_body.tools:
            response = await _handle_stream_tools(
                engine,
                model,
                request_body,
                complexity_info,
                app_config=config,
                bus=getattr(request.app.state, "bus", None),
                memory_service=getattr(request.app.state, "memory_service", None),
                client_system=client_system,
                latency=latency,
            )
            return measure_response(response, latency)
        with latency.phase("toolSetupMs"):
            tooling = await _chat_tooling_async(request.app.state, config)
        response = await _handle_stream(
            engine,
            model,
            request_body,
            complexity_info,
            trace_store=getattr(request.app.state, "trace_store", None),
            app_config=config,
            bus=getattr(request.app.state, "bus", None),
            memory_service=getattr(request.app.state, "memory_service", None),
            client_system=client_system,
            # Sans cette trousse, le chat du bureau parlait au moteur nu et
            # Diapason ne pouvait rien LIRE — ni l'heure, ni l'agenda, ni une
            # tâche Succès. C'est le fil qui manquait entre les 98 outils
            # enregistrés et la seule interface qui sert vraiment.
            tooling=tooling,
            latency=latency,
            routage=routage,
        )
        return measure_response(response, latency)

    # Non-streaming: use agent if available, otherwise direct engine call.
    #
    # EXCEPTION: when the client explicitly passed `tools`, they're asking
    # for raw OpenAI-compat function-calling — return the model's
    # tool_call decision verbatim. Routing through `_handle_agent` would
    # call `agent.run(input_text)`, which IGNORES `request_body.tools`,
    # runs the agent's own internal tool loop with its own (different)
    # tool spec, and returns only `result.content` — so the model's
    # tool_calls vanish and the user sees a generic acknowledgement
    # (e.g. "Understood. If you have another request...") that the
    # agent's re-prompted LLM produced. See #414.
    #
    # If a future caller needs agent orchestration WITH client-supplied
    # tools (e.g. injecting MCP tools through this endpoint and wanting
    # the agent to execute them), add an explicit opt-in header rather
    # than removing this guard — silent re-routing is what produced #414.
    # ``_handle_agent`` (sync ``agent.run()``) and ``_handle_direct`` (sync
    # ``engine.generate()``) both make blocking upstream calls; run them in a
    # worker thread so a slow/wedged non-streaming request can't stall the
    # event loop and every other concurrent request with it.
    if agent is not None and not request_body.tools:
        response = await asyncio.to_thread(
            _handle_agent,
            agent,
            model,
            request_body,
            complexity_info,
            trace_store=getattr(request.app.state, "trace_store", None),
            bus=getattr(request.app.state, "bus", None),
        )
    else:
        bus = getattr(request.app.state, "bus", None)
        response = await asyncio.to_thread(
            _handle_direct,
            engine,
            model,
            request_body,
            bus=bus,
            complexity_info=complexity_info,
            app_config=config,
            client_system=client_system,
        )

    if routage.substitue and isinstance(response, ChatCompletionResponse):
        response.routing = routage.public()

    # Hand the completed exchange to the background memory service.
    _remember_exchange(
        getattr(request.app.state, "memory_service", None),
        query_text_for_complexity,
        response,
        bus=getattr(request.app.state, "bus", None),
        source="server.chat",
    )
    return response

action_metrics async

action_metrics(request: Request)

Privacy-safe latency distribution for deterministic actions.

Source code in src/diapason/server/routes.py

@router.get("/v1/actions/metrics")
async def action_metrics(request: Request):
    """Privacy-safe latency distribution for deterministic actions."""
    from diapason.actions.metrics import METRICS

    return METRICS.snapshot()

list_models async

list_models(request: Request) -> ModelListResponse

List locally installed models (Ollama).

Cloud models are not included here — they live in the Cloud Models tab of the UI and are selected there, not from this endpoint. Embedding models are not included either — offering a model that cannot chat in a chat-model picker is a loaded footgun (see _est_modele_embedding). The server's own default model comes first: it is what the desktop app picks when nothing is selected yet.

Source code in src/diapason/server/routes.py

@router.get("/v1/models")
async def list_models(request: Request) -> ModelListResponse:
    """List locally installed models (Ollama).

    Cloud models are not included here — they live in the Cloud Models tab
    of the UI and are selected there, not from this endpoint. Embedding
    models are not included either — offering a model that cannot chat in
    a chat-model picker is a loaded footgun (see _est_modele_embedding).
    The server's own default model comes first: it is what the desktop app
    picks when nothing is selected yet.
    """
    from diapason.server.cloud_router import is_cloud_model, list_local_models

    # Prefer engine.list_models() so mock engines work in tests.
    # Filter out any cloud model IDs that may appear via MultiEngine.
    # Fall back to direct Ollama query only when the engine returns nothing.
    engine = request.app.state.engine
    all_ids = await asyncio.to_thread(engine.list_models)
    model_ids = [
        m for m in all_ids if not is_cloud_model(m) and not _est_modele_embedding(m)
    ]
    if not model_ids:
        model_ids = [
            m for m in await list_local_models() if not _est_modele_embedding(m)
        ]

    defaut = str(getattr(request.app.state, "model", "") or "")
    if defaut in model_ids:
        model_ids = [defaut, *[m for m in model_ids if m != defaut]]

    lengths = await asyncio.gather(*(_context_length_of(mid) for mid in model_ids))
    return ModelListResponse(
        data=[
            ModelObject(id=mid, context_length=length)
            for mid, length in zip(model_ids, lengths)
        ],
    )

prewarm_model async

prewarm_model(request: Request)

Keep a local Ollama model resident without generating any content.

Source code in src/diapason/server/routes.py

@router.post("/v1/models/prewarm")
async def prewarm_model(request: Request):
    """Keep a local Ollama model resident without generating any content."""
    body = await request.json()
    model_name = str(body.get("model") or "").strip()
    if not model_name:
        raise HTTPException(status_code=400, detail="model is required")

    engine = request.app.state.engine
    for _ in range(5):
        candidate = getattr(engine, "__dict__", {}).get("_inner")
        if candidate is None:
            break
        engine = candidate
    # MultiEngine can identify the concrete backend for this exact model.
    from diapason.engine.multi import MultiEngine

    if isinstance(engine, MultiEngine):
        selected = engine._engine_for(model_name)
        if selected is not None:
            engine = selected

    from diapason.core.local_mode import host_is_local

    if str(getattr(engine, "engine_id", "")).lower() != "ollama" or not host_is_local(
        str(getattr(engine, "_host", ""))
    ):
        raise HTTPException(
            status_code=409,
            detail="Model prewarm is available only for local Ollama models.",
        )
    prewarm = getattr(engine, "prewarm", None)
    if not callable(prewarm):
        raise HTTPException(status_code=501, detail="Engine cannot prewarm models.")
    loaded = await asyncio.to_thread(prewarm, model_name)
    if not loaded:
        raise HTTPException(
            status_code=503,
            detail="Ollama could not preload the model.",
        )
    return {
        "status": "ready",
        "model": model_name,
        "keep_alive": str(getattr(engine, "_keep_alive", "30m")),
    }

pull_model async

pull_model(request: Request)

Pull / download a model from the Ollama registry.

Source code in src/diapason/server/routes.py

@router.post("/v1/models/pull")
async def pull_model(request: Request):
    """Pull / download a model from the Ollama registry."""
    body = await request.json()
    model_name = body.get("model", "").strip()
    if not model_name:
        raise HTTPException(status_code=400, detail="'model' field is required")

    engine = request.app.state.engine
    engine_name = getattr(request.app.state, "engine_name", "")
    # Only Ollama supports pulling
    if engine_name != "ollama" and getattr(engine, "engine_id", "") != "ollama":
        raise HTTPException(
            status_code=501,
            detail="Model pulling is only supported with the Ollama engine",
        )

    import httpx as _httpx

    host = getattr(engine, "_host", "http://localhost:11434")
    try:
        from diapason.core.local_mode import LocalOnlyError, assert_may_leave

        assert_may_leave("the model pull request", destination=host)
        async with _httpx.AsyncClient(base_url=host, timeout=600.0) as client:
            resp = await client.post(
                "/api/pull",
                json={"name": model_name, "stream": False},
            )
        resp.raise_for_status()
    except LocalOnlyError as exc:
        raise HTTPException(status_code=403, detail=str(exc)) from exc
    except (_httpx.ConnectError, _httpx.TimeoutException) as exc:
        raise HTTPException(status_code=502, detail=f"Ollama unreachable: {exc}")
    except _httpx.HTTPStatusError as exc:
        raise HTTPException(
            status_code=exc.response.status_code,
            detail=f"Ollama error: {exc.response.text[:300]}",
        )

    return {"status": "ok", "model": model_name}

delete_model async

delete_model(model_name: str, request: Request)

Delete a model from Ollama.

Source code in src/diapason/server/routes.py

@router.delete("/v1/models/{model_name:path}")
async def delete_model(model_name: str, request: Request):
    """Delete a model from Ollama."""
    engine = request.app.state.engine
    engine_name = getattr(request.app.state, "engine_name", "")
    if engine_name != "ollama" and getattr(engine, "engine_id", "") != "ollama":
        raise HTTPException(status_code=501, detail="Only supported with Ollama engine")

    import httpx as _httpx

    host = getattr(engine, "_host", "http://localhost:11434")
    try:
        from diapason.core.local_mode import LocalOnlyError, assert_may_leave

        assert_may_leave("the model deletion request", destination=host)
        async with _httpx.AsyncClient(base_url=host, timeout=30.0) as client:
            resp = await client.request(
                "DELETE",
                "/api/delete",
                json={"name": model_name},
            )
        resp.raise_for_status()
    except LocalOnlyError as exc:
        raise HTTPException(status_code=403, detail=str(exc)) from exc
    except (_httpx.ConnectError, _httpx.TimeoutException) as exc:
        raise HTTPException(status_code=502, detail=f"Ollama unreachable: {exc}")
    except _httpx.HTTPStatusError as exc:
        raise HTTPException(
            status_code=exc.response.status_code,
            detail=f"Ollama error: {exc.response.text[:300]}",
        )

    return {"status": "deleted", "model": model_name}

reload_cloud_engine async

reload_cloud_engine(request: Request)

Hot-reload cloud API keys and (re-)initialize the cloud engine.

Called by the desktop app immediately after the user saves a cloud API key so that cloud models become available without a full app restart.

Source code in src/diapason/server/routes.py

@router.post("/v1/cloud/reload")
async def reload_cloud_engine(request: Request):
    """Hot-reload cloud API keys and (re-)initialize the cloud engine.

    Called by the desktop app immediately after the user saves a cloud API
    key so that cloud models become available without a full app restart.
    """
    import os

    submitted_keys: dict[str, str] | None = None
    try:
        body = await request.json()
        raw_keys = body.get("keys") if isinstance(body, dict) else None
        if isinstance(raw_keys, dict):
            submitted_keys = {
                str(k): str(v)
                for k, v in raw_keys.items()
                if str(k).endswith("_API_KEY")
            }
    except Exception:
        submitted_keys = None

    if submitted_keys is not None:
        for key, value in submitted_keys.items():
            if value:
                os.environ[key] = value
            else:
                os.environ.pop(key, None)
    else:
        # Compatibility fallback for non-desktop/manual configurations.
        keys_path = get_config_dir() / "cloud-keys.env"
        if keys_path.exists():
            for raw_line in keys_path.read_text().splitlines():
                line = raw_line.strip()
                if line and not line.startswith("#") and "=" in line:
                    k, v = line.split("=", 1)
                    os.environ[k.strip()] = v.strip()

    # Try to build a fresh CloudEngine.
    try:
        from diapason.engine.cloud import CloudEngine
        from diapason.engine.multi import MultiEngine

        cloud = CloudEngine()
        if not cloud.health():
            return {
                "status": "no_cloud",
                "message": "No cloud models available (check API keys)",
            }
    except Exception as exc:
        return {"status": "error", "message": str(exc)}

    # Locate the innermost engine, working through InstrumentedEngine layers.
    outer = request.app.state.engine
    inner = getattr(outer, "_inner", outer)

    if isinstance(inner, MultiEngine):
        # Replace or insert the cloud entry in the existing MultiEngine.
        new_engines = [(k, e) for k, e in inner._engines if k != "cloud"]
        new_engines.append(("cloud", cloud))
        inner._engines = new_engines
        inner._refresh_map()
    else:
        # Wrap the existing engine (which may be security-wrapped) with a new
        # MultiEngine that includes the cloud engine.
        engine_name = getattr(request.app.state, "engine_name", "local")
        new_multi = MultiEngine([(engine_name, inner), ("cloud", cloud)])
        if hasattr(outer, "_inner"):
            outer._inner = new_multi
        else:
            request.app.state.engine = new_multi
        request.app.state.engine_name = "multi"

    return {"status": "ok", "message": "Cloud engine reloaded"}

savings async

savings(request: Request)

Return savings summary compared to cloud providers.

Only includes telemetry from the current server session so that counters start at zero each time a new model + agent is launched.

Source code in src/diapason/server/routes.py

@router.get("/v1/savings")
async def savings(request: Request):
    """Return savings summary compared to cloud providers.

    Only includes telemetry from the current server session so that
    counters start at zero each time a new model + agent is launched.
    """
    from diapason.core.config import DEFAULT_CONFIG_DIR
    from diapason.server.savings import compute_savings, savings_to_dict
    from diapason.telemetry.aggregator import TelemetryAggregator

    db_path = DEFAULT_CONFIG_DIR / "telemetry.db"
    if not db_path.exists():
        empty = compute_savings(0, 0, 0)
        return savings_to_dict(empty)

    session_start = getattr(request.app.state, "session_start", None)

    agg = TelemetryAggregator(db_path)
    try:
        # current_methodology_only excludes pre-fix legacy rows from
        # the leaderboard's per-token efficiency numerator/denominator
        # — see the comment on _time_filter for the bimodal-Wh/token
        # background.
        summary = agg.summary(since=session_start, current_methodology_only=True)
        # Exclude cloud model tokens from savings — only local
        # inference counts toward cost savings.
        _cloud_prefixes = (
            "gpt-",
            "o1-",
            "o3-",
            "o4-",
            "claude-",
            "gemini-",
            "openrouter/",
        )
        local_models = [
            m
            for m in summary.per_model
            if not any(m.model_id.startswith(p) for p in _cloud_prefixes)
        ]
        result = compute_savings(
            prompt_tokens=sum(m.prompt_tokens for m in local_models),
            completion_tokens=sum(m.completion_tokens for m in local_models),
            total_calls=sum(m.call_count for m in local_models),
            session_start=session_start if session_start else 0.0,
            prompt_tokens_evaluated=sum(
                m.prompt_tokens_evaluated for m in local_models
            ),
        )
        return savings_to_dict(result)
    finally:
        agg.close()

reset_telemetry async

reset_telemetry()

Clear all stored telemetry records.

Useful after updating token-counting methodology — clears historical records that were computed under the old rules so that the savings dashboard and leaderboard submissions start fresh with corrected values.

Source code in src/diapason/server/routes.py

@router.post("/v1/telemetry/reset")
async def reset_telemetry():
    """Clear all stored telemetry records.

    Useful after updating token-counting methodology — clears
    historical records that were computed under the old rules so
    that the savings dashboard and leaderboard submissions start
    fresh with corrected values.
    """
    from diapason.core.config import DEFAULT_CONFIG_DIR
    from diapason.telemetry.aggregator import TelemetryAggregator

    db_path = DEFAULT_CONFIG_DIR / "telemetry.db"
    if not db_path.exists():
        return {"status": "ok", "records_cleared": 0}

    agg = TelemetryAggregator(db_path)
    try:
        count = agg.clear()
    finally:
        agg.close()
    return {"status": "ok", "records_cleared": count}

server_info async

server_info(request: Request)

Return server configuration: model, agent, engine.

Source code in src/diapason/server/routes.py

@router.get("/v1/info")
async def server_info(request: Request):
    """Return server configuration: model, agent, engine."""
    agent = getattr(request.app.state, "agent", None)
    agent_id = getattr(agent, "agent_id", None) if agent else None
    # Fall back to configured agent name if agent didn't instantiate
    if agent_id is None:
        agent_id = getattr(request.app.state, "agent_name", None)
    from diapason.engine.ollama import _default_num_ctx

    return {
        "model": getattr(request.app.state, "model", ""),
        "agent": agent_id,
        "engine": getattr(request.app.state, "engine_name", ""),
        # Effective context window for LOCAL models: the engine sends this
        # num_ctx on every Ollama call, so a model's theoretical maximum is
        # capped by it in practice.
        "num_ctx": _default_num_ctx(),
    }

health async

health(request: Request)

Health check endpoint.

Source code in src/diapason/server/routes.py

@router.get("/health")
async def health(request: Request):
    """Health check endpoint."""
    engine = request.app.state.engine
    healthy = await asyncio.to_thread(engine.health)
    if not healthy:
        raise HTTPException(status_code=503, detail="Engine unhealthy")
    return {"status": "ok"}

list_channels async

list_channels(request: Request)

List available messaging channels.

Source code in src/diapason/server/routes.py

@router.get("/v1/channels")
async def list_channels(request: Request):
    """List available messaging channels."""
    bridge = getattr(request.app.state, "channel_bridge", None)
    if bridge is None:
        return {"channels": [], "message": "Channel bridge not configured"}
    channels = bridge.list_channels()
    return {"channels": channels, "status": bridge.status().value}

channel_send async

channel_send(request: Request)

Send a message to a channel.

Source code in src/diapason/server/routes.py

@router.post("/v1/channels/send")
async def channel_send(request: Request):
    """Send a message to a channel."""
    bridge = getattr(request.app.state, "channel_bridge", None)
    if bridge is None:
        raise HTTPException(status_code=503, detail="Channel bridge not configured")

    body = await request.json()
    channel_name = body.get("channel", "")
    content = body.get("content", "")
    conversation_id = body.get("conversation_id", "")

    if not channel_name or not content:
        raise HTTPException(
            status_code=400,
            detail="'channel' and 'content' are required",
        )

    ok = bridge.send(channel_name, content, conversation_id=conversation_id)
    if not ok:
        raise HTTPException(status_code=502, detail="Failed to send message")
    return {"status": "sent", "channel": channel_name}

channel_status async

channel_status(request: Request)

Return channel bridge connection status.

Source code in src/diapason/server/routes.py

@router.get("/v1/channels/status")
async def channel_status(request: Request):
    """Return channel bridge connection status."""
    bridge = getattr(request.app.state, "channel_bridge", None)
    if bridge is None:
        return {"status": "not_configured"}
    return {"status": bridge.status().value}

security_scan async

security_scan()

Run a read-only security environment audit and return findings.

Source code in src/diapason/server/routes.py

@router.get("/v1/security/scan")
async def security_scan():
    """Run a read-only security environment audit and return findings."""
    from diapason.cli.scan_cmd import PrivacyScanner

    scanner = PrivacyScanner()
    results = scanner.run_all()
    return {
        "has_warnings": any(r.status == "warn" for r in results),
        "has_failures": any(r.status == "fail" for r in results),
        "findings": [
            {
                "name": r.name,
                "status": r.status,
                "message": r.message,
                "platform": r.platform,
            }
            for r in results
        ],
    }