A multi-agent meta-harness for coding agents — drives Claude Code, Gemini CLI, and OpenCode in one calm full-screen TUI.
Project description
Aegis
The meta-harness. Drive Claude Code, Gemini CLI, and OpenCode side by side from one calm terminal — and make them collaborate.
┌ aegis · 3 agents · ~/code/aegis ─────────────────────────────────────┐
│ ● 1 lucid-knuth ·opus· ● 2 wry-hopper ·gemini· ● 3 brisk-curie * │
│ │
│ › explain the retry path in worker.py │
│ │
│ ⠹ Thinking… (3.2s) │
│ ⏺ Read(worker.py) │
│ └ ok │
│ The retry path lives in _run_turn at line 142 … │
│ │
│ ⏺ aegis_handoff(target=wry-hopper) │
│ └ delivered to wry-hopper │
│ │
│ queues: tests ●1/2 ○0 ✓3 ✗0 last: brisk-curie │
│ lucid-knuth ·opus· opus·full ↑128k (94% cached) ↓1k │
│ ───────────────────────────────────────────────────────────────────── │
│ › ask something… │
└───────────────────────────────────────────────────────────────────────┘
Above the harness, not beside it
Most agentic frameworks (CrewAI, LangGraph, AutoGen, the whole long list) talk directly to LLM providers — they replace your coding agent and reimplement tool use, permissions, sandboxing, terminal integration. Aegis takes the opposite path:
- It sits above your existing coding agents and drives them over
their structured protocols —
stream-jsonfor Claude Code, the Agent Client Protocol (ACP) for Gemini CLI and OpenCode, and a clean driver seam for whatever lands next. - It doesn't reimplement the agent. Tool use, sandboxing, MCP hosting, model selection — that's the harness's job. Aegis's job is the layer above: tabs, routing, delegation, persistence, the things a single-conversation CLI was never built to do.
- It makes them collaborate. Six composable coordination primitives mean a Claude tab can hand off to a Gemini tab, dispatch an OpenCode worker, subscribe to a shared canvas, share a live terminal, fan a question out to a committee, or kick off a deterministic Python workflow that drives all three.
The harness wars are over. You probably already have your favorite (or two, or three). Aegis lets you keep them — and run them as a team.
Six primitives for agent coordination
Each primitive has one verb and lands the same way in the receiving
agent's transcript: as a ✉ block with a sender tag, timestamp, and a
short body preview. One delivery channel, six wake patterns.
→ Inbox — send context to a peer
Any agent can hand off to any other live agent. Fire-and-forget; the recipient gets a normal user-message turn tagged with the sender's handle. Use when you want a specific peer to pick up where you left off.
aegis_handoff(target_handle="reviewer", from_handle="impl",
context="PR ready at branch feat/x — please review")
# → reviewer's transcript:
# ✉ from agent:impl · 17:42:03Z
# PR ready at branch feat/x — please review
⏳ Queue — spawn a worker on demand
Enqueue a task to a named queue and the substrate spawns a fresh agent
of the queue's configured profile, runs the payload as its opening
turn, and (with callback=true) delivers the worker's final result
back to your inbox. Producer keeps working between enqueue and
callback. Generalizes delegation: parallelism, max-in-flight caps,
restart safety, all built in.
aegis_enqueue(queue="review", payload="…full self-contained prompt…",
from_handle="impl", callback=True)
# → {task_id: 01HK…, queued_position: 1}
# …minutes later, in impl's transcript:
# ✉ from queue:review · task#01HK… · ok · 17:46:11Z
# PR looks clean. Two nits flagged in the diff comments…
▦ Canvas — collaborate on a shared document
Open a shared markdown file. Multiple agents read it, write sections of it, subscribe to it. Each write wakes every other subscriber with a diff-aware notification. The classical blackboard pattern — terminal- native, MCP-driven, file-backed (you can grep it, commit it, open it in your editor).
# PM
aegis_canvas_open(name="report-q3", file="vault/reports/q3.md",
from_handle="pm")
aegis_canvas_subscribe(name="report-q3", from_handle="pm")
# Researcher (in another tab, after a handoff)
aegis_canvas_write_section(name="report-q3", section="data",
content="Q3 numbers came in stronger…",
from_handle="researcher")
# → PM's transcript:
# ✉ from canvas:report-q3 · 20:30:00Z
# section "data" · written by agent:researcher (+18 / -3 lines)
# ──
# Q3 numbers came in stronger than projected…
▮ Terminal — share a live shell
Spawn a PTY-backed shell that any agent (or Alex) can run commands on,
send raw keystrokes to, and subscribe to. Command boundaries are
detected from OSC 133 shell-integration markers; every finalized
command lands in an append-only JSONL ledger and wakes subscribers
through the same ✉ channel.
# PM
aegis_term_spawn(name="build", from_handle="pm")
aegis_term_subscribe(name="build", from_handle="pm")
# builder (after a handoff)
rec = aegis_term_run(name="build", cmd="pytest -q",
from_handle="builder")
# → PM's transcript:
# ✉ from term:build · 14:03:25Z
# $ pytest -q · run by agent:builder
# exit 0 · 4.20s
# ──
# 6 passed in 4.18s
▣ Groups — broadcast and gather
Form a named committee of agents that share one inbox-fanout channel
and one in-flight broadcast slot. Send one structured four-field
question — objective, output_format, tool_guidance,
boundaries — collect N parallel answers, reduce them into a single
result. Use when the same question has multiple useful perspectives,
or when you want to race providers and keep the fastest.
# spawn three reviewers with different lenses
aegis_group_spawn_mixed(name="audit", from_handle="pm",
profiles=["sec_reviewer", "style_reviewer", "logic_reviewer"])
aegis_group_broadcast(name="audit",
objective="audit PR #214 (branch feat/rate-limit)",
output_format="bullet list, each item severity-tagged (high/med/low)",
tool_guidance="prefer Read + Grep; avoid Bash and Edit",
boundaries="report only — no patches, no commits")
# collect every reply, keyed by reviewer
result = aegis_group_wait_all(name="audit",
timeout=300,
reducer="join_by_handle")
# → result.reduced = {"sec_reviewer": "…", "style_reviewer": "…",
# "logic_reviewer": "…"}
Switching to aegis_group_wait_any returns on the first reply and (by
default) sends a passive cancel envelope to the losers — useful when
the cheapest acceptable answer wins. Built-in reducers: concat,
join_by_handle, last_wins, majority_vote; custom reducers
register one function. Groups also have YAML presets in
.aegis.yaml (groups.presets.<name>.profiles: […]) and a dedicated
TUI tab with Members / Current broadcast / Recent broadcasts panels.
Reach for it when: multi-lens code audit, fastest-answer racing, cross-provider consensus, generate-and-pick (N candidates → one), role-persona panels (PM / eng / UX react to the same proposal). Full walk-through in docs/groups.md.
⟳ Workflow — deterministic Python orchestration
When the dance has to be reliable — TDD loops, bug triage, multi-step plans, anything where retries with feedback matter — wrap it in a workflow. Plain Python at the top of the stack. Calls agents, runs bash predicates, retries with feedback, captures structured output.
@workflow("tdd-cycle")
async def tdd_cycle(engine, *, feature: str) -> str:
impl = await engine.spawn("implementer")
await engine.send(impl, f"Write a failing test for: {feature}")
await engine.bash_predicate(
f"pytest tests/ -k {feature} 2>&1 | grep -E 'FAIL|ERROR'",
retry_with="The test should fail because the feature isn't built yet")
await engine.send(impl, "Now implement it.")
await engine.bash_predicate(
f"pytest tests/ -k {feature}",
retry_with="Tests are still failing. Output:\n{stdout}")
reviewer = await engine.spawn("reviewer")
return await engine.send(reviewer, "Final review of branch.")
Triggered by any agent: aegis_run_workflow(name="tdd-cycle", kwargs={"feature": "rate_limit"}). Workflows sit at the top of the
stack — they span agents, they own the loop, they're the right tool
when the spec is "follow this exact procedure" rather than "figure
it out."
The aegis.workflows package ships four seed workflows registered on
import: brainstorm_to_spec (Q/A → spec doc), execute_plan (parse
plan → dispatch implementer per task with durable resume),
review_branch (parallel reviewer fan-out → report), and tdd_cycle
(predicate-driven TDD loop). See docs/workflows.md.
What else is in the box
- Multi-tab TUI. Generated alliterating handles (
lucid-knuth,wry-hopper) for agents, purpose names (build,db) for terminals. State dots, sticky*, terminal bell when a backgrounded agent finishes. Click any block to copy it. - Honest metrics. True input (incl. cache) with cached %, output, tool calls, per-turn and per-session wall-clock. Provisional while streaming, exact at turn end. No log scraping anywhere.
- Queue dashboard. Always-on one-line strip above the status bar
shows live per-queue depth and the most recent in-flight worker.
Ctrl+Dexpands into a full-screen modal withQUEUES / IN-FLIGHT / QUEUED / RECENTbands and a live assistant-text tail. - Session persistence.
aegisreopens the last workspace by default — agent tabs, terminal tabs, profiles, order, with each underlying session genuinely resumed (model memory intact).aegis --cleanopts out. - Workflow catalog.
aegis.workflowsships four ready-to-use seeds (brainstorm_to_spec,execute_plan,review_branch,tdd_cycle); importing them registers. Engine offersask_human, explicitcheckpoint+ durable resume,bash_predicateretry loops, andparallelfan-out. - Headless + Telegram.
aegis serveruns the SessionManager + MCP plane without a TUI. Add a Telegram token to drive the team from your phone. - MCP plane. Every spawned agent is injected with the aegis MCP
server: orientation (
aegis_meta), session listing, handoff, queue dispatch, canvas ops, terminal ops, group broadcast/gather, workflow invocation. One consistent surface across providers. With--strict-mcp-config, aegis is the only MCP server the spawned agent sees.
Remote plane — cross-machine handoff
aegis serve can expose a second HTTP plane — distinct from the
loopback MCP plane, bound wherever you want it reachable — that other
aegis serve instances POST into. One agent on one machine can hand
a long task off to another without leaving the substrate:
aegis_enqueue(
queue="implementation",
payload="Implement the design at <path> with TDD…",
from_handle="lucid-knuth",
target="builder", # ← new — routes to a remote aegis
)
# → {task_id: "01J…", target: "builder",
# callback_note: "no wire return channel in v1; completion behavior
# is whatever the receiving serve is configured to do"}
Configuration lives in .aegis.yaml. Outbound — the peers this serve
can call:
remotes:
builder:
url: http://100.64.0.5:8556
# token: "<optional bearer>"
Inbound — opt-in receive side; default off:
remote_plane:
bind: 100.64.0.5:8556 # the address to listen on
accept_tokens: [] # optional bearer-token allowlist
accept_from: [] # optional source-IP allowlist
Bind the plane wherever it should be reachable from, and only from — typically a private overlay network (Tailscale/Headscale/WireGuard/ VPN) so the network itself is the trust anchor. Bearer-token and source-IP gates compose with AND on top. By default the call is fire-and-forget: the receiving serve runs the worker under its own config and whatever it does on completion (commit and push, message through its own bridge, write to a shared folder, nothing) is up to it.
Since v0.8.0, aegis_enqueue(target=…, callback=True) opts in to a
wire callback that delivers the remote worker's final message
back to the originating agent's inbox as a normal ✉ from queue:<peer>:<name> envelope (symmetric peers config required —
both sides define each other in remotes:). A small
/remote/v1/schedule/* control plane (with aegis_schedule_* MCP
tools and aegis schedule push --to <peer> / --remote <peer> CLI
verbs) lets one serve push schedules into a peer and inspect or
remove them remotely — useful for self-scheduling future work or
managing a fleet from one host. Full surface, error model, and
patterns in docs/remote.md.
Per-queue budgets
Declare rolling USD or output-token ceilings on any queue. The substrate enforces them at enqueue time — if admitting the task would push the queue over any configured ceiling, the enqueue is rejected immediately with a structured error that names the blocking constraint and gives an ETA for when the queue will unblock.
# .aegis.py
queues = {
"impl": {
"agent": "opus",
"max_parallel": 2,
"budgets": [
{"usd": 1.00, "window": "1h"},
{"usd": 10.00, "window": "24h"},
{"output_tokens": 500000, "window": "1h"}, # runaway belt
{"usd": 50.00, "window": "7d"},
],
},
"fast": {
"agent": "haiku-fast",
"max_parallel": 4,
# no budgets: key → no caps; behaves as before
},
}
All-must-allow: a task is admitted only if every budget entry is
under its ceiling. When rejected, the error names every blocking
constraint, how much was spent vs the limit, and an unblock_at ETA.
Budget state is visible via aegis budget list/show (with --remote <peer> for cross-host), the aegis_budget_status MCP tool, and the
read-only GET /remote/v1/budget endpoints on the remote plane. No
alerts — observability is pull-only; the rejection at enqueue time is
the only loud signal.
See docs/budget.md for the full model.
Scheduled workflows
Aegis runs a cron-style scheduler alongside QueueManager and the inbox
router. Schedules are declared in .aegis.yaml and can be split into
drop-in overlays under .aegis/schedules/<name>.yaml. Each entry names
a workflow (built-in or registered), a trigger (cron or fire_at),
a lifecycle (forever / once / {fires: N} / {until: <iso>}),
and an overlap policy (skip / queue / kill).
# .aegis.yaml
schedules:
morning-briefing:
workflow: prompt
cron: "0 6 * * *"
timezone: America/Havana
args: { agent: default, message: "Write today's briefing." }
ci-watch:
workflow: enqueue
cron: "*/5 * * * *"
lifecycle: forever
on_overlap: skip
args: { queue: ci, payload: "Check CI status and report failures." }
Two workflows ship in-tree: prompt (one-shot agent message) and
enqueue (scheduler → queue handoff).
The substrate writes a JSONL audit log per schedule under
.aegis/state/schedules/<name>.jsonl plus a derived
schedules.snapshot.json for dashboards. On boot it replays each log
to rebuild fire counts and closes any dangling fire_requested record
as failed:interrupted. Editing .aegis.yaml or any overlay file
hot-swaps the schedule table without a restart — entries that didn't
change keep their state.
aegis schedule list # current schedules + next fire
aegis schedule show morning-briefing
aegis schedule run morning-briefing # force-fire once
aegis schedule disable morning-briefing # comment-preserving YAML edit
aegis schedule logs morning-briefing -n 50
Install
pip install aegis-harness # or: uv pip install aegis-harness
Requires Python 3.13+ and at least one of: claude, gemini, or
opencode on your PATH, signed-in.
Quickstart
aegis init # interactive wizard — detects installed CLIs, writes .aegis.py
aegis # full-screen TUI
The wizard finds whichever agent CLIs you have installed and walks you
through picking a model, permission mode, and optional queues. The
generated .aegis.py is plain Python — edit it freely afterward.
Keys
| Key | Action |
|---|---|
Enter |
Send |
Ctrl+T / Ctrl+N |
New tab (default agent) / new tab (pick agent) |
Ctrl+E |
New terminal tab (term:<name>) |
Ctrl+W |
Close tab (last → quit) |
Ctrl+1..9 / Ctrl+Tab / Ctrl+←→ |
Switch tabs |
Ctrl+K |
Toggle terminal-tab input between run and raw mode |
Ctrl+D |
Open / close the queue dashboard |
Escape |
Interrupt the active turn (or dismiss a modal) |
Click on a block |
Copy that message / tool result to clipboard |
Ctrl+Q |
Quit |
A backgrounded tab that finishes shows a * and rings the bell.
Configuration
.aegis.py is plain Python. The wizard writes one for you; here's the
shape:
from aegis import Agent, ClaudeCode, GeminiCLI, OpenCode
agents = {
"default": Agent(provider=ClaudeCode(model="opus", effort="high",
permission="auto")),
"reviewer": Agent(provider=ClaudeCode(model="sonnet",
permission="read")),
"fast": Agent(provider=GeminiCLI(model="gemini-3-flash-preview",
permission="full")),
"oss": Agent(provider=OpenCode(model="opencode/kimi-k2.6",
permission="full")),
}
default_agent = "default"
queues = {
"review": {"agent": "reviewer", "max_parallel": 2},
"fast": {"agent": "fast", "max_parallel": 4},
}
Full reference: Configuration.
Headless + Telegram
aegis serve runs the SessionManager and MCP plane without the TUI; add
a Telegram token to drive it from your phone:
# .aegis.py
telegram_token = "…" # or set AEGIS_TELEGRAM_TOKEN
telegram_chat_id = 123456 # the single allowed chat
v0.10 ships a full substrate command surface alongside the original session-spawn verbs:
| Command | Action |
|---|---|
/new [agent] |
Spawn a new session |
/close [handle] |
Close a session |
/interrupt |
Interrupt the active turn |
/queue list |
Per-queue depth + in-flight (local) |
/queue show <name> |
Full queue detail (local) |
/schedule list [@peer] |
All schedules with next fire |
/schedule show <name> [@peer] |
Full schedule detail |
/schedule run <name> |
Force-fire a schedule now (local) |
/budget list [@peer] |
Budget state per queue |
/budget show <queue> [@peer] |
Per-constraint budget detail |
/peers |
Remotes with reachability probe |
/help [command] |
Registry-driven help |
A systemd unit template lives at scripts/aegis-serve.service.
Full setup, output examples, and FAQ: docs/telegram.md.
Docs
Full documentation: https://apiad.github.io/aegis/
- Install
- Usage
- Configuration
- Drivers — Claude / Gemini / OpenCode
- Queues — inter-agent delegation
- Canvas — shared markdown blackboard
- Terminals — shared live PTY
- Groups — broadcast-and-gather committees
- Remote plane — laptop ↔ VPS enqueue over HTTP
- Workflows — Python orchestration + catalog
- Budgets
- Telegram — substrate commands from the phone
- MCP plane — the tool surface
- Architecture
- Roadmap
- API reference
Status
Beta. Personal-infrastructure-grade, evolves fast. Expect change before 1.0. See the roadmap for what's next.
License
MIT — see LICENSE.
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Filter files by name, interpreter, ABI, and platform.
If you're not sure about the file name format, learn more about wheel file names.
Copy a direct link to the current filters
File details
Details for the file aegis_harness-0.10.0.tar.gz.
File metadata
- Download URL: aegis_harness-0.10.0.tar.gz
- Upload date:
- Size: 168.2 kB
- Tags: Source
- Uploaded using Trusted Publishing? Yes
- Uploaded via: uv/0.11.16 {"installer":{"name":"uv","version":"0.11.16","subcommand":["publish"]},"python":null,"implementation":{"name":null,"version":null},"distro":{"name":"Ubuntu","version":"24.04","id":"noble","libc":null},"system":{"name":null,"release":null},"cpu":null,"openssl_version":null,"setuptools_version":null,"rustc_version":null,"ci":true}
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
1975e1fd42f688df1dd023b684ee3ed1ff04a89054af4b443a1a429cb59c01b8
|
|
| MD5 |
c15b797fa360f444e0f3afe49897a2dd
|
|
| BLAKE2b-256 |
78e7fbdfd7e0b5c942a5389ed995c22f55cf6257cfb9e71b9981f7872c70d84f
|
File details
Details for the file aegis_harness-0.10.0-py3-none-any.whl.
File metadata
- Download URL: aegis_harness-0.10.0-py3-none-any.whl
- Upload date:
- Size: 203.4 kB
- Tags: Python 3
- Uploaded using Trusted Publishing? Yes
- Uploaded via: uv/0.11.16 {"installer":{"name":"uv","version":"0.11.16","subcommand":["publish"]},"python":null,"implementation":{"name":null,"version":null},"distro":{"name":"Ubuntu","version":"24.04","id":"noble","libc":null},"system":{"name":null,"release":null},"cpu":null,"openssl_version":null,"setuptools_version":null,"rustc_version":null,"ci":true}
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
7555b2cb69e71f836499bb2c0939bf29ef554815a80df19a734963b5cc7cead3
|
|
| MD5 |
327f32c6e9fcea4feb737aac09c7fcf7
|
|
| BLAKE2b-256 |
d2cad33f8b2a20eb1d685f14ff4c5bd178b77fe26522c650340f328e0e8df53f
|