add node-embedded custom tools

README.md

@@ -94,6 +94,32 @@ def run(inputs, ctx):
 
 The six built-in tools available to every node are `shell`, `read_file`, `write_file`, `edit_file`, `web_search`, and `web_fetch`. `read_file` also returns images (PNG/JPEG/GIF/WebP) as attachments for vision-capable models; `web_search` and `web_fetch` are backed by parallel.ai. Tools can be invoked either *agentically* — named in `ctx.agent(tools=[...])` so the node's own model decides when to call them — or *directly* via `ctx.tools.<name>(...)`, which runs them deterministically with no model in the loop (see [Design decisions](#design-decisions)).
 
+Nodes can also declare custom tools inside the same Python block—there is no separate node field or registry configuration. A lowercase `NodeTool` subclass registers automatically while the node source is loaded, and then uses the same quoted-name surfaces as built-ins:
+
+```python
+from app.runner.node_tools import NodeTool
+
+class lookup_order(NodeTool):
+    description = "Look up an order by id."
+    parameters = {
+        "type": "object",
+        "properties": {"order_id": {"type": "string"}},
+        "required": ["order_id"],
+    }
+
+    def execute(self, ctx, order_id):
+        return {"order_id": order_id, "status": "shipped"}
+
+def run(inputs, ctx):
+    result = ctx.agent(
+        prompt=f"Help with order {inputs['order_id']}",
+        tools=["lookup_order"],
+    )
+    return {"answer": result["content"]}
+```
+
+Custom tool names must use the same lowercase `snake_case` convention as native tools and cannot shadow built-in or MCP names. They are isolated to their node, included automatically in frozen run source, traced like native calls, and restored from that frozen source when an agent conversation is continued.
+
 ## Design decisions
 
 A few choices that shape how the system behaves in use:

backend/app/api/call_chats.py

@@ -86,6 +86,16 @@ def _node_name(db: Session, node_run: models.NodeRun) -> str:
     return node_run.node_id
 
 
+def _snapshot_node_code(db: Session, node_run: models.NodeRun) -> str:
+    """Frozen source for the node that produced ``node_run``."""
+
+    run = db.get(models.Run, node_run.run_id)
+    for node in ((run.workflow_snapshot if run else None) or {}).get("nodes", []) or []:
+        if node.get("id") == node_run.node_id:
+            return node.get("code") or ""
+    return ""
+
+
 def _build_continuation(db: Session, nr: models.NodeRun, call_id: str) -> models.CallChat:
     """Build — but do NOT persist — the continuation for one agent call,
     seeded from its recorded transcript. Raises 404 if the node run, the call,
@@ -255,6 +265,7 @@ def send_call_chat_turn(
         tools=list(chat.tools or []),
         model=model,
         child_env=child_env,
+        node_code=_snapshot_node_code(db, nr),
     )
     return schemas.CallChatTurnOut(turn_id=turn_id)
 

backend/app/orchestrator/prompt.py

@@ -159,6 +159,23 @@ def _format_node_tool_names() -> str:
 
 Every node defines a `run(inputs, ctx)` function. Top-level `import`s and small helper functions alongside `run` are fine — the whole code blob is `exec`'d into a fresh namespace per run, so reach for `json`, `re`, `pathlib`, etc. when they're cleaner than routing through an LLM.
 
+A node may also declare custom tools directly in the same code blob. Subclass `NodeTool`; the lowercase snake_case class name is the tool name and the runtime registers it automatically when the node source executes. Do not add a separate registry, decorator, or list. Custom tools are node-local, may call `ctx.tools.*`, and are available through the same name-based surfaces as native tools: agentically (`tools=["lookup_order"]`) and directly (`ctx.tools.lookup_order(...)`). Keep top-level code declarative (imports, helpers, tool classes, and `run`) because the frozen source is also loaded when continuing one of the node's agent calls.
+
+```python
+from app.runner.node_tools import NodeTool
+
+class lookup_order(NodeTool):
+    description = "Look up an order by id."
+    parameters = {
+        "type": "object",
+        "properties": {"order_id": {"type": "string"}},
+        "required": ["order_id"],
+    }
+
+    def execute(self, ctx, order_id):
+        return {"order_id": order_id, "status": "shipped"}
+```
+
 If a node imports a third-party package, make sure it's installed *before* the workflow runs.
 
 ```python
@@ -171,7 +188,7 @@ def run(inputs, ctx):
 
 `ctx` provides:
 
-- `ctx.agent(prompt, tools=[...])` — runs an LLM inside the node. Pass tool names ([[NODE_TOOL_NAMES]]) in the `tools` list; the LLM running inside the node decides when to invoke them. Returns a dict with keys `content` (str), `tool_calls_made` (list), `usage`, `cost`. Omit the `model` arg (see *# design conventions*). Optional `label` when a node makes several calls — keep it short and meaningful.
+- `ctx.agent(prompt, tools=[...])` — runs an LLM inside the node. Pass registered tool names ([[NODE_TOOL_NAMES]] plus any lowercase custom `NodeTool` class names) as strings; the LLM running inside the node decides when to invoke them. Returns a dict with keys `content` (str), `tool_calls_made` (list), `usage`, `cost`. Omit the `model` arg (see *# design conventions*). Optional `label` when a node makes several calls — keep it short and meaningful.
 - `ctx.tools.shell(...)` / `ctx.tools.read_file(...)` / `ctx.tools.web_fetch(...)` / … — direct (non-LLM) tool calls, same names, returning the same dicts the LLM-mediated form would produce. The agentic form above is the default; reserve direct calls for when there's nothing for a model to decide (see *# direct calls vs wrapping the tool in an agent*).
 - `ctx.log("...")` — appends a visible line to the run log.
 - `ctx.workdir` — `pathlib.Path` to a per-run scratch directory.

backend/app/runner/chat.py

@@ -176,6 +176,7 @@ def start_chat_turn(
     tools: list[str],
     model: str,
     child_env: dict[str, str],
+    node_code: str = "",
 ) -> None:
     """Begin a chat turn in the background. Returns immediately.
 
