Tower

An air-traffic-control-inspired coordination layer for multi-agent LLM systems that perform irreversible actions (charging a card, sending an email, writing to one customer's record). Instead of a generic lock, agents file a flight plan declaring intent before acting; Tower clears it immediately or holds it if another agent is already mid-action on the same resource. Handoffs between agents are verified with a mandatory readback step, so silently-dropped context is caught before it causes a wrong action.

For the design rationale, the data behind the problem, how this differs from related research, and benchmark results, see docs/DESIGN.md.

Features

• Flight plans -- declare intent on a resource before acting; get cleared or held, never a silent race.
• Handoff + readback verification -- the receiving agent restates a handoff's key facts; Tower checks it against the source before anything proceeds.
• Emergency priority -- a plan can preempt the normal queue and flag the audit log for immediate human attention.
• Two backends, one interface -- an embedded (in-process, zero infrastructure) backend and a distributed (FastAPI + Postgres) backend for agents spread across processes or machines.
• LangGraph adapter -- wrap a node with @tower_guarded and it files, awaits, and reports its own lifecycle automatically.
• Black box -- every filing, clearance, hold, handoff, and readback verdict is recorded, independent of any single agent run.

Requirements

• Python 3.12+
• Docker, only if you want the distributed (Postgres-backed) backend

Install

pip install tower-protocol                    # core protocol + embedded backend
pip install tower-protocol[langgraph]         # + the LangGraph adapter
pip install tower-protocol[distributed]       # + the FastAPI/Postgres backend
pip install tower-protocol[all]               # everything

From source, using uv:

git clone https://github.com/YamenHassan17/tower-protocol
cd tower-protocol
uv sync --all-extras

Quickstart -- embedded backend (no infrastructure needed)

import asyncio
from tower import EmbeddedTowerBackend, FlightPlan

async def main():
    tower = EmbeddedTowerBackend()
    plan = await tower.file_plan(
        FlightPlan(agent_id="agent-a", resource_key="stripe:charge:cust_1", action="charge_card")
    )
    print(plan.status)  # "cleared" -- nothing else is using this resource

asyncio.run(main())

Quickstart -- LangGraph

from tower.langgraph import tower_guarded

@tower_guarded(tower, agent_id="billing-agent", action="charge_card", resource_key="stripe:charge:cust_1")
async def charge_customer(state: dict) -> dict:
    ...  # your node's real logic -- unchanged, Tower wraps around it

See examples/research_scenario.py for a full runnable graph: two agents contend for a shared resource, and a deliberately incomplete handoff gets caught by readback verification.

uv run python examples/research_scenario.py

Running the distributed backend

docker compose up -d postgres
uv run uvicorn "tower.distributed.app:create_app" --factory --reload

Set TOWER_DATABASE_URL if you're not using the default docker-compose.yml Postgres (postgresql+asyncpg://tower:tower@localhost:5433/tower).

Development

docker compose up -d postgres   # required for the distributed backend tests
uv sync --all-extras
uv run ruff check .             # lint
uv run mypy                     # type check
uv run pytest -q                # full test suite

Regenerate the benchmark charts in docs/assets/ with:

uv run python benchmarks/run_benchmarks.py

Project layout

src/tower/
  core/         backend-agnostic protocol: FlightPlan, HandoffPacket, conflict
                detection, readback verification, the AuditLog interface
  backends/     embedded (in-process, asyncio.Lock) backend
  distributed/  FastAPI service + Postgres-backed store          [extra: distributed]
  langgraph/    tower_guarded() node decorator, send_handoff()/verify_handoff() [extra: langgraph]
examples/       a full runnable LangGraph scenario
benchmarks/     scripts that measure the real system and produce docs/assets/*.png
tests/          the test suite described above
docs/           design rationale, related work, results

Contributing

Issues and PRs welcome. Please run ruff check ., mypy, and pytest -q before submitting.

License

MIT © Yamen AlBarghouthi