Skip to content

Python SDK

mini-ork is usable from Python in two composable ways: primitives you import in-process (verification, dispatch, memory, routing), and an orchestrator client that drives the full classify → plan → execute → verify lifecycle. Most applications start with the primitives and add the orchestrator when they want the whole loop.

The import surface is lazy: import mini_ork stays cheap and only loads the dispatch, runtime and memory machinery when you reference it.

Install

pip install -e '.[full]'    # from a checkout; installs the CLI and runtime too
python3 -c "import mini_ork; print(sorted(dir(mini_ork)))"

A runnable tour: python3 examples/sdk/hello_world.py — exercises backends, a memory roundtrip, routing and the fail-fast dispatch contract, unconfigured.

Execution-anchored verification

Crucible runs a test against a patch in an isolated runtime and returns an ExecOutcome. The status is the signal — not a model's opinion of the patch.

from mini_ork import Crucible, RuntimeSpec

cru = Crucible(RuntimeSpec(image="python:3.11-slim", backend="auto"))
outcome = cru.run_test(
    test_src="def test_x():\n    assert add(2, 2) == 4\n",
    patch=my_patch,
)

backend="auto" picks from what is available; available_backends() reports the candidates (docker, subprocess, prime, modal, docker-cli).

Reading ExecOutcome

Status What it says What to do
passed The test ran and passed Evidence for the patch
failed The test ran and failed Evidence against the patch
test_defect The test itself is broken Repair the test; do not blame the patch
error The harness errored Investigate the harness
apply_fail The patch did not apply Fix the patch/base
no_run Nothing executed No evidence either way 
if outcome.informative:      # passed or failed — real evidence
    ship = outcome.passed
elif outcome.test_is_broken: # repair the probe, do not blame the patch
    ...

informative is true only for passed and failed. Treating error or no_run as a failure is a common and expensive mistake: it reports a harness problem as a patch problem and can send an agent chasing a bug that does not exist.

flowchart TD
    R[run_test] --> S{status}
    S -->|passed / failed| I[informative: evidence]
    S -->|test_defect| T[fix the test]
    S -->|apply_fail| A[fix patch or base]
    S -->|error / no_run| H[fix the harness]

Heterogeneous dispatch

dispatch_model sends one turn to any configured lane. An unroutable lane or a missing credential returns a structured ok=False result — it does not raise, stall, or leave partial state behind.

from mini_ork import DispatchRequest, dispatch_model

res = dispatch_model(DispatchRequest(model="codex", prompt="…"))
if res.ok:
    text, cost = res.text, res.cost_usd
else:
    handle(res.error, res.rc)

Check ok before reading text. The structured failure exists so a batch loop can branch instead of unwinding.

Verified-outcome memory

Scoped semantic memory backed by SQLite. The default HashEmbedder is stdlib only — no new dependency and no network for a basic roundtrip.

from mini_ork import memory

memory.add("ingest lane retry budget is 3", scope="proj", infer=False)
hits = memory.search("how many ingest retries?", scope="proj", top_k=3)

Scopes keep projects from bleeding into each other. The layer is meant to hold outcomes that passed verification, not every model assertion; see The learning loop.

Bandit routing

preferred_lane returns the lane that has been winning for a role, learned from verified outcomes at zero extra model calls:

from mini_ork import preferred_lane

lane = preferred_lane("code_fix", node_type="implementer")

It reads a warmed state database (mini-ork init plus a few runs). With no priors it returns the default. router and recompute_advantages expose the underlying pieces if you want to inspect or recompute the table.

Routing is not a correctness guarantee

Learned routing has been measured to reduce cost, but lower cost at equal correctness is not established. Use it to choose a starting lane, and keep the gates, not the router, as the correctness boundary.

Orchestrator

from mini_ork import MiniOrk, RunRequest

mo = MiniOrk()
result = mo.run(RunRequest(kickoff="kickoff.md", recipe="code-fix", mode="live"))
print(result.run_id, result.verdict, result.ok)

The client shells out to the mini-ork CLI behind a stable --json contract. RunResult is populated from one structured mini_ork_result={…} line, not scraped from human-readable output.

Provider policy and auto-init

RunRequest can carry a ProviderPolicy, written to .mini-ork/config/agents.yaml before the run:

