Multi-agent composition

The Corti Agent SDK is in alpha v2 private preview. The API may change between releases. Contact help@corti.ai to request access.

The Agent SDK provides three composition primitives for coordinating multiple agents. Start with workflow; only reach for stateGraph when you need cycles or branching that depends on accumulated state.

At a glance

The three patterns differ in topology:

Workflow: linear pipeline

Each step runs in order. A when predicate can skip a step, and transform can rewrite the input to the next step.

Parallel: concurrent fan-out

All agents run at the same time on the same input. Fulfilled results are collected; rejected steps are isolated.

StateGraph: cycles and routing

Nodes share a typed state object. Routing functions inspect the state after each node and pick the next node, END, or loop back. maxIterations bounds cycles.

Workflow

A Workflow is a fixed list of steps. Each step receives the previous step’s response (or a transform of it) and returns a new response.

import { workflow } from "@corti/agent-sdk";

const result = await workflow([
  summarizer,                               // step 1: bare agent
  classifier,                               // step 2: bare agent
  {                                         // step 3: full config
    agent: escalator,
    when: (prev) => (prev.text ?? "").toLowerCase().includes("urgent"),
    transform: () => note,
    retries: 2,
    retryDelay: 500,
  },
]).run(note);

console.log(result.output.text);    // final step's text
console.log(result.steps);          // per-step responses
console.log(result.stoppedEarly);   // true if a step failed

from corti_agent_sdk import workflow

result = await workflow([
    summarizer,                              # step 1: bare agent
    classifier,                              # step 2: bare agent
    {                                        # step 3: full config
        "agent": escalator,
        "when": lambda prev: "urgent" in (prev.text or ""),
        "transform": lambda _: note,
        "retries": 2,
        "retry_delay": 0.5,
    },
]).run(note)

print(result.output.text)     # final step's text
print(result.steps)           # per-step responses
print(result.stopped_early)   # True if a step failed

Step options

Each step can be a bare AgentHandle, a Parallel group (auto-wrapped), or a configuration object:

Field	Description
`agent`	Any object with a `run(input)` method: an `AgentHandle`, a `Parallel` group, or a custom runnable
`when`	Predicate on the previous response. If false, the step is skipped and the previous response passes forward unchanged
`transform`	Map the previous response to a new input for this step. Default: `prev.text`
`retries`	Additional attempts when a step returns `status: "failed"` (default 0)
`retryDelay` / `retry_delay`	Delay between retries in milliseconds (TypeScript) or seconds (Python), default 1000 ms / 1.0 s

WorkflowResult

Field	Type	Description
`output`	`MessageResponse`	The last executed response
`steps`	`MessageResponse[]`	Responses from every executed step (skipped steps excluded)
`stoppedEarly` / `stopped_early`	`boolean`	True when a step failed and stopped execution early

Parallel

Parallel runs multiple agents concurrently on the same input. Use it standalone or drop it into a workflow step list.

import { parallel, workflow } from "@corti/agent-sdk";

// Standalone
const fanout = await parallel([differential, redFlags, workup]).run(presentation);
fanout.fulfilled;  // MessageResponse[]: successful results
fanout.rejected;   // unknown[]: exceptions from failed steps

// Inside a workflow: fulfilled outputs are joined with newlines
const { output } = await workflow([
  parallel([differential, redFlags, workup]),
  synthesizer,
]).run(presentation);

from corti_agent_sdk import parallel, workflow

# Standalone
fanout = await parallel([differential, red_flags, workup]).run(presentation)
fanout.fulfilled  # List[MessageResponse]: successful results
fanout.rejected   # List: exceptions from failed steps

# Inside a workflow: fulfilled outputs are joined with newlines
result = await workflow([
    parallel([differential, red_flags, workup]),
    synthesizer,
]).run(presentation)

Per-step overrides

Pass per-step input or credentials when branches need different data:

parallel([
  { agent: differential, input: "specialised prompt for differential" },
  { agent: redFlags, input: "focus on red flags only" },
  workup,  // bare handle: uses shared input
]);

parallel([
    {"agent": differential, "input": "specialised prompt for differential"},
    {"agent": red_flags, "input": "focus on red flags only"},
    workup,  # bare handle: uses shared input
])

ParallelResult

Field	Type	Description
`results`	`PromiseSettledResult[]` / `List[Dict]`	One entry per step with `status` and `value` or `reason`
`fulfilled`	`MessageResponse[]`	Responses from steps that completed without raising
`rejected`	`unknown[]`	Exceptions from steps that raised

Parallel swallows individual failures. Inspect rejected if you need fail-fast semantics. When used inside a Workflow, if all parallel steps fail, the workflow throws.

StateGraph

When you need cycles or branching that depends on accumulated state, use StateGraph<TState>. Each node mutates a typed state object; edges route to the next node, including loops bounded by maxIterations.

Concepts

State: a plain typed object that accumulates across every node execution. Each node returns a Partial<S> that is shallow-merged in.
Nodes: async functions (state: S) => Promise<Partial<S>>. Use agentNode() to wrap an AgentHandle.
Edges: a static node name, END, or a routing function (state: S) => string | END that runs after the node updates the state.
END: sentinel that stops execution. A node with no registered edge also terminates the run.

