LangGraph construction

Status: Current | Last reviewed: 2026-10-08

How async-agent turns a workflow definition into a running LangGraph StateGraph.

The big picture

workflow_versions.config (JSONB)
  ─ nodes: array of { id, type, config, ... }
  ─ edges: array of { from, to, condition? }
       │
       ▼
workflow_graph_builder.py
  ─ create StateGraph(WorkflowState)
  ─ for each node: graph.add_node(id, registry[type].handler)
  ─ for each edge: graph.add_edge or add_conditional_edges
  ─ compile checkpointer → CompiledStateGraph
       │
       ▼
Runtime: graph.ainvoke(initial_state)

Key files

File	What it does
apps/async-agent/app/services/workflow_graph_builder.py	Builds the StateGraph from the JSON config
apps/async-agent/app/nodes/registry.py	Maps node type strings to handler functions
apps/async-agent/app/nodes/	One file per node type
apps/async-agent/app/workers/consumer.py	Pulls jobs from the queue and runs the graph
shared/workflow-node-types/	Cross-language registry of node types and their schemas

State

The shared WorkflowState (a TypedDict) carries:

• inputs — the trigger payload.
• messages — accumulated LLM messages (LangChain format).
• outputs — node-output values keyed by node ID.
• tenant_id, conversation_id, agent_id — context.
• metadata — run-level metadata (run ID, started_at, retry counts).

Each node handler receives the state, returns a partial state to merge in. LangGraph handles the merge.

Node handlers

A node handler is a function:

def my_node(state: WorkflowState, config: dict) -> dict:
    # ... do work ...
    return {"outputs": {**state["outputs"], "my_node_id": result}}

Handlers are registered by decorating with @register_node("type_name") in their module file. The registry imports every file under app/nodes/ at startup so all handlers register.

Built-in node types

Type	File	Purpose
tool	tool.py	Invoke an MCP tool via agent-host
llm	llm.py	One LLM call with structured output schema
foreach	foreach.py	Iterate a collection; fan out and gather
human_review	human_review.py	Pause; resume on approval row write
code	code.py	Run sandboxed Python via code-executor
condition	condition.py	Pure-data routing
branch	branch.py	Subgraph routing
pii_redaction	pii_redaction.py	Multi-provider PII scrub
webhook	webhook.py	POST output to an external URL
set_variable	set_variable.py	Mutate state explicitly

Edges and conditional routing

A simple linear edge becomes graph.add_edge("a", "b").

A conditional edge becomes graph.add_conditional_edges("a", router, {"yes": "b", "no": "c"}) where router(state) returns one of the keys.

The builder reads edges[].condition from the config — a string expression evaluated against state. The expression language is intentionally tiny: state["outputs"]["node_x"] == "value" style. No arbitrary code.

Checkpointing

LangGraph's checkpointer interface persists state between nodes. async-agent uses a Postgres-backed checkpointer that writes to workflow_run_checkpoints. Long-running workflows (anything with human_review or sleep nodes) survive process restarts.

Configure via WORKFLOW_CHECKPOINTER_URL (defaults to DATABASE_URL).

Per-node execution rows in workflow_node_executions are emitted by node_execution_tracer.py. The trace input.config is rendered against the current workflow state before it is stored, so the run details UI shows values such as {{input.ani}} as the concrete value used by that node.

Workflow test jobs that target the draft version force-refresh the workflow config in consumer.py before phase initialization and graph construction. That keeps a just-saved editor change, such as a node rename, aligned with the node execution trace that the Test tab reads back.

Template rendering also supports inline JSONata with {{jsonata: ...}}. Expressions evaluate against the same document as Assignment JSONata and End output mappings: upstream node IDs at top level, plus system, vars, input, trigger, context, iterator, node_results, and state. Object and list results render as formatted JSON text for prompt fields. The ai-manager prompt editor keeps workflow variable autocomplete active inside the JSONata directive and inserts selected fields as raw JSONata paths.

