python-sdk.md

Python SDK reference

The Python SDK ships as pip install skarta. The wheel bundles a platform-native runtime binary; a single import skarta is enough to run end to end. Source: sdks/python/.

High-level API (P54 wrapper)

The SDK ships a beginner-friendly wrapper layer on top of the low-level surface documented below. Every wrapper class compiles down to the same @worker, @tool, Extension, Client, run_workflow(...) shapes you read about further down; nothing is hidden permanently.

Wrapper class	Purpose	Reference
LLM	Model declaration with auth + tuning	agents.md
Agent	Single-LLM worker with auto tool-loop and typed I/O	agents.md
Tool	Alias for the existing tool decorator	agents.md
Step	Typed deterministic Process / Pipeline step body (no LLM, no ctx)	processes.md
Pipeline	Linear DAG sugar; type-checked at construction	processes.md
Process	General DAG with bindings, branches, loops, races	processes.md
Team	Multi-agent collaboration on the conversation primitive	teams.md
Room	Declarative multi-agent dialogue; participants can be Agent / Team / Process / bare @worker / pre-registered worker names	rooms.md
Orchestrator	LLM-driven planner over a registry of agents and tools	agents.md
Memory	Cross-call session storage (file / db / custom)	agents.md
Skill	By-name reference to a registered Agent Skill	agents.md
Schedule / Webhook / Event	Trigger decorators	triggers.md
Approval	HITL gate (Slack / email / custom) for @tool(requires_approval=True)	hitl.md
Budget / Retry	Per-agent cost / retry policy	agents.md
Sandbox / Permissions	Re-exported from low-level surface	below
App	Manual lifecycle control (async with App() as app: ...)	agents.md
idempotent	Declarative handler dedup keyed by an input field; durable across restart with App.idempotency_store("db")	idempotency.md
before_model, after_model, before_tool, after_tool, before_skill, on_context_change, on_compaction	Interceptor decorators (kwargs: priority, mode, filter)	Interceptor hook decorators

Hello agent in three lines:

from skarta import LLM, Agent
text = await Agent(llm=LLM(model="claude-haiku-4-5"),
                   instructions="You are concise.").execute("Hello.")

The fastest entry point is quickstart.md. The rest of this page is the low-level reference that the wrapper itself consumes.

LLM credentials

LLM(...) carries api_key, base_url, and provider. The runtime resolves them per call in this order:

1. Inline credentials. When LLM(...) carries api_key plus at least one of base_url or provider, the runtime builds (or cache-reuses) an ad-hoc provider per (provider_kind, base_url, api_key). No TOML editing required.
2. Static registry. When the inline fields are unset, the runtime falls back to the [providers.*] + [models.*] + built-in registry.

Custom OpenAI-compatible endpoints (Ollama, vLLM, LiteLLM, Azure OpenAI, etc.) work through the inline path with no extra wiring:

from skarta import LLM, Agent

ollama = LLM.openai_compat(
    model="llama3",
    api_key="ollama",
    base_url="http://localhost:11434/v1",
)
text = await Agent(llm=ollama, instructions="Be concise.").execute("Hi.")

Advanced (lower-level reference)

The rest of this page is the lower-level reference. You do not need any of the constructs below to ship the everyday case; reach for them when a wrapper surface does not give you the control you need. Admin RPCs (webhook / schedule / outbound event / conversation administration) live in this section because they call into the underlying Client.

Top-level imports

from skarta import (
    # clients
    Client,           # sync
    AsyncClient,      # async (alias for RuntimeClient)
    RuntimeClient,    # async, low-level

    # remote runs (dial another runtime and get a handle to the run on it)
    RemoteRuntime, RemoteRunHandle,           # async
    SyncRemoteRuntime, SyncRemoteRunHandle,   # blocking mirrors

    # runtime-to-runtime trust (per-runtime ed25519 identity)
    PeerIdentity, peer_id_for_public_key_b64,

    # extension building
    Extension,
    AgentContext,

    # decorators
    tool, worker, orchestrator,

    # binding helpers
    input_, literal,
    each, each_index, each_count,
    collect, collect_count,
    merge, concat, coalesce, default,
    when, on_failure, on_complete,
    loop_, loop_index, loop_previous,
    race, stream_,

    # interceptors / context providers / sessions
    InterceptorDef, InterceptorHandler, InterceptorHook, InterceptorMode,
    ContextProviderDef, ContextProviderHandler,
    SessionStorageHandler,
    ReconnectConfig,

    # outbound events (declarations + handler type alias for type hints)
    EventSubscriberDef, EventSubscriberHandler,

    # peer-message envelope (returned by ctx.recv_peer / ctx.try_recv_peer)
    PeerMessage,

    # conversation transcript types (returned by Conversation-handle
    # methods like `Conversation.get_transcript()`; the lower-level
    # `client.conversation_get_transcript(...)` returns the same shape
    # as a raw dict)
    ConversationStream, Transcript, TurnRecord,

    # capability declarations (advanced -- most users only set these via
    # the @worker / @tool kwargs and the Extension.add_* helpers)
    Sandbox, SandboxOverride,
    Permissions,                          # sandbox filesystem / network grants
    ScheduleDeclaration,

    # high-level wrappers (also used as @worker / @tool kwargs and
    # Extension.add_* helpers)
    Approval, ApprovalEvent,              # HITL gate and the audit-trail event
    Budget,                               # per-call cost / token ceilings
    Retry,                                # retry policy on tool / worker failure

    # low-level wire codec (advanced; only needed if you build a custom
    # transport or replay tool)
    Codec, MAX_MESSAGE_SIZE,

    # errors
    FrameworkError, FrameworkErrorCode, ValidationError,
    AdminRouteConfigError,                # raised at App(admin=...) construction
    RuntimeSpawnError,                    # raised when auto-spawn fails to bring up the runtime

    # auto-spawn target descriptor (returned by the runtime resolver)
    ResolvedTarget,
)

Two protocol types live one import deeper because they are submission-time overrides rather than registration-time declarations:

from skarta.protocol.types import RetryPolicy, NodeConfigOverride

RetryPolicy and NodeConfigOverride are passed to client.run_workflow(node_overrides=...) and to Team(config_overrides=...). See Per-run controls for the full call shape.

Clients

skarta.Client (sync)  (skarta package only)

import skarta

with skarta.Client() as client:
    print(client.list_extensions())

Client() resolves where to connect using these env vars in order:

1. SKARTA_SOCKET_PATH  an existing Unix socket.
2. Auto-spawn the bundled runtime on a per-process socket. Disable with SKARTA_AUTOSPAWN=0.

SKARTA_RUNTIME_BIN (test / power-user override) points at a custom runtime binary the SDK should launch in step 2 instead of the bundled one. Right for stub binaries that exercise crash-handling paths and for editable dev installs that build the runtime locally.

Client(target=...) short-circuits the env-var dance.

Methods (every method is blocking, returns a plain dict or list):

• list_extensions()
• registry_query()  returns {tools, workers, teams, skills, extensions} as lists of names (not full declarations). See Registry inspection for the exact shape.
• run_workflow(workers, input, ...)
• run_workflow_target(target, input, ...)
• run_workflow_stream(workers, input, ...)  returns a SyncWorkflowEventStream.
• run_workflow_target_stream(target, input, ...)
• check_team_health(team_name)  see Registry inspection below.
• visualize_team(team, format)  see Registry inspection below.
• list_models(), get_model(model_id), list_providers()
• save_session(session, *, expected_updated_at=None), load_session(id), list_sessions(), delete_session(id)
• branch_session(id, from_node_id, label=None), create_checkpoint(id, name, node_id=None), restore_checkpoint(id, name)  from_node_id is the node to fork at; label optionally names the new branch.

save_session returns {id, updated_at}. The updated_at is server-stamped (RFC3339); the caller's value is ignored. To opt into optimistic concurrency, pass expected_updated_at= with the value the server returned on the previous save (or with the updated_at of the loaded session). On a mismatch the runtime raises FrameworkError(code=SESSION_VERSION_CONFLICT) instead of overwriting; error.details carries expected_updated_at and actual_updated_at so the caller can reload, merge, and retry.

create_checkpoint returns {name, node_id, session} where session is the full updated session blob. Use it to refresh local state in one round trip; without it, a stale follow-up save_session would silently overwrite the new checkpoint map.

• cancel()  cancel every in-flight RPC future on this client (returns the count). Client-side only: the runtime keeps running the DAG. To abort the DAG itself mid-flight, call cancel_workflow(dag_id) (see Cancellation).
• cancel_workflow(dag_id)  abort an in-flight DAG by id. Returns {"cancelled": True, "dag_id"} on success; unknown / completed id raises FrameworkError(code=REGISTRY_NOT_FOUND). Sync mirror of AsyncClient.cancel_workflow.
• socket_path(), target(), mode(), is_connected().

Admin and inspection methods (full roster)

The headline list above covers the everyday paths. Every other admin RPC the runtime exposes is also reachable on Client / AsyncClient; the reference table below points at the per-feature section that documents the shape:

Method	What it does	Reference
describe_tool(name)	Full ToolDeclaration (name, params, advisory fields, sandbox).	Registry inspection
describe_worker(name)	Full WorkerDeclaration (input / output schema, bindings, advisory fields).	Registry inspection
webhook_list(...) / webhook_create(...) / webhook_delete(...)	Read / register / delete dynamic webhook ingress routes.	Webhook administration
schedule_list() / schedule_create(...) / schedule_delete(...) / schedule_trigger(...)	Read / register / delete / fire-now dynamic cron schedules.	Schedules admin
conversation_create(...) / conversation_list(...) / conversation_describe(...) / conversation_get_transcript(...) / conversation_resume(...) / conversation_terminate(...) / conversation_rpc(...)	Multi-agent conversation primitive admin.	conversations.md
event_publish(...) / event_subscribe_list(...) / event_subscribers_active(...)	Outbound bus publishing + subscriber inventory.	Outbound events
workflow_resume(run_id)	Resume a durable DAG run after restart.	Durability and resume

SyncWorkflowEventStream is a plain iterator. After the loop, the terminal state is on stream.final_result / stream.final_error:

stream = client.run_workflow_stream(workers=[...], input={...})
for event in stream:
    print(event["type"])
print(stream.final_result)

Use stream.drain_to_completion() if you only want the final DagResult and nothing in between.

The final DagResult dict carries status, node_results, node_errors, skipped_nodes, cancelled_nodes, and total_cost. skipped_nodes is a flat list[str] of every node the runtime skipped (gated by a false conditions expression, propagated from a failed upstream, etc.) so a caller that does not subscribe to events can still tell which workers ran vs. which were skipped without diffing requested workers against node_results.keys(). The post-hoc list intentionally carries names only; per-node reason strings live on the streaming node_skipped events (see concepts.md section Failure handling for the reason vocabulary). Non-streaming callers that need to know why each node was skipped can subscribe via client.run_workflow_stream(...) instead of run_workflow(...). cancelled_nodes lists every node the runtime cancelled mid-flight; today this only covers race losers when a downstream race(..., cancel_losers=True) aborted the slow candidate. A cancelled node has neither a node_results entry nor a node_errors entry; cancellation is not a failure.

skarta.AsyncClient

Same RPC surface as Client, every RPC method is async def. Use inside an async function or an asyncio entry point.

The only Client-side method without an AsyncClient peer is cancel() (sync-only, cancels every in-flight RPC future on this client). It exists on Client because the sync surface bridges concurrent.futures.Futures onto a private event-loop thread, and that bridge is what cancel() cancels. On AsyncClient you have direct access to the underlying asyncio.Tasks, so you cancel by cancelling the task you spawned (task.cancel()) rather than going through a client-level helper. cancel_workflow(dag_id) (abort the running DAG itself) is available on both surfaces.

