fix(mcp): annotate tools, confirm deletes, fence app output, expire sessions

docs/src/content/docs/configuration.md

@@ -98,7 +98,8 @@ in).
 
 Environment: `API_URL`, `AUTH_ENABLED`, `OIDC_ISSUER`, `OIDC_ISSUER_INTERNAL`,
 `OIDC_DEVICE_CLIENT_ID` (from the UI's OIDC Secret); optional `OIDC_JWKS_URL`,
-`OIDC_AUDIENCE`, `HOST`, `PORT`, `LOG_LEVEL`.
+`OIDC_AUDIENCE`, `HOST`, `PORT`, `LOG_LEVEL`, `MCP_SESSION_TTL_SECONDS` (default 28800:
+how long an idle device-flow login is kept).
 
 ## Logs and probes
 

docs/src/content/docs/mcp.md

@@ -59,15 +59,35 @@ anonymously.
 | `authenticate` | Start/complete the device-flow login, or confirm the cached token. |
 | `describe_cluster` | Source types, launchable namespaces, apps domain — call first to pick valid options. |
 | `launch_app` | Create + launch (static, or Python/pixi via `pixi_task`); optional `resources` (requests/limits, cluster defaults otherwise). **Idempotent on (namespace, name)** — re-launching your own app updates it instead of failing, so retries are safe; an app owned by someone else is refused, never overwritten. |
-| `list_apps` | Apps with phase, URL, owner; optional namespace filter. |
-| `get_app` | Full spec + status for one app. |
+| `list_apps` | Apps with phase, URL, owner; optional namespace filter. Inline sources list file `paths` only. |
+| `get_app` | Full spec + status for one app; inline files come back fenced (see below). |
 | `get_app_status` | Lightweight poll: phase, URL, replicas, conditions. |
-| `get_app_logs` | Recent pod logs — first stop when phase is `Failed`. |
+| `get_app_logs` | Recent pod logs (fenced, last 64 KiB) — first stop when phase is `Failed`. |
 | `update_app` | Patch spec (source, runtime, access, …). |
 | `stop_app` / `start_app` | Scale to zero / back up. |
 | `restart_app` | Roll the app's pods without changing the spec; fails if the app is stopped. |
 | `get_app_metrics` | Instantaneous CPU/memory per pod (`available:false` when the cluster has no metrics server). |
-| `remove_app` | Delete permanently (cascades; the tool description tells agents to confirm with the user). |
+| `remove_app` | Delete permanently (cascades). Refused unless called with `confirm: true`, which the agent sets only after the user agrees. |
+
+Every tool carries MCP annotations, so clients that auto-approve tools can tell them
+apart: `describe_cluster`, `list_apps`, `get_app`, `get_app_status`, `get_app_logs` and
+`get_app_metrics` are `readOnlyHint`; `launch_app` (it updates an existing app),
+`update_app`, `stop_app` and `remove_app` are `destructiveHint`.
+
+### Untrusted output
+
+Pod logs and inline source files are written by whoever launched the app, and they land in
+the agent's context. Tools return them inside markers with a random nonce
+(`<untrusted-1a2b…>` … `</untrusted-1a2b…>`) that the content cannot close early, next to
+a `notice` telling the agent to treat the content as data, never as instructions. Logs are
+capped to the last 64 KiB and each inline file to its first 16 KiB (`truncated` /
+`truncatedFiles` say when).
+
+### Sessions
+
+Device-flow logins are kept in the MCP server's memory per MCP session, and dropped after
+`MCP_SESSION_TTL_SECONDS` (default 8 hours) without use. A connection without a session
+id cannot log in. Because sessions are in memory, `mcp.replicaCount` must stay 1.
 
 Every tool returns structured JSON. Errors carry the API's human-readable `detail`, and a
 `401` includes `nextAction: call the authenticate tool`, so agents can self-recover.

mcp/src/nebari_apps_mcp/auth.py

@@ -28,6 +28,7 @@ class TokenState:
     user_code: str = ""
     interval: float = 5.0
     device_expires_at: float = 0.0
+    last_used: float = field(default_factory=time.time)
 
     @property
     def valid(self) -> bool:
@@ -39,7 +40,19 @@ class DeviceFlowAuth:
     _sessions: dict[str, TokenState] = field(default_factory=dict)
 
     def state(self, session_id: str) -> TokenState:
-        return self._sessions.setdefault(session_id or "anonymous", TokenState())
+        """This session's login. Sessions idle longer than the TTL are dropped first.
+
+        Without a session id there is nothing to key a login on, and a shared
+        fallback would hand one agent's token to another, so it is refused.
+        """
+        if not session_id:
+            raise MissingSessionError("this MCP connection has no session id; reconnect with a stateful client")
+        now = time.time()
+        for stale in [sid for sid, s in self._sessions.items() if now - s.last_used > settings.session_ttl]:
+            del self._sessions[stale]
+        state = self._sessions.setdefault(session_id, TokenState())
+        state.last_used = now
+        return state
 
     async def authenticate(self, session_id: str) -> dict[str, Any]:
         """Start, poll, or confirm a device-flow login for this session."""
@@ -55,7 +68,10 @@ async def authenticate(self, session_id: str) -> dict[str, Any]:
                 "keycloak.url (the OIDC secret provides the client id once the NebariApp reconciles)",
             }
 
