Services

The three entry points: the Tier-1 one-liner, the Tier-2 typed builder, and the base class for custom agents.

Korch

Korch(
    *,
    settings: Settings | None = None,
    model_gateway: IModelGateway | None = None,
    runtime: IDurableRuntime | None = None,
    router: BaseRouter | None = None,
    repository: GraphRepository | None = None,
    middleware: Sequence[Middleware] = (),
    connectors: Sequence[Connector]
    | ConnectorRegistry
    | None = None,
)

The one-liner entry point to durable multi-agent execution.

Constructing Korch wires the object graph: each collaborator is taken as given, or resolved from Settings once, here. With no arguments and no environment it runs locally with MockLM, the in-process runtime, and in-memory persistence.

Parameters:

Name	Type	Description	Default
`settings`	`Settings \| None`	Explicit settings; resolved from the environment when omitted.	`None`
`model_gateway`	`IModelGateway \| None`	The gateway to inject; resolved from config when omitted.	`None`
`runtime`	`IDurableRuntime \| None`	The durable runtime to inject; resolved from config when omitted.	`None`
`router`	`BaseRouter \| None`	The model router to inject; resolved from config when omitted.	`None`
`repository`	`GraphRepository \| None`	The graph repository to inject; resolved from config when omitted.	`None`
`middleware`	`Sequence[Middleware]`	Middleware fired around each superstep, in registration order.	`()`
`connectors`	`Sequence[Connector] \| ConnectorRegistry \| None`	Tools available to the Architect's plan (ADR 0015) — a :class:`~korchestrator.tools.ConnectorRegistry` this `Korch` owns, or connectors to build one from. Omit entirely if no agent needs tools; an agent whose planned `AgentConfig.tools` is non-empty with no `connectors` given raises `ConfigurationError` (P10.2).	`None`

Example

from korchestrator import Korch korch = Korch() korch.run("Summarize durable agent execution") # doctest: +SKIP

Store the (optionally injected) collaborators; resolution happens on first use.

cancel

cancel(run_id: str) -> None

Cancel run_id; it terminates with RunStatus.CANCELLED.

Raises:

Type	Description
`NotImplementedError`	On the local runtime.
`MissingExtraError`	If the Temporal runtime is selected without the `[temporal]` extra.

edit_resume

edit_resume(
    run_id: str,
    *,
    updates: Mapping[str, JSONValue] | None = None,
    trust_delta: float = 0.0,
) -> None

Apply an operator's context/trust edit to a paused run, then resume it.

Goes through the same reducer discipline the barrier itself uses (last-value merge for updates, a clamped fold for trust_delta) — an edit is as replayable and auditable as an agent's own update.

Parameters:

Name	Type	Description	Default
`run_id`	`str`	The paused run to edit and resume.	required
`updates`	`Mapping[str, JSONValue] \| None`	Context-channel values to merge into the paused state.	`None`
`trust_delta`	`float`	Folded into `trust_score`, clamped to `[0.0, 1.0]`.	`0.0`

Raises:

Type	Description
`NotImplementedError`	On the local runtime.
`MissingExtraError`	If the Temporal runtime is selected without the `[temporal]` extra.

on

on(event: str, handler: EventHandler) -> Self

Register handler for event (e.g. "superstep"); returns self.

pause

pause(run_id: str) -> None

Signal run_id to pause for human review (durable HITL, spec 06 §7).

A paused run consumes no compute while it awaits resume, edit_resume, or cancel, bounded by a 24h deadline after which it times out. Governance also triggers this automatically when trust drops below the configured threshold — this method is for an operator-initiated pause.

Raises:

Type	Description
`NotImplementedError`	On the local runtime, which is synchronous and has no in-flight run to pause. Use `KORCH_RUNTIME=temporal` for durable HITL.
`MissingExtraError`	If the Temporal runtime is selected without the `[temporal]` extra.

resume

resume(run_id: str) -> None

Lift a pause and let run_id continue from its checkpointed state.

Raises:

Type	Description
`NotImplementedError`	On the local runtime.
`MissingExtraError`	If the Temporal runtime is selected without the `[temporal]` extra.

run

run(
    objective: str, *, max_supersteps: int = 10
) -> RunResult

Run a swarm against objective and return the terminal :class:RunResult.

Classifies the objective, has the Architect plan a team of agents, and drives the plan through the kernel and the configured runtime.

Parameters:

Name	Type	Description	Default
`objective`	`str`	The goal, at least 10 characters.	required
`max_supersteps`	`int`	Hard halt bound. Defaults to 10.	`10`

Returns:

Type	Description
`RunResult`	The terminal :class:`RunResult`, including `final_answer`.

Raises:

Type	Description
`ValidationError`	If `objective` is shorter than 10 characters, or `max_supersteps` is outside 1-100.
`MissingExtraError`	If reasoning is used without the `[dspy]` extra.

Example

from korchestrator import Korch Korch().run("Summarize durable agent execution") # doctest: +SKIP

Swarm

