orchcore 2.0.0

Reusable orchestration core for AI coding agent CLI pipelines

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, Gemini, 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

Module	Purpose
stream/	4-stage pipeline (Filter → Parse → Monitor → Stall Detect) for 5 agent formats
pipeline/	Phase orchestration — sequential/parallel multi-agent execution in topological dependency order
runner/	Async subprocess management with stdout/stderr streaming and optional stdin prompt transport (prompt_via = "stdin")
registry/	Agent configurations as data (TOML/dict) with runtime lookup
config/	Layered configuration: TOML → env vars → CLI overrides → profiles
recovery/	Rate-limit detection, exponential backoff, git dirty-tree recovery
workspace/	Artifact lifecycle management
prompt/	Jinja2 template rendering with frontmatter stripping
ui/	UICallback protocol for pluggable display layers
signals/	Graceful SIGINT/SIGTERM shutdown
display/	Colored stderr logging (no Rich dependency in core)
observability/	Optional OpenTelemetry integration

Documentation

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

Document	Description
Installation	Prerequisites, install options, extras
Quick Start	Define agents, build phases, run pipelines
Configuration Reference	Full settings table, profiles, env vars
Stream Events Reference	StreamEvent fields, types, agent states
UICallback Reference	Protocol methods and built-in implementations
Architecture	Package layout, core abstractions, design decisions
Stream Pipeline	4-stage composable pipeline deep-dive
Design Document	Problem statement, requirements, proposed design
Writing a UICallback	Build custom display layers
Agent Registry	TOML config, adding new agents, ToolSets
Recovery & Retry	Rate limits, backoff, failure modes
Contributing	Dev setup, code standards, testing

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.