-        state = self.state(session_id)
+        try:
+            state = self.state(session_id)
+        except MissingSessionError as exc:
+            return {"status": "unavailable", "message": str(exc)}
 
         if state.valid:
             return {"status": "authenticated", "message": "already logged in; token is valid"}
@@ -160,6 +176,10 @@ async def bearer(self, session_id: str, passthrough: str = "") -> str:
 auth = DeviceFlowAuth()
 
 
+class MissingSessionError(Exception):
+    pass
+
+
 class TokenVerificationError(Exception):
     pass
 

mcp/src/nebari_apps_mcp/config.py

@@ -57,6 +57,11 @@ class Settings:
     # so a plain audience check cannot do this. Unset disables the check.
     oidc_allowed_clients: list[str] | None = field(default_factory=lambda: _client_list("OIDC_ALLOWED_CLIENTS"))
 
+    # Seconds a device-flow login (and its refresh token) is kept after the
+    # session last used it. Sessions live in this process's memory, which is
+    # why mcp.replicaCount must stay 1.
+    session_ttl: float = field(default_factory=lambda: float(os.environ.get("MCP_SESSION_TTL_SECONDS", "28800")))
+
     host: str = field(default_factory=lambda: os.environ.get("HOST", "0.0.0.0"))
     port: int = field(default_factory=lambda: int(os.environ.get("PORT", "8080")))
 

mcp/src/nebari_apps_mcp/server.py

@@ -20,10 +20,11 @@
 from starlette.requests import Request
 from starlette.responses import JSONResponse, Response
 
-from .auth import TokenVerificationError, auth, get_validator
+from .auth import MissingSessionError, TokenVerificationError, auth, get_validator
 from .client import ApiClient, ApiError
 from .config import settings
 from .logs import REQUEST_ID_HEADER, RequestIdMiddleware, configure_logging
+from .untrusted import fence_app, fence_logs, summarize_app
 
 if TYPE_CHECKING:
     import mcp_types as mt