import skarta

async def main():
    async with await skarta.AsyncClient.connect("/tmp/skarta.sock") as client:
        print(await client.list_extensions())

AsyncClient does not auto-spawn. Either pass an explicit target, or use RuntimeManager directly (see below).

Transports

The target string picks the transport. The wire protocol is identical on all of them, so the choice is purely operational.

Target form	Transport	When to use
/path/to/skarta.sock	Unix socket	Same host as the runtime (default, lowest overhead).
ws://host:port / wss://host:port	WebSocket	Remote runtime; works through any HTTP proxy or load balancer.
grpc://host:port / grpcs://host:port	gRPC over HTTP/2	Remote runtime where you want HTTP/2 multiplexing, native server-streaming, and standard gRPC tooling / load balancers.

async with await skarta.AsyncClient.connect("grpc://127.0.0.1:50071") as client:
    print(await client.list_extensions())

gRPC needs the optional grpcio dependency: pip install 'skarta[grpc]' (or pip install 'skarta-client[grpc]'). The runtime exposes gRPC when started with SKARTA_GRPC_PORT set (alongside the Unix-socket and WebSocket listeners, which are unaffected). grpcs:// uses the runtime's [tls] certificate; the same plaintext gate as WebSocket refuses a non-loopback grpc:// bind without TLS unless SKARTA_ALLOW_PLAINTEXT=1.

To pin a self-signed or private-CA certificate for grpcs://, pass tls=GrpcTls(ca_cert="/path/ca.pem") to connect(...) / skarta.Client(...) / Extension.run(...), or set SKARTA_GRPC_CA_CERT (PEM path) / SKARTA_GRPC_CA_CERT_PEM (inline PEM). GrpcTls(server_name="...") (or SKARTA_GRPC_SERVER_NAME) overrides the hostname checked against the certificate when you dial an IP or a name that differs from the cert subject. Full operator detail is in deployment.md.

Per-run controls

run_workflow, run_workflow_target, run_workflow_stream, and run_workflow_target_stream all accept the same per-run option set. Every option is keyword-only.

kwarg	Type	What it does
node_overrides	dict[str, NodeConfigOverride]	Per-node retry / timeout / concurrency / model / validation-policy overrides. Keyed by worker name. The runtime accepts a wire-shape dict ({"retry": {...}, "timeout_ms": N, "concurrency": M, "model": "...", "validation_policies": {...}}) for callers that prefer not to import NodeConfigOverride.
binding_overrides	dict[str, dict[str, str]]	Per-worker binding rewrites; keyed by worker name, then field name.
loop_configs	list[LoopConfig]	Loop pairs declared at submission time.
skip_policies	dict[str, SkipPolicy]	Per-node skip behaviour when an upstream fails. Three variants: SKIP_DOWNSTREAM (default; this node and all transitive descendants are skipped), SKIP_OPTIONAL (skip only if every predecessor failed/skipped, otherwise dispatch normally), USE_DEFAULT (mark this node completed with null without calling the worker so consumers can use default(...) / coalesce(...) / on_failure(...) fallbacks). Pass either the enum (from skarta.protocol.types import SkipPolicy) or the string form. An on_failure(...) / on_complete(...) binding on the same node always wins over the skip policy. See concepts.md section Failure handling.
dag_timeout_ms	int	Whole-DAG deadline. On expiry the runtime aborts every in-flight node and the result returns with status == "cancelled". No DAG_TIMEOUT error is added to node_errors; already-completed nodes stay in node_results. See concepts.md section DAGs for the full timeout / status table.
max_parallel	int	Cap on concurrent in-flight nodes across the DAG. Defaults to [dag] max_parallel_nodes (default 100).
orchestrator	str	Name of an @orchestrator to plan the DAG. The submission's workers / target argument becomes the planner's input.

from skarta.protocol.types import RetryPolicy, NodeConfigOverride

result = await client.run_workflow(
    workers=["fetch", "summarise"],
    input={"url": "https://api.example.com/v1/items"},
    node_overrides={
        "fetch": NodeConfigOverride(
            retry=RetryPolicy(
                max_attempts=4,
                backoff_ms=500,
                backoff_multiplier=2.0,
                retry_on=["PROVIDER_RATE_LIMITED", "PROVIDER_ERROR"],
            ),
            timeout_ms=10_000,
        ),
        "summarise": NodeConfigOverride(timeout_ms=30_000, concurrency=2),
    },
    dag_timeout_ms=120_000,
    max_parallel=8,
)

RetryPolicy(max_attempts=1, backoff_ms=1000, backoff_multiplier=2.0, retry_on=None):

• max_attempts  total attempts including the first call. 1 (default) disables retry.
• backoff_ms  delay before the second attempt; multiplied by backoff_multiplier for each subsequent attempt.
• backoff_multiplier  float multiplier applied to backoff_ms between attempts (default 2.0, exponential).
• retry_on  optional whitelist of FrameworkErrorCode strings (e.g. ["PROVIDER_RATE_LIMITED", "PROVIDER_ERROR"]). When set, only those codes retry. When unset, the runtime falls back to the error's own retryable flag (provider rate-limits and transient transport errors set this; validation errors do not).

Precedence on a given node, outermost wins: submission node_overrides > team config_overrides (outermost team on the closure path first) > worker decorator (concurrency only) > built-in defaults.

Remote runs

