Add the assistant: a chat that drives the MCP tools in-process

A provider set under Settings (OpenAI-compatible; key encrypted in
flow-data/ai.json) turns on an Assistant panel in the sidebar. Each turn runs
the FastMCP tools in-process as the person chatting, streams the rounds over
SSE and keeps one transcript per user. The panel floats on any screen and
shares the editor's lane: a settings panel keeps the top half, the chat the
bottom. The MCP client is now built once, for the endpoint and the chat.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

backend/fluksio/ai/__init__.py

@@ -0,0 +1,7 @@
+"""The assistant: a model driving the MCP tools from inside the dashboard.
+
+``config`` is the provider and the key, ``client`` the one call every
+OpenAI-compatible provider answers, ``agent`` the loop between the model and
+the tools. The tools are the MCP server's own, so what the assistant can do is
+exactly what an external agent can do.
+"""

backend/fluksio/ai/agent.py

@@ -0,0 +1,172 @@
+"""The turn: the model, the tools, and the rounds between them.
+
+The tools are the MCP server's own, called in-process rather than over the
+endpoint: same names, same schemas, same REST calls underneath. A tool added
+for external agents is there for the assistant the moment it exists, and the
+assistant can do nothing an external agent could not.
+
+The transcript is a list of chat-completion messages, mutated in place by
+:func:`run_turn` and persisted by the caller. It is also what the browser
+renders, so nothing is stored twice.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+from collections.abc import AsyncIterator
+from typing import Any
+
+import httpx
+from mcp.types import TextContent
+
+from fluksio.ai import client
+from fluksio.ai.config import AIConfig
+from fluksio.flow.schemas import FlowDef
+from fluksio.mcp.server import caller_authorization, mcp
+
+logger = logging.getLogger(__name__)
+
+_SYSTEM = """You are the assistant inside a Fluksio instance. You read, build, edit and \
+operate flows through the tools; they are the same API the editor uses, so \
+anything you save appears in the editor at once.
+
+How to work:
+- Read before you write: get_flow before save_flow, and pass back the version \
+you read. On a version conflict, read again and reapply the change.
+- save_flow writes a draft, and validate_flow tells you whether it is sound. \
+Publish only when the user asks for the change to go live; otherwise leave \
+the draft for review in the editor and say so.
+- list_node_types gives the node types and their parameters; get_node_source \
+shows how a Python node is written before you write one.
+- Keep answers short, in plain text without markdown. Name what you changed.
+
+The flow document `save_flow` takes, as a JSON schema:
+"""
+
+# ponytail: a tool's answer is replayed on every later round of the
+# conversation, so one big one is paid for again and again; cut it here rather
+# than summarise it. A per-tool digest is the upgrade if this bites.
+_RESULT_LIMIT = 20_000
+
+
+def system_prompt(flow: str | None) -> str:
+    prompt = _SYSTEM + json.dumps(FlowDef.model_json_schema())
+    if flow:
+        prompt += f"\n\nThe user is looking at the flow named {flow!r}."
+    return prompt
+
+
+async def tool_specs() -> list[dict[str, Any]]:
+    return [
+        {
+            "type": "function",
+            "function": {
+                "name": tool.name,
+                "description": tool.description or "",
+                "parameters": tool.inputSchema,
+            },
+        }
+        for tool in await mcp.list_tools()
+    ]
+
+
+def _arguments(raw: Any) -> dict[str, Any]:
+    if isinstance(raw, dict):
+        return raw
+    try:
+        parsed = json.loads(raw or "{}")
+    except ValueError:
+        return {}
+    return parsed if isinstance(parsed, dict) else {}
+
+
+async def _call_tool(name: str, arguments: dict[str, Any]) -> str:
+    """Run one tool; whatever goes wrong is text the model can read."""
+    try:
+        result = await mcp.call_tool(name, arguments)
+    except Exception as exc:  # noqa: BLE001 - an unknown tool, or refused arguments
+        return json.dumps({"error": str(exc)})
+    if isinstance(result, dict):
+        text = json.dumps(result)
+    else:
+        text = "".join(c.text for c in result if isinstance(c, TextContent))
+    if len(text) > _RESULT_LIMIT:
+        text = text[:_RESULT_LIMIT] + "… (truncated)"
+    return text
+
+
+def summarize(content: str | None) -> dict[str, Any]:
+    """What the browser shows for a tool's answer: whether it worked."""
+    try:
+        result = json.loads(content or "")
+    except ValueError:
+        return {"ok": True}
+    if isinstance(result, dict) and "error" in result:
+        return {"ok": False, "error": str(result["error"])[:300]}
+    return {"ok": True}
+
+
+async def run_turn(
+    config: AIConfig,
+    messages: list[dict[str, Any]],
+    text: str,
+    *,
+    flow: str | None,
+    authorization: str,
+) -> AsyncIterator[dict[str, Any]]:
+    """One user turn, yielding progress events as it goes.
+
+    A round is appended to *messages* only once every tool in it has answered,
+    so a turn cut off part-way leaves a transcript the provider still accepts.
+    """
+    messages.append({"role": "user", "content": text})
+    tools = await tool_specs()
+    system = {"role": "system", "content": system_prompt(flow)}
+    token = caller_authorization.set(authorization)
+    try:
+        for _ in range(config.max_rounds):
+            try:
+                response = await client.chat_completion(
+                    api_base=config.api_base,
+                    api_key=config.api_key,
+                    model=config.model,
+                    messages=[system, *messages],
+                    tools=tools,
+                )
+            except (client.ProviderError, httpx.HTTPError) as exc:
+                yield {"type": "error", "detail": f"The provider answered {exc}"}
+                return
+            choice = response["choices"][0]["message"]
+            content = choice.get("content") or ""
+            calls = choice.get("tool_calls") or []
+            assistant: dict[str, Any] = {"role": "assistant", "content": content}
+            if content:
+                yield {"type": "message", "content": content}
+            if not calls:
+                messages.append(assistant)
+                return
+            assistant["tool_calls"] = calls
+            answers: list[dict[str, Any]] = []
+            for call in calls:
+                function = call.get("function") or {}
+                name = str(function.get("name", ""))
+                arguments = _arguments(function.get("arguments"))
+                yield {"type": "tool_call", "name": name, "arguments": arguments}
+                result = await _call_tool(name, arguments)
+                answers.append(
+                    {
+                        "role": "tool",
+                        "tool_call_id": call.get("id", ""),
+                        "content": result,
+                    }
+                )
+                yield {"type": "tool_result", "name": name, **summarize(result)}
+            messages.append(assistant)
+            messages.extend(answers)
+        yield {
+            "type": "error",
+            "detail": f"Stopped after {config.max_rounds} rounds of tool calls.",
+        }
+    finally:
+        caller_authorization.reset(token)