Swarm(
    objective: str,
    *,
    settings: Settings | None = None,
    model_gateway: IModelGateway | None = None,
    runtime: IDurableRuntime | None = None,
    router: BaseRouter | None = None,
    repository: GraphRepository | None = None,
    middleware: Sequence[Middleware] = (),
    connectors: Sequence[Connector]
    | ConnectorRegistry
    | None = None,
)

A typed, explicitly constructed agent swarm.

Build the topology with :meth:add and :meth:edges (both return Self so chaining type-checks), then :meth:run it. Collaborators are optional and resolved from Settings at the composition root when omitted.

Parameters:

Name	Type	Description	Default
`objective`	`str`	The goal, at least 10 characters.	required
`settings`	`Settings \| None`	Explicit settings; resolved from the environment when omitted.	`None`
`model_gateway`	`IModelGateway \| None`	The gateway to inject; resolved from config when omitted.	`None`
`runtime`	`IDurableRuntime \| None`	The durable runtime to inject; resolved from config when omitted.	`None`
`router`	`BaseRouter \| None`	The model router to inject; resolved from config when omitted.	`None`
`repository`	`GraphRepository \| None`	The graph repository to inject; resolved from config when omitted.	`None`
`middleware`	`Sequence[Middleware]`	Middleware fired around each superstep, in registration order.	`()`
`connectors`	`Sequence[Connector] \| ConnectorRegistry \| None`	Tools available to agents in this swarm (ADR 0015) — a :class:`~korchestrator.tools.ConnectorRegistry` this `Swarm` owns, or connectors to build one from. Omit entirely if no agent needs tools; an agent whose `tools` is non-empty with no `connectors` given raises `ConfigurationError` (P10.2).	`None`

Example

from korchestrator import Agent, Swarm swarm = ( ... Swarm(objective="Review this PR for security and performance") ... .add(Agent(id="security", role="security-reviewer")) ... .add(Agent(id="lead", role="review-lead")) ... .edges([("security", "lead")]) ... ) swarm.size 2

Start an empty swarm for objective; store optionally injected collaborators.

size `property`

size: int

The number of agents added to the swarm.

add

add(agent: Agent) -> Self

Add an agent to the swarm and return self for chaining.

Raises:

Type	Description
`ValidationError`	If an agent with this `id` was already added — a duplicate would otherwise silently overwrite the earlier agent (spec 08 §7).

cancel

cancel(run_id: str) -> None

Cancel run_id; it terminates with RunStatus.CANCELLED.

Raises:

Type	Description
`NotImplementedError`	On the local runtime.
`MissingExtraError`	If the Temporal runtime is selected without the `[temporal]` extra.

edges

edges(edges: list[tuple[str, str]]) -> Self

Declare directed edges (from_id, to_id) and return self for chaining.

edit_resume

edit_resume(
    run_id: str,
    *,
    updates: Mapping[str, JSONValue] | None = None,
    trust_delta: float = 0.0,
) -> None

Apply an operator's context/trust edit to a paused run, then resume it.

Goes through the same reducer discipline the barrier itself uses (last-value merge for updates, a clamped fold for trust_delta) — an edit is as replayable and auditable as an agent's own update.

Parameters:

Name	Type	Description	Default
`run_id`	`str`	The paused run to edit and resume.	required
`updates`	`Mapping[str, JSONValue] \| None`	Context-channel values to merge into the paused state.	`None`
`trust_delta`	`float`	Folded into `trust_score`, clamped to `[0.0, 1.0]`.	`0.0`

Raises:

Type	Description
`NotImplementedError`	On the local runtime.
`MissingExtraError`	If the Temporal runtime is selected without the `[temporal]` extra.

on

on(event: str, handler: EventHandler) -> Self

Register handler for event (e.g. "superstep"); returns self.

pause

pause(run_id: str) -> None

Signal run_id to pause for human review (durable HITL, spec 06 §7).

A paused run consumes no compute while it awaits resume, edit_resume, or cancel, bounded by a 24h deadline after which it times out. Governance also triggers this automatically when trust drops below the configured threshold — this method is for an operator-initiated pause.

Raises:

Type	Description
`NotImplementedError`	On the local runtime, which is synchronous and has no in-flight run to pause. Use `KORCH_RUNTIME=temporal` for durable HITL.
`MissingExtraError`	If the Temporal runtime is selected without the `[temporal]` extra.

resume

resume(run_id: str) -> None

Lift a pause and let run_id continue from its checkpointed state.

Raises:

Type	Description
`NotImplementedError`	On the local runtime.
`MissingExtraError`	If the Temporal runtime is selected without the `[temporal]` extra.

run

run(*, max_supersteps: int = 10) -> RunResult

Run the swarm's declared topology to a terminal :class:RunResult.

Parameters:

Name	Type	Description	Default
`max_supersteps`	`int`	Hard halt bound. Defaults to 10.	`10`

Returns:

Type	Description
`RunResult`	The terminal :class:`RunResult`, including `final_answer`.

Raises:

Type	Description
`ValidationError`	If the objective is too short, `max_supersteps` is outside 1-100, or the topology is invalid (no agents, or an edge referencing an unknown agent).
`MissingExtraError`	If reasoning is used without the `[dspy]` extra.

