adeelahmad · Oct 3, 2026 · Oct 3, 2026 · Oct 3, 2026 · Oct 3, 2026 · Oct 3, 2026
diff --git a/fastapi_backend/app/api/v1/router.py b/fastapi_backend/app/api/v1/router.py
@@ -10,6 +10,7 @@
     collections,
     comments,
     content_types,
+    custom_nodes,
     entities,
     entity_setup,
     fields,
@@ -72,6 +73,7 @@
     pipelines,
     content_types,
     workflows,
+    custom_nodes,
     chats,
     collections,
     batches,

diff --git a/fastapi_backend/app/api/v1/routes/custom_nodes.py b/fastapi_backend/app/api/v1/routes/custom_nodes.py
@@ -0,0 +1,79 @@
+"""Custom nodes: bodies of nodes saved under a name, kept private, shared with namespaces or with everyone."""
+
+from __future__ import annotations
+
+import contextlib
+from collections.abc import Iterator
+
+from fastapi import APIRouter, HTTPException
+
+from app.api.deps import CurrentUser, Db, Principal, Writer, domain_errors
+from app.domain import auth, custom_nodes
+from app.schemas.common import Created, Ok
+from app.schemas.custom_nodes import CustomNode, CustomNodeCreate, CustomNodeUpdate, CustomNodeVersionCreate
+from app.schemas.templates import VersionSaved
+
+router = APIRouter(prefix="/custom-nodes", tags=["workflows"])
+
+
+def me(user: Principal) -> dict:
+    return custom_nodes.who(user.id, user.email, user.admin and user.via != "oauth", user.roles)
+
+
+@contextlib.contextmanager
+def _errors() -> Iterator[None]:
+    with domain_errors():
+        try:
+            yield
+        except PermissionError as e:
+            raise HTTPException(403, str(e)) from None
+
+
+@router.get("")
+def list_custom_nodes(user: CurrentUser, db: Db, scope: str | None = None) -> list[CustomNode]:
+    """The custom nodes you can use: yours, the ones shared with your namespaces or with everyone (admins: all)."""
+    return custom_nodes.visible(db, me(user), scope)
+
+
+@router.post("")
+def create_custom_node(body: CustomNodeCreate, user: Writer, db: Db) -> Created:
+    with _errors():
+        nid = custom_nodes.create(
+            db, me(user), body.name, body.graph.model_dump(), body.params, body.description, body.icon, body.color,
+            body.visibility, body.namespaces,
+        )  # fmt: skip
+    auth.audit(db, user.as_audit(), "custom_node.create", f"custom_node:{nid}")
+    return Created(id=nid)
+
+
+@router.get("/{nid}")
+def get_custom_node(nid: int, user: CurrentUser, db: Db, version: int | None = None) -> CustomNode:
+    """One version (default: the current one) and the list of versions."""
+    with _errors():
+        return custom_nodes.visible_one(db, me(user), nid, version)
+
+
+@router.post("/{nid}/versions")
+def create_custom_node_version(nid: int, body: CustomNodeVersionCreate, user: Writer, db: Db) -> VersionSaved:
+    with _errors():
+        n = custom_nodes.save_version(db, me(user), nid, body.graph.model_dump(), body.params, body.notes)
+    auth.audit(db, user.as_audit(), "custom_node.save", f"custom_node:{nid}", {"version": n})
+    return VersionSaved(version=n)
+
+
+@router.patch("/{nid}")
+def update_custom_node(nid: int, body: CustomNodeUpdate, user: Writer, db: Db) -> Ok:
+    """Its name, look, and who sees it."""
+    with _errors():
+        custom_nodes.update(db, me(user), nid, **body.model_dump(exclude_unset=True))
+    auth.audit(db, user.as_audit(), "custom_node.update", f"custom_node:{nid}")
+    return Ok()
+
+
+@router.delete("/{nid}")
+def delete_custom_node(nid: int, user: Writer, db: Db) -> Ok:
+    """Takes it off the palette; workflows that use it keep running the version they pinned."""
+    with _errors():
+        custom_nodes.remove(db, me(user), nid)
+    auth.audit(db, user.as_audit(), "custom_node.delete", f"custom_node:{nid}")
+    return Ok()
diff --git a/fastapi_backend/app/api/v1/routes/workflows.py b/fastapi_backend/app/api/v1/routes/workflows.py
@@ -4,8 +4,9 @@
 
 from fastapi import APIRouter, HTTPException
 
