stroblme
dde3d0af0d
The portal's copy of the dashboard is the `appshell` target of backend/Dockerfile (18 MB instead of the 252 MB nginx image); the nginx frontend Dockerfile, its confs and csp.check.ts are gone, since the app and the hub both hash the CSP from the index.html they serve. The runtime image copies the 4 MB git binary alone: the store's init, add, commit, gc and rev-parse are builtins, and a build-time round trip fails the build if that ever changes. About 100 MB of perl and helpers less. The bundled Redis's overcommit warning and FastMCP's lifespan warning no longer greet every `docker logs`.
376 lines
12 KiB
Python
376 lines
12 KiB
Python
"""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
|
|
import warnings
|
|
from contextvars import ContextVar
|
|
from typing import Any
|
|
|
|
import httpx
|
|
from mcp.server.fastmcp import FastMCP
|
|
from mcp.server.lowlevel.server import request_ctx
|
|
|
|
from fluksio.sdk.client import seg as _seg
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# FastMCP's own settings model leaves `lifespan` a forward reference, which
|
|
# pydantic-settings warns about on every start. Nothing reads that field from
|
|
# the environment, so the warning says nothing about this app.
|
|
with warnings.catch_warnings():
|
|
warnings.filterwarnings("ignore", message="Field 'lifespan' has an incomplete")
|
|
mcp = FastMCP("fluksio")
|
|
|
|
#: 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
|
|
|
|
|
|
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]},
|
|
)
|
|
)
|
|
|
|
|
|
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:
|
|
source = None
|
|
authorization = getattr(source, "headers", {}).get("authorization") or 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/{_seg(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/{_seg(name)}/nodes/{_seg(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/{_seg(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/{_seg(name)}/history/{_seg(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/{_seg(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. A parameter added
|
|
or dropped there adds or drops the input port or setting in the draft,
|
|
which moves its version on: read the flow again before saving it.
|
|
"""
|
|
return await _call(
|
|
"PUT", f"/flows/{_seg(name)}/nodes/{_seg(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/{_seg(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/{_seg(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/{_seg(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/{_seg(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/{_seg(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/{_seg(name)}/nodes/{_seg(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/{_seg(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/{_seg(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/{_seg(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/{_seg(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/{_seg(name)}/nodes/{_seg(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)
|