app/backend/fluksio/mcp/server.py

"""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")


@mcp.tool()
async def get_graph() -> Any:
    """Every flow as one graph, with nodes talking to the same thing merged.

    Nodes on the same broker topic, URL or bucket come back as a single entry,
    which is how wiring that runs between flows shows up.
    """
    return await _call("GET", "/flows/graph")


# -----------------------------------------------------------------------------
# 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 and
    one per setting, 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")


@mcp.tool()
async def cancel_node(name: str, node_id: str) -> Any:
    """Stop a node's code while it is running. Nothing to stop is not an error."""
    return await _call("POST", f"/flows/{name}/nodes/{node_id}/cancel")


# -----------------------------------------------------------------------------
# Modules
# -----------------------------------------------------------------------------


@mcp.tool()
async def get_modules() -> Any:
    """The python packages node code can import, and the manifest asking for them."""
    return await _call("GET", "/modules/")


@mcp.tool()
async def apply_modules(requirements: str) -> Any:
    """Install exactly these requirements, one pip line each.

    This replaces the whole manifest: a package left out is uninstalled. A
    manifest that does not resolve changes nothing.
    """
    return await _call("POST", "/modules/apply", json={"requirements": requirements})


# -----------------------------------------------------------------------------
# Observability
# -----------------------------------------------------------------------------


@mcp.tool()
async def get_health() -> Any:
    """How the engine is doing right now: flows, nodes, queue and loop lag."""
    return await _call("GET", "/observability/summary")


@mcp.tool()
async def get_metrics(
    flow: str | None = None, node: str | None = None, hours: int = 24
) -> Any:
    """Executions, errors and timings per minute over the last few hours."""
    params: dict[str, Any] = {"hours": hours}
    if flow:
        params["flow"] = flow
    if node:
        params["node"] = node
    return await _call("GET", "/observability/timeseries", params=params)


@mcp.tool()
async def list_failures(flow: str | None = None, limit: int = 50) -> Any:
    """Recent failures with their tracebacks, newest first."""
    params: dict[str, Any] = {"kind": "failure", "limit": limit}
    if flow:
        params["flow"] = flow
    return await _call("GET", "/observability/events", params=params)


@mcp.tool()
async def list_runs(flow: str | None = None, limit: int = 50) -> Any:
    """Recent cascades: what triggered them, how long they took, how they ended.

    A page of rows plus the total number matching, which says whether the limit
    cut anything off.
    """
    params: dict[str, Any] = {"limit": limit}
    if flow:
        params["flow"] = flow
    return await _call("GET", "/observability/runs", params=params)