aegis-gov: a small Python library for multi-agent task graphs and circuit breakers

Multi-agent LLM systems have a coordination problem that most tutorials skip past. You can string together a few asyncio.gather calls or a list of prompts, but once you need three or four agents to hand work to each other in a defined order — and you need the whole thing to degrade gracefully when one call fails — the scaffolding grows quickly and gets tangled with provider-specific SDK code.

I wrote aegis-gov to separate that coordination scaffolding from the business logic. It is a small Python library (one hard dependency: requests) that provides:

• A provider-agnostic adapter protocol for Anthropic, OpenAI, and Ollama
• A DAG task scheduler that resolves dependencies, runs independent tasks in parallel, and cascades skip signals when an upstream task fails
• A circuit-breaker pool that bounds concurrency and stops sending work to an endpoint that is already returning errors

This article walks through each of those pieces, shows you the actual code, and is honest about what is not there yet.

The problem in concrete terms

Suppose you have four agents: a researcher, a writer, a translator, and a publisher. The writer depends on the researcher. The translator and publisher both depend on the writer but can run in parallel. The publisher should not run at all if the writer failed.

Without a scheduler, you write this by hand every time. The failure-cascade logic especially tends to become a set of nested conditionals that grows with each new dependency edge. And if your LLM provider returns a string of 429s or 5xx errors, there is nothing to stop the loop from hammering the endpoint until you kill the process.

aegis-gov addresses both problems with two focused classes: TaskQueue and AgentPool.

Installation

pip install aegis-gov                      # requests only
pip install "aegis-gov[anthropic]"         # + Anthropic SDK
pip install "aegis-gov[openai]"            # + OpenAI SDK
pip install "aegis-gov[all]"               # both LLM SDKs

Python 3.10+ is required.

The adapter layer

The LLMAdapter protocol has two methods: generate() returns a string, stream() yields string chunks. All three concrete adapters satisfy this protocol:

from aegis_gov import AnthropicAdapter, OpenAIAdapter, OllamaAdapter

# Anthropic
adapter = AnthropicAdapter(model="claude-sonnet-4-6")

# OpenAI or any compatible endpoint (LM Studio, vLLM, Azure, etc.)
adapter = OpenAIAdapter(model="gpt-4o-mini", base_url="http://localhost:1234/v1")

# Ollama — no extra package needed, communicates over HTTP
adapter = OllamaAdapter(model="qwen2.5:14b")

The adapter is a field on AgentConfig, so switching providers for a single agent is a one-line change. The rest of your orchestration code does not need to know which provider is in use.

Running a single agent

from aegis_gov import OpenMultiAgent, AgentConfig, AnthropicAdapter

oma = OpenMultiAgent()
result = oma.run_agent(
    AgentConfig(
        name="analyst",
        system_prompt="You are a concise market analyst.",
        adapter=AnthropicAdapter(model="claude-haiku-4-5-20251001"),
    ),
    task="Top 3 open-source multi-agent frameworks in 2026?",
)
print(result)

The DAG task scheduler

TaskQueue takes a list of Task objects, validates the dependency graph for cycles at construction time (raises CyclicDependencyError if one is found), and exposes a ready() method that returns the tasks whose dependencies are all in done state.

from aegis_gov import OpenMultiAgent, Task
from aegis_gov import AgentConfig, AnthropicAdapter

team = OpenMultiAgent.create_team("pipeline", [
    AgentConfig(name="researcher",  system_prompt="Research topics thoroughly."),
    AgentConfig(name="writer",      system_prompt="Write clear reports."),
    AgentConfig(name="translator",  system_prompt="Translate to Japanese."),
    AgentConfig(name="publisher",   system_prompt="Format output as Markdown."),
])

tasks = [
    Task(id="research",   description="Research AI trends",        agent="researcher"),
    Task(id="draft",      description="Write a report draft",      agent="writer",     depends_on=["research"]),
    Task(id="translate",  description="Translate to Japanese",     agent="translator", depends_on=["draft"]),
    Task(id="publish",    description="Format as Markdown",        agent="publisher",  depends_on=["draft"]),
]

oma = OpenMultiAgent()
results = oma.run_tasks(tasks, team=team)

translate and publish both depend only on draft, so they execute in parallel once draft completes.

Cascade failure

When stop_on_failure=False (the default), only tasks that directly or transitively depend on a failed task are skipped. Independent branches continue:

from aegis_gov import TaskQueue, Task

q = TaskQueue([
    Task(id="fetch",   description="Fetch data"),
    Task(id="process", description="Process data",  depends_on=["fetch"]),
    Task(id="report",  description="Write report",  depends_on=["process"]),
], stop_on_failure=False)

q.complete("fetch", success=False)
print(q.skipped_tasks())  # ["process", "report"]
print(q.summary())        # {"failed": 1, "skipped": 2}

Setting stop_on_failure=True halts the entire queue on the first failure.

The circuit breaker

AgentPool wraps a threading.Semaphore to bound how many agents run simultaneously, and tracks consecutive failures to open the circuit:

from aegis_gov import AgentPool, OpenMultiAgent

pool = AgentPool(
    max_concurrent=4,
    consecutive_failure_limit=5,
    recovery_timeout_s=30.0,
)
oma = OpenMultiAgent(pool=pool)
print(oma.get_status())
# {"pool_state": "closed", "pool_consecutive_failures": 0, ...}

The state machine has three states: CLOSED (normal), OPEN (rejecting new work, raising CircuitOpenError), and HALF_OPEN (sending one probe call after recovery_timeout_s elapses). A successful probe returns the circuit to CLOSED; another failure reopens it.

The tool registry

Agents can be given callable tools:

from aegis_gov import ToolRegistry

registry = ToolRegistry()
# five built-ins are registered automatically:
# file_read, http_get, shell, memory_store, memory_retrieve

registry.define_tool(
    name="search_web",
    description="Search the web for recent information",
    fn=my_search_fn,
    schema={"type": "object", "properties": {"query": {"type": "string"}}, "required": ["query"]},
)

The built-in tools are thin wrappers. They are not hardened for production use — treat them as stubs you replace with your own implementations.

What it does not do

Being honest about scope matters:

• No async runtime. All execution is synchronous with threading.Semaphore and ThreadPoolExecutor. If you need asyncio-native agents, this is not the right library yet.
• No retry logic built in. The circuit breaker protects against cascading failures but does not automatically retry failed calls with backoff. You handle retries in your task logic or adapter.
• No streaming through the orchestrator. Individual adapters support stream(), but run_agent() and run_tasks() collect the full response before returning.
• No persistence. Task state lives in memory for the duration of a process. There is no checkpoint or resume capability.
• The built-in tools are minimal. shell, http_get, etc. are thin wrappers without sandboxing, rate limiting, or error enrichment.
• Alpha status. The public API may change before 1.0. Treat it accordingly.

Roadmap

Areas I plan to work on next, in rough priority order:

1. Async support (asyncio.Semaphore, async def generate())
2. Configurable retry with exponential backoff at the pool level
3. Streaming pass-through in run_agent()
4. Better observability hooks — callbacks on task state transitions
5. Solidifying the built-in tool implementations

Contributions and issue reports are welcome. The test suite uses pytest; see pyproject.toml for the dev extras.

Links

• GitHub: https://github.com/aegis-os/aegis-gov
• PyPI: https://pypi.org/project/aegis-gov/
• Changelog: https://github.com/aegis-os/aegis-gov/blob/main/CHANGELOG.md
• Install: pip install "aegis-gov[all]"