Orchestrator Python MCP Server
It runs a machine-verified 4-phase workflow MCP server for AI coding assistants, enforcing DESIGN → PLAN → EXECUTE → VERIFY → COMPLETE with human approval gates.
Initialize sessions:
orchestrate_initstarts a new workflow with a task description, sets phase to DESIGN, and returns SOP instructions with HMAC anti-tamper security.Check status:
orchestrate_statusreports whether a session is active and its current phase.Approve deliverables:
orchestrate_approvegrants the required human gate before verification for DESIGN/PLAN phases.Verify and advance:
orchestrate_verifyvalidates phase deliverables (e.g., design.md headings, plan.md task schema, checked tasks, file sizes, test command) and advances to the next phase with SOP instructions.Plan execution:
orchestrate_get_dag_batchesparses plan.md tasks into ordered parallel batches with dependency/file-collision handling.List agents:
orchestrate_get_agentsreturns specialized agent personas usable as subagents.Archive:
orchestrate_archivemoves session deliverables to.orchestrator/archive/<session_id>/and releases the session lock.
Click on "Deploy Server".
Wait a few minutes for the server to deploy. Once ready, it will show a "Started" state.
In the chat, type
@followed by the MCP server name and your instructions, e.g., "@Orchestrator Python MCP Server Start a new orchestration session to build a user CRUD API"
That's it! The server will respond to your query, and you can continue using it as needed.
Here is a step-by-step guide with screenshots.
Orchestrate Python MCP Server
Machine-verified 4-phase workflow MCP server for AI coding assistants. Enforces DESIGN → PLAN → EXECUTE → VERIFY → COMPLETE with human approval gates, state integrity checks, and DAG task scheduling.
Quickstart
Installation & Run
# Sync environment
uv sync
# Run tests (Unit + 26 BDD Scenarios)
uv run pytest
# Start MCP server (stdio transport)
uv run python -m orchestrate_mcp.serverMCP Client Configuration
Example OpenCode config (opencode.json):
{
"mcp": {
"orchestrate": {
"type": "local",
"command": ["uv", "run", "--project", "/path/to/orchestrate", "python", "-m", "orchestrate_mcp.server"]
}
}
}The server runs in stdio mode; cwd resolves to the session directory where the client launched the server. See the distribution guide for other clients (Claude Desktop, Cursor, Windsurf).
Related MCP server: Vibe Engineering MCP
Table of Contents
Workflow Lifecycle
[Start] ──> orchestrate_init(task="...")
│
▼ (DESIGN Phase)
Write `.orchestrator/design.md`
│
▼
orchestrate_approve() ──> orchestrate_verify()
│
┌───────────────────────┘
▼ (PLAN Phase)
Write `.orchestrator/plan.md`
│
▼
orchestrate_approve() ──> orchestrate_verify()
│
┌───────────────────────┘
▼ (EXECUTE Phase)
orchestrate_get_dag_batches()
Subagents implement tasks & mark [x]
│
▼
orchestrate_verify()
│
▼ (VERIFY Phase)
orchestrate_verify() (runs automated test command)
│
▼ (COMPLETE Phase)
orchestrate_archive() ──> [Done]MCP Tools Reference
Tool Name | Parameters | Description |
|
| Initializes new session in |
| (none) | Returns |
| (none) | Human gate approval. Unlocks verification for |
| (none) | Validates phase deliverables. Advances to next phase on pass, returns next SOP prompt. |
| (none) | Parses |
|
| Releases lock and moves session files into |
Deliverable Requirements per Phase
DESIGN Phase:
Deliverable:
.orchestrator/design.mdRequired Headings:
## Requirements,## Architecture,## Self-Confidence Audit.Gate: Requires
orchestrate_approvebefore verification.
PLAN Phase:
Deliverable:
.orchestrator/plan.mdTasks Schema:
- [ ] **<id>**: <desc> (Agent: <role>, Target: <file>, blocked_by: [<deps>])Detailed Specs:
### <id>section for every task.Final Barrier: Final task MUST be assigned to
Agent: implementation-reviewerblocked by all prior tasks.Test Command: Valid executable
Test command: <cmd>under## Verification.Gate: Requires
orchestrate_approvebefore verification.
EXECUTE Phase:
Rule: All tasks in
.orchestrator/plan.mdmarked checked- [x].Rule: Target files must exist on disk and have size > 0 bytes.
VERIFY Phase:
Rule: Runs plan's test command via shell subprocess (120s timeout). Passes if exit code is 0.
Example Deliverables
Below are minimal, realistic examples demonstrating valid syntax and required structure for phase deliverables.
Example .orchestrator/design.md
# Design — Collinear-feature handling in corrected CFI estimators
## Goal
Fix `run_cfi_corrected` crash when input feature matrix contains near-collinear or duplicate pairs (`DatasetException: |rho|=1 is effectively 1`).
## Requirements
### Functional
| # | Requirement |
|---|---|
| F1 | `corrected_mutual_information(a, b)` returns `1.0` when pair is collinear (`rho**2 >= 1 - 1e-12`). |
| F2 | `corrected_variation_of_information(a, b)` returns `0.0` when pair is collinear. |
| F3 | `BaseCorrectedCfiConfig` gains opt-in `drop_collinear_features: bool = False`. |
### Non-Functional
- **No silent fallbacks**: Constant/NaN inputs still raise `DatasetException`.
- **Zero regression**: Unaffected estimator paths remain byte-identical.
## Architecture
```python
_EFFECTIVELY_COLLINEAR_RHO2 = 1 - 1e-12
def _is_effectively_collinear(rho: float) -> bool:
return rho**2 >= _EFFECTIVELY_COLLINEAR_RHO2
```
In `corrected_mutual_information`, short-circuit before histogram binning:
- If `np.isfinite(rho)` and `_is_effectively_collinear(rho)`: return `1.0`.
## Self-Confidence Audit
- Guessed paths: 0% (inspected `dependence.py`, `config.py`, `impl.py`)
- Unresolved assumptions: 0% (analytic limits Cover & Thomas Thm 2.4.1)
- Missed edge cases: 0% (constant inputs delegate to existing validation)
- Unchecked config: 0% (pydantic & numpy dependencies verified)
**Score: 97%** (>= 95% gate pass)Example .orchestrator/plan.md
# Implementation Plan — Collinear-feature handling in corrected CFI estimators
## Overview
Implement 2-layer collinear handling: (1) estimator analytic limit guard in `dependence.py`, (2) opt-in feature dedup in `config.py` / `impl.py`.
## Tasks
- [ ] **T1**: Estimator guard in dependence.py (Agent: coder, Target: src/research/cfi/dependence.py, blocked_by: [])
- [ ] **T2**: Add drop_collinear_features config field (Agent: coder, Target: src/research/cfi/config.py, blocked_by: [])
- [ ] **T3**: Unit tests for collinear limits & config (Agent: tester, Target: tests/test_cfi.py, blocked_by: [T1, T2])
- [ ] **T4**: Final Implementation Verification Audit (Agent: implementation-reviewer, Target: .orchestrator/plan.md, blocked_by: [T1, T2, T3])
## Detailed Task Specifications
### T1: Estimator guard in dependence.py
- **Target**: `src/research/cfi/dependence.py`
- **Signatures & Contracts**:
- Add `_EFFECTIVELY_COLLINEAR_RHO2: float = 1 - 1e-12`
- Add `_is_effectively_collinear(rho: float) -> bool`
- Update `corrected_mutual_information(a: ArrayLike, b: ArrayLike, n_bins: int | None = None) -> float`
- **Old → New Implementation**:
```python
# Before
x, y = _as_paired_arrays(a, b)
hx, hy, hxy = _binning_and_entropies(x, y, n_bins)
# After
x, y = _as_paired_arrays(a, b)
if n_bins is None:
rho = float(np.corrcoef(x, y)[0, 1])
if np.isfinite(rho) and _is_effectively_collinear(rho):
return 1.0
hx, hy, hxy = _binning_and_entropies(x, y, n_bins)
```
- **Acceptance Criteria**:
- `corrected_mutual_information(2*a+1, a) == 1.0` without raising `DatasetException`.
### T2: Add drop_collinear_features config field
- **Target**: `src/research/cfi/config.py`
- **Signatures & Contracts**:
- Extend `BaseCorrectedCfiConfig(BaseModel)` with new schema fields.
- **Old → New Implementation**:
```python
# Before
class BaseCorrectedCfiConfig(BaseModel):
scoring: Any = log_loss
seed: int = 42
# After
class BaseCorrectedCfiConfig(BaseModel):
scoring: Any = log_loss
seed: int = 42
drop_collinear_features: bool = False
collinear_threshold: float = Field(default=0.999, ge=0.0, lt=1.0)
```
### T3: Unit tests for collinear limits & config
- **Target Test File**: `tests/test_cfi.py`
- **Test Scenarios**: MI limit = 1.0, VI limit = 0.0, validation constraint `lt=1.0`.
- **Copy-Paste Implementation**:
```python
def test_mi_collinear_returns_one():
a = np.arange(100.0)
b = 2.0 * a + 1.0
assert corrected_mutual_information(b, a) == pytest.approx(1.0)
def test_vi_collinear_returns_zero():
a = np.arange(100.0)
b = 2.0 * a + 1.0
assert corrected_variation_of_information(b, a, normalize=True) == 0.0
def test_collinear_threshold_validation():
with pytest.raises(ValidationError):
cfi_config("onc_vi", collinear_threshold=1.5)
```
### T4: Final Implementation Verification Audit
- **Target**: `.orchestrator/plan.md`
- **Verification Specialist Duties**:
- Inspect `git diff` against specifications in T1–T3.
- Verify all unit tests pass with zero regressions.
## File Inventory
| File | Status | Purpose |
|---|---|---|
| `src/research/cfi/dependence.py` | Modified | Add collinear short-circuit |
| `src/research/cfi/config.py` | Modified | Add `drop_collinear_features` schema field |
| `tests/test_cfi.py` | Modified | Collinear regression test suite |
## Verification
Test command: uv run pytest tests/test_cfi.py -v
## Confidence Self-Audit
- Guessed paths: 0%
- Unresolved assumptions: 0%
- Missed edge cases: 0%
- Unchecked config: 0%
**Score: 98%** (>= 95% gate pass)Available Tools
7 toolsorchestrate_approveC
Grant human approval for the current phase deliverable.
Inputs: workspace_root. Returns: error, phase, message, success
orchestrate_archiveC
Archive current orchestration deliverables and release session lock.
Inputs: force, workspace_root. Returns: error, message, success, archived_session_id
orchestrate_get_agentsA
List all specialized orchestrator agent personas. Each name is a registered OpenCode subagent ID, spawnable by name via the subagent tool even if hidden from the advertised subagent catalog.
Inputs: no named parameters. Returns: error, agents, success
orchestrate_get_dag_batchesB
Compute topological execution batches from plan.md tasks.
Inputs: workspace_root. Returns: error, batches, success, total_tasks
orchestrate_initB
Initialize a new orchestration session with HMAC anti-tamper security.
Inputs: task_description (required), workspace_root. Returns: error, phase, success, session_id, +1 more
orchestrate_statusB
Query current session status and active phase.
Inputs: workspace_root. Returns: phase, message, active_session
orchestrate_verifyB
Run machine verification on current phase deliverables and advance phase on success.
Inputs: workspace_root. Returns: phase, errors, success, previous_phase, +1 more
Tool Schema Changelog
Recent tool additions, removals, and schema changes observed during successful MCP inspections.
Loading tool schema changelog…
TDQS
Scored across 7 tools
Each tool targets a distinct operation: session init, status, human approval, machine verification, archiving, DAG batch computation, and agent listing. There is no meaningful overlap, and the approve versus verify distinction is clear because one is human-driven and the other is machine-driven.
All tools share the orchestrate_ prefix and lowercase snake_case, which makes the set easy to predict. However, status reads as a noun rather than an action verb, and the get_* retrieval tools stand slightly apart from the other lifecycle verbs.
Seven tools is well-scoped for an orchestration lifecycle server, covering session creation, status, approval, verification, archival, DAG computation, and agent discovery without redundancy. The count feels appropriate for the domain.
The core lifecycle is present, but the human-in-the-loop workflow lacks explicit reject, cancel, or rollback paths, and verification only advances on success. Additionally, the DAG batch and agent listing tools are auxiliary and do not include an execution tool, leaving a notable gap in the orchestration loop.
Maintenance
Related MCP Connectors
Design, build and run teams of AI agents with checked handoffs between them.
AI work orchestration for plans, tasks, teams, and coding-agent dispatch.
AI-native programming language and toolchain for accountable, verifiable agent-driven development.
Build, test and publish agentic workflows from your AI client: approve steps, read tables, see runs.
Related MCP Servers
- AlicenseAqualityCmaintenanceEnables coding agents to run persistent, budgeted work sessions with dependency graphs, human decision queues, required validation, and clean handoffs for autonomous or human-in-the-loop execution.16MIT
- FlicenseBqualityCmaintenanceEnables autonomous software development by managing requirements, architecture, task dependency DAGs, isolated worker workspaces, execution, contract and integration verification, and checkpoints. It provides observable evidence gates so AI workers can build and verify software against reproducible contracts.28-
- AlicenseNot gradedqualityAmaintenanceEnables AI coding providers to run structured multi-agent workflows with specifications, tracked tasks, handoffs, approvals, and health-gated completion across supported coding environments.29 npmMIT
- AlicenseNot gradedqualityAmaintenanceEnables AI coding providers to run structured multi-agent workflows with specs, tracked tasks, handoffs, health gates, and audited repair mode.29 npmMIT