@@ -189,7 +190,7 @@ def start_chat_turn(
     ev_mod.get_or_create(turn_id)
     threading.Thread(
         target=_run_turn,
-        args=(turn_id, chat_id, messages, tools, model, child_env),
+        args=(turn_id, chat_id, messages, tools, model, child_env, node_code),
         daemon=True,
     ).start()
 
@@ -215,13 +216,18 @@ def _run_turn(
     tools: list[str],
     model: str,
     child_env: dict[str, str],
+    node_code: str = "",
 ) -> None:
     workdir = tempfile.mkdtemp(prefix="wfchat-")
     try:
         payload = {
             "messages": messages,
             "tools": tools,
             "model": model,
+            # Frozen source from the run that produced this conversation.
+            # The chat child re-executes it to recover embedded NodeTool
+            # subclasses without storing a second copy on CallChat.
+            "node_code": node_code,
             "workdir": workdir,
             "env": child_env,
         }

backend/app/runner/chat_child.py

@@ -7,7 +7,8 @@
 
 Reads a JSON payload from stdin::
 
-    {"messages": [...], "tools": [...], "model": "...", "workdir": "...", "env": {...}}
+    {"messages": [...], "tools": [...], "model": "...", "node_code": "...",
+     "workdir": "...", "env": {...}}
 
 Emits the same per-call event contract a run does — ``llm_call_started``,
 ``llm_round_started``, ``llm_call_chunk``, ``tool_call_started/finished``,
@@ -36,6 +37,7 @@ def main() -> None:
     payload = _read_payload()
     messages = payload.get("messages") or []
     tools = payload.get("tools") or []
+    node_code = payload.get("node_code") or ""
     model = payload.get("model") or payload.get("default_model") or ""
     workdir = Path(payload.get("workdir") or ".")
     workdir.mkdir(parents=True, exist_ok=True)
@@ -55,6 +57,14 @@ def main() -> None:
 
     cancelled = False
     try:
+        # A continuation exposes the exact custom tools from the run it
+        # continues, not whatever the live node contains now. The API sends
+        # frozen node source; loading it registers NodeTool subclasses but
+        # never invokes the node's ``run`` function.
+        if node_code:
+            from app.runner.node_tools import execute_node_source
+
+            execute_node_source(node_code, ctx.register_node_tool)
         result = ctx.agent(model=model, prompt=messages, tools=tools)
     except KeyboardInterrupt:
         cancelled = True

backend/app/runner/child.py

@@ -141,8 +141,9 @@ def _emit_node_skipped(node_id: str, inputs: dict, output_ports: list[dict]) ->
 def _execute_node(node: dict, ctx, inputs: dict) -> dict:
     """Exec the node's user code and return its (port-normalised) outputs.
     Raises if the code is malformed or returns the wrong shape."""
-    ns: dict = {}
-    exec(node.get("code") or "", ns, ns)
+    from app.runner.node_tools import execute_node_source
+
+    ns = execute_node_source(node.get("code") or "", ctx.register_node_tool)
     run_fn = ns.get("run")
     if not callable(run_fn):
         raise RuntimeError("node code must define `run(inputs, ctx)` function")