muxswarm 0.3.2

Python SDK that fully drives the Mux-Swarm engine binary (stdio NDJSON + WebSocket).

muxswarm — Python SDK for the Mux-Swarm engine

A Python library that fully drives the Mux-Swarm engine binary — a .NET CLI-native agentic swarm runtime. The SDK runs the compiled engine as a subprocess (or talks to its WebSocket bridge) and speaks the engine's NDJSON protocol, giving you typed events, streaming, interactive sessions, headless one-shots, prefilled-config bootstrapping, and binary management.

Status. The full stack is implemented and verified offline against a protocol mock (tests/, 9/9) and live against a real downloaded engine binary. The SDK wires and validates the engine's entire CLI + slash-command surface (see muxswarm.commands) and drives the entire Config.json / Swarm.json schema via a fluent editor (muxswarm.manage.MuxConfig, 1:1 with the engine's C# config classes). Core SDK is pure stdlib; the WebSocket adapter needs the optional websockets extra and the guarded shell needs mcp at runtime.

---

Why this exists

Mux-Swarm runs as a console app. With --stdio it emits NDJSON (one JSON object per line) on stdout and reads plain text lines on stdin. With --serve <port> it bridges that same NDJSON contract over a WebSocket. This SDK wraps that protocol so you can drive the whole runtime from Python without touching the CLI by hand.

---

Install

pip install -e .            # core SDK (stdlib only)
pip install -e ".[ws]"      # + WebSocket adapter (websockets)
pip install -e ".[dev]"     # + pytest for the test suite

You also need the engine binary. Either:

• set MUXSWARM_BINARY=/path/to/MuxSwarm(.exe), or
• pass binary="..." to MuxSwarm(...), or
• let the SDK fetch a release: MuxSwarm(..., auto_download=True, install_dir="./engine").

Release assets (from the GitHub Releases tab) are per-platform: mux-swarm-win-x64.zip, mux-swarm-linux-x64.tar.gz, mux-swarm-osx-x64.tar.gz, mux-swarm-osx-arm64.tar.gz. Each zip/tarball contains the exe plus the sibling dirs it needs (Configs/, Prompts/, Context/, Sessions/, Skills/, Runtime/).

---

Quick start (auth + prefilled config → running)

import asyncio
from muxswarm import MuxSwarm

async def main():
    mux = MuxSwarm.quickstart(
        endpoint="https://openrouter.ai/api/v1",
        api_key="sk-...",                 # raw key (in-process only) ...
        # api_key_env_var="OPENROUTER_API_KEY",   # ... or reference an env var
        work_dir="./.mux",                # writes Configs/Config.json + Swarm.json
        sandbox_path="./.mux/sandbox",
        binary="/path/to/MuxSwarm.exe",   # or auto_download=True
    )
    result = await mux.run_goal("Summarize this repo", mode="agent", agent="MuxAgent")
    print(result.final_summary or result.streamed_text)

asyncio.run(main())

quickstart writes a complete Config.json + Swarm.json with setupCompleted=true, an enabled provider, and model IDs auto-resolved from the endpoint (OpenRouter / Anthropic / OpenAI / Ollama), then launches the engine with --cfg/--swarmcfg. No interactive engine setup is triggered — that's the "natural code-side get-up-and-running" path.

Even shorter:

from muxswarm import run
res = await run("What is 2+2?", endpoint="https://openrouter.ai/api/v1",
                api_key="sk-...", binary="/path/MuxSwarm.exe", mode="agent", agent="MuxAgent")

---

Models

quickstart resolves model IDs from the endpoint (OpenRouter / Anthropic / OpenAI / Ollama). For anything else — including a local OpenAI-compatible proxy (e.g. CLIProxyAPI on 127.0.0.1) — the endpoint default is the ollama placeholder llama3, which most providers reject (HTTP 502: unknown provider for model llama3). Set the model explicitly:

# simplest: one id for the entire swarm (orchestrator + all agents + light + compaction)
mux = MuxSwarm.quickstart(endpoint, api_key_env_var="CLI_PROXY_TOKEN", model="claude-sonnet-4-6")

# per role (orchestrator / agent / light / compaction)
mux = MuxSwarm.quickstart(endpoint, ..., models={
    "orchestrator": "claude-sonnet-4-6",
    "agent":        "claude-sonnet-4-6",
    "light":        "claude-haiku-4-5-20251001",
    "compaction":   "claude-haiku-4-5-20251001",
})