from mini_ork import MiniOrk, ProviderPolicy, RunRequest

result = MiniOrk().run(RunRequest(
    kickoff="kickoff.md",
    recipe="docs",
    mode="dry-run",
    provider_policy=ProviderPolicy.codex_only(),
    auto_init=True,
))

auto_init=True bootstraps .mini-ork/ on first use, matching mini-ork init. Set it to False when your application manages initialization itself.

What RunResult carries

RunResult is a transparency contract, not just a return code:

Field Meaning
command The exact argv that ran
output Full combined output
events Line-by-line events, for streaming adapters
plan_path The plan written by the planner
retained_home The .mini-ork home the run used
run_id, task_class, verdict, returncode Result identity and outcome
init_ran, init_output Whether bootstrap ran

Recursive delegation

A run can spawn child runs. spawn is parent-centric: you name the parent, and mini-ork enforces recursion limits before approving a child.

from mini_ork import MiniOrk, SpawnRequest

child = MiniOrk().spawn(SpawnRequest(
    parent_run_id="run-root-123",
    kickoff="child-task.md",
    recipe="code-fix",
    allow_child_spawn=True,
    mode="dry-run",
))
print(child.spawn_id, child.child_workspace, child.spawn_status)

Limits, with defaults:

Variable Default Bounds
MINI_ORK_RECURSIVE_MAX_DEPTH 2 Delegation depth
MINI_ORK_RECURSIVE_MAX_CHILDREN 4 Children per parent
MINI_ORK_RECURSIVE_MAX_DESCENDANTS 16 Total descendants
MINI_ORK_RECURSIVE_MAX_PARALLEL 4 Concurrent children

Children run under .mini-ork/runs/<parent>/children/<child>/worktree/ and share the parent's state database. The parent owns merge and publish.

Defining recipes in code

RecipeBuilder targets the same recipes/<name>/ contract the CLI reads. Code and YAML are two front-ends to one format.

from mini_ork import ExtensionRegistry, NodeSpec, RecipeBuilder

registry = ExtensionRegistry()

recipe = (
    RecipeBuilder("invoice-audit", "invoice_audit", "Audit invoices")
    .keywords("invoice", "tax", "vat")
    .node(NodeSpec(name="planner", type="planner", model_lane="planner"))
    .node(NodeSpec(name="verifier", type="verifier",
                   verifier_ref="verifiers/check.sh"))
    .edge("planner", "verifier", "verifies")
    .build()
)
registry.recipe(recipe)

@registry.verifier("invoice-total")
def verify_invoice_total(artifact_path, plan_path):
    return artifact_path.exists() and plan_path.exists()

Use code when a host application owns its recipes; use YAML when operators edit them without shipping code.

Prompt optimization with GEPA

GEPA (Genetic-Pareto reflective prompt evolution) mutates a prompt by reflecting on an evaluator's natural-language feedback. mini-ork wraps the external gepa framework (pip install gepa) in recipe mode (optimize a recipe's node prompts against real runs) or external mode (optimize any seed prompt against any critic, with no recipe and no state database).

The critic is a command in any language. It reads a JSON object on stdin and prints its verdict as the last non-blank line of stdout:

{"score": 0.8, "feedback": "the answer omits the tax rate"}

Log freely above that line. The feedback is forwarded verbatim into GEPA's reflection — that diagnosis is the signal GEPA turns into targeted mutations.

python -m mini_ork.gepa.run_gepa \
    --seed-file seed_prompt.json \
    --eval-cmd "node dist/eval-c4.js" \
    --trainset fixtures/*.json \
    --reflection-lm anthropic/opus \
    --budget 30 --out out/gepa-c4

The seam is three composable pieces:

from mini_ork.gepa import (
    SubprocessRunBackend,   # cross-process / cross-language critic
    CallbackRunBackend,     # in-process: eval_fn(candidate, task) -> (score, feedback)
    GEPARunBackendAdapter,  # composes a RunBackend into gepa's adapter
)

Fail-loud. A non-zero exit, unparseable stdout, a missing score, or a score outside [0, 1] raises EvaluatorError and halts the run. A broken evaluator is never silently scored 0.0 — that would poison the Pareto front with a fake gradient.

Next