backend/fluksio/ai/client.py

@@ -0,0 +1,76 @@
+"""One OpenAI-compatible provider call, and the model listing beside it.
+
+Chat completions with function tools is the dialect every hosted provider and
+every local server speaks — OpenAI, Anthropic's compatibility endpoint,
+OpenRouter, Ollama, vLLM — so one client covers them without a provider SDK.
+The admin types the URL; the model never chooses where a request goes.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+from urllib.parse import urlparse
+
+import httpx
+
+
+class ProviderError(Exception):
+    """The provider answered, and the answer was a refusal."""
+
+
+def check_base(api_base: str) -> None:
+    parsed = urlparse(api_base)
+    if parsed.scheme not in ("http", "https") or not parsed.hostname:
+        raise ValueError("The provider URL must be an http(s) URL")
+
+
+def _headers(api_key: str) -> dict[str, str]:
+    return {"Authorization": f"Bearer {api_key}"}
+
+
+def _detail(response: httpx.Response) -> str:
+    try:
+        error = response.json().get("error")
+    except ValueError:
+        error = None
+    if isinstance(error, dict):
+        error = error.get("message")
+    text = str(error or response.text or response.reason_phrase)
+    return f"{response.status_code}: {text[:300]}"
+
+
+async def chat_completion(
+    *,
+    api_base: str,
+    api_key: str,
+    model: str,
+    messages: list[dict[str, Any]],
+    tools: list[dict[str, Any]],
+    timeout: float = 120.0,
+) -> dict[str, Any]:
+    """``POST {api_base}/chat/completions``, the whole answer at once."""
+    body: dict[str, Any] = {"model": model, "messages": messages}
+    if tools:
+        body["tools"] = tools
+    async with httpx.AsyncClient(timeout=timeout) as http:
+        response = await http.post(
+            api_base.rstrip("/") + "/chat/completions",
+            json=body,
+            headers=_headers(api_key),
+        )
+    if response.is_error:
+        raise ProviderError(_detail(response))
+    result: dict[str, Any] = response.json()
+    return result
+
+
+async def list_models(api_base: str, api_key: str, timeout: float = 15.0) -> list[str]:
+    """The ids ``GET {api_base}/models`` offers, for the settings screen."""
+    async with httpx.AsyncClient(timeout=timeout) as http:
+        response = await http.get(
+            api_base.rstrip("/") + "/models", headers=_headers(api_key)
+        )
+    if response.is_error:
+        raise ProviderError(_detail(response))
+    data = response.json().get("data") or []
+    return sorted(str(m["id"]) for m in data if isinstance(m, dict) and "id" in m)

backend/fluksio/ai/config.py

@@ -0,0 +1,98 @@
+"""The assistant's provider: where the model is, and the key that opens it.
+
+One file on the data volume beside ``cloud.json``, for the same reasons: it is
+a credential, it must survive a rebuild, and it must never be in the flows
+repository. The key is encrypted with the cipher the secrets store uses, so
+both turn unreadable together when ``SECRET_KEY`` changes. Nothing here exists
+until a superuser writes it, and its absence is what keeps every AI surface
+off.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import os
+from dataclasses import asdict, dataclass
+from typing import Any
+
+from cryptography.fernet import InvalidToken
+
+from fluksio.core.config import settings
+from fluksio.flow.secrets import fernet
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass(frozen=True)
+class AIConfig:
+    #: An OpenAI-compatible base, e.g. ``https://api.openai.com/v1``.
+    api_base: str
+    api_key: str
+    model: str
+    #: How many model-and-tools rounds one turn may take before it is stopped.
+    max_rounds: int = 12
+
+    @property
+    def ready(self) -> bool:
+        return bool(self.api_base and self.api_key and self.model)
+
+    def public(self) -> dict[str, Any]:
+        """The secret-free view the settings screen gets."""
+        key = self.api_key
+        masked = None
+        if key:
+            masked = f"{key[:3]}…{key[-4:]}" if len(key) > 8 else "••••"
+        return {
+            "api_base": self.api_base,
+            "model": self.model,
+            "max_rounds": self.max_rounds,
+            "api_key_set": bool(key),
+            "api_key_masked": masked,
+        }
+
+
+def load() -> AIConfig | None:
+    path = settings.AI_CONFIG_FILE
+    try:
+        data = json.loads(path.read_text())
+    except OSError:
+        return None
+    except ValueError:
+        logger.exception("Could not read %s; the assistant stays off", path)
+        return None
+    sealed = data.pop("api_key_enc", "")
+    key = ""
+    if sealed:
+        try:
+            key = fernet(settings.SECRET_KEY).decrypt(sealed.encode()).decode()
+        except InvalidToken:
+            logger.error(
+                "%s was written with a different SECRET_KEY; enter the key again",
+                path,
+            )
+    try:
+        return AIConfig(api_key=key, **data)
+    except TypeError:
+        logger.exception("Could not read %s; the assistant stays off", path)
+        return None
+
+
+def save(config: AIConfig) -> None:
+    path = settings.AI_CONFIG_FILE
+    path.parent.mkdir(parents=True, exist_ok=True)
+    data = asdict(config)
+    key = data.pop("api_key")
+    data["api_key_enc"] = (
+        fernet(settings.SECRET_KEY).encrypt(key.encode()).decode() if key else ""
+    )
+    # Written through a temporary file so a crash mid-write cannot leave a
+    # half-parsed credential behind, and never group- or world-readable.
+    tmp = path.with_suffix(".tmp")
+    tmp.write_text(json.dumps(data, indent=2))
+    os.chmod(tmp, 0o600)
+    tmp.replace(path)
+
+
+def delete() -> None:
+    settings.AI_CONFIG_FILE.unlink(missing_ok=True)