# per named agent (mix with role keys; a per-agent entry wins for that agent)
mux = MuxSwarm.quickstart(endpoint, ..., models={
    "agent":     "claude-sonnet-4-6",      # base for every agent
    "CodeAgent": "claude-opus-4-6",        # ...except CodeAgent
    "WebAgent":  "gpt-5.2-2025-12-11",     # ...and WebAgent
})

Precedence (low → high): endpoint default → model= shorthand → models= role key / per-agent-name. If you ship a placeholder id (llama3/model-id-here) with no override, the SDK emits a UserWarning pointing you here. model=/models= are accepted by quickstart, run(), build_quickstart_config, and write_quickstart.

---

Driving the config (model swap · agent creation · whole-schema edits)

MuxConfig is a fluent, round-trip-safe editor over the (Config.json, Swarm.json) pair. Its dataclasses map the engine's C# config classes 1:1 (every [JsonPropertyName]), so it can drive the entire config surface deterministically — which is the reliable alternative to the racy interactive REPL config commands (/setmodel, /swap, /provider, /maxp, …; see below).

from muxswarm import MuxSwarm, MuxConfig

mux = MuxSwarm.quickstart(endpoint, api_key_env_var="CLI_PROXY_TOKEN",
                          model="claude-sonnet-4-6", work_dir="./.mux")

# edit the live config this client points at, then it applies on next launch
(mux.config()
    .set_all_models("claude-sonnet-4-6")                  # every slot at once
    .set_model("compaction", "claude-haiku-4-5-20251001") # one slot
    .set_agent_model("WebAgent", "gpt-5.2-2025-12-11")    # one named agent
    .add_agent("DataAgent",                               # create a new agent
               description="Crunches CSV/parquet and emits summaries.",
               model="claude-sonnet-4-6",
               mcp_servers=["Filesystem", "ReplShellMcp"])
    .set_single_agent("CodeAgent")                        # bind the /agent loop
    .save())

Model swapping

cfg = MuxConfig.from_dir("./.mux")          # or .from_paths(cfg, swarm)
cfg.get_models()                            # {orchestrator, compaction, vision,
                                            #  singleAgent, WebAgent, CodeAgent, ...}
cfg.set_model("orchestrator", "id")         # role or agent-name slot
cfg.set_models({"orchestrator": "a", "CodeAgent": "b"})
cfg.set_all_models("id", include_light=True)
cfg.set_model_opts("orchestrator", {"temperature": 0.2, "reasoning": {"effort": "high"}})
cfg.save()

Agent creation / management

cfg.add_agent("DataAgent", description="...", model="...",
              prompt_path="Prompts/Agents/dataagent.md",   # default: <name>.md
              can_delegate=False, mcp_servers=["Filesystem"],
              tool_patterns=["*"], skill_patterns=["*"])
cfg.update_agent("DataAgent", description="v2", model="...")
cfg.remove_agent("DataAgent")
cfg.list_agents(); cfg.get_agent("CodeAgent")
cfg.set_orchestrator(model="...", prompt_path="...")

Bind the single-agent loop (the reliable /swap)

The interactive /swap picker is a racy stdin handshake over piped I/O. Instead, set Swarm.json.singleAgent — the engine reads it directly, so the interactive /agent loop runs exactly that agent:

cfg.set_single_agent("CodeAgent").save()    # copies the CodeAgent def into singleAgent
# then: async with mux.session(mode="agent") as s: await s.ask("...")   # runs CodeAgent

Provider · MCP servers · paths · limits · security

cfg.set_provider("https://api.openai.com/v1", api_key_env_var="OPENAI_API_KEY")
cfg.add_server("ChromaDB", command="uvx", args=["chroma-mcp"])
cfg.enable_server("Playwright"); cfg.disable_server("ChromaDB"); cfg.list_servers()
cfg.set_allowed_paths(["/work"]); cfg.add_allowed_path("/extra"); cfg.set_sandbox("/work")
cfg.set_user_info(name="Jane", role="SRE", timezone="America/Chicago")
cfg.set_execution_limits(maxOrchestratorIterations=20, activityTimeoutSeconds=600)
cfg.set_docker_exec(True); cfg.set_telemetry(enabled=True, endpoint="http://otel:4317")
cfg.set_security("trusted")                 # rewrites only managed shell/browser + CodeAgent
cfg.save()

Every mutator returns self (chainable). The underlying typed models (AppConfig, SwarmConfig, AgentConfig, ModelOpts, …) are exported too if you prefer to build/serialize the pair by hand.

---

Headless one-shot (-p style)

