---
title: External Workflows
description: Build Smithers workflows from non-TSX sources — including Python scripts — using the host node JSON protocol and Pydantic schema auto-discovery.
---

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

```ts
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

```ts
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:

```ts
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.

```ts
// 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).

```ts
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

```ts
type ExternalSmithersConfig<S extends Record<string, z.ZodObject<any>>> = {
  schemas: S;
  agents: Record<string, AgentLike>;
  buildFn: (ctx: SerializedCtx) => HostNodeJson;
  dbPath?: string;
};
```

| Option | Type | Default | Description |
|---|---|---|---|
| `schemas` | `Record<string, ZodObject>` | required | Zod schemas for output tables |
| `agents` | `Record<string, AgentLike>` | required | Agent registry for ref resolution |
| `buildFn` | `(ctx: SerializedCtx) => HostNodeJson` | required | Synchronous build function |
| `dbPath` | `string` | ephemeral temp dir | Path 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:

```ts
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:

```ts
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](https://github.com/astral-sh/uv) to run Python scripts. Install uv and ensure it is on `PATH`:

```bash
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:

```python
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

```ts
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](#schema-auto-discovery) below). Pass explicit Zod schemas to skip discovery:

```ts
const workflow = createPythonWorkflow({
  scriptPath: "./workflow.py",
  agents: { claude: myClaudeAgent },
  schemas: {
    analysis: z.object({ summary: z.string(), score: z.number() }),
  },
});
```

### Configuration

```ts
type PythonWorkflowConfig = {
  scriptPath: string;
  agents: Record<string, AgentLike>;
  schemas?: Record<string, z.ZodObject<any>>;
  dbPath?: string;
  cwd?: string;
  timeoutMs?: number;
  env?: Record<string, string>;
};
```

| Option | Type | Default | Description |
|---|---|---|---|
| `scriptPath` | `string` | required | Path to the Python script (relative to `cwd`) |
| `agents` | `Record<string, AgentLike>` | required | Agent registry |
| `schemas` | `Record<string, ZodObject>` | auto-discovered | Zod schemas; omit to auto-discover from Pydantic |
| `dbPath` | `string` | ephemeral | SQLite database path |
| `cwd` | `string` | `process.cwd()` | Working directory for subprocess |
| `timeoutMs` | `number` | `30000` | Per-invocation timeout in milliseconds |
| `env` | `Record<string, string>` | `process.env` | Additional 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:

```json
{ "kind": "text", "text": "hello" }
```

Or an element node:

```json
{
  "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:

```python
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.

```ts
import { pydanticSchemaToZod } from "smithers-orchestrator/external";

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

### Supported Patterns

| Pydantic Pattern | Zod Output |
|---|---|
| `type: "string"` with `minLength`/`maxLength`/`pattern` | `z.string().min().max().regex()` |
| `type: "number"` / `type: "integer"` with `minimum`/`maximum` | `z.number().int().min().max()` |
| `type: "boolean"` | `z.boolean()` |
| `type: "array"` with `items` | `z.array(...)` |
| `type: "object"` with `properties` + `required` | `z.object(...)` with optional non-required fields |
| `enum: [...]` on a string field | `z.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

```ts
// 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! },
});
```

```python
# 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)))
```