Let agents drive the flow API over MCP

The engine now speaks MCP at /mcp, with a built-in OAuth 2.1 authorization
server in front of it: an agent registers itself, sends a human to the browser
to approve it, and exchanges the resulting code for a token. PKCE is required,
codes are single-use and stored only as hashes, the browser is redirected to
the URI that was registered rather than the one asked for, and refresh tokens
rotate so that replaying a spent one revokes the whole line.

Twenty tools cover reading, building, publishing and running flows, and each
one calls the same REST endpoint the dashboard calls, in-process, carrying the
caller's own token. That keeps one description of what a flow is and what may
be done to it — validation, the draft/publish split, the version check — and
means an agent can do nothing a person could not do in the browser.

Agent tokens are RS256 with a keypair of their own rather than the secret that
signs browser sessions, so deleting the key withdraws every agent without
logging anyone out, and deps.decode_token grew the branch that trusting a
second issuer will need when the hosted login arrives.

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

backend/app/mcp/server.py

@@ -0,0 +1,240 @@
+"""The tools an agent can call, each one a request to the flow API.
+
+Tools do not reach into the engine: they call the same REST endpoints the
+dashboard calls, over an in-process ASGI transport. That keeps one description
+of what a flow is and how it may be changed — validation, the draft/publish
+split, the version check that stops two clients overwriting each other — and it
+means an agent cannot do anything a person could not do in the browser.
+
+The caller's token rides along on every hop, so the API sees the agent's own
+identity rather than some service account.
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+import httpx
+from mcp.server.fastmcp import FastMCP
+from mcp.server.lowlevel.server import request_ctx
+
+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.
+_client: httpx.AsyncClient | None = 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.
+
+    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.
+    """
+    try:
+        source = request_ctx.get().request
+    except LookupError:
+        return
+    authorization = getattr(source, "headers", {}).get("authorization")
+    if authorization:
+        request.headers["authorization"] = authorization
+
+
+async def _call(method: str, path: str, **kwargs: Any) -> Any:
+    """One REST call, with failures handed back as data rather than raised.
+
+    An agent can read an error and try something else; a transport fault just
+    ends the conversation.
+    """
+    if _client is None:
+        return {"error": "The MCP endpoint is not running."}
+    try:
+        response = await _client.request(method, f"/api/v1{path}", **kwargs)
+    except httpx.HTTPError as exc:
+        logger.exception("MCP call to %s failed", path)
+        return {"error": f"Could not reach the flow API: {exc}"}
+
+    if response.status_code >= 400:
+        detail: Any = response.text
+        try:
+            detail = response.json().get("detail", detail)
+        except ValueError:
+            pass
+        return {"error": detail, "status": response.status_code}
+    if not response.content:
+        return {"ok": True}
+    return response.json()
+
+
+# -----------------------------------------------------------------------------
+# Reading
+# -----------------------------------------------------------------------------
+
+
+@mcp.tool()
+async def list_flows() -> Any:
+    """Every flow, with its node count, errors, and whether it is running."""
+    return await _call("GET", "/flows/")
+
+
+@mcp.tool()
+async def get_flow(name: str) -> Any:
+    """One flow: its definition, the state of its nodes, and its problems.
+
+    The definition is the working copy — unpublished edits included — which is
+    what to base a change on.
+    """
+    return await _call("GET", f"/flows/{name}")
+
+
+@mcp.tool()
+async def list_node_types() -> Any:
+    """The node types that can be placed, with their parameter schemas."""
+    return await _call("GET", "/flows/node-types")
+
+
+@mcp.tool()
+async def get_node_source(name: str, node_id: str) -> Any:
+    """The Python source of one node."""
+    return await _call("GET", f"/flows/{name}/nodes/{node_id}/source")
+
+
+@mcp.tool()
+async def get_flow_state(name: str) -> Any:
+    """The last value seen on every message of a flow."""
+    return await _call("GET", f"/flows/{name}/state")
+
+
+@mcp.tool()
+async def get_message_history(name: str, message: str) -> Any:
+    """Recent numeric values of one message, oldest first."""
+    return await _call("GET", f"/flows/{name}/history/{message}")
+
+
+@mcp.tool()
+async def list_secrets() -> Any:
+    """The names of stored secrets, for pointing a node parameter at one.
+
+    Values are never returned. Reference one from a node parameter as
+    ``{"$secret": "<name>"}``.
+    """
+    return await _call("GET", "/secrets/")
+
+
+@mcp.tool()
+async def list_shared_nodes() -> Any:
+    """Node sources shared across flows, and which nodes use each."""
+    return await _call("GET", "/flows/library")
+
+
+# -----------------------------------------------------------------------------
+# Building
+# -----------------------------------------------------------------------------
+
+
+@mcp.tool()
+async def save_flow(name: str, definition: dict[str, Any]) -> Any:
+    """Save a flow as an unpublished draft.
+
+    ``definition`` is a whole flow document — the shape `get_flow` returns
+    under ``definition``. It must carry the ``version`` that was read, so an
+    edit someone else made in between is refused rather than overwritten; on a
+    conflict, read the flow again and reapply the change.
+
+    Nothing here reaches the engine until `publish_flow`.
+    """
+    return await _call("PUT", f"/flows/{name}", json=definition)
+
+
+@mcp.tool()
+async def save_node_source(name: str, node_id: str, code: str) -> Any:
+    """Save a node's Python source and report whether it compiles.
+
+    A node defines ``process(...)``, taking one argument per input port plus
+    ``params``, and returns a dict keyed by output port.
+    """
+    return await _call(
+        "PUT", f"/flows/{name}/nodes/{node_id}/source", json={"code": code}
+    )
+
+
+@mcp.tool()
+async def publish_flow(name: str, version: int) -> Any:
+    """Deploy a flow's unpublished changes: this is what puts them live."""
+    return await _call("POST", f"/flows/{name}/publish", json={"version": version})
+
+
+@mcp.tool()
+async def discard_draft(name: str) -> Any:
+    """Throw unpublished changes away and go back to what is running."""
+    return await _call("POST", f"/flows/{name}/discard-draft")
+
+
+@mcp.tool()
+async def delete_flow(name: str) -> Any:
+    """Delete a flow and the code of its nodes."""
+    return await _call("DELETE", f"/flows/{name}")
+
+
+@mcp.tool()
+async def validate_flow(name: str) -> Any:
+    """What would keep this flow from running: loops, unconnected inputs."""
+    return await _call("POST", f"/flows/{name}/validate")
+
+
+# -----------------------------------------------------------------------------
+# Running
+# -----------------------------------------------------------------------------
+
+
+@mcp.tool()
+async def run_flow(name: str, inputs: dict[str, Any] | None = None) -> Any:
+    """Run every node of a flow once and return the resulting state.
+
+    With unpublished changes this runs the draft, so a change can be tried
+    before it is published.
+    """
+    return await _call("POST", f"/flows/{name}/run", json={"inputs": inputs or {}})
+
+
+@mcp.tool()
+async def trigger_node(
+    name: str, node_id: str, values: dict[str, Any] | None = None
+) -> Any:
+    """Feed values into one node and run everything downstream of it."""
+    return await _call(
+        "POST", f"/flows/{name}/nodes/{node_id}/trigger", json={"values": values or {}}
+    )
+
+
+@mcp.tool()
+async def start_flow(name: str) -> Any:
+    """Let the engine run this flow: subscriptions, schedules and webhooks."""
+    return await _call("POST", f"/flows/{name}/start")
+
+
+@mcp.tool()
+async def stop_flow(name: str) -> Any:
+    """Take a flow off the engine. Nothing of it stays subscribed or scheduled."""
+    return await _call("POST", f"/flows/{name}/stop")
+
+
+@mcp.tool()
+async def pause_flow(name: str) -> Any:
+    """Hold a flow's nodes while its incoming values keep arriving."""
+    return await _call("POST", f"/flows/{name}/pause")
+
+
+@mcp.tool()
+async def resume_flow(name: str) -> Any:
+    """Let a paused flow carry on, running whatever was held back."""
+    return await _call("POST", f"/flows/{name}/resume")