Like claude -p, you can fire a single goal and get a structured result back without ever opening an interactive session. run_goal is the full headless driver: it maps every launch-relevant engine flag to a typed keyword and validates it against the engine surface before spawning.

res = await mux.run_goal(
    "Audit the logs in ./var and summarize anomalies",
    mode="pswarm",            # --parallel
    max_parallelism=8,        # --max-parallelism 8
    goal_id="nightly-audit",  # --goal-id nightly-audit
    continuous=True,          # --continuous
    min_delay=600,            # --min-delay 600
    watchdog=True,            # --watchdog
    mcp_strict=False,         # --mcp-strict false
    plan=False,
    timeout=900,
)
print(res.ok, res.exit_code, res.final_summary)

Every one of those kwargs is validated; a bad value raises muxswarm.commands.CliValidationError before the engine launches. For a fully dynamic call use the escape hatch:

res = await mux.run_cli(goal="do it", parallel=True, delimiter="---", watchdog=True)

run_cli(**flags) accepts any snake_case param from CLI_FLAGS and validates the whole argv via build_cli_args. Lifecycle flags get dedicated methods instead (below).

Lifecycle launches

proc = await mux.serve(6723)          # --serve 6723  (long-lived; NDJSON over ws://.../ws)
proc = await mux.daemon()             # --daemon      (file watch / cron / status triggers)
rc   = await mux.register_service()   # --register    (OS service, then exits)
rc   = await mux.remove_service()     # --remove
await proc.exit()                     # stop a long-lived launch

---

Package layout

Module	Purpose
muxswarm.events	NDJSON event model: MuxEvent, EventType, RequestKind, parse_line, request_kind.
muxswarm.driver	Low-level async stdio driver: MuxProcess (spawn --stdio, stream events, auto-answer blocking requests, cancel, exit).
muxswarm.client	High-level ergonomic client: MuxSwarm, MuxSession, RunResult, module-level run().
muxswarm.config	Typed Config.json/Swarm.json models + setup helpers (build_quickstart_config, write_quickstart, resolve_model_ids). Dataclasses map the engine C# config classes 1:1.
muxswarm.manage	Fluent whole-config editor MuxConfig: model swap, agent create/update/remove, single-agent bind, provider/MCP/paths/limits/security.
muxswarm.binary	Locate or download the engine release binary + dir structure (find_binary, ensure_binary, download_release, BinaryManager).
muxswarm.commands	Authoritative, validated map of the entire engine surface: CLI_FLAGS, SLASH_COMMANDS, build_cli_args, validate_cli_args, is_known_slash, normalize_slash.
muxswarm.guard	Claude-Code-style guarded shell + security policy (ShellPolicy, DEFAULT_DENY_COMMANDS).
muxswarm.ws	Optional WebSocket adapter + REST helpers for --serve mode (MuxWebSocketClient, MuxRestClient). Needs muxswarm[ws].

---

The protocol, in brief

Outbound events ({"type": "...", ...}): splash, info/success/warning/error/debug/body/markup, prompt, step, rule, panel, table, stream / stream_end, thinking_start/update/end, agent_turn_start/end, delegation, tool_call, tool_result, task_start/done, task_complete, user_input.

Blocking request events — the engine waits for a stdin reply:

event	reply
input_request	a line of text (empty → engine default)
confirm_request	y / n
select_request	1-based index
multiselect_request	comma-separated 1-based indices

The driver auto-answers these via a pluggable request_handler (default: accept/yes/first) so the engine never deadlocks. Supply your own handler to take control.

Inbound (you → engine): slash command to enter a mode (/agent, /swarm, /pswarm, /stateless), then plain text per turn. __CANCEL__ cancels the active turn; /exit or EOF shuts down.

Modes (run_goal(mode=...)): swarm (default, full orchestration), pswarm (--parallel), agent (--agent <name>), stateless (one-shot interactive).

---

Full CLI flag reference

Mirrored 1:1 from the engine's argument parser (muxswarm.commands.CLI_FLAGS, engine v0.10.3). All are validated by the SDK. * = changes process lifecycle (handled by a dedicated method, not build_cli_args).

