Agentic Bus

Reference implementation of the Liquid Interfaces Protocol
A dynamic, negotiation-driven, multi-agent coordination runtime where interfaces are not static contracts — they are ephemeral relational events.

Quick Start • Architecture • Key Concepts • Dashboard UI • REST API • CLI Reference • Testing • Contributing • License

🧭 Overview

Agentic Bus introduces a coordination paradigm in which interfaces are not persistent technical artifacts, but ephemeral relational events that emerge through intention articulation and semantic negotiation at runtime.

Instead of pre-wired API contracts, a requesting agent simply states its intent in natural language — e.g., "deliver this container within 200 km of the closed port, optimizing for cost and time" — and the runtime discovers capable agents, negotiates terms, composes an execution graph, and dissolves everything once the task is complete, leaving zero technical debt.

📄 Read the full paper: lip.md — "Liquid Interfaces: A Dynamic Ontology for the Interoperability of Autonomous Systems"

✨ Key Concepts

How is this different?

🏗️ Architecture

                          ┌──────────────┐
                          │   Dashboard  │  (Next.js 16 — ui/)
                          │   React UI   │
                          └──────┬───────┘
                                 │ REST
                                 ▼
Requester ──WebSocket──► Coordinator ──WebSocket──► Provider Agents
                              │  │
                         ┌────┘  └────┐
                    ┌────┴────┐  Admin REST API
                    │ LangGraph│  (FastAPI :8766)
                    └────┬────┘
                         │
               IBAC ◄────┤────► Registry
                         │
                    Telemetry (OTel)
                         │
                    Persistence (SQLAlchemy)

The Coordinator implements the full Agentic Bus session lifecycle:

1. Accept & authenticate WebSocket connections (OIDC)
2. Open intent sessions
3. Discover eligible agents via semantic adjudication
4. Request offers from matching agents
5. Evaluate offers through IBAC governance
6. Negotiate & compose offers into an execution plan
7. Build a LangGraph dynamically
8. Supervise execution with failure handling
9. Dissolve the session — all artifacts are ephemeral

Project Layout

agentic-bus/
│
├── docker-compose.yml          # Full stack: Ollama + coordinator + dashboard
├── Dockerfile                  # Coordinator image
├── docker/                     # Container entrypoint
├── schemas/                    # Generated LIP JSON Schemas (see CONTRIBUTING)
│
├── agentic_bus/                # Python package
│   ├── cli.py                  # CLI entry point (agbus command)
│   ├── core/                   # Shared infrastructure
│   │   ├── protocol/           #   Message model & envelope
│   │   ├── transport/          #   WebSocket server / client
│   │   ├── session/            #   Session lifecycle management
│   │   ├── registry/           #   Dynamic capability registry
│   │   ├── ibac/               #   Intention-Based Access Control engine
│   │   ├── telemetry/          #   OpenTelemetry instrumentation
│   │   ├── auth/               #   OIDC authentication & admin auth
│   │   ├── llm/                #   Multi-provider LLM factory
│   │   └── persistence/        #   SQLAlchemy models & repositories
│   │       ├── models.py       #     DB models (agents, tenants, users, IBAC rules, LLM configs)
│   │       ├── repository.py   #     Agent repository
│   │       ├── tenant_repository.py
│   │       ├── user_repository.py
│   │       ├── ibac_repository.py
│   │       ├── llm_repository.py
│   │       └── managed_agent_repository.py
│   │
│   ├── coordinator/            # Coordination runtime
│   │   ├── server.py           #   Server entry point (WS + REST)
│   │   ├── runtime.py          #   Core coordinator runtime
│   │   ├── intent/             #   Intent admission & decomposition
│   │   ├── negotiation/        #   Offer collection, scoring, composition
│   │   ├── graph/              #   Dynamic LangGraph synthesis
│   │   ├── execution/          #   Supervised execution & failure handling
│   │   └── admin/              #   Admin REST API (FastAPI)
│   │       ├── api.py          #     All REST endpoints
│   │       ├── service.py      #     Business logic
│   │       ├── schemas.py      #     Pydantic DTOs
│   │       ├── serializers.py  #     Model → DTO serializers
│   │       └── audit.py        #     Audit logging
│   │
│   └── agents/                 # Agent SDK & examples
│       ├── base/               #   Base agent framework
│       ├── factory.py          #   Agent factory (CrewAI integration)
│       ├── managed_server.py   #   Managed agent server
│       ├── requester.py        #   Intent requester client
│       └── examples/           #   Sample provider agents
│           ├── logistics_agent/
│           └── intent_client_example.py
│
├── ui/                         # Admin Dashboard (Next.js 16)
│   ├── src/
│   │   ├── app/                #   App Router pages
│   │   │   ├── page.tsx        #     Dashboard home (stats overview)
│   │   │   ├── agents/         #     Agent management (persistent & managed)
│   │   │   ├── intent/         #     Intent session inspector
│   │   │   ├── ibac/           #     IBAC rule management
│   │   │   ├── audit/          #     Audit log viewer
│   │   │   ├── tenants/        #     Multi-tenant management
│   │   │   ├── users/          #     User administration
│   │   │   └── settings/       #     Coordinator & LLM settings
│   │   ├── components/         #   Reusable UI components (shadcn/ui)
│   │   ├── hooks/              #   Custom React hooks
│   │   │   ├── use-async.ts    #     Async data fetching
│   │   │   └── use-intent-ws.ts#     WebSocket intent streaming
│   │   └── lib/                #   Shared utilities
│   │       ├── api.ts          #     REST API client
│   │       ├── protocol.ts     #     Protocol type definitions
│   │       └── types.ts        #     TypeScript types
│   └── package.json
│
└── tests/                      # Test suite (22 modules, 411 tests)
    ├── test_admin.py
    ├── test_auth.py
    ├── test_cli.py
    ├── test_graph.py
    ├── test_ibac.py
    ├── test_ibac_rules.py
    ├── test_intent_client.py
    ├── test_llm_config.py
    ├── test_llm_factory.py
    ├── test_managed_agents.py
    ├── test_negotiation.py
    ├── test_persistence.py
    ├── test_protocol.py
    ├── test_registry.py
    ├── test_session.py
    ├── test_telemetry.py
    └── test_tenants_users.py

