Snipara Orchestrator
Production-first agentic orchestrator that uses Snipara for context-aware validation, hierarchical task coordination, and explicit production gates. Tasks aren't done until they work in production.
Advanced package
Most users do not need snipara-orchestrator. Start with Hosted MCP for agents or API / SDK for product integrations unless you specifically need production validation, proof-based checks, drift detection, or explicit htask orchestration.
For normal coding work, prefer Hosted MCP plus optional snipara-companion and Snipara Sandbox first. Use orchestrator only when you need a stronger control plane around validation and release workflows.
snipara-companion may suggest this package for FULL + ORCHESTRATED work, but it does not install orchestrator or spawn workers automatically. Install it explicitly through pip install snipara-orchestrator or npx create-snipara repair --with-orchestrator.
Philosophy
Traditional CI/CD pipelines stop at "tests pass, deploy". But production is the real test. Snipara Orchestrator implements prod-first validation:
Proof-Based Verification
Every check produces a proof with endpoint, user_tested, and result. No guessing.
Single Gatekeeper
One authority decides pass/fail. No conflicting validation layers.
Live Production Checks
HTTP health checks, UI tests with Playwright, schema validation against live endpoints.
Fail-Fast on Drift
Detect route changes, schema drift, or unhealthy services before wasting time.
Installation
pip install snipara-orchestratornpx create-snipara repair --with-orchestratorOr with all optional dependencies (Playwright for UI tests):
pip install snipara-orchestrator[all]Quick Start
Basic Usage
import asynciofrom snipara_orchestrator import Orchestrator, Task, ValidationCriteriafrom snipara_orchestrator.models import LiveCheck, OrchestratorConfigconfig = OrchestratorConfig( snipara_api_key="snp-your-key", snipara_project="my-project", prod_url="https://api.example.com",)orchestrator = Orchestrator(config)await orchestrator.initialize()task = Task( id="deploy-auth", title="Deploy Authentication", criteria=ValidationCriteria( local_tests=["pnpm test", "pnpm lint"], live_checks=[ LiveCheck(url="/health", expected_status=200), LiveCheck(url="/auth/login", method="POST", expected_status=200), ], required_proofs=2, ),)result = await orchestrator.execute_task(task)print(f"Status: {result.status.value}")CLI Usage
snipara-orchestrator initCreated orchestrator.yaml with default configurationsnipara-orchestrator run "deploy-auth"Task completed: DONE (3/3 proofs passing)Task Lifecycle
Tasks flow through a strict state machine:
PENDING → IN_PROGRESS → LOCAL_OK → VALIDATING → PROD_OK → DONE ↘ ENV_DRIFT (fail-fast)| State | Description |
|---|---|
PENDING | Task created, waiting to start |
IN_PROGRESS | Local tests running |
LOCAL_OK | Local tests passed, ready for validation |
VALIDATING | Running live checks against production |
PROD_OK | All live checks passed |
DONE | Task completed successfully |
ENV_DRIFT | Environment changed unexpectedly (routes, schema, health) |
FAILED | Task failed validation |
Snipara Integration
The orchestrator uses Snipara for context-aware validation. It queries your documentation to understand expected behavior, stores decisions in memory, and coordinates htask-based multi-agent workflows when you explicitly choose that pattern.
# Query context for task understandingcontext = await snipara.query("authentication API endpoints")# Remember decisions across sessionsawait snipara.remember( content="Used JWT for auth, not session cookies", type="decision",)# Coordinate with other agents via htasksfeature = await snipara.create_htask_feature(swarm_id, "Billing rollout", "Coordinate API, frontend, QA, and deploy validation")Drift Detection
The orchestrator automatically detects environment drift:
Route Drift
Expected endpoints missing or returning different status codes.
Schema Drift
OpenAPI/JSON Schema mismatches between expected and actual responses.
Health Drift
Services unhealthy or returning degraded status.
Multi-Agent Workflows
Use Snipara htasks to coordinate multiple agents. The orchestrator provides wrappers for feature creation, task creation, queue recommendations, completion evidence, and tree inspection. It still leaves actual Codex, Claude, or CI worker spawning explicit.
# Coordinator creates a feature and workstreamsawait snipara.create_swarm(name="deployment")feature = await snipara.create_htask_feature(swarm_id, "Auth rollout", "Ship auth safely")await snipara.create_htask(swarm_id, "Run unit tests", parent_id=feature["task_id"], priority=10)# Workers request the next safe batch and report evidencebatch = await snipara.recommend_htask_batch(swarm_id, limit=3)result = await execute_task(batch["tasks"][0])await snipara.complete_htask(swarm_id, batch["tasks"][0]["task_id"], result=result)snipara-orchestrator htask-next --swarm-id <swarm> --limit 3snipara-orchestrator htask-tree --swarm-id <swarm> --task-id <feature>API Reference
Orchestrator
| Method | Description |
|---|---|
initialize() | Initialize orchestrator and connect to Snipara |
execute_task(task) | Run full task lifecycle with validation |
check_drift() | Check for environment drift |
get_status(task_id) | Get current task status |
Gatekeeper
| Method | Description |
|---|---|
gate_local_to_validation(task) | Decide if local tests passed |
gate_validation_to_prod(task) | Decide if live checks passed |
generate_cutover_checklist(task) | Generate deployment checklist |
Examples
Basic Usage
Simple task with local tests and live checks.
Multi-Agent Workflow
Coordinator and workers using Snipara swarms.