@@ -104,7 +105,10 @@ async def on_call_tool(
         if context.fastmcp_context is not None:
             session_id = context.fastmcp_context.session_id or ""
         passthrough = get_http_headers().get("authorization", "")
-        bearer = await auth.bearer(session_id, passthrough)
+        try:
+            bearer = await auth.bearer(session_id, passthrough)
+        except MissingSessionError as exc:
+            raise ToolError(str(exc)) from exc
         if not bearer:
             raise ToolError("not authenticated: call the authenticate tool and follow its instructions")
         token = bearer.removeprefix("Bearer ").strip()
@@ -120,7 +124,7 @@ async def on_call_tool(
 
 
 def _session_id(ctx: Context) -> str:
-    return ctx.session_id or "anonymous"
+    return ctx.session_id or ""
 
 
 async def _api(ctx: Context) -> ApiClient:
@@ -129,14 +133,18 @@ async def _api(ctx: Context) -> ApiClient:
     return ApiClient(bearer, request_id=_request_id())
 
 
+# Hints for clients that auto-approve tools: these only read.
+READ_ONLY: dict[str, Any] = {"readOnlyHint": True, "openWorldHint": False}
+
+
 def _error(exc: ApiError) -> dict[str, Any]:
     out: dict[str, Any] = {"error": exc.detail, "status": exc.status}
     if exc.status == 401:
         out["nextAction"] = "call the authenticate tool and follow its instructions"
     return out
 
 
-@mcp.tool
+@mcp.tool(annotations={"readOnlyHint": False, "destructiveHint": False, "idempotentHint": True, "openWorldHint": False})
 async def authenticate(ctx: Context) -> dict[str, Any]:
     """Log in to the cluster via the Keycloak device flow, or confirm the cached login.
 
@@ -147,7 +155,7 @@ async def authenticate(ctx: Context) -> dict[str, Any]:
     return await auth.authenticate(_session_id(ctx))
 
 
-@mcp.tool
+@mcp.tool(annotations=READ_ONLY)
 async def describe_cluster(ctx: Context) -> dict[str, Any]:
     """Describe what this cluster supports: available source types, namespaces apps may be
     launched into, and the apps domain (apps get https://<subdomain>.<appsDomain>).
@@ -160,7 +168,7 @@ async def describe_cluster(ctx: Context) -> dict[str, Any]:
         return _error(exc)
 
 
-@mcp.tool
+@mcp.tool(annotations=READ_ONLY)
 async def list_apps(
     ctx: Context,
     namespace: Annotated[str, Field(description="Filter to one namespace; empty = all visible namespaces")] = "",
@@ -170,12 +178,13 @@ async def list_apps(
     """
     try:
         async with await _api(ctx) as api:
-            return await api.get("/apps", params={"namespace": namespace} if namespace else None)
+            apps = await api.get("/apps", params={"namespace": namespace} if namespace else None)
+            return [summarize_app(a) for a in apps]
     except ApiError as exc:
         return _error(exc)
 
 
-@mcp.tool
+@mcp.tool(annotations={"readOnlyHint": False, "destructiveHint": True, "idempotentHint": True, "openWorldHint": False})
 async def launch_app(
     ctx: Context,
     name: Annotated[str, Field(description="Kubernetes name: lowercase letters, digits, hyphens")],
@@ -294,17 +303,17 @@ async def launch_app(
         return _error(exc)
 
 
-@mcp.tool
+@mcp.tool(annotations=READ_ONLY)
 async def get_app(ctx: Context, namespace: str, name: str) -> Any:
     """Full spec and status for one app: source, runtime, access, URL, and conditions."""
     try:
         async with await _api(ctx) as api:
-            return await api.get(f"/apps/{namespace}/{name}")
+            return fence_app(await api.get(f"/apps/{namespace}/{name}"))
     except ApiError as exc:
         return _error(exc)
 
 
-@mcp.tool
+@mcp.tool(annotations=READ_ONLY)
 async def get_app_status(ctx: Context, namespace: str, name: str) -> Any:
     """Lightweight status poll: phase, URL, replica readiness, conditions, message.
     Poll this after launch_app until phase=Running (typically seconds; image pulls can
@@ -317,7 +326,7 @@ async def get_app_status(ctx: Context, namespace: str, name: str) -> Any:
         return _error(exc)
 
 
-@mcp.tool
+@mcp.tool(annotations=READ_ONLY)
 async def get_app_logs(
     ctx: Context,
     namespace: str,
@@ -329,12 +338,13 @@ async def get_app_logs(
     """
     try:
         async with await _api(ctx) as api:
-            return await api.get(f"/apps/{namespace}/{name}/logs", params={"lines": lines})
+            resp = await api.get(f"/apps/{namespace}/{name}/logs", params={"lines": lines})
+            return fence_logs(str(resp.get("logs", "")))
     except ApiError as exc:
         return _error(exc)
 
 
-@mcp.tool
+@mcp.tool(annotations={"readOnlyHint": False, "destructiveHint": True, "idempotentHint": True, "openWorldHint": False})
 async def update_app(
     ctx: Context,
     namespace: str,
@@ -349,47 +359,49 @@ async def update_app(
     """Update an existing app's spec. The operator rolls the workload automatically."""
     try:
         async with await _api(ctx) as api:
-            return await api.patch(f"/apps/{namespace}/{name}", json=patch)
+            return fence_app(await api.patch(f"/apps/{namespace}/{name}", json=patch))
     except ApiError as exc:
         return _error(exc)
 
 
-@mcp.tool
+@mcp.tool(annotations={"readOnlyHint": False, "destructiveHint": True, "idempotentHint": True, "openWorldHint": False})
 async def stop_app(ctx: Context, namespace: str, name: str) -> Any:
     """Scale an app to zero replicas (phase becomes Stopped). Routing and configuration
     are kept; start_app brings it back.
     """
     try:
         async with await _api(ctx) as api:
-            return await api.post(f"/apps/{namespace}/{name}/stop")
+            return fence_app(await api.post(f"/apps/{namespace}/{name}/stop"))
     except ApiError as exc:
         return _error(exc)
 
 
-@mcp.tool
+@mcp.tool(annotations={"readOnlyHint": False, "destructiveHint": False, "idempotentHint": True, "openWorldHint": False})
 async def start_app(ctx: Context, namespace: str, name: str) -> Any:
     """Scale a stopped app back up to one replica."""
     try:
         async with await _api(ctx) as api:
-            return await api.post(f"/apps/{namespace}/{name}/start")
+            return fence_app(await api.post(f"/apps/{namespace}/{name}/start"))
     except ApiError as exc:
         return _error(exc)
 
 
-@mcp.tool
+@mcp.tool(
+    annotations={"readOnlyHint": False, "destructiveHint": False, "idempotentHint": False, "openWorldHint": False}
+)
 async def restart_app(ctx: Context, namespace: str, name: str) -> Any:
     """Roll the app's pods without changing its spec (like `kubectl rollout restart`).
     Use after external content or config changes, or to recover a wedged pod. Fails if the
     app is stopped (start_app first).
     """
     try:
         async with await _api(ctx) as api:
-            return await api.post(f"/apps/{namespace}/{name}/restart")
+            return fence_app(await api.post(f"/apps/{namespace}/{name}/restart"))
     except ApiError as exc:
         return _error(exc)
 
 
-@mcp.tool
+@mcp.tool(annotations=READ_ONLY)
 async def get_app_metrics(ctx: Context, namespace: str, name: str) -> Any:
     """Instantaneous CPU (millicores) and memory (Mi) usage per pod, from metrics.k8s.io.
     Returns {available, pods:[{name, cpu, memory}]}; available=false when the cluster has no
@@ -402,11 +414,22 @@ async def get_app_metrics(ctx: Context, namespace: str, name: str) -> Any:
         return _error(exc)
 
 
-@mcp.tool
-async def remove_app(ctx: Context, namespace: str, name: str) -> dict[str, Any]:
+@mcp.tool(annotations={"readOnlyHint": False, "destructiveHint": True, "idempotentHint": True, "openWorldHint": False})
+async def remove_app(
+    ctx: Context,
+    namespace: str,
+    name: str,
+    confirm: Annotated[
+        bool, Field(description="Must be true: set it only after the user has agreed to delete this app")
+    ] = False,
+) -> dict[str, Any]:
     """Delete an app permanently. Cascades: workload, routing, TLS certificate, and OIDC
-    client are all removed. There is no undo - confirm with the user first.
+    client are all removed. There is no undo - ask the user first, then call with confirm=true.
     """
+    if not confirm:
+        raise ToolError(
+            f"remove_app deletes {namespace}/{name} permanently; ask the user to confirm, then call again with confirm=true"
+        )
     try:
         async with await _api(ctx) as api:
             await api.delete(f"/apps/{namespace}/{name}")