🚀 Quick Start

Try it with no API keys

git clone https://github.com/draiven-io/agentic-bus.git && cd agentic-bus && docker compose up

That brings up a local model (Ollama), the coordinator with the paper's four logistics agents already seeded and running, and the dashboard:

Open the dashboard, go to Intent, and submit something like "a storm has closed the port — find me an alternative route and tell me what it costs". You'll watch discovery, negotiation, plan approval, execution and dissolution happen live.

On the local model. The compose stack defaults to qwen2.5:3b so the first run is a ~2 GB download rather than a signup. It is enough to watch the full lifecycle, but negotiation quality scales with the model. For results worth judging the paradigm on, point the coordinator at a hosted model — set AGBUS_BOOTSTRAP_LLM_PROVIDER, AGBUS_BOOTSTRAP_LLM_MODEL and AGBUS_BOOTSTRAP_LLM_API_KEY in docker-compose.yml, or pick a larger local one with AGBUS_DEMO_MODEL=qwen2.5:14b docker compose up.

Write an agent

pip install agentic-bus

That is a small install — pydantic, websockets and OpenTelemetry — because writing an agent should not require a web framework, an ORM and an LLM stack. An agent is two methods:

from agentic_bus import AgentCapability, BaseAgent


class WeatherAgent(BaseAgent):
    def capabilities(self):
        return [AgentCapability(
            capability_id="forecast",
            description="Weather forecast for a city",
        )]

    async def execute_task(self, payload, context):
        return {"forecast": "sunny"}


WeatherAgent(agent_id="weather-01").run_forever()

