swarmgraph-sdk 0.2.1

SwarmGraph SDK: deterministic multi-agent orchestration primitives

SwarmGraph SDK

Deterministic multi-agent orchestration primitives, packaged as a standalone Python distribution (swarmgraph-sdk, import name swarmgraph).

Status: stable (0.2.1). The wheel builds and passes twine check. Paid provider execution is gated and disabled by default. There is no production provider-backed runtime in this package; provider adapters are pluggable seams with deterministic defaults.

What SwarmGraph is

• A small, typed, deterministic, offline-first runtime for multi-agent orchestration.
• Pluggable providers, decomposition strategies, consensus protocols, guardrails, retry policies, and notification hooks — all with clear failure behavior.
• Up-front graph validation (TaskGraph / TaskScheduler).
• Durable JSON checkpoints + faithful resume (event continuity preserved).
• A typed event stream with bounded retention + optional sink.
• An HITL approval store (in-memory + atomic JSON file).
• A single-process worker service with submit/signal/query.
• A deterministic record/replay provider for paid-provider fixtures.
• An OpenTelemetry exporter behind the [otel] optional extra.

What SwarmGraph is not

• Not a distributed-execution engine (single-process today).
• Not a Temporal/LangGraph deployment server (no managed runs, no signals, no horizontal scaling). It is a library.
• Not a sandbox. Untrusted code/prompts should be isolated by the host (ARC Studio provides isolation; the SDK does not).
• Not a production-graded Raft/BFT implementation. The raft and bft protocols are deterministic approximations for offline consensus reasoning — see the consensus docstrings.

Install / build

# Build wheel + sdist from the monorepo
uv build python/packages/swarmgraph-sdk