Flag (aliases)	Value	Description
--goal	text | file	Explicit goal (inline or a path to a goal file).
--goal-id	str	Attach a goal/session id (groups continuous runs).
--continuous	—	Continuous autonomous mode (loop the goal).
--parallel	—	Parallel swarm (concurrent batch dispatch).
--max-parallelism	uint (4)	Max concurrent agent tasks in parallel mode.
--agent	str	Single-agent mode with the named swarm agent.
--plan	—	Plan mode (confirm before executing).
--workflow (--wf)	file	Run a deterministic, replayable workflow file.
--provider	str	Set the active LLM provider on launch.
--model	str	Override the model id for this run.
--min-delay	uint (300)	Min seconds between continuous loops.
--persist-interval	uint (60)	Persist the session every N seconds.
--session-retention	uint (10)	Keep the last N sessions.
--stdio	—	NDJSON output / line stdin (driver injects this).
--delimiter	str	Multi-line input delimiter (e.g. ---).
--watchdog	—	External watchdog (auto-restart on crash).
--mcp-strict	true|false (true)	Require all MCP servers to connect.
--docker-exec	true|false	Route execution through docker skills.
--prod	—	Production mode.
--report *	str (optional)	Generate audit report(s) and exit.
--cfg	path	Override Config.json path (wired by quickstart).
--swarmcfg	path	Override Swarm.json path.
--clear	—	Clear the screen and continue (interactive).
--serve *	port (6723)	Embedded web UI / WebSocket bridge — mux.serve().
--daemon *	—	Daemon mode (file watch / cron / status) — mux.daemon().
--register *	—	Register as an OS service, then exit — mux.register_service().
--remove *	—	Unregister the OS service, then exit — mux.remove_service().
--help (-h) *	—	Print help and exit (not driven by the SDK).

Full slash-command reference

Mirrored 1:1 from the engine's interactive switch + /help text (muxswarm.commands.SLASH_COMMANDS). Send any of these in a live session via sess.send("/cmd") or proc.send_command("/cmd") — aliases are auto-normalized and unknown commands warn.

Command (aliases)	Description
/swarm	Interactive multi-agent swarm loop.
/pswarm	Parallel swarm (concurrent batch dispatch).
/agent	Interactive single-agent loop (/agent <name>).
/onboard	Create/update operator profile (BRAIN.md + MEMORY.md).
/stateless	Stateless single-agent loop (one-off tasks).
/subagents (/sub)	Enable sub-agent delegation in the single-agent loop.
/parasubagents (/psub)	Enable parallel sub-agent delegation.
/addcontext	Configure context injected into each agent.
/plan	Toggle plan mode.
/continuous (/cont)	Toggle continued autonomous execution.
/workflow	Run a workflow file (/workflow <file>).
/resume	Resume a previous single-agent session.
/compact	Compact current session context (single-agent loops).
/maxp ‡	Set max parallel agents (/maxp <n>).
/model	View current swarm models.
/setmodel ‡	Change a model for an agent/orchestrator/compaction.
/swap ‡	Swap the active single-agent.
/provider ‡	View or switch the active LLM provider.
/limits †	Display execution limits.
/tools †	List MCP tools across enabled servers.
/skills †	List local skills.
/memory †	View the knowledge graph.
/sessions †	List saved sessions.
/dockerexec	Toggle Docker execution mode.
/delimiter	Toggle the multi-line input delimiter.
/dbg / /nodbg	Enable / disable tool-call output (stdio only).
/disabletools	Toggle disabling of tool calls.
/setup	Run initial setup / reconfigure.
/reloadskills	Refresh the skills directory.
/refresh	Full system refresh (config + MCP + skills).
/report	Generate session audit reports (/report [id]).
/clear	Clear the screen.
/status †	System status: provider, models, tools, skills, sessions.
/qc / /qm	Quick-config / quick-model helpers.
/help	Show the command reference.
/exit	Exit Mux-Swarm (driver sends this on graceful shutdown).

† Read-only info commands are exposed as typed MuxSwarm methods that drive a transient outer-REPL process and return the resulting events: list_tools(), list_skills(), list_sessions(), status(), show_memory(), show_models(), show_limits(). (list_tools/status take warmup= — default 6s — because MCP servers connect in the background after boot.) These run at the outer REPL only; inside an /agent//swarm loop a line is consumed by the agent as a turn.

‡ Interactive config commands emit their own input_request sub-prompts and are racy to drive over piped stdio — they are not orchestrated. Use MuxConfig (above) for deterministic equivalents: /setmodel→set_model, /swap→set_single_agent, /provider→set_provider, /maxp→ set_execution_limits/config, /dockerexec→set_docker_exec, /delimiter→--delimiter, /onboard//setup//addcontext→config-side.

Programmatic access:

from muxswarm import CLI_FLAGS, SLASH_COMMANDS, build_cli_args, validate_cli_args, normalize_slash
normalize_slash("/cont")            # -> "/continuous"
build_cli_args(goal="x", parallel=True, max_parallelism=8)   # validated argv tail