backend/fluksio/api/main.py

@@ -1,6 +1,7 @@
 from fastapi import APIRouter
 
 from fluksio.api.routes import (
+    ai,
     alerts,
     artifacts,
     cloud,
@@ -51,3 +52,6 @@ api_router.include_router(oauth.router)
 # Always mounted like oauth, so the generated SDK keeps its shape; the
 # endpoints refuse to work unless the private API is explicitly enabled.
 api_router.include_router(private.router)
+# The assistant. Always mounted for the same reason; without a provider
+# configured the status says so and the chat answers 404.
+api_router.include_router(ai.router)

backend/fluksio/api/routes/ai.py

@@ -0,0 +1,250 @@
+"""The assistant: its provider, and the conversation with it.
+
+The provider is instance-wide and superuser-only to set, because the key it
+holds is spent by everyone who chats. The conversation is per person and runs
+as them: every tool call carries their own token, so the assistant can do
+nothing they could not do in the browser.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import uuid
+from collections.abc import AsyncIterator
+from typing import Any
+
+import httpx
+from fastapi import APIRouter, Depends, HTTPException, Request
+from fastapi.responses import StreamingResponse
+from pydantic import BaseModel, Field
+
+from fluksio.ai import agent, client
+from fluksio.ai import config as ai_config
+from fluksio.api.deps import CurrentUser, TokenDep, get_current_active_superuser
+from fluksio.core.config import settings
+from fluksio.flow.schemas import NAME_PATTERN
+from fluksio.mcp import server as tools
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(prefix="/ai", tags=["ai"])
+
+#: Whose turn is running: a second message while one is answered is refused
+#: rather than interleaved into the same transcript.
+_busy: set[uuid.UUID] = set()
+
+
+class AIConfigPublic(BaseModel):
+    api_base: str
+    model: str
+    max_rounds: int
+    api_key_set: bool
+    api_key_masked: str | None
+
+
+class AIConfigUpdate(BaseModel):
+    api_base: str = Field(min_length=1, max_length=512)
+    #: Blank or absent keeps the stored key.
+    api_key: str | None = Field(default=None, max_length=512)
+    model: str = Field(default="", max_length=128)
+    max_rounds: int = Field(default=12, ge=1, le=50)
+
+
+class AIModelsBody(BaseModel):
+    api_base: str = Field(min_length=1, max_length=512)
+    #: Absent tries the stored key, so a saved provider can be re-listed.
+    api_key: str | None = Field(default=None, max_length=512)
+
+
+class AIStatus(BaseModel):
+    available: bool
+    model: str | None
+
+
+class ChatItem(BaseModel):
+    role: str
+    content: str | None = None
+    tool_calls: list[dict[str, Any]] | None = None
+    #: On a tool's answer: whether it worked, and the error if not.
+    ok: bool | None = None
+    error: str | None = None
+
+
+class TurnBody(BaseModel):
+    message: str = Field(min_length=1, max_length=20_000)
+    #: The flow open in the editor, when the chat is opened from one.
+    flow: str | None = Field(default=None, pattern=NAME_PATTERN.pattern)
+
+
+def _public(config: ai_config.AIConfig | None) -> AIConfigPublic:
+    if config is None:
+        return AIConfigPublic(
+            api_base="",
+            model="",
+            max_rounds=12,
+            api_key_set=False,
+            api_key_masked=None,
+        )
+    return AIConfigPublic(**config.public())
+
+
+@router.get(
+    "/config",
+    dependencies=[Depends(get_current_active_superuser)],
+    response_model=AIConfigPublic,
+)
+def read_ai_config() -> Any:
+    return _public(ai_config.load())
+
+
+@router.put(
+    "/config",
+    dependencies=[Depends(get_current_active_superuser)],
+    response_model=AIConfigPublic,
+)
+def update_ai_config(body: AIConfigUpdate) -> Any:
+    try:
+        client.check_base(body.api_base)
+    except ValueError as exc:
+        raise HTTPException(status_code=400, detail=str(exc)) from exc
+    current = ai_config.load()
+    key = body.api_key or (current.api_key if current else "")
+    config = ai_config.AIConfig(
+        api_base=body.api_base.rstrip("/"),
+        api_key=key,
+        model=body.model.strip(),
+        max_rounds=body.max_rounds,
+    )
+    ai_config.save(config)
+    return _public(config)
+
+
+@router.delete("/config", dependencies=[Depends(get_current_active_superuser)])
+def delete_ai_config() -> Any:
+    ai_config.delete()
+    return _public(None)
+
+
+@router.post(
+    "/config/models",
+    dependencies=[Depends(get_current_active_superuser)],
+    response_model=list[str],
+)
+async def list_ai_models(body: AIModelsBody) -> Any:
+    """What the provider offers, tried with the key given or the one stored."""
+    try:
+        client.check_base(body.api_base)
+    except ValueError as exc:
+        raise HTTPException(status_code=400, detail=str(exc)) from exc
+    current = ai_config.load()
+    key = body.api_key or (current.api_key if current else "")
+    try:
+        return await client.list_models(body.api_base.rstrip("/"), key)
+    except (client.ProviderError, httpx.HTTPError) as exc:
+        raise HTTPException(
+            status_code=400, detail=f"The provider answered {exc}"
+        ) from exc
+
+
+@router.get("/status", response_model=AIStatus)
+def read_ai_status(_: CurrentUser) -> Any:
+    """Whether there is an assistant to show. Readable by anyone signed in."""
+    config = ai_config.load()
+    if config is None or not config.ready:
+        return AIStatus(available=False, model=None)
+    return AIStatus(available=True, model=config.model)
+
+
+def _chat_file(user_id: uuid.UUID) -> Any:
+    return settings.CHATS_DIR / f"{user_id}.json"
+
+
+def _load_chat(user_id: uuid.UUID) -> list[dict[str, Any]]:
+    try:
+        messages: list[dict[str, Any]] = json.loads(_chat_file(user_id).read_text())
+    except (OSError, ValueError):
+        return []
+    return messages
+
+
+def _save_chat(user_id: uuid.UUID, messages: list[dict[str, Any]]) -> None:
+    path = _chat_file(user_id)
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(json.dumps(messages))
+
+
+@router.get("/chat", response_model=list[ChatItem])
+def read_chat(user: CurrentUser) -> Any:
+    """The transcript, with each tool's answer reduced to whether it worked."""
+    items = []
+    for message in _load_chat(user.id):
+        if message.get("role") == "tool":
+            items.append(
+                ChatItem(role="tool", **agent.summarize(message.get("content")))
+            )
+        else:
+            items.append(
+                ChatItem(
+                    role=message["role"],
+                    content=message.get("content"),
+                    tool_calls=message.get("tool_calls"),
+                )
+            )
+    return items
+
+
+@router.delete("/chat", response_model=list[ChatItem])
+def clear_chat(user: CurrentUser) -> Any:
+    _chat_file(user.id).unlink(missing_ok=True)
+    return []
+
+
+def _sse(event: dict[str, Any]) -> str:
+    return f"data: {json.dumps(event)}\n\n"
+
+
+@router.post("/chat")
+def chat(
+    body: TurnBody, request: Request, user: CurrentUser, token: TokenDep
+) -> StreamingResponse:
+    """One turn, streamed as server-sent events.
+
+    ``message``, ``tool_call`` and ``tool_result`` events as the model works,
+    ``error`` when something stops it, and ``done`` last either way.
+    """
+    config = ai_config.load()
+    if config is None or not config.ready:
+        # 404 rather than 403: an unconfigured instance says nothing about it.
+        raise HTTPException(status_code=404, detail="The assistant is not set up")
+    if user.id in _busy:
+        raise HTTPException(status_code=409, detail="Still answering the last one")
+    tools.connect(request.app)
+    messages = _load_chat(user.id)
+
+    async def generate() -> AsyncIterator[str]:
+        _busy.add(user.id)
+        try:
+            yield _sse({"type": "start"})
+            async for event in agent.run_turn(
+                config,
+                messages,
+                body.message,
+                flow=body.flow,
+                authorization=f"Bearer {token}",
+            ):
+                yield _sse(event)
+        except Exception as exc:  # noqa: BLE001 - the stream is the error channel
+            logger.exception("The assistant's turn failed")
+            yield _sse({"type": "error", "detail": f"{type(exc).__name__}: {exc}"})
+        finally:
+            # What happened so far, even if the browser left part-way.
+            _save_chat(user.id, messages)
+            _busy.discard(user.id)
+        yield _sse({"type": "done"})
+
+    return StreamingResponse(
+        generate(),
+        media_type="text/event-stream",
+        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
+    )

backend/fluksio/core/config.py

@@ -27,6 +27,8 @@ DERIVED_PATHS = {
     "OAUTH_PRIVATE_KEY_FILE": "oauth-key.pem",
     "CLOUD_CONFIG_FILE": "cloud.json",
     "PROVISIONERS_FILE": "provisioners.json",
+    "AI_CONFIG_FILE": "ai.json",
+    "CHATS_DIR": "chats",
 }
 
 
@@ -102,6 +104,12 @@ class Settings(BaseSettings):
     # Written only when someone enrols this instance with a portal.
     # Its absence is what keeps remote access off.
     CLOUD_CONFIG_FILE: Path = Path("flow-data/cloud.json")
+    # The assistant's provider: written from Settings, key encrypted with the
+    # app key like the secrets store. Absent until someone configures it, and
+    # its absence is what keeps every AI surface off.
+    AI_CONFIG_FILE: Path = Path("flow-data/ai.json")
+    # One transcript per user, beside the config it talks through.
+    CHATS_DIR: Path = Path("flow-data/chats")
     OAUTH_CODE_EXPIRE_SECONDS: int = 60
     # Short, because an agent's token is a bearer secret held by a program
     # rather than a person, and it can refresh unattended.

backend/fluksio/flow/secrets.py

@@ -34,14 +34,18 @@ class SecretNotFound(KeyError):
         return f"No secret named '{self.name}' — add it under Secrets."
 
 
+def fernet(key: str) -> Fernet:
+    """The cipher a store keyed on the app key uses, so every one of them
+    turns unreadable together when ``SECRET_KEY`` changes."""
+    return Fernet(base64.urlsafe_b64encode(hashlib.sha256(key.encode()).digest()))
+
+
 class SecretsStore:
     """Named credentials, encrypted at rest with a key derived from the app key."""
 
     def __init__(self, path: Path, key: str) -> None:
         self._path = path
-        self._fernet = Fernet(
-            base64.urlsafe_b64encode(hashlib.sha256(key.encode()).digest())
-        )
+        self._fernet = fernet(key)
 
     def _read(self) -> dict[str, str]:
         if not self._path.exists():

backend/fluksio/main.py

@@ -3,6 +3,7 @@ import contextlib
 import hashlib
 import inspect
 import logging
+import sys
 import tempfile
 from collections.abc import AsyncIterator, Callable, Coroutine
 from contextlib import AbstractAsyncContextManager, asynccontextmanager
@@ -392,10 +393,11 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
         raise
     finally:
         await _release()
-        if settings.MCP_ENABLED:
-            from fluksio.mcp.http import aclose
-
-            await aclose()
+        # Imported by the endpoint when MCP is on and by the assistant on its
+        # first turn; either way the client it built is closed here.
+        tools = sys.modules.get("fluksio.mcp.server")
+        if tools is not None:
+            await tools.aclose()
 
 
 # The schema enumerates every endpoint this instance serves, including the

backend/fluksio/mcp/http.py

@@ -11,7 +11,6 @@ from __future__ import annotations
 import logging
 from urllib.parse import urlparse
 
-import httpx
 import jwt
 from fastapi.concurrency import run_in_threadpool
 from mcp.server.auth.provider import AccessToken
@@ -27,9 +26,6 @@ from fluksio.mcp import server
 
 logger = logging.getLogger(__name__)
 
-#: The tool calls never leave the process, so the host is a label, not a route.
-_INTERNAL_BASE = "http://fluksio-mcp.internal"
-
 
 class _JWTVerifier:
     """Accept only tokens minted for the MCP channel, by agents still registered.
@@ -112,20 +108,5 @@ def build_http_app(app: object) -> Starlette:
     # module-level so the tools can be declared at import time.
     server.mcp._token_verifier = _JWTVerifier()  # noqa: SLF001
 
-    server.set_client(
-        httpx.AsyncClient(
-            transport=httpx.ASGITransport(app=app),  # type: ignore[arg-type]
-            base_url=_INTERNAL_BASE,
-            # Running a flow is synchronous work on a threadpool, so a slow
-            # flow must not look like a dead endpoint.
-            timeout=120.0,
-            event_hooks={"request": [server.forward_caller_auth]},
-        )
-    )
+    server.connect(app)
     return server.mcp.streamable_http_app()
-
-
-async def aclose() -> None:
-    if server._client is not None:  # noqa: SLF001
-        await server._client.aclose()  # noqa: SLF001
-        server.set_client(None)

backend/fluksio/mcp/server.py

@@ -13,6 +13,7 @@ identity rather than some service account.
 from __future__ import annotations
 
 import logging
+from contextvars import ContextVar
 from typing import Any
 
 import httpx
@@ -23,28 +24,63 @@ logger = logging.getLogger(__name__)
 
 mcp = FastMCP("fluksio")
 
-#: Set by ``http.build_http_app``; the ASGI client that carries tool calls into
-#: the REST API without a network hop.
+#: The tool calls never leave the process, so the host is a label, not a route.
+_INTERNAL_BASE = "http://fluksio-mcp.internal"
+
+#: The ASGI client that carries tool calls into the REST API without a network
+#: hop. Built once by :func:`connect`, for the endpoint and the assistant alike.
 _client: httpx.AsyncClient | None = None
 
+#: The ``Authorization`` value for tool calls made from inside this process,
+#: where no MCP request carries one: the assistant sets it to the token of the
+#: person chatting, for the length of their turn.
+caller_authorization: ContextVar[str | None] = ContextVar(
+    "caller_authorization", default=None
+)
+
 
 def set_client(client: httpx.AsyncClient | None) -> None:
     global _client
     _client = client
 
 
-async def forward_caller_auth(request: httpx.Request) -> None:
-    """Carry the calling agent's bearer token onto the REST call.
+def connect(app: object) -> None:
+    """Wire the tools to *app*'s API, once."""
+    if _client is not None:
+        return
+    set_client(
+        httpx.AsyncClient(
+            transport=httpx.ASGITransport(app=app),  # type: ignore[arg-type]
+            base_url=_INTERNAL_BASE,
+            # Running a flow is synchronous work on a threadpool, so a slow
+            # flow must not look like a dead endpoint.
+            timeout=120.0,
+            event_hooks={"request": [forward_caller_auth]},
+        )
+    )
 
-    The transport attaches the originating HTTP request to every JSON-RPC
-    message, and a tool handler runs inside that request's context, so the
-    token is read per message rather than per connection.
+
+async def aclose() -> None:
+    if _client is not None:
+        await _client.aclose()
+        set_client(None)
+
+
+async def forward_caller_auth(request: httpx.Request) -> None:
+    """Carry the caller's bearer token onto the REST call.
+
+    Over the endpoint, the transport attaches the originating HTTP request to
+    every JSON-RPC message, and a tool handler runs inside that request's
+    context, so the token is read per message rather than per connection. A
+    call from inside the process has no such request and uses what
+    :data:`caller_authorization` holds.
     """
+    authorization = caller_authorization.get()
     try:
         source = request_ctx.get().request
     except LookupError:
-        return
-    authorization = getattr(source, "headers", {}).get("authorization")
+        source = None
+    authorization = getattr(source, "headers", {}).get("authorization") or authorization
     if authorization:
         request.headers["authorization"] = authorization
 

backend/tests/test_ai.py

@@ -0,0 +1,138 @@
+"""The assistant: who may set it up, and that it acts as the person chatting."""
+
+import json
+from collections.abc import Iterator
+from pathlib import Path
+from typing import Any
+
+import pytest
+from fastapi.testclient import TestClient
+
+from fluksio.ai import agent
+from fluksio.core.config import settings
+
+
+@pytest.fixture
+def ai_files(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
+    monkeypatch.setattr(settings, "AI_CONFIG_FILE", tmp_path / "ai.json")
+    monkeypatch.setattr(settings, "CHATS_DIR", tmp_path / "chats")
+    return tmp_path
+
+
+def _events(response: Any) -> list[dict[str, Any]]:
+    return [
+        json.loads(line[5:])
+        for line in response.iter_lines()
+        if line.startswith("data:")
+    ]
+
+
+def test_the_provider_is_a_superuser_setting_and_the_key_stays_sealed(
+    client: TestClient,
+    superuser_token_headers: dict[str, str],
+    normal_user_token_headers: dict[str, str],
+    ai_files: Path,
+) -> None:
+    assert client.get(
+        "/api/v1/ai/status", headers=normal_user_token_headers
+    ).json() == {
+        "available": False,
+        "model": None,
+    }
+    body = {"api_base": "http://llm.test/v1", "api_key": "sk-test-123456", "model": "m"}
+    refused = client.put(
+        "/api/v1/ai/config", json=body, headers=normal_user_token_headers
+    )
+    assert refused.status_code == 403
+
+    saved = client.put("/api/v1/ai/config", json=body, headers=superuser_token_headers)
+    assert saved.status_code == 200, saved.text
+    assert saved.json()["api_key_masked"] == "sk-…3456"
+    assert "sk-test" not in (ai_files / "ai.json").read_text()
+    assert client.get(
+        "/api/v1/ai/status", headers=normal_user_token_headers
+    ).json() == {
+        "available": True,
+        "model": "m",
+    }
+    # A save that names no key keeps the one there is.
+    kept = client.put(
+        "/api/v1/ai/config",
+        json={"api_base": "http://llm.test/v1", "model": "m2"},
+        headers=superuser_token_headers,
+    )
+    assert kept.json()["api_key_set"] is True
+    assert kept.json()["model"] == "m2"
+
+
+@pytest.mark.usefixtures("ai_files")
+def test_a_turn_runs_the_tools_as_the_person_chatting(
+    client: TestClient,
+    superuser_token_headers: dict[str, str],
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    client.put(
+        "/api/v1/ai/config",
+        json={"api_base": "http://llm.test/v1", "api_key": "k", "model": "m"},
+        headers=superuser_token_headers,
+    )
+    seen: list[list[dict[str, Any]]] = []
+    answers: Iterator[dict[str, Any]] = iter(
+        [
+            {
+                "role": "assistant",
+                "content": None,
+                "tool_calls": [
+                    {
+                        "id": "call_1",
+                        "type": "function",
+                        "function": {"name": "list_flows", "arguments": "{}"},
+                    }
+                ],
+            },
+            {"role": "assistant", "content": "Counted them."},
+        ]
+    )
+
+    async def fake_completion(**kwargs: Any) -> dict[str, Any]:
+        seen.append(kwargs["messages"])
+        assert any(t["function"]["name"] == "save_flow" for t in kwargs["tools"])
+        return {"choices": [{"message": next(answers)}]}
+
+    monkeypatch.setattr(agent.client, "chat_completion", fake_completion)
+
+    with client.stream(
+        "POST",
+        "/api/v1/ai/chat",
+        json={"message": "How many flows are there?"},
+        headers=superuser_token_headers,
+    ) as response:
+        assert response.status_code == 200
+        events = _events(response)
+
+    assert [e["type"] for e in events] == [
+        "start",
+        "tool_call",
+        "tool_result",
+        "message",
+        "done",
+    ]
+    # The tool went through the REST API with the caller's own token: an
+    # unauthenticated call would have come back as an error the model reads.
+    assert events[2] == {"type": "tool_result", "name": "list_flows", "ok": True}
+    answered = seen[-1][-1]
+    assert answered["role"] == "tool"
+    assert "count" in json.loads(answered["content"])
+    assert seen[-1][0]["role"] == "system"
+
+    transcript = client.get("/api/v1/ai/chat", headers=superuser_token_headers).json()
+    assert [m["role"] for m in transcript] == ["user", "assistant", "tool", "assistant"]
+    assert transcript[2] == {
+        "role": "tool",
+        "content": None,
+        "tool_calls": None,
+        "ok": True,
+        "error": None,
+    }
+    client.delete("/api/v1/ai/chat", headers=superuser_token_headers)
+    assert client.get("/api/v1/ai/chat", headers=superuser_token_headers).json() == []