It connects to a coordinator (AGBUS_COORDINATOR_URI, default ws://localhost:8765), registers its capabilities, and from then on participates in discovery, negotiation, IBAC governance and execution. You never write an endpoint, a schema or a route.

The runtime handles the parts that bite in production:

• Reconnects with exponential backoff and jitter, and re-registers on every reconnect — a coordinator restart doesn't leave the agent silently orphaned.
• Runs tasks concurrently (bounded by max_concurrent_tasks, default 8), so one slow task doesn't stop the agent answering anything else.
• Cancels in-flight work on dissolve, so execute_task receives CancelledError and can clean up — the protocol's ephemerality guarantee is actually enforced, not just documented.

For a coordinator with real OIDC, supply a token provider. It's called on every reconnect, so short-lived tokens refresh rather than going stale:

WeatherAgent(
    agent_id="weather-01",
    token_provider=lambda: my_oidc_client.access_token(),   # may be async
)

Submitting an intent is the other half:

from agentic_bus import submit_intent

result = await submit_intent("what's the weather in Lisbon?")

Test it without any of the infrastructure

agentic_bus.testing ships a stand-in coordinator, so testing an agent needs no Docker, no model provider and no network:

from agentic_bus.testing import LocalBus

async def test_forecast():
    async with LocalBus() as bus:
        agent = await bus.add_agent(WeatherAgent(agent_id="weather-01"))

        result = await bus.execute(agent.agent_id, {"city": "Lisbon"})

        assert result.status == "success"
        assert result.artifacts[0]["forecast"] == "sunny"

It speaks LIP over a real socket rather than calling your handlers directly, so serialisation, the receive loop and concurrency all take part — faking those out is what lets connection-level bugs survive a green suite. You can also drive intents (send_intent), tear sessions down (dissolve), inspect the transcript (messages, events), check the token your agent sent (auth_headers), and simulate a coordinator that refuses registration or predates LIP 0.2.0.

It is not a coordinator: discovery and negotiation are LLM-driven in the real runtime and are not reproduced, so tests stay deterministic. Use it to check what your agent does, not how a coordinator would choose it.

Run a coordinator

The coordinator is a much heavier thing — LangGraph, FastAPI, SQLAlchemy, the LLM providers — so it lives behind an extra:

pip install "agentic-bus[server]"

agbus install && agbus serve

agbus install is an interactive wizard: it writes a .env for the server and database settings and stores your LLM provider in the database. To skip the wizard, see Configuration below.

Install matrix

Commands that need an extra you don't have say so, and name the extra.

Develop against a checkout

git clone https://github.com/draiven-io/agentic-bus.git
cd agentic-bus
pip install -e ".[dev]"

agbus serve

cd ui && npm install && npm run dev

python -m agentic_bus.agents.examples.logistics_agent.agent

⚠️ Note: run agbus from the directory containing your .env.

Configuration

Configuration is split in two, deliberately:

agbus install writes the .env and stores your first LLM provider in the database. Add or switch providers later without touching either by hand:

agbus llm add --name prod --provider anthropic --model claude-sonnet-4-20250514 --api-key sk-ant-... --activate

agbus llm list

The .env covers the runtime itself:

AGBUS_HOST=0.0.0.0
AGBUS_PORT=8765
AGBUS_DATABASE_URL=sqlite:///agbus_agents.db
AGBUS_AGENT_AUTO_APPROVE=false

Run agbus config show to display the resolved runtime configuration and the active LLM provider.

🖥️ Admin Dashboard (UI)

The Admin Dashboard is a full-featured Next.js 16 application that provides a visual management interface for the entire Agentic Bus runtime. Built with React 19, Tailwind CSS 4, shadcn/ui, and Recharts.

Pages

Running the UI

cd ui
npm install
npm run dev       # Development mode (http://localhost:3000)
npm run build     # Production build
npm run start     # Production server

🔌 Admin REST API

The coordinator exposes a FastAPI admin REST API on port 8766 (configurable via AGBUS_API_PORT). Interactive Swagger documentation is available at /api/docs, and an unauthenticated liveness probe at /health.

Endpoints

💻 CLI Reference

agbus install                          # Interactive setup wizard
agbus serve                            # Start the coordinator server

agbus db init                          # Create / migrate database tables

agbus agent list                       # List all registered agents
agbus agent show  <id>                 # Inspect a single agent
agbus agent approve <id>               # Approve a pending enrolment
agbus agent reject  <id>               # Reject a pending enrolment
agbus agent revoke  <id>               # Revoke an approved agent
agbus agent delete  <id>               # Permanently remove an agent
agbus agent create                     # Create a managed agent (interactive)
agbus agent activate <id>              # Activate a managed agent
agbus agent disable <id>               # Disable a managed agent
agbus agent add-capability <id>        # Add capability to a managed agent
agbus agent remove-capability <id> <c> # Remove a capability
agbus agent tools                      # List available CrewAI tools

agbus llm list                         # List LLM configurations
agbus llm show <name>                  # Inspect one configuration
agbus llm add                          # Add a provider configuration
agbus llm activate <name>              # Make a configuration current
agbus llm update <name>                # Update a configuration
agbus llm remove <name>                # Delete a configuration

agbus config show                      # Display resolved configuration
agbus config init                      # Write a starter .env file

agbus help                             # Comprehensive documentation
agbus help quickstart                  # Step-by-step setup guide

🧪 Testing

The project includes a comprehensive test suite — 411 tests across 22 modules — covering every subsystem.

The suite is hermetic: it ignores your .env, requires no API keys, and never touches the network. Each test gets a scrubbed environment and its own migrated database (see tests/conftest.py), so a green run on your machine means a green run in CI.

# Run all tests
pytest

# Run a specific test file
pytest tests/test_negotiation.py

# Run with verbose output
pytest -v

Test Coverage

🗺️ Roadmap

• Admin Dashboard (Web UI)
• Admin REST API (FastAPI)
• Multi-tenant & user management
• IBAC rule management
• Audit logging
• LLM configuration management
• Managed agent lifecycle (CrewAI integration)
• Distributed coordinator clustering
• Persistent session replay & auditing
• Agent marketplace & trust scoring
• Plugin system for custom negotiation strategies
• Multi-modal intent support (voice, image, structured data)

🤝 Contributing

Contributions are welcome! Whether it's bug reports, feature requests, documentation improvements, or code contributions — we'd love your help.

1. Fork the repository
2. Create a feature branch (git checkout -b feature/amazing-feature)
3. Commit your changes (git commit -m 'Add amazing feature')
4. Push to the branch (git push origin feature/amazing-feature)
5. Open a Pull Request

Please make sure all tests pass before submitting:

pip install -e ".[dev]"
pytest
ruff check .

📖 Citation

If you use Agentic Bus in your research, please cite:

@misc{desá2026liquidinterfacesdynamicontology,
      title={Liquid Interfaces: A Dynamic Ontology for the Interoperability of Autonomous Systems}, 
      author={Dhiogo de Sá and Carlos Schmiedel and Carlos Pereira Lopes},
      year={2026},
      eprint={2601.21993},
      archivePrefix={arXiv},
      primaryClass={cs.AI},
      url={https://arxiv.org/abs/2601.21993}, 
}

📄 License

This project is licensed under the MIT License — see the LICENSE file for details.

Made with ❤️ by Draiven