orchcore

Reusable orchestration core for AI coding agent CLI pipelines.

What is orchcore?

orchcore is an async-first Python 3.12+ library that provides unified infrastructure for launching, monitoring, and managing multiple AI coding agent CLIs (Claude, Codex, Antigravity, Copilot, OpenCode) as subprocesses through phase-based pipelines. It was extracted from four production orchestration systems — Planora, Articles, Finvault, and Raven — eliminating 60-70% of duplicated infrastructure so consuming projects only implement domain-specific logic.

Features

• Multi-agent subprocess orchestration — async launch, stream capture, concurrency control
• Unified stream processing — 4-stage pipeline normalizes 5 JSONL formats into a single StreamEvent model
• Phase pipelines — sequential/parallel execution with dependency checks and resume
• Rate-limit recovery — automatic detection, timezone-aware reset parsing, exponential backoff
• Layered configuration — TOML files, env vars, CLI overrides, named profiles
• Protocol-based UI — UICallback decouples engine from display (Rich, Textual, headless)
• Registry-as-data — new agent support via TOML config alone, zero code changes
• Graceful shutdown — SIGINT/SIGTERM with subprocess cleanup and state preservation
• Safe subprocess boundaries — filtered agent environments by default, explicit cwd support, and opt-in git recovery

Installation

uv pip install orchcore

From source:

git clone https://github.com/AbdelazizMoustafa10m/orchcore.git
cd orchcore
uv pip install -e ".[dev]"

Requirements: Python 3.12+

Quick Start

1. Define Agents

# agents.toml
[agents.claude]
binary = "claude"
model = "claude-sonnet-4-20250514"
subcommand = "-p"
stream_format = "claude"

# Flag profiles — names are your project's workflow vocabulary, not orchcore's.
[agents.claude.flags]
plan = ["--think", "--verbose"]

[agents.claude.output_extraction]
strategy = "jq_filter"
jq_expression = ".content[0].text"

2. Run a Pipeline

"""Minimal orchcore pipeline setup.

Requires a matching ``agents.toml`` file and the configured agent CLI on
``PATH``. Use ``dry_run=True`` at the AgentRunner layer for artifact-level
smoke tests; a real pipeline run launches the configured CLI.
"""

from __future__ import annotations

import asyncio
from pathlib import Path

from orchcore.pipeline import Phase, PhaseRunner, PipelineRunner
from orchcore.registry import AgentRegistry, ToolSet
from orchcore.runner import AgentRunner
from orchcore.ui import NullCallback


async def main() -> None:
    registry = AgentRegistry()
    registry.load_from_toml(Path("agents.toml"))

    phase = Phase(
        name="planning",
        agents=("claude",),
        # Selects [agents.<name>.flags].plan — a name YOUR project defines.
        flag_profile="plan",
        tools=ToolSet(internal=("Read", "Glob", "Grep"), permission="read-only"),
    )

    runner = AgentRunner()
    phase_runner = PhaseRunner(runner, registry, max_concurrency=4)
    pipeline = PipelineRunner(phase_runner)

    result = await pipeline.run_pipeline(
        phases=[phase],
        prompts={"planning": "Analyze the codebase and create a plan."},
        ui_callback=NullCallback(),
    )

    print(f"Success: {result.success} | Cost: ${result.total_cost_usd}")


if __name__ == "__main__":
    asyncio.run(main())

By default, agent subprocesses receive a filtered environment: common API keys and provider-specific variables are not inherited unless you set env_policy = "inherit", pass env_passlist, or provide explicit env_vars in your agent config.

Modules

Documentation

Full documentation is available at abdelazizmoustafa10m.github.io/orchcore.

Contributing

See CONTRIBUTING.md for development setup, testing instructions, and code standards.

Please also review CODE_OF_CONDUCT.md and SECURITY.md.

License

orchcore is released under the MIT License.