Skip to main content
Smithers workflows are normally written in TSX. The external workflow API lets you drive the same engine from any process that can read stdin and write JSON to stdout. The TypeScript side handles agents, schemas, and database setup; the external process owns the build logic. Python is the first-class external runtime. The createPythonWorkflow function wires everything together automatically.

​
Import

import {
  createExternalSmithers,
  createPythonWorkflow,
  pydanticSchemaToZod,
  serializeCtx,
  hostNodeToReact,
} from "smithers-orchestrator/external";

​
Host Node JSON Protocol

The bridge between an external process and the Smithers engine is the HostNodeJson type. Every time the engine calls the build function, it passes a serialized context on stdin and expects a HostNodeJson tree on stdout.

​
HostNodeJson

type HostNodeJson =
  | {
      kind: "element";
      tag: string;
      props: Record<string, string>;
      rawProps: Record<string, any>;
      children: HostNodeJson[];
    }
  | { kind: "text"; text: string };
Each element node maps 1:1 to a JSX component (Task, Approval, Signal, etc.). The tag field is the component name as a string. rawProps carries the full prop values including non-string types; props carries the string-serialized version used for display.

​
SerializedCtx

The engine serializes the current SmithersCtx before invoking the build function: 
type SerializedCtx = {
  runId: string;
  iteration: number;
  iterations: Record<string, number>;
  input: any;
  outputs: OutputSnapshot;
};
The external process receives this as JSON on stdin, uses it to decide which nodes to emit, and writes a HostNodeJson tree to stdout.

​
Agent Reference Resolution

String agent references in rawProps.agent are resolved back to live AgentLike objects before the tree reaches the engine. If a referenced agent name is not in the registry, the engine throws UNKNOWN_AGENT with the available agent names. 
// External process emits:
{ kind: "element", tag: "Task", rawProps: { agent: "claude" }, ... }

// TypeScript side resolves "claude" → actual AgentLike before rendering

​
createExternalSmithers

The low-level factory. Use this when your build function is already written in TypeScript (e.g., wrapping a non-Python subprocess or a WASM module). 
import { createExternalSmithers } from "smithers-orchestrator/external";

const workflow = createExternalSmithers({
  schemas: {
    analysis: z.object({ summary: z.string(), score: z.number() }),
  },
  agents: { claude: myClaudeAgent },
  buildFn: (ctx: SerializedCtx): HostNodeJson => {
    // Return a host node tree based on ctx
    return {
      kind: "element",
      tag: "Task",
      props: { id: "analyze" },
      rawProps: { id: "analyze", agent: myClaudeAgent },
      children: [],
    };
  },
});

​
ExternalSmithersConfig

type ExternalSmithersConfig<S extends Record<string, z.ZodObject<any>>> = {
  schemas: S;
  agents: Record<string, AgentLike>;
  buildFn: (ctx: SerializedCtx) => HostNodeJson;
  dbPath?: string;
};
OptionTypeDefaultDescription
schemasRecord<string, ZodObject>requiredZod schemas for output tables
agentsRecord<string, AgentLike>requiredAgent registry for ref resolution
buildFn(ctx: SerializedCtx) => HostNodeJsonrequiredSynchronous build function
dbPathstringephemeral temp dirPath for the SQLite database

​
Ephemeral SQLite Database

When dbPath is omitted, createExternalSmithers provisions an ephemeral SQLite database in a temp directory (os.tmpdir()/smithers-ext-*/smithers.db). WAL mode and a 5-second busy timeout are applied automatically. The database is closed on process exit. Pass an explicit dbPath for durable storage across restarts.

​
serializeCtx

Serialize a live SmithersCtx to a SerializedCtx for passing to the build function or an external process: 
import { serializeCtx } from "smithers-orchestrator/external";

const serialized = serializeCtx(ctx);
// { runId, iteration, iterations, input, outputs }

​
hostNodeToReact

Convert a HostNodeJson tree to React elements, resolving string agent references: 
import { hostNodeToReact } from "smithers-orchestrator/external";

const element = hostNodeToReact(hostNode, agents);
Throws UNKNOWN_AGENT if a referenced agent name is not present in the agents map.

​
Python Integration

createPythonWorkflow is the recommended entry point for Python-defined workflows. It combines schema auto-discovery, subprocess management, and the host node protocol into a single call.

​
Setup

Smithers uses uv to run Python scripts. Install uv and ensure it is on PATH: 
curl -LsSf https://astral.sh/uv/install.sh | sh
Your Python script must read a SerializedCtx JSON from stdin and write a HostNodeJson tree to stdout: 
import json
import sys

def run(ctx: dict) -> dict:
    return {
        "kind": "element",
        "tag": "Task",
        "props": {"id": "analyze"},
        "rawProps": {"id": "analyze", "agent": "claude"},
        "children": [],
    }

if __name__ == "__main__":
    ctx = json.loads(sys.stdin.read())
    print(json.dumps(run(ctx)))

​
createPythonWorkflow

import { createPythonWorkflow } from "smithers-orchestrator/external";

const workflow = createPythonWorkflow({
  scriptPath: "./workflow.py",
  agents: { claude: myClaudeAgent },
});
Schemas are auto-discovered from the Python script’s Pydantic models (see Schema Auto-Discovery below). Pass explicit Zod schemas to skip discovery: 
const workflow = createPythonWorkflow({
  scriptPath: "./workflow.py",
  agents: { claude: myClaudeAgent },
  schemas: {
    analysis: z.object({ summary: z.string(), score: z.number() }),
  },
});