-from app.api.deps import Acl, AdminWriter, CurrentUser, Db, Writer, domain_errors
-from app.domain import auth, jobs, workflows
+from app.api.deps import Acl, AdminWriter, Cfg, CurrentUser, Db, Writer, domain_errors
+from app.api.v1.routes.custom_nodes import me
+from app.domain import auth, custom_nodes, flow, jobs, store, workflows
 from app.schemas.common import Created, Ok
 from app.schemas.pipelines import JobQueued
 from app.schemas.templates import VersionSaved
@@ -15,39 +16,62 @@
     WorkflowCatalog,
     WorkflowCreate,
     WorkflowRunRequest,
+    WorkflowTry,
+    WorkflowTryRequest,
     WorkflowUpdate,
     WorkflowVersionCreate,
 )
 
 router = APIRouter(prefix="/workflows", tags=["workflows"])
 
 
-def _ports(t):
-    if t in workflows.TERMINAL:
-        return []
-    return ["yes", "no"] if t == "condition" else ["out"]
+DYNAMIC = {"switch": "outputs", "set": "inputs", "group": "both", "custom": "both"}
+
+
+def _node_type(t):
+    kit = next(k for k in workflows.KITS.values() if t in k.all_types)
+    ins, outs, many = flow.ports({"type": t, "config": {}}, kit)
+    return NodeType(
+        type=t,
+        settings=sorted(workflows.CONFIG[t]),
+        scopes=[sc for sc, types in workflows.SCOPES.items() if t in types],
+        inputs=0 if not ins else -1 if many else len(ins),
+        outputs=outs,
+        input_ports=ins,
+        dynamic=DYNAMIC.get(t),
+        primitive=t in flow.PRIMITIVES,
+        keeps=t in workflows.TERMINAL,
+    )
 
 
 @router.get("")
 def list_workflows(user: CurrentUser, db: Db) -> WorkflowCatalog:
-    """Saved workflows, plus the nodes a workflow can be built from."""
+    """Saved workflows, plus the nodes a workflow can be built from: the primitives, each scope's own nodes, and the
+    custom nodes you can use."""
     return WorkflowCatalog(
-        node_types=[
-            NodeType(
-                type=t,
-                settings=sorted(workflows.CONFIG[t]),
-                scopes=[sc for sc, types in workflows.SCOPES.items() if t in types],
-                inputs=0 if t == "input" else -1 if t == "merge" else 1,
-                outputs=_ports(t),
-            )
-            for t in workflows.NODE_TYPES
-        ],
+        node_types=[_node_type(t) for t in workflows.NODE_TYPES],
         operators=list(workflows.OPS),
         entity_types=list(workflows.ENTITY_TYPES),
         workflows=workflows.list_workflows(db),
+        custom_nodes=custom_nodes.visible(db, me(user)),
     )
 
 
+@router.post("/test")
+def try_workflow(body: WorkflowTryRequest, user: AdminWriter, acl: Acl, db: Db, cfg: Cfg) -> WorkflowTry:
+    """Run a graph (saved or not) once without keeping anything: what each node passed on, for the canvas. Nodes
+    that would save something say what they would save; models are still asked."""
+    spaces = None
+    if body.scope == "recording":
+        if body.recording is None:
+            raise HTTPException(400, "choose a recording to try it on")
+        acl.recording(body.recording, "editor")
+    else:
+        spaces = [acl.namespace(n) for n in body.namespaces] if body.namespaces else sorted(store.space_names(db))
+    with domain_errors():
+        return WorkflowTry(**workflows.try_graph(db, cfg, body.graph.model_dump(), body.scope, body.recording, spaces, user.email))
+
+
 @router.post("")
 def create_workflow(body: WorkflowCreate, user: AdminWriter, db: Db) -> Created:
     with domain_errors():