# Import + run in an isolated environment (no monorepo on the path)
WHL=python/packages/swarmgraph-sdk/dist/*.whl
uv run --isolated --with "$WHL" swarmgraph run "Explain consensus" --json

Quickstart

from swarmgraph import SwarmGraphRunner, SwarmGraphConfig

runner = SwarmGraphRunner(config=SwarmGraphConfig(max_rounds=1))
result = runner.run("Explain consensus")
print(result["status"])  # "completed"

Typed event stream

async for event in runner.stream("Explain consensus"):
    print(event.kind, event.id)

Durable checkpoints + resume

from swarmgraph import JsonFileCheckpointStore, SwarmGraphRunner

store = JsonFileCheckpointStore("./checkpoints")
runner = SwarmGraphRunner(checkpoint_store=store)
runner.run("Explain consensus")

# Resume continues from the saved round/tasks, not from round 0.
checkpoint_id = store.list_ids()[-1]
runner.resume(checkpoint_id)

CLI

swarmgraph --version
swarmgraph version

# Run a prompt
swarmgraph run "Explain consensus" --json
swarmgraph run "Explain consensus" --stream            # JSONL event stream

# Checkpoint + resume
swarmgraph run "Explain consensus" --checkpoint-dir ./ckpts --json
swarmgraph list-checkpoints --checkpoint-dir ./ckpts
swarmgraph inspect-checkpoint <ckpt-id> --checkpoint-dir ./ckpts
swarmgraph render-graph <ckpt-id> --checkpoint-dir ./ckpts
swarmgraph run --resume <ckpt-id> --checkpoint-dir ./ckpts --json

# Config knobs (all offline-safe)
swarmgraph run "task" --consensus quorum --max-rounds 3 --max-parallel-workers 4
swarmgraph run "task" --budget-limit-usd 0.10        # fail-fast on cost gate
swarmgraph run "task" --execution-mode gated_local   # deny paid calls by default

Exit codes:

Code	Meaning
0	Run finished with status == "completed".
1	Bad arguments, missing/corrupt checkpoint, configuration error.
2	Run finished with a non-completed status (failed, cancelled, …).

Provider-backed execution

provider_backed runs each worker through an injected Provider. With the offline EchoProvider it stays deterministic and network-free, which is the default for tests:

from swarmgraph import EchoProvider, SwarmGraphConfig, SwarmGraphRunner
from swarmgraph.config import ExecutionMode

cfg = SwarmGraphConfig(execution_mode=ExecutionMode.provider_backed)
runner = SwarmGraphRunner(config=cfg, provider=EchoProvider())
runner.run("Explain consensus")

gated_local still denies paid provider calls unless allow_paid_calls=True. provider_backed only requires a provider to be injected. To gate a paid provider in provider_backed, wrap it in GatedProvider:

from swarmgraph import GatedProvider, HTTPChatProvider

paid = HTTPChatProvider(base_url="https://api.example.com/v1", model="gpt-test", transport=transport)
runner = SwarmGraphRunner(config=cfg, provider=GatedProvider(paid))            # denied by default
runner = SwarmGraphRunner(config=cfg, provider=GatedProvider(paid, allow_paid_calls=True))  # opt-in

A denied call surfaces as a failed task (raising PaidCallDeniedError inside the worker), never a silent paid request.

Retry policy

from swarmgraph import BackoffPolicy, RetryPolicy, SwarmGraphConfig

cfg = SwarmGraphConfig(
    max_rounds=5,
    retry=RetryPolicy(
        max_attempts=3,
        backoff=BackoffPolicy.exponential,
        initial_delay_seconds=0.05,
        max_delay_seconds=1.0,
        retryable_error_substrings=("timeout", "5xx"),
        retry_guardrail_rejections=False,  # opt-in
    ),
)

Default (max_attempts=1) preserves the historical "one try, fail fast" behavior. Tests with initial_delay_seconds=0.0 never sleep.

Graph validation

from swarmgraph import GraphValidationError, SwarmTask, TaskGraph

graph = TaskGraph.from_iterable([
    SwarmTask(id="root", prompt="root"),
    SwarmTask(id="child", prompt="child", parent_task_id="root"),
])
graph.validate()                  # raises GraphValidationError if invalid
order = graph.topological_order() # deterministic

Detected problems: duplicate_id, self_dependency, missing_dependency, cycle. Each problem carries a structured task_ids field so multi-error reports surface every issue at once.

Bounded event buffer + sink

sink_events = []
cfg = SwarmGraphConfig(event_buffer_size=128)
runner = SwarmGraphRunner(config=cfg, event_sink=sink_events.append)

The in-memory runner.events is trimmed to 128 entries; every emitted event is delivered to the sink regardless. runner.dropped_events counts trim losses; runner.event_sink_errors records sink exceptions.

Approval store (HITL)

from swarmgraph import (
    ApprovalRecord, ApprovalStatus, InMemoryApprovalStore, JsonFileApprovalStore
)
from swarmgraph.models import ApprovalDecision

store = JsonFileApprovalStore("./approvals")
store.create(ApprovalRecord(token_id="tok-1", task_id="t", swarm_id="s"))
# Operator decides asynchronously:
store.decide("tok-1", ApprovalDecision(approved=True, reason="ok"))

OpenTelemetry export (optional extra)

pip install swarmgraph-sdk[otel]

from swarmgraph.otel import OTelExporter
from swarmgraph import SwarmGraphRunner, SwarmGraphConfig

exporter = OTelExporter(service_name="my-app")
runner = SwarmGraphRunner(config=SwarmGraphConfig(), on_event=exporter.on_event)

You configure the OpenTelemetry tracer provider yourself; the exporter only translates events into spans.

Record / replay provider fixtures (alpha)

from swarmgraph.replay import RecordingProvider, ReplayProvider

# Record once against the real provider:
rec = RecordingProvider(real_provider, fixture_path="fixtures/api.json")
# ... run ...
rec.flush()

# Replay forever in CI:
replay = ReplayProvider(fixture_path="fixtures/api.json")

A cache miss raises ReplayMissError — there is no silent fallthrough to a real network call.

Terminal status semantics

The SwarmStatus enum on the wire is stable: pending | running | completed | failed | cancelled. The runner reports each honestly:

• completed — and only completed — when every task reached a terminal state via successful execution.
• failed for any non-clean termination, including:
  • error_code = "budget_exhausted" — budget gate tripped.
  • error_code = "max_rounds_exceeded" — max_rounds ran out with ready pending work.
  • error_code = "dependency_deadlock" — pending work has no resolvable dependencies (failed or missing).
• cancelled when the injected cancellation token is observed.

The structured error_code lives in state.metadata["error_code"] and in the terminal error event in the stream:

{"kind": "error",
 "data": {"error_code": "max_rounds_exceeded", "message": "...", "pending_tasks": [...]}}

Prior to the v0.1.0a1 review pass, the runner always returned status == "completed" regardless of pending work; that bug is fixed and covered by test_max_rounds_exhausted_with_pending_tasks_does_not_claim_completed.

Typed errors

from swarmgraph import (
    SwarmGraphError,            # base class for all SDK errors
    ConfigurationError,         # invalid runtime config
    CheckpointError,            # base for checkpoint store errors
    CheckpointNotFoundError,    # also a FileNotFoundError (back-compat)
    CheckpointCorruptError,     # also a ValueError (back-compat)
    PaidCallDeniedError,        # gated provider denied a paid call
)

except SwarmGraphError is the safe catch-all for SDK-originated failures without accidentally swallowing user-code exceptions raised from callbacks.

Provider adapters

The SDK ships provider adapters that implement swarmgraph.providers.Provider with no ARC coupling:

• EchoProvider — fully deterministic, offline. Echoes the last user message.
• HTTPChatProvider — OpenAI-style chat-completions shape. It performs no network I/O on its own; the caller injects an async transport. Without a transport, complete() raises, so tests can never make a surprise live call.

from swarmgraph import EchoProvider, HTTPChatProvider

provider = EchoProvider()

async def transport(url, headers, json_body):
    ...  # back with httpx / aiohttp / a recorded fixture

http_provider = HTTPChatProvider(
    base_url="https://api.example.com/v1",
    model="gpt-test",
    transport=transport,
)

Paid provider execution still requires allow_paid_calls=True and is denied by default.

Consensus

Multiple deterministic consensus protocols are available (majority, quorum, raft, bft, confidence_weighted, critic_verifier, gossip, selective_debate, hitl_signoff). For fan-out worker groups, per-worker vote confidence is derived from observable worker-result signals (output substance, artifacts, errors) rather than a fixed value, so confidence-weighted protocols can differentiate strong and weak outputs.

Event history and checkpoint continuity

Events emitted during a run are persisted in SwarmCheckpoint.events (as serializable dicts) and mirrored on SwarmState.events. On resume, the runner seeds its event list from the checkpoint before emitting the resume audit event, so runner.get_events() provides a continuous audit trail across checkpoint/resume boundaries:

from swarmgraph import JsonFileCheckpointStore, SwarmGraphConfig, SwarmGraphRunner

store = JsonFileCheckpointStore("./checkpoints")
cfg = SwarmGraphConfig(max_rounds=2)
runner = SwarmGraphRunner(config=cfg, checkpoint_store=store)
runner.run("Explain consensus")

checkpoint_id = store.list_ids()[-1]
checkpoint = store.load(checkpoint_id)
# checkpoint.events contains all events emitted during the original run.

runner2 = SwarmGraphRunner(config=cfg, checkpoint_store=store)
runner2.resume(checkpoint_id)
# runner2.get_events() starts with the checkpoint events, then adds the resume
# audit event and any new events from the resumed run.

Source ownership lives here. ARC (agent_runtime_cockpit.swarmgraph) is a thin compatibility bridge that re-exports this package so existing agent_runtime_cockpit.swarmgraph.* imports keep working unchanged. The repo is a uv workspace, so uv run --package swarmgraph-sdk ... targets this member directly while sharing one lockfile with ARC.

Further reading

• CHANGELOG.md — chronological feature / fix log, one section per review iteration.
• SECURITY.md — threat model, safety guarantees, embedder hardening checklist.
• docs/event-schema.md — wire-format reference for every event kind.
• docs/swarmgraph-alpha-limits.md — honest list of what is not in this alpha.
• docs/swarmgraph-runtime-review.md — competitive matrix vs. LangGraph / AutoGen / CrewAI / OpenAI Agents / Temporal (no fabricated benchmarks).
• docs/swarmgraph-distributed-runner.md — why the distributed runner is deferred and what shipping it requires.
• docs/swarmgraph-mcp-design.md — planned swarmgraph-mcp package (design only, not yet shipped).

Release

See .github/workflows/swarmgraph-sdk-release.yml. The release workflow builds, verifies isolated import/CLI, and runs twine check as a dry run only. Publishing is intentionally disabled until external release is approved.

---

Standalone repository

This is the standalone SwarmGraph SDK repository, extracted from the ARC Studio monorepo (arc-theia-studio) to be developed independently. It contains the SDK package (swarmgraph/), the pure-SDK test suite (tests/), examples, benchmarks, API snapshots, and the design ADRs (docs/adr/). At extraction the package was at v0.1.0rc4 (iteration 10, provider-call hooks — see REVIEW_REPORT_ITERATION_10.md).

The SDK package has no runtime dependency on ARC Studio — the agent_runtime_cockpit.swarmgraph bridge depends on this SDK, not the reverse. The ARC-bridge/event-bus integration tests (≈18 files that import agent_runtime_cockpit) intentionally remain in the ARC monorepo; only the pure-SDK tests travel with this repository.

Develop with uv

uv venv && uv pip install -e .       # editable install into a local venv
uv run pytest -q                     # run the SDK test suite
uv run ruff check swarmgraph         # lint
uv run mypy swarmgraph               # type-check

Build & publish

uv build                             # → dist/swarmgraph_sdk-<version>-py3-none-any.whl + .tar.gz
uv publish                           # publish to PyPI (token via --token or env); gated — see docs/beta-exit-criteria

Publishing is not automatic. See the SDK's release-gating policy before any uv publish.