Coordination

Queue/event agent communication layer for multi-agent systems.

Overview

Coordination provides the messaging and task infrastructure for agent-to-agent communication in multi-agent systems. It runs as an async FastAPI server with SQLite-backed persistence, supporting:

• Task queue — priority-ordered task dispatch with claim/dead-letter/work-steal semantics
• Event stream — publish/subscribe event bus for inter-agent signals
• Agent registry — service discovery for agent capabilities and routing
• Pre-execution memory — strategic context recall before task assignment

The package is lightweight by design (~300 KB, ~1,800 LOC across 10 source files). It uses FastAPI+Uvicorn for the async server, Pydantic for message validation, and aiosqlite for persistence with zero external infrastructure.

Components

Installation

pip install coordination

Or from source:

git clone <repo-url>
cd autonomous-ventures/coordination
pip install -e .

Usage

Start the coordination server:

coordination
# or
python -m coordination

With custom config:

COORDINATION_HOST=0.0.0.0 COORDINATION_PORT=8080 coordination

The server serves:

• POST /tasks — enqueue a new task
• GET /tasks/claim/{agent_id} — claim next available task for an agent
• POST /tasks/{task_id}/complete — mark task as done
• POST /tasks/{task_id}/fail — mark task as failed
• POST /events — publish an event to the event bus
• GET /events/{agent_id} — poll events for an agent
• GET /agents — list registered agents
• GET /health — health check endpoint

Python API

Task Queue

from coordination.task_queue import TaskQueue
from coordination.message import Task

queue = TaskQueue(storage=...)

# Enqueue a task with priority (0 = highest)
task = Task(queue="default", payload={"action": "research"}, priority=5)
await queue.enqueue(task)

# Claim the next available task for an agent
claimed = await queue.claim("agent-1")

Event Stream

from coordination.event_stream import EventStream

stream = EventStream(storage=...)

# Subscribe to event types
await stream.subscribe("agent-1", ["task.assigned", "agent.updated"])

# Publish an event (delivered to all matching subscribers)
await stream.publish({
    "type": "task.assigned",
    "source": "coordinator",
    "payload": {"task_id": "...", "assignee": "agent-1"}
})

# Poll for new events
events = await stream.poll("agent-1")

Agent Registry

from coordination.agent_registry import AgentRegistry

registry = AgentRegistry(storage=...)

# Register an agent with capabilities
await registry.register("agent-1", capabilities=["research", "summarize"])

# Find agents by capability
agents = await registry.find_by_capability("research")

Configuration

Configure via environment variables:

Architecture

┌──────────────────────────────────────────────┐
│              Coordination Server              │
│  ┌──────────┐ ┌──────────┐ ┌──────────────┐  │
│  │Task Queue│ │   Event  │ │Agent Registry│  │
│  │Priority  │ │   Stream │ │Capability    │  │
│  │+ Claim   │ │ Pub/Sub  │ │Discovery     │  │
│  └──────────┘ └──────────┘ └──────────────┘  │
│  ┌────────────────────────────────────────┐   │
│  │        SQLite Storage (aiosqlite)      │   │
│  └────────────────────────────────────────┘   │
│  ┌────────────────────────────────────────┐   │
│  │  Pre-Execution Memory (strategic ctx)  │   │
│  └────────────────────────────────────────┘   │
└──────────────────────────────────────────────┘

Development

pip install -e ".[dev]"
pytest tests/ -v

All 103 tests pass across task queue, event stream, agent registry, SQLite storage, and integration tests.

License

MIT — see LICENSE.

Maintainer

Published by Autonomous Ventures. Initial release: v0.1.0 (May 2026). Beta — Development Status :: 4 - Beta.