RemoteRuntime lets one runtime dial another, submit a workflow to it, and get a first-class RemoteRunHandle for the run on the far side. It rides the same client transports as Client (grpc:// / grpcs:// / ws:// / wss:// / unix:); the dialing runtime is just another authenticated client of the receiving runtime. There is no new transport and no new listener. The narrative walkthrough (the await / callback / poll patterns) lives in remote-runs.md; this is the API surface.

Two forms, matching the rest of the SDK: RemoteRuntime / RemoteRunHandle are async, and SyncRemoteRuntime / SyncRemoteRunHandle are blocking mirrors that own a private event-loop thread (exactly like Client mirrors AsyncClient). SyncRemoteRuntime refuses construction from inside a running event loop.

Async:

from skarta import RemoteRuntime

rt = await RemoteRuntime.connect("wss://runtime-b.internal:9944", api_key=KEY)
handle = await rt.run(target="summarise", input={"topic": "rain"})
result = await handle.await_result(timeout_ms=60_000)   # the DagResult
print(result["status"])                                 # completed / failed / cancelled
await rt.close()

Blocking:

from skarta import SyncRemoteRuntime

with SyncRemoteRuntime("grpcs://runtime-b.internal:50051", api_key=KEY) as rt:
    handle = rt.run(["researcher"], input={"topic": "rain"}, detached=True)
    result = handle.await_result(timeout_ms=60_000)

RemoteRuntime

Member	Returns / does
await RemoteRuntime.connect(url, *, api_key=None, tls=None)	Open a connection and wrap it. url accepts the same targets as RuntimeClient.connect. Classmethod.
await rt.run(workers=None, *, target=None, input=None, callback=None, detached=False, orchestrator=None, dag_id=None)	Submit a run and return a RemoteRunHandle. Pass exactly one of workers (a worker list) or target. detached=True spawns the run on the far side and returns immediately; a callback object asks the receiver to notify you once, when the run reaches a terminal state.
await rt.run_stream(...)	Same submission shape; the returned handle's stream() async-iterates the live DagEvent frames.
rt.attach(run_id)	Re-attach to an already-submitted run by id. The handle's poll / result / await_result reach the runtime for state. Use after a caller restart.
rt.client	The underlying RuntimeClient, if you need the low-level surface.
await rt.close()	Close the connection. RemoteRuntime is also an async context manager.

SyncRemoteRuntime(url, *, api_key=None, tls=None) mirrors all of the above without await, and is a plain (non-async) context manager.

RemoteRunHandle

Member	Returns / does
handle.run_id	The run id on the far side. Persist it to re-attach later.
await handle.await_result(timeout_ms=None)	The final DagResult. Synchronous submissions return inline; detached submissions poll workflow.status until terminal, then fetch workflow.result. timeout_ms bounds the wait and raises TimeoutError when it elapses (the run keeps executing on the far side); None waits indefinitely.
await handle.poll()	The current workflow.status snapshot, no waiting.
await handle.result()	workflow.result now, without waiting for a terminal state.
await handle.cancel()	Abort the run mid-flight via workflow.cancel.
handle.stream()	Async-iterate the live DagEvent dicts. Valid only for handles from run_stream.

SyncRemoteRunHandle exposes the same members as blocking calls.

Runtime trust (peer identity)

When a runtime dials another, the receiver can require that the caller present a signed peer token from a runtime it trusts, using a per-runtime ed25519 identity. It is progressive: with no peers configured the receiver stays open; once any peer is registered the remote-run data plane is peer-gated, and an unknown or under-scoped caller is refused with PeerUntrusted. The full model (the [peers] registry, the skarta identity / skarta peers admin commands, scope intersection, and the open-vs-enforced posture) lives in runtime-trust.md.

The Python surface is small; most operators manage identity and the trust registry from the CLI rather than in code.

Member	Returns / does
PeerIdentity.generate()	A fresh ed25519 identity.
PeerIdentity.from_seed(seed)	Reconstruct an identity from its 32-byte seed.
identity.seed	The 32-byte private seed. Store it as the runtime's secret; never publish it.
identity.peer_id	The derived peer_id ("peer_" + sha256(public_key)[:8].hex()). Safe to publish and register on a trusting peer.
identity.public_key_b64	The base64 public key. Safe to publish; this is what an operator registers on the receiving runtime.
peer_id_for_public_key_b64(public_key_b64)	Derive the peer_id for a base64 public key without constructing a full identity.

Registry inspection

Three read-only RPCs let a dashboard or hot-reload pipeline inspect the live registry without running anything.

registry_query()

Runs the registry.query RPC and returns a snapshot of every name currently in the registry, grouped by kind. Use it to drive a dashboard, sanity-check what registered after spawning extensions, or diff before / after a hot-reload.

Return shape (every value is a list[str] of names, not full declarations):

{
    "tools": ["expensive_lookup", "fetch_snapshot", ...],
    "workers": ["fetch_snapshot", "analyse", ...],
    "teams": ["support_ops", ...],
    "skills": ["customer-tone", ...],
    "extensions": ["my-extension", "audit-extension", ...],
}

Note: registry_query ships names only. Full ToolDeclaration / WorkerDeclaration / TeamDeclaration records (with description, parameters schema, bindings, and the advisory classifier fields like latency_class, cost_class, model_preference, idempotent, requires_approval, streaming, steerable) are kept in the runtime's in-memory ExtensionEntry.capabilities. To pull the full record over the wire, use describe_tool and describe_worker below.

list_extensions() is a thin convenience over registry_query() that returns just the extensions slice.

describe_tool(name)

Returns the full ToolDeclaration for a registered tool. Use this when a planner, router, or dashboard needs the metadata that drives dispatch routing - registry_query only returns names.

Return shape mirrors the ToolDeclaration wire schema:

{
    "name": "lookup_policy_internal",
    "description": "Internal-DB policy lookup, free + idempotent.",
    "parameters": {"type": "object", "properties": {"policy_id": {"type": "string"}}, "required": ["policy_id"]},
    "return_schema": {"type": "object", "properties": {"active": {"type": "boolean"}}},
    "latency_class": "instant",
    "cost_class": "cheap",
    "idempotent": True,
    "requires_approval": False,
    "streaming": False,
    "steerable": False,
}

Unknown name raises FrameworkError(code=REGISTRY_NOT_FOUND) with error.details = {"tool": "<name>"}.

describe_worker(name)

Returns the full WorkerDeclaration: input/output schemas, the full bindings map, validation_policies, tools_required, skills_required, model_preference, concurrency, conditions, plus advisory metadata (latency_class, cost_class, streaming, steerable).

Unknown name raises FrameworkError(code=REGISTRY_NOT_FOUND) with error.details = {"worker": "<name>"}.

Both methods are available on Client (sync) and AsyncClient (async); the wire shape and error semantics are identical.

check_team_health(team_name)

Re-validates the team against the current set of registered worker schemas, surfacing the same incompatibilities the runtime would emit on the push path (registry.schema_incompatibility, see concepts.md section Hot reload). Use it to detect schema drift after a worker hot-swap before you submit work that would fail at Point 2.

• team_name  the same string you passed to add_team(...) or set on TeamDeclaration.name. The closure walker re-validates sub-teams transparently.
• Unknown team raises FrameworkError(code=REGISTRY_NOT_FOUND) with error.message = "Team '<team_name>' is not registered" and error.details = { "team": "<team_name>" }. The same shape is used by every registry-read RPC that takes a team name (see Common error codes).

Return shape:

{
    "team": "support_ops",
    "healthy": False,
    "incompatibilities": [
        {
            "worker": "summarise",
            "change": "Field 'topic' removed from input schema",
            "affected_bindings": ["summarise.topic <- fetch.topic"],
            "suggestion": "Re-bind 'summarise.topic' or restore the field on 'summarise'.",
            "scope": ["support_ops"],   # outermost team first; empty for root-level workers
        },
    ],
}

healthy is True exactly when incompatibilities is empty. Each entry covers one worker that no longer fits its current bindings, including the schema-evolution case the design block calls out: when an extension hot-swaps a worker to a v2 schema with a renamed field, the renamed field shows up as a change description on that worker's incompatibility entry, with the binding that referenced the old name listed under affected_bindings.

The pull-side check_team_health and the push-side registry.schema_incompatibility event share the same incompatibility shape: check_team_health is on-demand, the event fires once at register-time per affected team. Subscribe to events for live alerts; call check_team_health to audit the current state on demand.

visualize_team(team, format="mermaid")

Renders the team's transitive closure as a graph in one of four formats. The CLI subcommand skarta dag-viz <team> is a thin wrapper around this RPC.

• team  the registered team name (same string used for check_team_health).
• format  one of "mermaid" (default), "dot", "json", "ascii". Any other value raises FrameworkError(code=PROTO_MALFORMED_MESSAGE) with error.details = { team, requested_format, supported_formats }.
• Unknown team raises FrameworkError(code=REGISTRY_NOT_FOUND) with the same canonical shape check_team_health uses: error.message = "Team '<team>' is not registered" and error.details = { "team": "<team>" }.

Return shape:

{
    "team": "support_ops",
    "format": "mermaid",
    "content": "graph TD\n  fetch --> summarise\n  ...",
}

content is a string for "mermaid", "dot", and "ascii". For "json" it is a structured object with nodes and edges arrays. Render the string formats by writing them to a .mmd / .dot / plain text file; consume "json" directly to drive a custom UI.

Webhook administration (P42)

Three admin RPCs let a client author dynamic webhooks at runtime, on top of the declared webhooks an extension publishes via ext.add_webhook(...). Available on both AsyncClient (async) and Client (sync); the sync surface delegates to the async transport via an internal event-loop thread per ADR-024.

webhook_create(...)

Registers a new dynamic webhook. Skarta core stores routing metadata only; signature verification, HMAC checks, and any per-route secret handling live in the worker behind the webhook.

resp = await client.webhook_create(
    name="stripe-events",
    target="process_stripe_event",          # XOR with worker_list
    response_mode="async",                  # or "sync" (timeout_ms required)
    timeout_ms=None,
    dedupe_header="stripe-signature",       # optional idempotency key
    dedupe_ttl_secs=86_400,
    description="Stripe events",
)
# {"name": "stripe-events",
#  "public_url": "http://127.0.0.1:9100/webhooks/stripe-events",
#  "created_at": "2026-05-01T08:03:03.582228+00:00"}

target and worker_list are XOR (per ADR-045): exactly one must be set.

Conflict policy (returns FrameworkError(code=REGISTRY_CONFLICT)):

• The name matches an existing declared webhook → error.details = {name, source: "declared"}. Declared webhooks win on collision.
• The name matches an existing dynamic webhook → error.details = {name, source: "dynamic"}.

webhook_delete(name)

Removes a dynamic webhook. Returns {"deleted": bool, "name"}. deleted=False means nothing existed under the supplied name.

webhook_list(*, source=None)

Merged listing of declared (P41) and dynamic (P42) webhooks. Returns {"webhooks": [...]}; each entry carries a source discriminator ("declared" or "dynamic"), the public URL, the response mode, the dedupe header (when set), and last_fired_at (dynamic rows only).

listing = await client.webhook_list()                # source defaults to "all"
listing = await client.webhook_list(source="dynamic") # only dynamic rows

Sync vs async

Identical surface on AsyncClient (await calls) and Client (blocking calls). The sync Client runs its own internal event-loop thread per ADR-024 and refuses construction from inside a running asyncio loop -- if you want to call the sync surface from inside an async function (e.g. a script that already runs asyncio.run(...)), wrap the sync call in asyncio.to_thread(...).

Persistence opt-in

By default dynamic webhooks live in process memory and disappear on restart -- right for tests and developer iteration. To persist across restart, set webhook.storage = "db" in framework.toml (or SKARTA_WEBHOOK_STORAGE=db in env) and supply SKARTA_DATABASE_URL. Rows persist to the same agent_storage::Storage handle the session / budget / permission / audit areas use; the dynamic_webhooks table holds routing metadata only (no signing-secret columns per ADR-049).

Outbound events (P48)

The dual of webhooks. Workers, agents, and admin callers publish typed events with ctx.publish(...) / client.event_publish(...); every extension that registered a matching subscriber gets a fan-out call from the runtime over its existing connection. See outbound-events.md for the full contract.

Subscriber side: Extension.add_event_subscriber(...)

from skarta import Extension

async def on_urgent_email(envelope: dict) -> None:
    payload = envelope["payload"]
    print(f"urgent: {payload['subject']}")

ext = Extension("notifier")
ext.add_event_subscriber(
    name="page-on-urgent",
    pattern="assistant.urgent_email",
    handler=on_urgent_email,
    description="Pages the on-call when the inbox-watcher flags an urgent email.",
)
await ext.run("/tmp/skarta.sock")

Pattern grammar: literal dot-segments, * matches one segment, ** (only at the end) matches any trailing segments. Subscriber names must be unique within an extension; a duplicate name raises ValueError at register time.

Type aliases on the public surface: EventSubscriberHandler is the callable type the handler argument accepts (an async function that takes the envelope dict and returns None); EventSubscriberDef is the underlying declaration record the SDK builds from your kwargs. Both are top-level imports from skarta so you can use them in your own type hints (e.g. handler: EventSubscriberHandler); building an EventSubscriberDef by hand is rare and only needed when programmatically generating subscribers outside the add_event_subscriber(...) helper.

Publisher side: ctx.publish(...)

async def watcher_worker(input, ctx):
    await ctx.publish(
        "assistant.urgent_email",
        {"subject": input["subject"], "from": input["from"]},
        idempotency_key=input["message_id"],
    )

Returns the bus's {"accepted", "event_id"?, "deduped"} summary. A deduped: true response signals that the same (name, idempotency_key) pair was suppressed by replay-dedupe.

Admin client: event_publish / event_subscribe_list / event_subscribers_active

# fire from outside any extension (e.g. from an admin script)
client.event_publish(
    "assistant.urgent_email",
    {"subject": "Quarterly board prep", "from": "ceo@example.com"},
    idempotency_key="msg-2026-05-04-0001",
)

# merged listing of subscribers + recent events
listing = client.event_subscribe_list(name="assistant.urgent_email", limit=20)

# "who fires when I emit X right now?"
active = client.event_subscribers_active("assistant.urgent_email")

Available identically on AsyncClient (await calls) and Client (blocking calls).

Schedules (cron triggers)

Wall-clock as a trigger. Each schedule fires its bound target on a cron-like timer; every fire becomes a fresh DAG submission carrying a scheduled_fire envelope. See schedules.md for the full contract.

Declared side: Extension.add_schedule(...)

from skarta import Extension

ext = Extension("morning-briefing")
ext.add_schedule(
    name="briefing.morning",
    cron_expr="0 7 * * 1-5",
    tz="Asia/Kolkata",
    target="briefing_worker",
    input={"audience": "ceo"},
    max_inflight=1,
    missed_fire_policy="run_one",
)

Schedules are runtime-fired (not extension-fired); declaring one does not register a callback in the extension. The runtime owns the timer driver and submits a fresh DAG against target (or worker_list) on every fire. Declared schedules disappear when the extension disconnects; pin survival across SDK process restarts by registering dynamic schedules instead (see below) and turning on [schedule] storage = "db".

Underlying type: each add_schedule(...) call builds a ScheduleDeclaration record (top-level import from skarta) from your kwargs. Building one by hand is rare; the helper exists so you do not have to. The class is exposed for callers that programmatically generate schedule lists (e.g. driving add_schedule from a configuration file).

Admin client: schedule_create / schedule_delete / schedule_list / schedule_trigger

from skarta import Client

with Client("/tmp/skarta.sock") as client:
    client.schedule_create(
        name="hourly_recap",
        cron_expr="@every 1h",
        target="recap_worker",
        max_inflight=1,
        missed_fire_policy="skip",
    )
    rows = client.schedule_list()           # merged declared + dynamic
    only_dyn = client.schedule_list(source="dynamic")
    client.schedule_trigger("hourly_recap")  # force one fire (manual=true)
    client.schedule_delete("hourly_recap")

Available identically on AsyncClient (await calls). Authorization: schedule.create / delete / trigger require schedule_admin (also accepted as schedule:admin to match the wire docs); schedule.list accepts read.

Sandbox declarations

Sandbox and SandboxOverride (top-level imports from skarta) are the typed declarations tools and workers use to advertise the capabilities they need (network egress, filesystem reads, env-var access, etc.) and what runtime policy applies. The SDK reference does not duplicate the capability catalogue here; see sandbox.md for the full capability set and policy precedence. In Python, these types appear when you set sandbox=Sandbox(...) on @worker / @tool or build a per-run narrow override via SandboxOverride(...). Skills do not declare capabilities in SKILL.md: the agentskills.io 1.0 manifest carries no Skarta-specific field, and the per-skill sandbox surface lives in framework.toml under [sandbox.skills.<name>].

Durability and resume (P43 / ADR-071)

When the runtime is configured [durability] enabled = true, every accepted submission is persisted before the first node dispatches and every node-state transition appends one row to the journal. A later await client.workflow_resume(run_id) reconstructs the executor from the journal and drives the run to completion without re-running already-completed nodes.

# Fresh run with a stable dag_id so resume can find it.
result = await client.run_workflow(
    workers=["pipeline"],
    input={...},
    dag_id="research-2026-05-10-001",
)

# After a runtime restart, pick the run up by its id.
resumed = await client.workflow_resume("research-2026-05-10-001")
# resumed["was_resumed"] == True
# resumed["node_attempt_counts"][...] >= 1, cumulative across the resume boundary

The result dict gains two wire-additive fields when durability is on:

• was_resumed: bool -- True iff this run was reconstructed from the journal (or the auto-resume sweep on startup).
• node_attempt_counts: Dict[str, int] -- per-node cumulative attempt counter across resume boundaries.

Workers can declare an idempotency posture so the runtime knows how to handle them on resume: pass idempotency="external_side_effect" on @Step or @worker to refuse re-dispatch on resume (the runtime records a synthetic failure with WORKER_EXTERNAL_SIDE_EFFECT_AT_RESUME so the downstream on_failure(<worker>) binding can take the recovery path). Default is "idempotent". The sync mirror is client.workflow_resume(run_id). See durability.md for the full surface.

Cancellation

Skarta exposes three cancellation surfaces. Pick by what you need to abort:

Surface	What it cancels	When to use
client.cancel()	Every in-flight RPC future on this client. Client-side only: the transport stays open, no message is sent to the runtime, and the runtime keeps running the DAG. The blocked caller's future re-raises CancelledError.	A caller wants to stop waiting (timeouts in your code, user pressed Ctrl+C in a CLI, FastAPI request was abandoned). The DAG itself continues; if you need its result later you cannot reattach.
client.cancel_workflow(dag_id)	The runtime signals the cancel token of an in-flight DAG; every running node task is aborted; the executor closes with DagResult.status == "cancelled" and the streaming surface emits a final dag_cancelled event with reason = "user_cancelled". Already-completed nodes stay in node_results; aborted-mid-flight nodes land in cancelled_nodes.	A user clicks "stop", a budget watcher detects an overrun, an ops dashboard kills a runaway run. Returns {"cancelled": true, "dag_id"} on success; unknown / already-finished id raises FrameworkError(code=REGISTRY_NOT_FOUND) with details = {"dag_id": id}.
Submission dag_timeout_ms	The runtime aborts every in-flight node and returns DagResult.status == "cancelled". The streaming surface emits a dag_cancelled event with reason = "timeout". Already-completed nodes stay in node_results; nothing is added to node_errors.	A hard ceiling on total wall-clock time. Pass dag_timeout_ms= on run_workflow(...) / run_workflow_stream(...) at submission time.

cancel_workflow(dag_id) needs a stable dag_id. Two ways to get one:

• Streaming: stream.dag_id is populated from the dag_started event (the first event the runtime emits) before the first next(stream) returns.
• Pin it at submission: pass dag_id="my-stable-id" as a keyword to any of the four submit methods (run_workflow, run_workflow_target, run_workflow_stream, run_workflow_target_stream). The runtime registers the cancel token under that exact id; collisions with an in-flight run raise FrameworkError(code=REGISTRY_CONFLICT). Empty / absent value mints a fresh dag-<uuid>.

dag_cancelled is the terminal streaming event for any cancelled run; the optional reason field discriminates "timeout" vs "user_cancelled". It does not fire on race-loser cancellation (those land on DagResult.cancelled_nodes, not as a top-level DAG cancel) and it does not fire from client.cancel() (which never reaches the runtime).

Teams

TeamDeclaration packages a multi-worker subgraph as a callable with its own typed input / output contract, then Extension.add_team(...) registers it with the runtime. Submit a team by passing its name to client.run_workflow_target(...); introspect a registered team with check_team_health(team_name) and visualize_team(team). The conceptual model and the field-by-field semantics live in concepts.md section Teams; this section is the Pydantic construction reference.

from skarta import Extension
from skarta.protocol.capabilities import TeamDeclaration

ext = Extension("support", "0.1.0")
ext.add_worker(classify_ticket)
ext.add_worker(draft_reply)
ext.add_team(
    TeamDeclaration(
        name="support_ops",
        description="Triage a ticket then draft a reply.",
        input_schema={
            "type": "object",
            "properties": {
                "ticket_id": {"type": "string"},
                "body": {"type": "string"},
            },
            "required": ["ticket_id", "body"],
        },
        output_schema={"type": "object"},
        nodes=["classify_ticket", "draft_reply"],
        entry="classify_ticket",
        exit="draft_reply",
        entry_bindings={},        # leave empty (runtime support pending)
        exit_bindings={},         # leave empty (runtime support pending)
        binding_overrides={},
        loop_configs=[],
        config_overrides={},
    )
)

TeamDeclaration is a Pydantic BaseModel exported from skarta.protocol.capabilities. Field types and required vs optional:

Field	Required	Type
name	yes	str
description	yes	str
input_schema	yes	dict (JSON Schema)
output_schema	yes	dict (JSON Schema)
nodes	yes	`list[str
entry	yes	str (a node id present in nodes)
exit	yes	str (a node id present in nodes)
entry_bindings	no, default {}	dict[str, str]  runtime support pending; must stay empty. Setting non-empty makes every submission that reaches this team fail with DAG_VALIDATION_FAILED.
exit_bindings	no, default {}	dict[str, str]  same constraint as entry_bindings.
binding_overrides	no, default {}	dict[str, dict[str, str]]  { "<node_id>": { "<field>": "<binding>" } }. Replaces a node's declared bindings for the duration of this team's closure.
loop_configs	no, default []	list[LoopConfig]  same shape as the per-run loop_configs kwarg on run_workflow(...). Loops apply only when this team is reached.
config_overrides	no, default {}	dict[str, NodeConfigOverride]  per-node overlay on the validation-policy and node-config ladders. Outermost team on the closure path wins.

Submitting a team:

result = await client.run_workflow_target(
    "support_ops",
    {"ticket_id": "ZD-49102", "body": "..."},
)

run_workflow(workers=[...]) is the worker-list path; run_workflow_target("<name>", input) is the team / single-target path. Streaming uses the same split (run_workflow_stream vs run_workflow_target_stream). Both go through the same closure walker, so a team's loop_configs and config_overrides apply identically on both surfaces.

A few rules the constructor will not catch (the runtime enforces them at submission time):

• entry and exit must each appear in nodes. An entry or exit that names a node not in nodes is reported by check_team_health as a structural incompatibility.
• A sub-team referenced by {"id": "...", "team": "..."} must already be registered when the parent team is submitted; an unknown team raises REGISTRY_NOT_FOUND.
• entry_bindings and exit_bindings must be empty (see the table). The wire shape accepts them so the protocol can ship the rewiring path without a version bump, but submissions touching a team with either field non-empty fail today.
• A name that exists in both the worker registry and the team registry is rejected with DAG_VALIDATION_FAILED at submission time. Pick one namespace.

Extension

from skarta import Extension

ext = Extension("my-extension", "0.1.0", api_key="skarta_…")  # api_key optional, see below
ext.add_worker(my_worker)
ext.add_tool(my_tool)
ext.add_orchestrator(MyOrchestrator)
ext.add_interceptor(InterceptorDef.observer(...))
ext.add_context_provider(ContextProviderDef(...))
ext.add_webhook(name="customer-events", target="handle_customer_event",
                response_mode="async", dedupe_header="x-event-id")
ext.set_skills_directory("./skills")
ext.set_session_storage(my_storage_handler)  # see "Session storage backend" below
ext.set_permissions(perms)
ext.on_event("budget.*", on_budget)          # see "Event subscription" below

await ext.run("/tmp/skarta.sock")           # blocks until shutdown
await ext.run_reconnecting("/tmp/skarta.sock")  # auto-reconnect; see "Reconnecting" below

Extension.add_webhook(...)

ext.add_webhook(
    name="customer-events",
    target="handle_customer_event",       # XOR with worker_list
    worker_list=None,                     # XOR with target
    response_mode="async",                # "async" | "sync"
    timeout_ms=None,                      # required when response_mode="sync"
    dedupe_header=None,                   # e.g. "x-event-id"; replay-dedupe key
    dedupe_ttl_secs=86_400,               # max 7 days
    description="Optional human-facing description",
)

The runtime exposes POST /webhooks/<name> on the webhook ingress port (default 9100; override via SKARTA_WEBHOOK_HOST + SKARTA_WEBHOOK_PORT or the [webhook_ingress] TOML block). Each fire submits a fresh DAG with the HTTP request bundled as input. See webhooks.md for the input bundle shape, response semantics, and dynamic-webhook admin RPCs.

Authentication via api_key

Pass api_key="skarta_…" to Extension(...) to authenticate the Register frame against the runtime's api_keys table (P44 / ADR-050). Falls back to the SKARTA_API_KEY environment variable when the constructor argument is None. Required when the runtime's [auth] required = true; ignored when the runtime is in default local-dev mode. The same convention applies to skarta.Client(target, api_key=...) and skarta.AsyncClient.connect(target, api_key=...).

Useful introspection:

• ext.id  the extension id.
• ext.handle()  None until register_ack arrives, then the runtime-issued handle string.
• ext.runtime_version()  set on register_ack.
• ext.worker_names(), ext.tool_names().

Reconnecting

await ext.run(target) blocks until the runtime sends a clean shutdown message; on any other terminal condition (transport drop, runtime crash, registration failure, malformed frame, internal SDK exception) it raises. await ext.run_reconnecting(target, config=None) wraps run in a retry loop:

from skarta import Extension, ReconnectConfig

ext = Extension("my-extension", "0.1.0")
# ... register capabilities ...

await ext.run_reconnecting(
    "/tmp/skarta.sock",
    config=ReconnectConfig(
        initial_backoff_s=0.1,   # first retry waits this many seconds
        max_backoff_s=10.0,      # cap on the doubling backoff
        max_attempts=2**31,      # effectively unlimited; lower it to fail fast in tests
    ),
)

ReconnectConfig is a dataclass exported from the top-level skarta package. All three fields are optional and use the defaults shown above; pass config=None (or omit it) to take all defaults. Backoff doubles on each retry, capped at max_backoff_s.

What counts as "transport error" worth retrying. The retry loop catches every Exception that escapes run, not just transport drops. That means it will retry on:

• TCP / Unix-socket disconnects, WebSocket drops.
• Runtime-side errors returned during the register handshake (including REGISTRY_CONFLICT when this extension's tool / worker / orchestrator name collides with an earlier extension, PROTO_VERSION_INCOMPATIBLE when the SDK protocol version is outside the runtime's supported range).
• Any FrameworkError the runtime returns for a malformed register message.
• SDK-side exceptions in the read loop (codec errors, JSON parse failures, handler crashes that escape).

It will not retry on a clean shutdown (the runtime sent shutdown, run returned cleanly). Loud-fail registration errors like REGISTRY_CONFLICT therefore loop forever by default. To fail fast on those, lower max_attempts (e.g. max_attempts=3) and inspect the raised FrameworkError from the loop's last attempt. There is no per-error-code allowlist today.

Each retry logs at warning level with the attempt count, target, exception, and computed backoff (Transport error on attempt N (target=...): <exc>. Reconnecting in S.SSs). Watch this log line if you want to spot looping registration failures.

Event subscription

ext.on_event(pattern, handler) registers an async callback for runtime-broadcast events. Multiple handlers per pattern are supported (each registration appends to a list).

async def on_budget(event: dict) -> None:
    print(event["category"], event["event_type"], event["data"])

ext.on_event("budget.*", on_budget)   # all budget events
ext.on_event("*", lambda e: print(e)) # firehose
ext.on_event("agent.model_call", on_call)  # one exact event

Handler signature: Callable[[dict], Awaitable[None]]. The handler receives the full event payload { "category", "event_type", "data" } and runs as a fire-and-forget asyncio task; an exception inside the handler is logged but does not terminate the read loop.

Pattern syntax (matches the runtime's filter exactly, after the bug-for-parity fix):

Pattern	Matches
"*"	every event the runtime broadcasts to this extension.
"<prefix>.*" (e.g. "webhook.*", "schedule.*", "conversation.*")	every event whose lowercased <category>.<event_type> starts with <prefix>..
Exact "<category>.<event_type>" (e.g. "webhook.webhook_received", "schedule.schedule_fired")	only that one event.

The match key is the lowercased <category>.<event_type> string, where <category> is the protocol's EventCategory rendered in snake_case. The full enum today (twelve variants): agent, dag, extension, budget, session, skill, registry, webhook, outbound_bus, schedule, sandbox, conversation. Of those, only five currently broadcast event envelopes that events_subscribe patterns can match: webhook, outbound_bus, sandbox, schedule, conversation. The other seven categories are reserved for runtime-internal telemetry that today only feeds Prometheus and the recent_events ring buffer (next paragraph explains why); subscribing to "agent.*", "dag.*", "budget.*", etc. is currently a no-op. Other pattern shapes (substring, regex, glob) are not supported either; they will simply never match.

events_subscribe capability. Each on_event(pattern, ...) call appends pattern to the extension's events_subscribe list, sent to the runtime at register time. The runtime keeps an index from pattern to extension handle and broadcasts only matching events to this extension. Calling on_event after await ext.run(...) has started does not retroactively subscribe at the runtime; the registered list is captured at register time.

What events actually flow over this channel. This is the important caveat. The runtime's events_subscribe channel only re-broadcasts events that come into the runtime as Message::Event envelopes from other connected extensions. Today, the runtime's own internal telemetry events (budget, model_call, dag node started / completed, interceptor fires, etc.) are not relayed over this channel. Those events live on the runtime's internal TelemetryBus, which feeds Prometheus metrics and the recent_events ring buffer but does not write Message::Event back to subscribed extensions. Concretely:

• A BUDGET_SOFT_LIMIT breach lights up the skarta_budget_trips_total{kind="soft"} Prometheus counter and lands on the internal ring buffer; an extension subscribed to "budget.*" will not see it. (See section Common error codes and concepts.md section Budget events are runtime-internal, not on the workflow stream.)
• DAG-node lifecycle events (dag_node_started, dag_node_completed, etc.) are delivered to the submission's run_workflow_stream consumer as DagEvent chunks on the per-run stream; they are not duplicated onto events_subscribe.
• Interceptor invocations are dispatched as RPCs (interceptor.invoke), not as Message::Event broadcasts.

So on_event is effectively an extension-to-extension pub/sub channel. Extension A posts a Message::Event envelope on the wire; the runtime forwards it to every other extension whose events_subscribe patterns match. Useful for cross-extension coordination ("the audit-log extension wants to hear app.payment.* from the checkout extension"), less useful as a window into the runtime's own behaviour. The current SDK has no public ext.emit_event(...) helper either, so emitting an event today requires writing a Message::Event envelope through the codec by hand.

The relationship to interceptors: interceptors are synchronous, RPC-style hooks that observe or modify a specific runtime operation (LLM call, tool call, context assembly), and can veto the operation by raising FrameworkError. Events are fire-and-forget broadcasts, never block, never veto, and only carry whatever the publishing extension chose to put in data.

Session storage backend

ext.set_session_storage(handler) registers this extension as the runtime's session-storage backend. While the extension is connected the runtime routes session_storage.{save, load, list, delete} RPCs to handler instead of writing to the built-in file directory or database. When the extension disconnects the runtime falls back to the built-in backend on the very next call (no cached selection).

Selection rule when several backends are available:

1. The first connected extension that declared session_storage=True wins (ordered by registered_at; first-registered is preferred so two extensions accidentally claiming the role do not flip-flop).
2. Otherwise, the runtime uses whatever built-in backend is configured: [session].storage = "file" (default) or [session].storage = "db".

The handler signature:

from typing import Any, Awaitable, Callable

SessionStorageHandler = Callable[[str, Any], Awaitable[Any]]

That is, an async def taking exactly two arguments: the sub-method name ("save", "load", "list", or "delete") and the raw params dict received on the wire. It must return a JSON-serializable response in the shape below. Raising FrameworkError(SessionStorageError, ...) (or SessionCorrupted, SessionNotFound) propagates back to the caller; raising any other exception is converted to INTERNAL_ERROR.

Wire contract for the four sub-methods:

Sub-method	Params dict	Response dict
"save"	{"session": <Session JSON>}	{}
"load"	{"id": "<session_id>"}	{"session": <Session JSON>}
"list"	None (the runtime sends JSON null)	{"sessions": [<SessionMeta JSON>, ...]}
"delete"	{"id": "<session_id>"}	{}

<Session JSON> is the full serialized session tree; <SessionMeta JSON> is the lightweight summary returned by list. Field shapes:

// Session
{
  "id": "string",
  "created_at": "RFC 3339 timestamp",
  "updated_at": "RFC 3339 timestamp",
  "nodes": {                            // keyed by node id
    "<node_id>": {
      "id": "string",
      "parent_id": "string | null",
      "role": "user | assistant | system | tool",
      "content": [ <ContentBlock>, ... ],
      "metadata": {
        "timestamp": "RFC 3339 timestamp",
        "model": "string | null",
        "usage": { "tokens_in": 0, "tokens_out": 0, "cost_usd": 0.0 } | null,
        "branch_label": "string | null"
      }
    }
  },
  "root_id": "string | null",           // id of the root node, when the tree is non-empty
  "active_branch": "string | null",     // id of the leaf new appends extend from
  "checkpoints": { "<name>": "<node_id>" }   // named pointers into the tree
}

// SessionMeta (one per row in the list response)
{
  "id": "string",
  "created_at": "RFC 3339 timestamp",
  "updated_at": "RFC 3339 timestamp",
  "active_branch": "string | null",
  "node_count": 0,
  "checkpoint_count": 0
}

Branching and checkpointing are not routed through the handler. They are in-memory mutations on the Session object the runtime already holds (active_branch repointed, a new entry added to checkpoints); the runtime then persists the mutated Session via a regular "save". The handler does not need to model branches or checkpoints separately, only to round-trip the full Session blob.

Minimal JSON-file backend example:

import asyncio
import json
import os
from pathlib import Path
from skarta import Extension

DIR = Path(os.environ.get("MY_SESSION_DIR", "./.sessions"))
DIR.mkdir(parents=True, exist_ok=True)

async def storage(method: str, params):
    if method == "save":
        session = params["session"]
        (DIR / f"{session['id']}.json").write_text(json.dumps(session))
        return {}
    if method == "load":
        body = (DIR / f"{params['id']}.json").read_text()
        return {"session": json.loads(body)}
    if method == "list":
        out = []
        for path in DIR.glob("*.json"):
            s = json.loads(path.read_text())
            out.append({
                "id": s["id"],
                "created_at": s["created_at"],
                "updated_at": s["updated_at"],
                "active_branch": s.get("active_branch"),
                "node_count": len(s.get("nodes", {})),
                "checkpoint_count": len(s.get("checkpoints", {})),
            })
        return {"sessions": out}
    if method == "delete":
        (DIR / f"{params['id']}.json").unlink(missing_ok=True)
        return {}
    raise ValueError(f"unknown session_storage sub-method: {method}")

ext = Extension("json-session-store", "0.1.0")
ext.set_session_storage(storage)
asyncio.run(ext.run("/tmp/skarta.sock"))

Bring the extension up alongside the runtime and client.save_session(...) / client.list_sessions() / client.load_session(...) will land on the JSON files. Drop the extension and the runtime continues from the built-in backend without any reconfiguration.

Interceptor hook decorators

Seven decorators register an interceptor handler on the default extension. Each maps to one of the runtime's InterceptorHook points; payload shapes for each hook live in concepts.md section Interceptors.

from skarta import before_model, after_model, before_tool, after_tool
from skarta import before_skill, on_context_change, on_compaction

Decorator	Fires on
@before_model	BEFORE_MODEL_CALL -- right before llm.complete dispatches to the provider.
@after_model	AFTER_MODEL_CALL -- after the provider response lands, before budget accounting.
@before_tool	BEFORE_TOOL_EXECUTION -- before a tool dispatch runs the handler.
@after_tool	AFTER_TOOL_EXECUTION -- after the handler returns (or after a vetoed call's error path).
@before_skill	BEFORE_SKILL_CALL (P53). after_skill is intentionally absent until the runtime hook ships.
@on_context_change	ON_CONTEXT_MUTATION -- when the runtime mutates the assembled context (skills inlining, context providers).
@on_compaction	ON_COMPACTION -- on context.notify_compaction or auto-trigger when the assembled-context estimate crosses [compaction] auto_trigger_threshold_tokens.

The handler must be async def; non-async callables raise TypeError at decoration time.

Decorator kwargs

All seven decorators accept the same three keyword arguments:

@before_model(priority=50, mode="modify", filter={"model": "claude-opus-4-5"})
async def cap_opus_temperature(payload):
    payload["context"]["temperature"] = min(payload["context"].get("temperature", 0.7), 0.3)
    return payload

Kwarg	Type	Default	Effect
priority	int	100	Order when several handlers register on the same hook. Lower runs first.
mode	str	"observe"	"observe" (read-only) or "modify" (may rewrite the payload). Returning a dict from a "modify" handler replaces the payload the runtime forwards; returning None is the observer no-op.
filter	Optional[Any]	None	Scopes the handler. Forwarded as-is into InterceptorDef.filter; the runtime applies it against the hook's payload before invoking the handler.

Returning (or raising) a FrameworkError (or any subclass like ApprovalDeniedError) vetoes the underlying tool call or LLM call -- the caller sees the error and the action never runs.

The Rust equivalents are InterceptorDef::observer(...).with_priority(...).with_filter(...) and InterceptorDef::modifier(...); see rust-sdk.md section Interceptors.

AgentContext

The handle a worker uses to talk to the runtime.

@worker(...)
async def my_worker(input: In, ctx: AgentContext) -> Out:
    response = await ctx.llm(model="...", messages=[...], system="...")
    result = await ctx.tool("lookup_customer", arguments={"id": "123"})
    sub = await ctx.invoke_worker("other_worker", {"x": 1})
    skill = await ctx.load_skill("style-guide")
    await ctx.send_peer("peer_worker", {"hint": "go fast"})
    msg = await ctx.recv_peer(timeout=2.0)
    await ctx.stream_partial({"token": "hello"})
    usage = ctx.usage()
    return Out(...)

Methods:

Method	What it does
await ctx.llm(model, messages, system=None, *, tools=None, max_tokens=None, temperature=None, thinking_enabled=False, stop_sequences=None, endpoint_name=None, api_key=None)	One LLM completion. endpoint_name resolves a registered endpoint by name; api_key passes a one-shot key. Without either, the runtime resolves the model from the registry.
await ctx.llm_stream(model, messages, system=None, *, tools=None, max_tokens=None, temperature=None, thinking_enabled=False, stop_sequences=None, endpoint_name=None, api_key=None)	Async iterator of streaming chunks. Mirrors the ctx.llm options surface so streamed calls accept temperature, tools, endpoint_name, api_key, etc. Tool calls arrive as tool_call_start / tool_call_delta / tool_call_end chunks; the worker assembles them and runs ctx.tool(...) itself, then makes a follow-up llm_stream call with the tool result appended (no auto-multi-turn). See concepts.md section LLM stream chunks for the full per-chunk shape.
await ctx.tool(name, arguments)	Call a registered tool. arguments is a dict whose keys must match the tool's declared input fields verbatim. For a Python tool declared with @tool and a Pydantic args_model, the keys are that model's field names. For a Rust tool declared with #[tool] async fn fetch(customer_id: String, limit: u32), the macro builds a hidden __ToolArgs struct from the function's typed parameters, so the wire shape is {"customer_id": "...", "limit": 10} (the bare parameter names). Cross-language calls work as long as the dict keys match the Rust parameter names; no positional / by-position dispatch.
await ctx.invoke_worker(name, input)	Call another worker as a subroutine.
await ctx.send_peer(to_node, data)	Fire-and-forget peer message inside the same DAG run. See Peer messaging below.
await ctx.recv_peer(timeout=None)	Await the next peer message targeted at this worker execution. Returns PeerMessage(sender, data). See Peer messaging below.
ctx.try_recv_peer()	Non-blocking peer receive. Returns PeerMessage(sender, data) or None. See Peer messaging below.
await ctx.conversation_create(name=, participants=, shape=, ...)	Open a multi-agent conversation room (P52 / ADR-061). Returns the resolved ConversationConfig. See Conversations.
await ctx.conversation_join(name, worker?)	Late-join an open room. The runtime emits participant_joined.
await ctx.conversation_resume(name, since=0)	Re-attach to a persisted room after runtime restart. Returns {config, transcript}.
ctx.conversation_handle(name, worker?)	Build an ergonomic Conversation handle: speak / address / reply / pushback / agree / pass_turn / give_floor / terminate / run / on_turn / poll / get_transcript / next_speaker / describe.
await ctx.conversation_leave(name, worker?)	Leave an open room without terminating it. Mirrors conversation_join.
await ctx.load_skill(name)	Load a skill by name; returns the skill body.
await ctx.gather_context(request)	Ask context providers for context.
await ctx.notify_context_mutation(payload)	Tell interceptors the context changed.
await ctx.stream_partial(data)	Push one partial-result chunk to subscribers of this worker's stream. data is free-form JSON; it is not validated against the worker's output schema. Chunks are delivered in send order. The runtime opens the stream lazily on the first call, so streaming=True on the decorator is metadata, not a gate.
await ctx.close_stream(status="completed")	Terminate this worker's partial stream. Valid status values: "completed", "cancelled", "error". Optional: the SDK auto-closes with "completed" when the handler returns. Status is for telemetry / replay hygiene; consumers see the same clean iterator exit on every status.
await ctx.read_stream(source_id)	Subscribe to an upstream worker's partial-result stream and return an async iterator of chunks. source_id is the value the runtime injected into the consumer's input under <bound_field>["__stream_source"] from a stream("<producer>") binding; do not construct it by hand. The iterator yields each chunk the producer pushed and exits cleanly (StopAsyncIteration) when the producer closes the stream. Late subscribers replay from the start. See concepts.md section Worker partial streams.
ctx.usage()	Aggregate token / cost usage so far.
ctx.worker_name, ctx.dag_id	Properties for the running worker / DAG.

Peer messaging

Workers running in the same DAG can talk to each other through ctx.send_peer / ctx.recv_peer / ctx.try_recv_peer. The runtime forwards each message as a peer_message frame over the extension transport.

Wire shape sent by send_peer:

{
  "dag_id": "<dag_id or empty string when no DAG context>",
  "from_node": "<sending worker's name>",
  "to_node": "<target worker's name>",
  "data": <any JSON-serializable value>
}

to_node resolution. The string is the @worker name (the same name you registered with, or the function name when name= is omitted). Cross-extension delivery is supported: the runtime looks up which extension owns the worker via its registry and forwards the frame to that extension's transport. Node ids, fan-out instance ids, and team-relative names are not addressable in the current build (the runtime explicitly notes this in its routing code, and a fan-out worker has all its parallel instances sharing one inbox keyed by worker name).

recv_peer / try_recv_peer return value. Both return a PeerMessage(sender, data) (P52 / ADR-061 Slice A) so the receiver can read who sent a message from the wire envelope without senders embedding their own names in the payload. The sender field carries the originating worker's name; data is the original send_peer payload.

Call	Returns
await ctx.recv_peer()	A PeerMessage(sender, data). Blocks indefinitely until one arrives.
await ctx.recv_peer(timeout)	A PeerMessage(sender, data), or None if timeout seconds elapse first. Never raises asyncio.TimeoutError.
ctx.try_recv_peer()	The next buffered PeerMessage(sender, data) or None if the inbox is empty. Never raises.

When the worker handler is executing in a context built without an inbox (for example, a unit test that constructs AgentContext directly), all three methods return None immediately rather than raising.

Inbox semantics.

• Scope  the inbox is keyed by (dag_id, worker_name). One inbox per worker per DAG run. Direct (non-DAG) dispatches share a single dag_id="" bucket per worker name.
• Lifetime  created lazily on the first matching peer_message frame or when worker.dispatch runs, whichever comes first. Removed when the worker handler returns. Late-arriving messages after that point are warned about by the read loop and dropped.
• Order  FIFO. Messages from a single sender arrive in send order. Order across multiple senders is the order the runtime saw frames on the wire.
• Backpressure  the inbox is an unbounded asyncio.Queue. send_peer is fire-and-forget on the sender side; producers never block waiting for the consumer.
• Delivery guarantees  at-most-once. The runtime drops the frame (with a warn-level log line) if the target worker is not registered at the moment the frame is processed. This typically means the target hasn't dispatched yet and the runtime cannot find any extension owning that worker name. The sender does not get an error.
• No cross-DAG-run delivery  a peer_message from dag_id=A will not reach a recv loop running with dag_id=B; the inbox keys do not match. Direct dispatches without a DAG (dag_id="") share their own bucket.
• Same-instance and fan-out  multiple parallel instances of the same worker name (concurrency > 1, or fan-out via each(...)) share the same inbox. Whichever instance happens to call recv_peer first wins. There is no instance-targeting yet.

Minimal example.

@worker(description="Driver", bindings={})
async def driver(input: In, ctx: AgentContext) -> Out:
    await ctx.send_peer("listener", {"hint": "go fast"})
    return Out(...)

@worker(description="Listener", bindings={})
async def listener(input: In, ctx: AgentContext) -> Out:
    msg = await ctx.recv_peer(timeout=2.0)   # returns {"hint": "go fast"} or None
    return Out(...)

Both workers must run in the same DAG submission for the message to land; if listener is dispatched in a separate run it will never see the hint.

Steering

Steering is the conceptual mechanism for sending control signals (cancel, priority bump, additional context, "you've produced enough output") into a long-running or streaming worker. The current Skarta build has the type system and protocol surface for steering but does not ship a worker-handler API for sending or receiving steer signals from inside an @worker function. Treat the rest of this section as the inventory of what is and is not on the wire today.

Shipped surface:

• The SteerSignal Pydantic model under skarta.protocol.types. Construct one with SteerSignal.cancel(reason=...), SteerSignal.priority(level=...), SteerSignal.context(data=...), or SteerSignal.enough(message=...). The kind field tags the variant on the wire ("cancel" | "priority" | "context" | "enough").
• The steerable: bool = False flag on the @worker and @tool decorators. Setting steerable=True sets the same flag on the WorkerDeclaration / ToolDeclaration registered with the runtime. It is currently advertisement-only; the runtime forwards it through registration and stores it on the registry, but no built-in policy branches on its value yet.
• The stream_steer protocol frame ({"stream_id": "<id>", "signal": <SteerSignal>}). The runtime dispatches inbound stream_steer frames to the same per-stream_id router that handles stream_chunk and stream_close, so a subscriber receiving a stream from this stream id will see the signal interleaved with chunks.

Not shipped:

• There is no ctx.steering() method on AgentContext. A streaming=True worker has no inbox-style API for reading inbound steer signals from inside its handler.
• There is no ctx.steer(target, signal) method either. The Python SDK does not expose a way to construct and send a stream_steer frame from worker code; the only way to push one onto the wire today is to bypass the SDK and write the frame directly through the codec.
• The cancel, priority, enough, context signals therefore have no defined effect on a target worker handler. cancel does not raise CancelledError into a running handler, priority does not reorder the executor's dispatch queue, enough does not stop a streaming response, and context does not feed extra messages into the next LLM call. These are the semantics one would expect, but they are not wired up.
• Steer signals do not currently emit a corresponding event on the DAG event log (no node_steered event). If a steer signal is delivered to a stream subscriber, only that subscriber sees it; there is no broadcast through events.subscribe.

If a demo bullet says "mid-stream steering of a streaming=True worker", what is actually needed today is one of: (a) build the worker around a peer-message channel (ctx.recv_peer is a fully shipped path that gives the same control-channel pattern); (b) hold the build until the steering API lands; or (c) have the user explicitly ask for the SDK steering work to be scheduled.

Decorators

@tool

from pydantic import BaseModel
from skarta import tool
from skarta.protocol.types import LatencyClass, CostClass

class LookupArgs(BaseModel):
    customer_id: str

@tool(
    description="Fetch a customer record.",
    latency_class=LatencyClass.SECONDS,
    cost_class=CostClass.CHEAP,
    idempotent=False,
    requires_approval=False,
    streaming=False,
    steerable=False,
)
async def lookup_customer(args: LookupArgs) -> dict:
    return {"id": args.customer_id, "plan": "pro"}

The first argument's Pydantic model is used to derive the JSON schema the model sees. The return value is normalised to a JSON-serializable dict (Pydantic models are dumped automatically). Pass args_model=... and result_model=... to override the inference.

All @tool kwargs:

Kwarg	Type	Default	What it does
name	str | None	function name	Wire-level tool name. Must be globally unique across all extensions; collision returns REGISTRY_CONFLICT at register time.
description	str | None	function docstring	Surfaced in registry snapshots and the orchestrator planning prompt. The model uses it to decide when to call the tool.
args_model	Type[BaseModel] | None	inferred from first arg	Override Pydantic input model used to derive parameters schema.
result_model	Type[BaseModel] | None	none	Pydantic model for the return value's JSON schema. Optional; the runtime still accepts dict / str / ToolResult returns.
streaming	bool	False	Advisory: the tool emits incremental output via ctx.stream_partial. Today the runtime opens the stream lazily on the first chunk regardless of this flag, so it is metadata for reviewers and orchestrator hints, not a gate.
steerable	bool	False	Advisory: the tool can accept mid-execution SteerSignal frames. Currently advertisement-only (no ctx.steering() is shipped). See Steering.
latency_class	LatencyClass	LatencyClass.SECONDS	One of INSTANT, SECONDS, MINUTES, HOURS. Advisory hint for orchestrator planning ("don't pick a HOURS tool for an interactive turn") and dashboards. The runtime does not enforce timeouts based on this field; per-call timeouts come from RetryPolicy.timeout_ms overrides.
cost_class	CostClass	CostClass.CHEAP	One of FREE, CHEAP, MODERATE, EXPENSIVE. Same status: advisory hint for orchestrator planning and dashboards, not a budget gate. Hard cost gates live under [budget] config and surface as BUDGET_* errors.
idempotent	bool	False	Advisory: re-running with the same arguments is safe. Future retry policies will read this; today no built-in retry path branches on it.
requires_approval	bool	False	Advisory: this tool should be gated behind a human approval step. Currently advertisement-only; no built-in interceptor or runtime policy enforces it. To actually gate a call, register an InterceptorDef on before_tool_call that checks the declaration and vetoes when needed.
sandbox	Sandbox | None	None	Per-tool execution-sandbox override (P49 / ADR-058). Sandbox is the typed manifest from skarta describing filesystem / network / subprocess permissions; the value is attached to the registered ToolDeclaration and consumed by the runtime when the tool runs in a sandbox-enabled extension. None (default) inherits the extension-level policy. See sandbox.md for the manifest fields and the runtime modes (enforce / audit / bypass).

LatencyClass and CostClass are (str, Enum) types from skarta.protocol.types, so LatencyClass.MINUTES == "minutes" and either form serialises identically. They are not re-exported from the top-level skarta package; import them explicitly.

The four advisory fields (latency_class, cost_class, idempotent, requires_approval) are part of the registered ToolDeclaration. The runtime stores them in the in-memory ExtensionEntry.capabilities and reads them when assembling the orchestrator's planning context. They are also reachable over the wire via describe_tool(name), which returns the full declaration including every advisory field. Use that RPC from a planner, dashboard, or routing layer that needs to branch on cost / latency / approval before submitting work. registry_query() still returns names only.

@worker

@worker(
    description="...",
    bindings={"x": "upstream.y"},
    tools_required=["lookup_customer"],
    skills_required=["customer-tone"],
    model_preference="claude-sonnet-4-6",   # advisory; see kwarg table below
    concurrency=4,
    streaming=True,
    steerable=False,                 # advertisement-only flag; see "Steering" above
    validation_policies={"x": "coerce"},
)
async def my_worker(input: In, ctx: AgentContext) -> Out: ...

Required: the first arg must be annotated with a Pydantic BaseModel or the literal type str, and the return type must be the same. The decorator extracts both schemas.

Plain-str shorthand (P52.6 / ADR-063)

For workers whose input or output is a single string (LLM transforms, formatters, classifiers that return a label), annotate the parameter or the return type as the literal str instead of writing a one-field Pydantic model:

from skarta import worker, AgentContext

@worker(description="Uppercase the input.")
async def shout(text: str, ctx: AgentContext) -> str:
    return text.upper()

The two sides are independent: str -> str, str -> SomeModel, and SomeModel -> str are all accepted. The decorator transparently swaps in a synthetic one-field wrap model (_PlainStringInputModel / _PlainStringOutputModel) so the wire payload is the canonical {"value": "<the string>"} object every Skarta worker emits. The field name is locked at value, so downstream workers bind on <upstream>.value like any other one-field shape:

@worker(bindings={"text": "shout.value"})
async def append_bang(text: str, ctx: AgentContext) -> str:
    return f"{text}!"

Anything other than the literal str (int, dict, list[str], custom type aliases) keeps the existing Pydantic-only rule and raises TypeError from the decorator. Mixed shapes (a str input feeding a Pydantic-output worker, or vice versa) work the same way; the wrap is per-side.

All @worker kwargs:

Kwarg	Type	Default	What it does
name	str | None	function name	Globally unique worker name. Collision returns REGISTRY_CONFLICT at register time.
description	str | None	function docstring	Surfaced in registry snapshots, the orchestrator planning prompt, and dashboards.
input_model	Type[BaseModel] | None	inferred from first arg type	Override the Pydantic input model.
output_model	Type[BaseModel] | None	inferred from return annotation	Override the Pydantic output model.
bindings	dict[str, Binding] | None	none	Field-level binding DSL. See bindings_dsl and concepts.md section Bindings.
tools_required	list[str] | None	[]	Names of tools the worker calls via ctx.tool(...). Visible to orchestrator planning. Advisory: the runtime does not verify these are registered before dispatch (a missing tool surfaces at call time as REGISTRY_NOT_FOUND).
skills_required	list[str] | None	[]	Names of skills the worker loads via ctx.load_skill(...). Same advisory status as tools_required.
model_preference	str | None	None	Advisory: a hint string the worker would prefer a specific model when it calls ctx.llm(...). Currently advertisement-only: it is registered, threaded through the DAG node, exposed in the visualised graph and NodeConfigOverride.model ladder, but the runtime does not auto-rewrite a worker's ctx.llm(model=...) argument based on it. The worker handler can read its own model_preference from the registry snapshot if needed. The submission-time node_overrides[<worker>].model field uses the same plumbing for per-run model swaps; that override wins when both are set.
concurrency	int	1	Max in-flight instances of this worker. Enforced by a per-worker semaphore that the runtime acquires before dispatch. Submission and team config_overrides cannot lower this; they can only override retry / timeout_ms / validation_policies.
conditions	str | None	None	A boolean expression on input fields; nodes whose condition evaluates false are skipped (skip reason: condition_false).
latency_class	LatencyClass	LatencyClass.SECONDS	One of INSTANT, SECONDS, MINUTES, HOURS. Advisory hint for orchestrator planning and dashboards. Not enforced as a timeout.
cost_class	CostClass	CostClass.MODERATE	One of FREE, CHEAP, MODERATE, EXPENSIVE. Advisory hint. Hard cost gates live under [budget] and surface as BUDGET_* errors.
streaming	bool	False	Documents that the worker calls ctx.stream_partial(...). The runtime opens the stream lazily on the first call regardless of this flag; the bit is metadata for reviewers and orchestrator hints.
steerable	bool	False	Advertisement-only flag; see Steering.
validation_policies	dict[str, str] | None	None	Per-field coercion policies ("fail" | "coerce" | "default" | "drop_field").
sandbox	Sandbox | None	None	Per-worker execution-sandbox override (P49 / ADR-058). Same Sandbox manifest type used on @tool. None (default) inherits the extension-level policy. See sandbox.md.
idempotency	str | None	None	Durability-replay posture for the worker (P43 / ADR-071). Accepts "idempotent" (alias "pure" / "safe") for handlers that are safe to re-execute on resume, or "external_side_effect" for handlers that must run at most once. Anything else raises TypeError from the decorator. None leaves the handler unannotated. See durability.md and idempotency.md.

@worker accepts concurrency (default 1, the maximum number of in-flight instances of this worker) but not retry or timeout_ms. Retry policy and per-node timeout are submission-time overrides, set via node_overrides={"<worker>": NodeConfigOverride(retry=..., timeout_ms=...)} on run_workflow(...) (see Per-run controls above) or via a team's config_overrides. Submission overrides win over team overrides; team overrides win over the worker's concurrency default.

The advisory classifier fields (latency_class, cost_class, model_preference, streaming, steerable) are recorded on the worker declaration. The runtime stores them in the in-memory ExtensionEntry.capabilities and threads model_preference into the derived DagNodeInfo. To pull the full declaration (including every advisory field plus bindings, validation_policies, tools_required, skills_required, concurrency, conditions) over the wire, use describe_worker(name). Visualised graphs (visualize_team(...) with format="json") carry model_preference per node; describe_worker is the path for everything else.

validation_policies is a per-field map: {"<field>": "fail" | "coerce" | "default" | "drop_field"}. Strings are accepted everywhere on the wire; the same values also live as the ValidationPolicy enum if you prefer typed code:

from skarta.protocol.types import ValidationPolicy, NodeConfigOverride

@worker(..., validation_policies={"x": ValidationPolicy.COERCE})
async def my_worker(input: In, ctx: AgentContext) -> Out: ...

# Same map shape on a team's config_overrides and on submission overrides:
NodeConfigOverride(validation_policies={"y": "fail"})

ValidationPolicy is a (str, Enum), so ValidationPolicy.COERCE == "coerce" and either form serialises the same. It is exported from skarta.protocol.types (not from the top-level skarta package).

The same validation_policies map is accepted on all three layers of the precedence ladder: worker decoration, team config_overrides[<worker>].validation_policies, and submission node_overrides[<worker>].validation_policies. Layers are merged per field (not whole-map replace): a submission saying {"y": "fail"} only overrides y; any other policies the worker or teams declared are preserved. See concepts.md section Validation for the full ladder, the policy semantics (default reads the JSON Schema default keyword; drop_field only applies to declared optional fields), and the error codes raised when a policy can't salvage a value (WORKER_INPUT_VALIDATION_FAILED at Point 3, WORKER_OUTPUT_VALIDATION_FAILED at Point 4, never BINDING_SCHEMA_MISMATCH, which is reserved for the static Point 2 binding-edge check).

@orchestrator

@orchestrator(
    model="claude-sonnet-4-6",
    system_prompt="Compose DAGs for the support ops team.",
    temperature=0.2,
    max_tokens=2048,
    thinking_enabled=False,
    stop_sequences=["END"],
    endpoint_name=None,
)
class MyOrchestrator:
    """A short docstring; used as the declared description."""

Marks a class as an orchestrator declaration. The decorator attaches an OrchestratorSpec to the class; pass the class to Extension.add_orchestrator(...). The decoration carries the model and LLM-call settings the runtime uses when a submission names this orchestrator. There is no plan(...) method to write: the runtime composes a registry snapshot, makes one tool-using LLM call against the resolved model, parses the chosen plan, and dispatches it.

Resolution order at dispatch (highest precedence first): CLI flags, process env, .env, framework.toml [orchestrators.<name>], the decoration above, then built-in defaults. Operators can swap the model or system prompt without an SDK redeploy. system_prompt and system_prompt_path are mutually exclusive; set at most one. When endpoint_name is set the runtime resolves a named LLM endpoint (provider + model + key + default params) per ADR-019; the decoration's model still overrides the endpoint's model when both are set.

Submit through an orchestrator by passing its name on the submission:

result = await client.run_workflow_target(
    "summarise",
    {"topic": "rain"},
    orchestrator="MyOrchestrator",
)

Streaming clients see two extra DagEvent chunks before any node-level event: orchestrator_planning (emitted before the LLM call) and orchestrator_plan_complete (emitted with a dag_summary once the plan parses cleanly). Plan failures (no model resolved, conflicting prompt fields, no submit_dag tool call returned, schema-violating plan) all surface as OrchestratorPlanError with structured details.

Binding helpers

Helper	Returns
input_(path)	Read from the workflow's outer input.
literal(value)	Constant.
each(path)	Fan out one instance per element.
each_index()	Position inside a fan-out.
each_count()	Total fan-out count.
collect(name)	Fan in: every fan-out instance's output.
collect_count(name)	Number of fan-out instances.
merge(*sources)	Combine sources field-by-field.
concat(*sources)	Concatenate strings or lists.
coalesce(*sources)	First non-null wins.
default(source, fallback)	Use source, fall back to fallback.
when(condition, then_expr, else_expr)	Pick then_expr when condition is truthy, otherwise else_expr. Both branches are full bindings.
on_failure(agent, accessor=FailureAccessor.ERROR)	Recovery binding. agent is the worker name string (not a source binding). Fires only when that worker actually errored (raised, validation-failed, or timed out); it does not fire on cascade-skips or race cancellations. Three accessors (import from skarta.protocol.bindings import FailureAccessor): ERROR (default; the upstream's FrameworkError serialised as {code, message, details?, retryable}), PARTIAL_OUTPUT (currently always resolves to None; no runtime path populates it), ATTEMPT_COUNT (the integer attempt that failed). A consumer whose every binding is on_failure(...) auto-skips when no targeted upstream failed; mix with at least one regular binding to keep the consumer alive on the success path (the on_failure field then resolves to None). Always wins over skip_policies. See concepts.md section Failure handling.
on_complete(agent, accessor=CompletionAccessor.STATUS)	agent is the worker name string (not a source binding). Runs regardless of upstream outcome (success, failure, or cancellation) and blocks skip-cascade like on_failure. Three accessors (import from skarta.protocol.bindings import CompletionAccessor): STATUS (default; one of "completed" / "failed" / "cancelled", but returns nothing if the upstream was skipped because the runtime doesn't stamp #status for skipped nodes), RESULT_OR_NULL (upstream output on success, null otherwise), ERROR_OR_NULL (the FrameworkError envelope on failure, null otherwise). Use RESULT_OR_NULL / ERROR_OR_NULL if you need a guaranteed value across every terminal state including skip.
loop_(...)	Loop a worker until a condition.
loop_index() / loop_previous()	Loop position and prior iteration's output. loop_previous() returns the value the runtime tracks in loop state and takes no arguments.
race(*sources, cancel_losers=False)	First worker to complete wins. The bound field receives a {name, output} envelope (declare the consumer's input as e.g. winner: RaceWinner where RaceWinner has fields name: str and output: <winner_schema>). Resolves on every dispatch path, including loop heads and tails, so "race two providers per round inside a refine loop" works as a single binding on the loop head. With cancel_losers=True, the still-running candidates are aborted the moment the winner reports; they end up in DagResult.cancelled_nodes, not node_results or node_errors.
stream_(source)	Subscribe to a producer worker's partial-result stream. source is a producer worker name only (e.g. "summarize"), not a dotted path. The bound field receives a marker dict {"__stream_source": "<dag_id>::<producer>"}; declare it on the consumer's Pydantic input as dict[str, Any] (or with model_config = ConfigDict(extra="allow")) and read it via await ctx.read_stream(input.<field>["__stream_source"]). See concepts.md section Worker partial streams.

You can also write the bare string form "worker.field" for a simple read.

RuntimeManager (auto-spawn)

For async scripts that want to manage the bundled runtime explicitly:

from skarta.runtime import RuntimeManager, ResolutionMode, resolve_target

socket = "/tmp/skarta-myapp.sock"
mgr = RuntimeManager.spawn(socket)
try:
    ...  # connect, run workflows
finally:
    mgr.stop()

mgr.stop() sends SIGTERM, waits up to a timeout, then SIGKILL. An atexit hook also catches it on interpreter exit.

Errors

from skarta import FrameworkError, FrameworkErrorCode, ValidationError

try:
    await client.run_workflow(...)
except FrameworkError as e:
    if e.code is FrameworkErrorCode.BINDING_SCHEMA_MISMATCH:
        ...

ValidationError is a FrameworkError subclass tagged with the failing field path. The wrapper splits validation into two paths so a beginner can except ValidationError without thinking about wire shape:

• skarta.ValidationError (and its three subclasses InputValidationError, OutputValidationError, ToolValidationError from skarta.exceptions) is the typed wrapper a beginner catches. Carries the parsed field_path, the failing value, and the expected schema where the runtime supplied it.
• skarta.errors.ValidationError is the wire shape; it is re-exported from the top-level package as LowLevelValidationError so the two names never collide. Reach for it only when you parse the wire envelope yourself (custom transports, replay tools, audit pipelines).

Catch ValidationError for the everyday case; catch LowLevelValidationError only when you are working below the wrapper.

Common error codes

Code	When it fires	Where to look
BINDING_SCHEMA_MISMATCH	A binding edge's source and target schemas don't match at Point 2 (DAG validation).	The submission aborts; check error.message for the field path.
REGISTRY_NOT_FOUND	A registry-read RPC was called with a name (team / worker / tool / provider / model) that is not registered. Today the team-shaped sites are check_team_health(team_name), visualize_team(team), and run_workflow_target("<team>", ...).	error.message = "<Noun> '<name>' is not registered"; error.details always carries the structured key for the kind being looked up: { "team": "<name>" } for team RPCs, { "id": "<model_id>" } for get_model, etc. Branch on the structured key rather than parsing the message.
WORKER_INPUT_VALIDATION_FAILED	A node's assembled input doesn't match its declared input schema (Point 3).	error.details carries { worker, errors: [{ field, message, expected?, actual? }] }.
WORKER_OUTPUT_VALIDATION_FAILED	A worker returned a value that doesn't match its declared output schema (Point 4).	Same shape as input.
WORKER_EXECUTION_ERROR	The code a handler should raise FrameworkError(code=FrameworkErrorCode.WORKER_EXECUTION_ERROR, message=...) with for any deliberate, domain-level failure. The runtime preserves the worker-supplied message and details and lands the node in node_errors. A handler that raises a non-FrameworkError exception (or a FrameworkError with a code the runtime does not accept on this path) is rewrapped as INTERNAL_ERROR with message "Handler raised unhandled exception: <exception>", so always raise this code explicitly when the failure is part of the worker's contract.	Lands in DagResult.node_errors[<node>].
WORKER_TIMEOUT	Per-node timeout_ms elapsed.	Lands in DagResult.node_errors[<node>]; the in-flight handle is aborted.
DAG_TIMEOUT	A loop_(...) exhausted max_iterations with on_max="fail". Not fired by submission dag_timeout_ms; that path returns status="cancelled" instead.	Lands on the loop tail in node_errors.
TOOL_TIMEOUT	The runtime → extension RPC for tool.call did not get a reply within 30 seconds (hardcoded transport-level cap; no user-facing knob today: no kwarg on ctx.tool, no field on @tool, no TOML entry). Fires when the owning extension is stuck or unreachable, not when a live tool exceeds a per-call deadline.	Surfaces inside the calling worker as a FrameworkError; recovery is up to the worker's handler.
WORKER_PERMISSION_DENIED / TOOL_PERMISSION_DENIED	A dispatch payload referenced a path / host / env-var outside the extension's declared Permissions.	error.details carries { subject, kind, value, declared }. See concepts.md section Permissions.
BUDGET_HARD_LIMIT	Cumulative cost exceeded [budget].hard_limit_usd after a provider call lands. The crossing call's response is replaced with this error; the provider was still billed.	error.details = { hard_limit_usd }
BUDGET_PER_CALL_LIMIT	The estimated tokens for a proposed ctx.llm / ctx.llm_stream call (max_tokens requested + estimated context tokens) exceed [budget].per_call_max_tokens. Gated before the provider call.	error.details = { estimated_tokens, per_call_max_tokens, model }
BUDGET_WORKER_LIMIT	The per-worker ledger for scope.worker_name (the literal @worker name) crossed worker_limits_usd[<name>]. Same call that crosses returns this code.	error.details = { worker, worker_cost_usd }
BUDGET_DAG_LIMIT	The per-DAG ledger crossed dag_limits_usd[<name>]. Today the runtime always mints dag_id as a random dag-<uuid> and exposes no submission knob to override it; the <name> key on dag_limits_usd is therefore un-triggerable from a regular run_workflow(...) call. Wired end-to-end in the runtime; awaiting a caller-supplied dag_id on submission.	error.details = { dag_id }
BUDGET_SOFT_LIMIT	Not raised as a FrameworkError. Soft-limit breach emits a Budget telemetry event of type "warning" and lets the call succeed. The event lives on the runtime's internal TelemetryBus only; it is not delivered to run_workflow_stream consumers, not broadcast to extensions over events_subscribe, and not exposed as an RPC. The user-visible signal is the Prometheus counter skarta_budget_trips_total{kind="soft"} on the /metrics endpoint, or a log scrape.	n/a (no error is raised).
SESSION_VERSION_CONFLICT	save_session(..., expected_updated_at=...) mismatched.	error.details carries expected_updated_at and actual_updated_at.
REGISTRY_CONFLICT	An extension tried to register a tool, worker, orchestrator, provider, or model name that is already claimed by an earlier-registered extension. The runtime is first-write-wins: the second extension's register request gets this error as its response and the connection closes; the second extension never receives register_ack. The earlier extension keeps the slot and is unaffected.	error.message reads "<Noun> '<name>' is already registered by another extension". The same code covers tool / worker / orchestrator / provider / model name collisions, so branch on the noun in the message only when you need to (the structured error.details is empty for this code today).
ORCHESTRATOR_AMBIGUOUS	A workflow.run / workflow.run_stream submission did not name an orchestrator (orchestrator= left unset) AND more than one orchestrator (each with a distinct name) is currently registered. The runtime explicitly refuses to silently pick one. Unrelated to duplicate-name registrations, which are rejected up-front with REGISTRY_CONFLICT.	error.details = { "choices": ["<name1>", "<name2>", ...] } (sorted). Pass one of the choices on the next submission as orchestrator="<name>". Naming an orchestrator that is not in the registry returns REGISTRY_NOT_FOUND with details = { "orchestrator": "<name>" }.
ORCHESTRATOR_PLAN_ERROR	The runtime-resident orchestrator's planner LLM call failed, returned no submit_dag tool call, or returned a plan that fails closure validation.	See concepts.md section Orchestrators.
WORKER_NOT_FOUND	A worker name supplied to run_workflow_target(...), ctx.invoke_worker(...), or any RPC that resolves a worker by name is not in the registry.	error.message reads "Worker '<name>' is not registered".
TOOL_NOT_FOUND	ctx.tool("<name>", ...) referenced a tool that no extension has registered, or the runtime received a tool.call envelope without a tool_name.	error.message reads "Tool '<name>' not found in registry" (or "Missing 'tool_name' parameter" for the malformed-envelope path).
TOOL_EXECUTION_ERROR	A tool handler failed in a non-recoverable way (script exited non-zero, spawn failed, handler raised a non-FrameworkError exception). Distinct from TOOL_VALIDATION_FAILED (input/output schema) and TOOL_TIMEOUT (transport-level cap).	error.message carries the script path and stderr (or the underlying spawn error). The auto tool-loop in Agent (see agents.md) feeds this back to the LLM rather than aborting; raise it explicitly when the failure is part of the tool's contract.
PROVIDER_AUTH_FAILED	The LLM provider returned 401 Unauthorized (bad / missing / revoked API key). Mapped from upstream HTTP status by the OpenAI- and Anthropic-compatible providers.	error.details carries the upstream's full body and HTTP status; error.message surfaces the upstream's structured error string when one is present. Sibling codes for the other upstream-failure paths: PROVIDER_RATE_LIMITED (429) and PROVIDER_ERROR (everything else).
SKILL_NOT_FOUND	ctx.load_skill("<name>") referenced a skill that is not in the registry, or the runtime received a skill.load envelope without a name.	error.message reads "Missing 'name' parameter" for the malformed-envelope path; the registry-miss path surfaces the skill name in the message.
SKILL_INVALID_FORMAT	A skill.install envelope was missing its path, the on-disk skill failed parsing, or the parsed manifest failed JSON Schema validation. Same code covers all three install-time validation failures.	error.message carries the parser error or the validator's structured errors list.
SKILL_LOAD_ERROR	Activating a registered skill failed at runtime (filesystem read error, manifest fields disappeared between install and load). Distinct from SKILL_INVALID_FORMAT, which fires up front at install.	error.message reads "Failed to activate skill '<name>': <underlying error>".
CONVERSATION_PARTICIPANT_NOT_IN_ROOM	A worker called a conversation verb (speak / address / reply / pushback / agree / pass_turn / give_floor) but is not a participant of the named room (or is registered only as an observer). The runtime validates the calling worker's role before dispatching.	error.details = { conversation, worker, reason }. Recovery is to rejoin the room as a participant rather than an observer, or to address the room from a worker that was added as a member at room construction.

FrameworkErrorCode is an enum exported from skarta. The wire form is SCREAMING_SNAKE_CASE (e.g. "WORKER_TIMEOUT"); the Python enum value equals the wire string, so e.code is FrameworkErrorCode.WORKER_TIMEOUT and e.code == "WORKER_TIMEOUT" both work. Branch on the enum, not the message; message text is human-facing and may evolve.

Environment variables

Variable	Purpose
SKARTA_SOCKET_PATH	Existing Unix socket. Wins over auto-spawn.
SKARTA_AUTOSPAWN	0 to refuse auto-spawn (clearer error than silent fallback). Default 1.
SKARTA_AUTOSPAWN_LOG	1 to forward the auto-spawned runtime's stdout/stderr to the Python logger.
SKARTA_SHUTDOWN_TIMEOUT_SECS	Seconds the atexit hook waits before SIGKILL (default 4).
SKARTA_READINESS_TIMEOUT_SECS	Seconds the auto-spawn manager waits for the socket to accept (default 5).

Sync vs async

Pick sync if your surrounding code is sync (scripts, CLIs, simple servers). Pick async if you are inside an asyncio app, FastAPI, or a streaming consumer. The same RPCs ship on both classes; switching later does not require rewriting workers.

Client refuses to be created from inside a running event loop. Use AsyncClient there.