---

Security profiles

quickstart / run generate configs under a security profile (default standard). The shell tool is a Claude-Code-style guarded MCP server: a command denylist plus path containment to the sandbox allow-list.

profile	shell	python exec	browser	code agent
locked	none	off	no	no
standard (default)	guarded	off	yes	yes
trusted	guarded	subprocess (cwd-pinned)	yes	yes
yolo	raw (mcp-async-repl)	in-process	yes	yes

mux = MuxSwarm.quickstart(endpoint, api_key="...", security="trusted")
# granular overrides:
mux = MuxSwarm.quickstart(endpoint, api_key="...", security="standard",
                          allow_browser=False, deny_commands=["npm"], allow_commands=["git","ls"])

Caveat: the guard is strict on the shell tool surface (denylist + path containment), not a kernel sandbox. trusted Python exec is a cwd-pinned subprocess, still not a hard jail. For true isolation, run the engine in a container. The guarded shell launches via uvx --with mcp, so the engine host needs uv/uvx; install the SDK with the guard runtime dep available if you ship shell_guard.py yourself.

---

API highlights

# headless one-shot -> aggregated result
res = await mux.run_goal(goal, mode="swarm", plan=False, continuous=False,
                         on_event=lambda ev: ...)   # RunResult: .ok .exit_code
                                                    #  .final_summary .streamed_text .errors

# multi-turn interactive session (runs the configured singleAgent)
async with mux.session(mode="agent") as sess:
    events = await sess.ask("do a thing")           # list[MuxEvent] for that turn
    await sess.send("/plan")                         # raw line if you need it
    await sess.cancel()                              # __CANCEL__

# drive the whole config; read-only info commands at the outer REPL
mux.config().set_single_agent("CodeAgent").set_all_models("claude-sonnet-4-6").save()
tools = await mux.list_tools()                       # -> panel event w/ tool list

# low-level, full control
from muxswarm import MuxProcess
async with MuxProcess(binary_path, args=[...], cfg=..., swarmcfg=...) as p:
    await p.send_command("/agent")
    await p.send("goal")
    async for ev in p.events(): ...

---

Examples

Runnable scripts in examples/:

1. 01_quickstart.py — auth + prefilled config → running.
2. 02_streaming_events.py — classify every event type live.
3. 03_interactive_session.py — multi-turn session + custom request handler.
4. 04_prefilled_config_and_auth.py — build/tweak the config pair explicitly.
5. 05_headless_batch.py — parallel swarm + the run() one-liner.
6. 06_websocket_serve.py — drive a --serve engine over WebSocket + REST file ops.

---

Tests

Fully offline (no binary, no key) via a protocol mock:

python tests/test_sdk.py     # plain runner
pytest tests/test_sdk.py     # or under pytest

Covers event parsing, quickstart config generation, the stdio driver end-to-end (including blocking-request auto-reply), and multi-turn interactive sessions. The mock (tests/mock_engine.py) re-implements the engine's NDJSON handshake.

Note: the mock is a Python script, so tests build the driver with stdio_flag=False (so --stdio isn't injected before the script path). The real engine uses the default stdio_flag=True.

---

Notes & known follow-ups

• Full surface, validated: muxswarm.commands mirrors the engine's ParseArgs + slash switch + /help text 1:1. run_goal/run_cli validate every flag before launch; send_command normalizes slash aliases and warns on unknown commands.
• Whole config, deterministically: muxswarm.manage.MuxConfig (+ the typed muxswarm.config models) map the engine's C# config classes 1:1 and drive the entire Config.json/Swarm.json schema. Prefer it over the interactive /setmodel//swap//provider//maxp commands, which emit sub-prompts that are racy over piped stdio.
• Interactive single-agent selection: session(mode="agent") runs whatever Swarm.json.singleAgent names. Bind it with mux.config().set_single_agent("CodeAgent").save() — session(agent=...) does NOT drive the racy /swap.
• Setup surface: setupCompleted=true in Config.json is the gate that makes the engine skip its interactive 7-step setup. The config layer always emits it, so a prefilled cfg pair runs immediately.
• Auth: a raw key is never written to config — it's passed in the launch env as MUXSWARM_SESSION_API_KEY (matching the engine's own behavior); or reference an existing env var by name via api_key_env_var.
• Binary resolution gotcha: find_binary/ensure_binary resolve an installed engine on PATH/MUXSWARM_BINARY first. Pass binary= explicitly to force a specific exe.
• See SCAN_REPORT.md for the full engine interface map this SDK was built against.