supervisors.rs

A composable multi-agent framework powered by Rust and Python.

Build, orchestrate, and scale intelligent agents with a high-performance Rust core and a flexible Python API.

Getting Started | Agent Types | Examples | Extensions | API Reference | Contributing

Overview

supervisors.rs is a cross-language agent framework designed for building production-grade AI agent systems. The core is written in Rust for performance and safety, while the Python API provides ergonomic abstractions for rapid development.

The framework supports four fundamental agent patterns that can be used independently or composed together to solve complex, real-world problems:

All four patterns are composable -- a SupervisorsAgent can manage MultiAgent teams as sub-agents, a LoopAgent can embed pipeline logic, and any agent type can be registered on the same Supervisors instance.

Getting Started

Installation

Build and install with maturin:

pip install maturin
maturin develop          # editable install for development

Quick Start

from supervisors import Agent, Supervisor, Message

class GreeterAgent(Agent):
    def handle_message(self, msg: Message) -> None:
        print(f"Hello from '{self.name}'! You said: {msg.content}")

sup = Supervisor()
greeter = GreeterAgent("greeter")
greeter.register(sup)

sup.send(Message("user", "greeter", "Hi there!"))
sup.run_once()
# Output: Hello from 'greeter'! You said: Hi there!

Agent Types

Loop Agent

The LoopAgent runs an iterative reasoning loop over each incoming message. Override the step() method to define per-iteration logic.

from supervisors import LoopAgent, Message

class ReasoningAgent(LoopAgent):
    def step(self, state):
        state["count"] = state.get("count", 0) + 1
        if state["count"] >= 3:
            state["done"] = True
            state["result"] = "Reasoning complete."
        return state

agent = ReasoningAgent("reasoner", max_iterations=10)
state = agent.run_loop(Message("user", "reasoner", "Think about X"))
print(state["result"])

Pipeline Agent

The PipelineAgent processes messages through an ordered sequence of stage functions.

from supervisors import PipelineAgent, Message

def parse(ctx):
    ctx["tokens"] = ctx["input"].split()
    return ctx

def analyse(ctx):
    ctx["count"] = len(ctx["tokens"])
    return ctx

agent = PipelineAgent("analyser", stages=[parse, analyse])
result = agent.run_pipeline(Message("user", "analyser", "hello world"))
print(result["count"])  # 2

Supervisor + SubAgents

The SupervisorAgent routes incoming messages to specialised sub-agents based on a routing function.

from supervisors import Agent, SupervisorAgent, Message, Supervisor

class Worker(Agent):
    def handle_message(self, msg):
        print(f"[{self.name}] handling: {msg.content}")

sup = Supervisor()
manager = SupervisorAgent(
    "manager",
    router=lambda msg: "worker_a" if "urgent" in msg.content else "worker_b",
)
manager.add_sub_agent(Worker("worker_a"))
manager.add_sub_agent(Worker("worker_b"))
manager.register(sup)

sup.send(Message("user", "manager", "urgent: fix production"))
sup.run_once()
# Output: [worker_a] handling: urgent: fix production

Multi-Agent

The MultiAgent coordinates a group of peer agents that collaborate to solve problems.

from supervisors import Agent, MultiAgent, Message, Supervisor

class Researcher(Agent):
    def handle_message(self, msg):
        print(f"[{self.name}] researching: {msg.content}")

sup = Supervisor()
team = MultiAgent(
    "research_team",
    members=[Researcher("alice"), Researcher("bob")],
    max_rounds=5,
)
team.register(sup)

sup.send(Message("user", "research_team", "Investigate topic X"))
sup.run_once()

Composing Agents

The real power emerges when you combine patterns. For example, a SupervisorAgent can manage MultiAgent teams as sub-agents:

from supervisors import (
    Agent, MultiAgent, SupervisorAgent, Message, Supervisor,
)

class Specialist(Agent):
    def handle_message(self, msg):
        print(f"[{self.name}] working on: {msg.content}")

# Build a collaborative team.
team = MultiAgent("dev_team", members=[
    Specialist("frontend"),
    Specialist("backend"),
])

# Wrap it in a hierarchical supervisor.
manager = SupervisorAgent("manager", router=lambda msg: "dev_team")
manager.add_sub_agent(team)

sup = Supervisor()
manager.register(sup)
sup.send(Message("cto", "manager", "Build feature X"))
sup.run_once()

Examples

Each example is a self-contained project in the examples/ directory:

Run any example:

cd examples/loop
python main.py

Extensions

Extensions add capabilities to any agent via the plugin system. Load extensions with agent.use(extension):

RAG (Retrieval-Augmented Generation)

from supervisors.ext.rag import RAGExtension

class MyRAG(RAGExtension):
    def __init__(self):
        super().__init__(auto_retrieve=True, top_k=5)
        self._store = []

    def retrieve(self, query, top_k=None):
        return [d for d in self._store if query.lower() in d.lower()]

    def add_documents(self, docs, **kwargs):
        self._store.extend(docs)

agent.use(MyRAG())

Function Calling

from supervisors.ext.function_calling import FunctionCallingExtension

fc = FunctionCallingExtension()

@fc.tool(description="Add two numbers")
def add(a: int, b: int) -> int:
    return a + b

agent.use(fc)
result = fc.call_tool("add", a=1, b=2)  # returns 3

MCP (Model Context Protocol)

from supervisors.ext.mcp import MCPExtension

mcp = MCPExtension(server_url="http://localhost:8080")

@mcp.mcp_tool(description="Reverse a string")
def reverse(text: str) -> str:
    return text[::-1]

agent.use(mcp)

Skills

from supervisors.ext.skills import SkillsExtension

skills = SkillsExtension()

@skills.skill
def summarise(agent, msg):
    return f"Summary: {msg.content[:50]}..."

agent.use(skills)

A2A (Agent-to-Agent)

Beyond the built-in A2A messaging (agent.send()), the A2A extension adds broadcast, discovery, and request/reply patterns:

from supervisors.ext.a2a import A2AExtension

a2a = A2AExtension()
agent.use(a2a)

a2a.broadcast(agent, "Hello everyone!")
names = a2a.discover_agents(agent)

API Reference

Core Types

Supervisor Methods

Agent Methods

Architecture

+-------------------------------------------+
|              Python Layer                 |
|                                           |
|  LoopAgent   PipelineAgent               |
|  SupervisorAgent   MultiAgent            |
|  Agent + Extensions (RAG, MCP, ...)      |
+-------------------------------------------+
|            Rust Core (PyO3)               |
|                                           |
|  Message    Supervisor    ToolRegistry    |
|  (tokio async runtime for concurrency)   |
+-------------------------------------------+

The Rust core provides:

• Message routing with fault-tolerant delivery
• Async dispatch via tokio for concurrent agent execution
• Tool registry for high-performance tool specification storage

The Python layer provides:

• Four composable agent patterns for any business scenario
• Extension plugin system for modular capabilities
• Rich lifecycle hooks for logging, metrics, and customisation

Contributing

Contributions are welcome! Please read the contributing guidelines before submitting a pull request.

# Development setup
git clone https://github.com/HsiangNianian/supervisors.rs.git
cd supervisors.rs
python -m venv .venv && source .venv/bin/activate
pip install maturin pytest
maturin develop --skip-install
pip install -e . --no-build-isolation

# Run tests
python -m pytest tests/ -v

License

AGPLv3