​
Configuration

type PythonWorkflowConfig = {
  scriptPath: string;
  agents: Record<string, AgentLike>;
  schemas?: Record<string, z.ZodObject<any>>;
  dbPath?: string;
  cwd?: string;
  timeoutMs?: number;
  env?: Record<string, string>;
};
OptionTypeDefaultDescription
scriptPathstringrequiredPath to the Python script (relative to cwd)
agentsRecord<string, AgentLike>requiredAgent registry
schemasRecord<string, ZodObject>auto-discoveredZod schemas; omit to auto-discover from Pydantic
dbPathstringephemeralSQLite database path
cwdstringprocess.cwd()Working directory for subprocess
timeoutMsnumber30000Per-invocation timeout in milliseconds
envRecord<string, string>process.envAdditional environment variables

​
Build Subprocess

Each time the engine calls the build function, Smithers spawns uv run <scriptPath> synchronously. The serialized context is passed on stdin; the process must write HostNodeJson to stdout and exit with code 0. Exit code non-zero, no output, or invalid JSON all throw EXTERNAL_BUILD_FAILED. Timeout throws EXTERNAL_BUILD_FAILED with a timeout message. Stderr is captured and included in the error details.

​
Build Output Validation

The host node output is validated for a kind field before reaching the engine. The minimal valid output is: 
{ "kind": "text", "text": "hello" }
Or an element node: 
{
  "kind": "element",
  "tag": "Task",
  "props": {},
  "rawProps": { "id": "step1", "agent": "claude" },
  "children": []
}

​
Schema Auto-Discovery

When schemas is omitted from createPythonWorkflow, Smithers runs the script with --schemas and parses the JSON output as a map of schema names to JSON Schema objects. In your Python script, handle --schemas to emit Pydantic model schemas: 
import json
import sys
from pydantic import BaseModel

class Analysis(BaseModel):
    summary: str
    score: float

SCHEMAS = {"analysis": Analysis}

if __name__ == "__main__":
    if "--schemas" in sys.argv:
        print(json.dumps({
            name: model.model_json_schema()
            for name, model in SCHEMAS.items()
        }))
    else:
        ctx = json.loads(sys.stdin.read())
        # ... build and print HostNodeJson
Schema discovery runs once at startup. The discovered schemas are converted to Zod using pydanticSchemaToZod and passed to createExternalSmithers.

​
Pydantic Schema Conversion

pydanticSchemaToZod converts a Pydantic v2 JSON Schema (from model.model_json_schema()) to a Zod object schema. 
import { pydanticSchemaToZod } from "smithers-orchestrator/external";

const zodSchema = pydanticSchemaToZod(analysis.model_json_schema());

​
Supported Patterns

Pydantic PatternZod Output
type: "string" with minLength/maxLength/patternz.string().min().max().regex()
type: "number" / type: "integer" with minimum/maximumz.number().int().min().max()
type: "boolean"z.boolean()
type: "array" with itemsz.array(...)
type: "object" with properties + requiredz.object(...) with optional non-required fields
enum: [...] on a string fieldz.enum([...])
anyOf: [T, {type: "null"}] (Optional)T.nullable()
allOf: [A, B]z.intersection(A, B)
oneOf: [A, B, ...]z.union([A, B, ...])
$ref: "#/$defs/ModelName"Resolved inline (circular refs become z.any())
default: value.default(value)
description: "...".describe("...")

​
$ref Resolution

Pydantic places nested models under $defs. pydanticSchemaToZod resolves #/$defs/ModelName references inline using a JSON Pointer walk. Circular references are detected and collapsed to z.any() to prevent infinite recursion.

​
nullable anyOf Collapse

Pydantic represents Optional[T] as anyOf: [T, {type: "null"}]. The converter detects this two-variant pattern and collapses it to T.nullable() for clean column mapping in the SQLite schema.

​
allOf Intersection

allOf with a single entry is unwrapped directly. Multiple entries produce z.intersection(A, z.intersection(B, ...)).

​
Full Example

// workflow.ts
import { createPythonWorkflow } from "smithers-orchestrator/external";
import { Anthropic } from "@anthropic-ai/sdk";

const claude = new Anthropic();

export default createPythonWorkflow({
  scriptPath: "./workflow.py",
  cwd: import.meta.dir,
  agents: { claude },
  timeoutMs: 60_000,
  env: { ANTHROPIC_API_KEY: process.env.ANTHROPIC_API_KEY! },
});

# workflow.py
import json
import sys
from pydantic import BaseModel

class BugReport(BaseModel):
    title: str
    severity: str
    description: str

SCHEMAS = {"bugReport": BugReport}

def build(ctx: dict) -> dict:
    has_report = bool(ctx["outputs"].get("bugReport"))
    return {
        "kind": "element",
        "tag": "Task",
        "props": {"id": "triage"},
        "rawProps": {
            "id": "triage",
            "agent": "claude",
            "output": "bugReport",
        },
        "children": [
            {
                "kind": "text",
                "text": f"Triage this issue: {ctx['input'].get('description', '')}",
            }
        ],
    }

if __name__ == "__main__":
    if "--schemas" in sys.argv:
        print(json.dumps({k: v.model_json_schema() for k, v in SCHEMAS.items()}))
    else:
        ctx = json.loads(sys.stdin.read())
        print(json.dumps(build(ctx)))