Example

from korchestrator import Agent, Swarm swarm = Swarm(objective="Summarize the design").add(Agent(id="lead", role="lead")) swarm.run(max_supersteps=5) # doctest: +SKIP

Agent

Agent(
    id: str,
    *,
    role: str,
    model: str | None = None,
    tools: tuple[str, ...] = (),
    goal: str = "",
    backstory: str = "",
    max_react_steps: int = 3,
    hitl_threshold: float | None = None,
    timeout_seconds: float = 120.0,
)

A single agent in a swarm — declarative by default, subclassable for custom behaviour.

Construct it declaratively and the framework supplies the reasoning::

Agent(id="security", role="security-reviewer", model="gpt-4o")

or subclass it and override :meth:think for a fully custom agent::

class WordCountAgent(Agent):
    async def think(self, state: AgentState) -> StateUpdate:
        ...

Frozen-snapshot contract (spec 06 §5): :meth:think MUST NOT mutate state (it is a frozen model, so mutation raises), MUST NOT call datetime.now() — use :attr:clock — MUST NOT read another agent's output from the same superstep, and MUST return within timeout_seconds. Returning halt=True permanently deactivates the node.

Parameters:

Name	Type	Description	Default
`id`	`str`	The agent's identifier, matching `^[a-z0-9][a-z0-9_-]{0,63}$`.	required
`role`	`str`	The persona's role, e.g. `"security-reviewer"`.	required
`model`	`str \| None`	The model to route this agent to, or `None` to let routing decide.	`None`
`tools`	`tuple[str, ...]`	Names of tools mounted on this agent.	`()`
`goal`	`str`	The persona's goal.	`''`
`backstory`	`str`	The persona's backstory.	`''`
`max_react_steps`	`int`	Bound on the ReAct loop, 0-10.	`3`
`hitl_threshold`	`float \| None`	Per-agent trust threshold for human-in-the-loop, 0.0-1.0, or `None`.	`None`
`timeout_seconds`	`float`	Per-agent wall-clock budget.	`120.0`

Raises:

Type	Description
`ValidationError`	If any input fails validation (e.g. an `id` that does not match the allowed pattern). Always `korchestrator.ValidationError` — no raw pydantic error escapes (spec 08 §2.2).

Example

from korchestrator import Agent agent = Agent(id="lead", role="review-lead") agent.id 'lead' agent.config.persona.role 'review-lead'

Validate the inputs into an immutable :class:AgentConfig; start with no clock bound.

clock `property`

clock: _BoundClock

The injected replay-safe clock; use self.clock.now() inside :meth:think.

Raises:

Type	Description
`ConfigurationError`	If no clock has been bound. The runtime binds one via :meth:`bind` before a run; in a unit test, call `agent.bind(clock=FakeClock())` first.

config `property`

config: AgentConfig

The immutable :class:AgentConfig this agent was validated into.

id `property`

id: str

The agent's identifier.

persona `property`

persona: AgentPersona

The agent's static persona.

bind

bind(
    *,
    clock: Clock,
    gateway: IModelGateway | None = None,
    tool_invoker: IToolInvoker | None = None,
) -> Self

Inject the replay-safe clock and optional collaborators; return self.

Called by the composition root before a run. The clock is the same injected Callable[[], datetime] the kernel uses, so an agent's timestamps stay replay-safe. The gateway is used by reasoning agents (e.g. :class:~korchestrator.agents.WorkerAgent); tool_invoker is used by a reasoning agent with AgentConfig.tools mounted (P10.2). Agents that do not need either ignore them.

is_complete

is_complete(state: AgentState) -> bool

Whether this agent considers its work finished, given state.

The default is False; reasoning agents override it to signal they have reached a final answer and should halt. It never mutates state.

think `async`

think(state: AgentState) -> StateUpdate

Turn a frozen state snapshot into this agent's :class:StateUpdate delta.

The base implementation raises — a declaratively-constructed agent has no reasoning of its own until the façade supplies the default reasoning agent (P4.9), and a custom agent must override this method.

Parameters:

Name	Type	Description	Default
`state`	`AgentState`	An immutable snapshot of the shared state. MUST NOT be mutated.	required

Returns:

Type	Description
`StateUpdate`	This agent's :class:`StateUpdate` delta for the current superstep.

to_node

to_node() -> Node

Materialise this agent as a kernel :class:~korchestrator.core.graph.Node.

Returns:

Name	Type	Description
`A`	`Node`	class:`Node` binding this agent's :class:`AgentConfig` to its :meth:`think` callable,
	`Node`	ready to place in an :class:`~korchestrator.core.graph.AgentGraph`.

Services

Korch

cancel

edit_resume

on

pause

resume

run

Swarm

size property

add

cancel

edges

edit_resume

on

pause

resume

run

Agent

clock property

config property

id property

persona property

bind

is_complete

think async

to_node

size `property`

clock `property`

config `property`

id `property`

persona `property`

think `async`