Modular Python Agent Framework
A lightweight, asyncio-based library for research and development of multi-agent systems
PyPI | Read the Docs | GitHub | mango@offis.de
mango (modular python agent framework) is a Python library for building and simulating multi-agent systems (MAS) on top of asyncio. It targets researchers and engineers who need reproducible agent behaviour, structured agent architectures, and controlled experimental conditions — without sacrificing usability for prototyping.
| Feature | Description |
|---|---|
| Container-based messaging | Agents are co-located in containers that short-circuit local message exchange without serialisation overhead |
| FIPA ACL message format | Standardised message envelope with built-in JSON and protobuf codecs |
| Role decomposition | Complex agent behaviour is factored into loosely coupled, independently testable role objects |
| TCP and MQTT transport | Agents communicate across process and machine boundaries via TCP sockets or an external MQTT broker |
| Discrete-event simulation | A clock-driven SimulationWorld provides deterministic, reproducible runs with configurable communication delay and loss, spatial environments, and structured data recording |
pip install mango-agentsRequires Python 3.10 or later.
Every agent is a subclass of Agent. Incoming messages are handled by handle_message; two lifecycle hooks signal when the agent has joined its container (on_register) and when all containers are active and external messaging is safe (on_ready).
The following complete script creates two agents, starts them in a shared container, sends a message from one to the other, and shuts everything down cleanly:
import asyncio
from mango import Agent, create_tcp_container, activate
class ReportingAgent(Agent):
def on_ready(self):
print(f"{self.aid}: ready")
def handle_message(self, content, meta):
print(f"{self.aid}: received '{content}' from {meta['sender_id']}")
async def main():
# All agents on a single host share one container.
container = create_tcp_container(addr=("127.0.0.1", 5555))
sender = container.register(ReportingAgent())
receiver = container.register(ReportingAgent())
async with activate(container):
await sender.send_message("Hello!", receiver.addr)
await asyncio.sleep(0.1) # allow time for message processing
asyncio.run(main())
# agent0: ready
# agent1: ready
# agent1: received 'Hello!' from agent0activate starts the container and guarantees a clean shutdown when its block exits, even if an exception is raised.
Agents are not limited to reacting to messages. schedule_periodic_task registers a coroutine that is called repeatedly at a fixed interval — useful for polling, broadcasting, or any time-driven behavior:
class SensorAgent(Agent):
def __init__(self, target_addr):
super().__init__()
self.target_addr = target_addr
def on_ready(self):
# Broadcast a reading every 5 seconds of (simulated) time.
self.schedule_periodic_task(self._broadcast, delay=5.0)
async def _broadcast(self):
await self.send_message("temperature: 22.5 C", self.target_addr)Running agents over a real network introduces timing variability that makes experiments hard to reproduce. The SimulationWorld eliminates this by replacing the network layer with a controlled, clock-driven execution model: time advances only when the simulation explicitly steps forward, there are no open sockets, and every run with the same inputs produces the same outputs.
┌──────────────────────────────────────────────────────┐
│ SimulationWorld │
│ │
│ ExternalClock ──► step_simulation() │
│ | │
│ ┌─────────────────┼─────────────────┐ │
│ v v v │
│ on_step() message delivery environment │
│ (all agents) (delay / loss model) .step() │
└──────────────────────────────────────────────────────┘
Each step proceeds in order:
- All agents receive
on_step(env, clock, step_size_s). - The environment behavior advances by
step_size_s. - The internal clock is set to the new time, waking any sleeping scheduled tasks.
- A convergence loop delivers all messages whose simulated arrival time has passed and waits for agents to settle between deliveries.
The following self-contained script runs two agents for 60 simulated seconds and records how many readings the monitor received over time:
import asyncio
from mango import Agent, create_world, discrete_step_until, record_agent
class SensorAgent(Agent):
"""Periodically sends a reading to a monitor agent."""
def __init__(self):
super().__init__()
self.monitor_addr = None # set after both agents are registered
self.readings_sent = 0
def on_ready(self):
self.schedule_periodic_task(self._broadcast, delay=10.0)
async def _broadcast(self):
await self.send_message(f"reading {self.readings_sent}", self.monitor_addr)
self.readings_sent += 1
class MonitorAgent(Agent):
"""Counts incoming readings."""
def __init__(self):
super().__init__()
self.received = 0
def handle_message(self, content, meta):
self.received += 1
print(f" t={self.current_timestamp:.0f} s monitor received: {content}")
async def run():
world = create_world(start_time=0.0)
sensor = world.register(SensorAgent())
monitor = world.register(MonitorAgent())
# Set the target address before on_ready is called (inside async with).
sensor.monitor_addr = monitor.addr
# Record the running count of received messages after every step.
record_agent(world, "received", lambda a: a.received,
filter_fn=lambda a: isinstance(a, MonitorAgent))
async with world:
await discrete_step_until(world, max_advance_time_s=60.0)
print(f"\nTotal readings received: {monitor.received}")
# world.data_agent_collections["received"].timeseries holds the full time series.
asyncio.run(run())
# t=10 s monitor received: reading 0
# t=20 s monitor received: reading 1
# t=30 s monitor received: reading 2
# t=40 s monitor received: reading 3
# t=50 s monitor received: reading 4
# t=60 s monitor received: reading 5
#
# Total readings received: 6discrete_step_until automatically determines each step size as the time until the next scheduled event — a message arrival or a task wakeup — and stops when no further events remain within the time budget.
For experiment designs that require fixed, uniform time increments, call step_simulation directly:
from mango import step_simulation
async with world:
for _ in range(10):
result = await step_simulation(world, step_size_s=1.0)
print(f"t = {world.clock.time:.1f} s messages delivered: {result.messages_delivered}")By default, messages are delivered instantly with no loss. SimpleCommunicationSimulation adds configurable one-way latency and independent packet loss, with optional per-link overrides:
from mango import create_world, SimpleCommunicationSimulation
world = create_world(
start_time=0.0,
communication_sim=SimpleCommunicationSimulation(
default_delay_s=0.1, # baseline one-way latency
loss_percent=0.01, # independent loss probability per message
delay_s_directed_edge_dict={
("agent0", "agent1"): 0.5, # directed per-link override
},
),
)For delays drawn from a distribution at runtime, use DelayProviderCommunicationSimulation:
import random
from mango import create_world, DelayProviderCommunicationSimulation
world = create_world(
start_time=0.0,
communication_sim=DelayProviderCommunicationSimulation(
default_delay_s_provider=lambda: random.gauss(0.1, 0.01),
),
)Custom communication models are implemented by subclassing CommunicationSimulation and overriding calculate_communication.
Agents can be embedded in a continuous 2-D area. Agents without a pre-assigned position are placed uniformly at random within the area during initialisation, and their positions can be recorded as a time series:
from mango import (
create_world, Area2D, DefaultEnvironment,
record_position, position_history, discrete_step_until,
)
env = DefaultEnvironment(space=Area2D(width=100.0, height=100.0))
world = create_world(start_time=0.0, environment=env)
# ... register agents ...
record_position(world)
async with world:
await discrete_step_until(world, max_advance_time_s=10.0)
history = position_history(world)
# history.timeseries["agent0"] -> [Position2D(x=..., y=...), ...]
# history.time -> [0.0, 5.0, 10.0, ...]Custom spaces and environment behaviors are implemented by subclassing Space and Behavior respectively.
record_world and record_agent capture any scalar as a time series after every simulation step, without modifying agent code. Results are plain Python lists compatible with numpy, pandas, and matplotlib:
from mango import create_world, record_world, record_agent
world = create_world(start_time=0.0)
# ... register agents ...
# World-level: total messages delivered so far
record_world(world, "msg_count", lambda: len(world.recorded_messages))
# Per-agent: an arbitrary attribute
record_agent(world, "energy_kwh", lambda a: a.energy_kwh)
async with world:
await discrete_step_until(world, max_advance_time_s=3600.0)
world.data_collections["msg_count"].timeseries # list of scalars
world.data_agent_collections["energy_kwh"].timeseries # dict[aid -> list]
world.data_agent_collections["energy_kwh"].time # shared time axisAll exchanged messages are also logged automatically in world.recorded_messages as MessageTransaction objects, each carrying sender, receiver, send time, and arrival time.
Any Agent or Role subclass may override these methods to participate in the simulation lifecycle:
class MyAgent(Agent):
def on_step(self, env, clock, step_size_s: float):
"""Invoked at the start of every simulation step, before the clock advances."""
def on_global_event(self, event):
"""Invoked when the environment broadcasts an event to all agents."""
def on_agent_event(self, event):
"""Invoked when the environment targets this specific agent."""| Resource | Link |
|---|---|
| Full documentation | mango-agents.readthedocs.io |
| Issue tracker | github.com/OFFIS-DAI/mango/issues |
| Contact | mango@offis.de |
Distributed under the MIT License.