Skip to content

Programmatic API

The CLI (uv run python -m ksi.cli) is one front door; ksi.run(...) is the other. Use it to drive a generational run from Python or a notebook without argparse.

import os
from pathlib import Path

import ksi
from ksi.benchmarks import ArcSessionEvaluator
from ksi.runtime import KsiContainerExecutor
from ksi.runtime.llm import build_llm_caller
from ksi.tasks.loaders import load_tasks_for_source

tasks = load_tasks_for_source(task_source="arc", tasks_path=my_tasks_path)

knowledge_db_path = str(Path("runtime_state") / "programmatic_api_knowledge.sqlite")
config = ksi.GenerationConfig(num_generations=1, num_agents=1, knowledge_db_path=knowledge_db_path)
llm = build_llm_caller(provider="anthropic", model="claude-sonnet-4-6")
evaluator = ArcSessionEvaluator()

# Provider env is REQUIRED — the container host validates provider auth and
# raises before running if it is missing. Unlike the CLI, `ksi.run` does NOT
# load a provider profile for you, so pass the env explicitly. This minimal
# dict is what the anthropic/api path needs; supply your own key via the
# environment (e.g. `export ANTHROPIC_API_KEY=...`).
provider_env = {
    "MODEL_PROVIDER": "anthropic",
    "MODEL": "claude-sonnet-4-6",
    "MODEL_AUTH_MODE": "api",
    "ANTHROPIC_API_KEY": os.environ["ANTHROPIC_API_KEY"],
}

runtime = KsiContainerExecutor(
    command=["npx", "--yes", "--prefix", "runtime_runner", "tsx", "runtime_runner/src/main.ts"],
    working_dir=".",
    knowledge_db_path=config.knowledge_db_path,
    env=provider_env,
)

traces = ksi.run(config, tasks, runtime=runtime, evaluator=evaluator, llm=llm)

For OpenAI, use {"MODEL_PROVIDER": "openai", "MODEL": "gpt-5.4-mini", "MODEL_AUTH_MODE": "api", "OPENAI_API_KEY": os.environ["OPENAI_API_KEY"]} (anthropic subscription auth uses MODEL_AUTH_MODE="subscription" with CLAUDE_CODE_OAUTH_TOKEN instead of ANTHROPIC_API_KEY).

Rather than hand-building this dict, you can reuse the exact helper the CLI uses to load a provider profile — the local .env.* files under configs/ksi/ (copy a committed *.template, add your real key, keep it untracked; same files --provider-profile reads):

from ksi.providers import load_provider_profile

provider_env = load_provider_profile("configs/ksi/.env.sonnet")
runtime = KsiContainerExecutor(..., knowledge_db_path=config.knowledge_db_path, env=provider_env)

A complete runnable script is in examples/programmatic/run_arc_demo.py.

GenerationConfig requires num_generations and num_agents; pass those explicitly, so the minimal valid form is GenerationConfig(num_generations=1, num_agents=1) (add knowledge_db_path=... to enable the knowledge substrate). The remaining fields carry the same defaults as the CLI flags (single source of truth) — notably cross_task_forum_rounds=2, so a minimal knowledge-enabled config such as GenerationConfig(num_generations=1, num_agents=1, knowledge_db_path=...) runs two cross-task forum rounds per generation (the engine's old programmatic fallback was 1). Set fields explicitly to trim phase-3 container/LLM cost, e.g. GenerationConfig(num_generations=1, num_agents=1, knowledge_db_path=..., cross_task_forum_rounds=1). To disable forums programmatically, zero both per_task_forum_rounds and cross_task_forum_rounds. (The legacy forum_rounds field was removed along with the --forum-rounds CLI flag, which now hard-errors.) One migration note: experiment_name now defaults to "ksi" everywhere (previously the programmatic default was "default" and the CLI default was "swarms_v2"). Resuming an experiment created under either old default requires passing the old name explicitly (experiment_name="default" or --experiment-name swarms_v2) — otherwise resume=True looks up the new name, finds nothing, and starts fresh.

The contract

ksi.run(config, tasks, *, runtime, evaluator, llm, persistence=None, working_dir=".") -> list[TaskTrace] is a thin, behavior-preserving wrapper around GenerationalOrchestrator(...).run(tasks=...) — the orchestrator construction and run match the CLI path. CLI-only conveniences are not applied: provider profile / environment loading and signal handling are your responsibility. You construct runtime / evaluator / llm yourself, which keeps every run explicit about what it depends on.

Building blocks

Piece Where Notes
run ksi.run the entry point
GenerationConfig ksi.GenerationConfig run parameters
GenerationalOrchestrator ksi.GenerationalOrchestrator the engine (use run instead of constructing directly unless you need to)
LLM caller ksi.runtime.llm.build_llm_caller provider/model → LLMCaller
Evaluators ksi.eval (NoopEvaluator) / ksi.benchmarks (e.g. ArcSessionEvaluator) or build via the evaluator registry
Runtimes ksi.runtime.KsiContainerExecutor or build via the runtime registry
Tasks ksi.tasks.loaders.load_tasks_for_source or construct TaskSpec(...) directly
Persistence/callbacks ksi.protocols.PersistenceObserver optional; the CLI's SqlitePersistence is one implementation

Custom tasks

load_tasks_for_source and CommandEvaluator work the same way for the built-in custom task source as for any benchmark — swap task_source="arc" + ArcSessionEvaluator for task_source="custom" + CommandEvaluator:

from ksi.eval.command import CommandEvaluator
from ksi.tasks.loaders import load_tasks_for_source

tasks = load_tasks_for_source(task_source="custom", tasks_path="tasks.jsonl")
evaluator = CommandEvaluator()

tasks.jsonl is the same JSON/JSONL record file the CLI's --task-source custom --tasks-path expects; you can also skip the file and construct TaskSpec objects directly. See Your own tasks for the full record schema, the command evaluator's scoring contract, and both forms side by side.

Typed package

ksi ships a PEP 561 py.typed marker, so type checkers (mypy/pyright) see its annotations when you build on the public API.

Prerequisites

ksi.run executes agents in containers, so the same prerequisites as the CLI apply: Docker running, the ksi-agent:bench image built, runtime_runner Node deps installed, and a provider API key in the environment. See the README and docs/architecture.md.