A Tool node's parameter-mapping form (ToolNodeParameterEditor) builds its fields from the referenced tool's parameter schema, and it reads that schema from three shapes depending on tool type: config.cached_tool_schemas[] per selected sub-tool (MCP / multi-tool OpenAPI), config.input_schema (workflow tools), or — as a fallback — the tool row's top-level input_schema column. That last fallback is load-bearing for authored tools whose schema lives in the tool_configs.input_schema/parameters columns rather than inside config, notably manual API tools (type_id:'openapi', subtype:'manual') and Python tools. Without it those tools always rendered "No parameter schema available for this tool" and their params could never be mapped, even though the schema was correctly stored. If a tool still shows that message, its stored schema is genuinely empty — re-save the tool (MCP: discover + save; authored: fill in Parameter Definitions + save) so a schema is persisted.

The workflow editor can persist a node-local config.output_schema for tools whose platform adapter can infer a richer result contract than generic MCP metadata declares. Merge-field, End output, and JSONata autocomplete consume that schema generically. The first native adapter is Salesforce MCP: for getRelatedRecords, ai-manager follows the configured relationship path via Salesforce describe child-relationship metadata, describes the child object, and writes a records-array schema back to the tool node.

Adding a new node type

1. Create apps/async-agent/app/nodes/my_node.py:

```python from .registry import register_node

@register_node("my_node") def handler(state, config): ... return {...} ```

1. Add the schema to shared/workflow-node-types/. This makes it surface in the ai-manager workflow editor.

2. Wire the editor: extend apps/ai-manager/app/(main)/workflows/[id]/... with the inspector form for the new type.

That is it — no changes needed in workflow_graph_builder.py. The registry pattern handles dispatch.

Comparing to the agent pipeline

Both use a JSON config + draft/published model. Where they differ:

	Agent pipeline	Async workflow
Runtime	Pipecat (real-time, audio frames)	LangGraph (durable, message-driven)
Stage edges	Implicit (linear chain)	Explicit (DAG with conditions)
State	Per-frame, ephemeral	Persisted via checkpointer
Latency budget	sub-100ms per stage	seconds to hours
Restartability	Drop the call on failure	Resume from last checkpoint

Some agents have both — a real-time pipeline that fires post-call workflows for analysis or follow-up. The link is in agent_versions.config.lifecycle.on_complete.

See also

• Async request lifecycle
• Workflow triggers
• async-agent service doc
• Universal pipeline builder

Decision nodes

router is the stored type for the Decision editor. Its ordered outputs retain {id, label, expression} plus optional editor_mode. Visual rules serialize to {{jsonata: ...}}; unsupported expressions remain raw. decision_mode defaults to rules. In ai mode, config.decision holds provider, model, shared input, optional inline images, named questions, timeout, and error/refusal routing policies. Credentials remain on the node envelope as integration_id.

The provider adapter validates answer names, types, allowed options, probability distributions, and numeric ranges before placing answers in the current node result. Branch evaluation receives that result in a temporary state, so <node>.answers.<question>.choice works in the same node. The chosen route and assessment are persisted together. Provider errors and refusals never become ordinary successful choices. Unconnected router outcomes map to END rather than the first available edge.

The JSONata Python binding embeds source inside a JavaScript string. workflow_expressions.py escapes that transport layer before invoking it; removing this step can turn a quoted value into executable expression syntax. The shared decision-rule fixtures run through both the visual parser and the Python evaluator. Regenerate the standalone export kernel after changing this boundary.

Single-node preview uses the same router handler. It restricts sample state to input, variables, prior node results, trigger and iterator context; tenant identity comes from authentication. AI assessment calls are refused in evaluation sandboxes pending a pinned mock contract. Provider usage is retained as run evidence; it is not currently booked into the general LLM cost ledger because Decisions has separate endpoint pricing.

Visual canvas inputs

Canvas configuration and testing must not require JSON entry when typed controls can represent the data. WorkflowValueEditor renders recursive object/list values with or without declared fields. The workflow Tester uses it for declared, nested, and ad-hoc inputs; saved-input and optional JSON-view round trips retain native types. GenericSchemaPanel renders typed array items and visual open objects instead of multiline or raw-JSON fallbacks.

DecisionSampleEditor builds the existing preview API's sample envelope from visual sections. Its field picker maps upstream outputs into node_results, including nested arrays, while inputs, variables, triggers, and loop context retain their runtime scopes. Samples are local UI state. Assessment inputs and rule operands offer field pickers. Image controls encode uploaded files as inline data URLs (2 MB combined upload limit) or store a selected field's merge token; authors never need to paste base64 or JSON. Raw expressions and read-only diagnostics remain optional.