Minimal example

import { stateGraph, agentNode, END } from "@corti/agent-sdk";

interface TriageState {
  note: string;
  severity: string;
  codes: string;
  approved: boolean;
}

const graph = stateGraph<TriageState>()
  .addNode("triage", agentNode(
    triageAgent,
    (s) => s.note,
    (r) => ({ severity: r.text ?? "" }),
  ))
  .addNode("coder", agentNode(
    coderAgent,
    (s) => s.note,
    (r) => ({ codes: r.text ?? "" }),
  ))
  .addNode("reviewer", agentNode(
    reviewerAgent,
    (s) => `Note: ${s.note}\n\nProposed codes: ${s.codes}`,
    (r) => ({ approved: (r.text ?? "").toLowerCase().startsWith("approved") }),
  ))
  .addEdge("triage", (s) =>
    s.severity.toLowerCase().includes("urgent") ? "coder" : END,
  )
  .addEdge("coder", "reviewer")
  .addEdge("reviewer", (s) => (s.approved ? END : "coder"));

const result = await graph.run("triage", { note, severity: "", codes: "", approved: false }, { maxIterations: 10 });

result.state;         // final TriageState
result.steps;         // per-node deltas
result.iterations;    // number of node executions
result.terminatedBy;  // "end" | "maxIterations" | "noEdge"

from corti_agent_sdk import stateGraph, agent_node, END

class TriageState(TypedDict):
    note: str
    severity: str
    codes: str
    approved: bool


graph = (
    stateGraph()
    .add_node("triage", agent_node(
        triage_agent,
        input_fn=lambda s: s["note"],
        merge_fn=lambda r: {"severity": r.text or ""},
    ))
    .add_node("coder", agent_node(
        coder_agent,
        input_fn=lambda s: s["note"],
        merge_fn=lambda r: {"codes": r.text or ""},
    ))
    .add_node("reviewer", agent_node(
        reviewer_agent,
        input_fn=lambda s: f"Note: {s['note']}\n\nProposed codes: {s['codes']}",
        merge_fn=lambda r: {"approved": (r.text or "").lower().startswith("approved")},
    ))
    .add_edge("triage", lambda s: "coder" if "urgent" in s["severity"].lower() else END)
    .add_edge("coder", "reviewer")
    .add_edge("reviewer", lambda s: END if s["approved"] else "coder")
)

result = await graph.run("triage", {"note": note, "severity": "", "codes": "", "approved": False}, max_iterations=10)

result.state          # final state dict
result.steps          # per-node deltas
result.iterations     # number of node executions
result.terminated_by  # "end" | "maxIterations" | "noEdge"

agentNode

Wraps an AgentHandle as a node function. Provide two callbacks: one to extract the agent’s input from state, and one to merge the response back.

agentNode(
  myAgent,
  (state) => state.input,                     // extract input from state
  (response) => ({ output: response.text ?? "" }), // merge response into state
)

agent_node(
    my_agent,
    input_fn=lambda s: s["input"],
    merge_fn=lambda r: {"output": r.text or ""},
)

Custom nodes

You can add non-agent nodes that transform state directly:

graph.addNode("normalize", (s) => ({ note: s.note.trim() }));

graph.add_node("normalize", lambda s: {"note": s["note"].strip()})

Result shape

Field	Type	Description
`state`	`S`	Final accumulated state after all nodes ran
`steps`	`StateGraphStep<S>[]`	Per-node history: `node` name, `delta`, post-delta `state`
`iterations`	`number`	Total node executions (including repeated nodes in cycles)
`terminatedBy` / `terminated_by`	`"end" \| "maxIterations" \| "noEdge"`	Why the graph stopped

Choosing between workflow and stateGraph

	`workflow()`	`stateGraph()`
Shape	Linear list of steps	Named nodes with explicit edges
Shared state	Previous response text only	Typed object, accumulated
Branching	`when` predicate (skip or run)	Routing function (pick any node)
Cycles	None	Supported, bounded by `maxIterations`
Best for	Known fixed pipelines	Conditional flows, review loops, dynamic routing

Decision guide

You want to…	Use
Run a fixed pipeline of agents (A to B to C)	`workflow`
Run agents concurrently on the same input	`parallel`
Conditionally branch, loop, or share typed state across agents	`stateGraph`

Quickstart

Concepts

Guides

Reference

Registry Connectors

Agent SDK

v1 Documentation (archived)

At a glance

Workflow

Step options

WorkflowResult

Parallel

Per-step overrides

ParallelResult

StateGraph

Concepts

Minimal example

agentNode

Custom nodes

Result shape

Choosing between workflow and stateGraph

Decision guide

​At a glance

​Workflow

​Step options

​WorkflowResult

​Parallel

​Per-step overrides

​ParallelResult

​StateGraph

​Concepts

​Minimal example

​agentNode

​Custom nodes

​Result shape

​Choosing between workflow and stateGraph

​Decision guide

At a glance

Workflow

Step options

WorkflowResult

Parallel

Per-step overrides

ParallelResult

StateGraph

Concepts

Minimal example

agentNode

Custom nodes

Result shape

Choosing between workflow and stateGraph

Decision guide