mango — modular python agent framework¶

asyncio-native framework for multi-agent systems in Python

mango provides containers, agents, role composition, scheduling, and a discrete-event simulation world — covering the full spectrum from small prototypes to large distributed deployments under a single, consistent API.

Get started

Tutorials

GitHub

pip install mango-agents

Features¶

Containers

Local, TCP, and MQTT transports; external-coupling container for co-simulation.

Agents

Reactive and proactive behaviour; full lifecycle callbacks (on_register, on_ready, on_stop).

Role system

Compose agent behaviour from small, reusable Role classes with shared state and event subscriptions.

Scheduling

Instant, periodic, timestamp, and conditional tasks in real-time (AsyncioClock) or simulation time (ExternalClock).

Simulation world

Discrete-event and fixed-step simulation; configurable message delay and loss; spatial environments and data recording.

Topologies

Distribute a networkx graph to agents so every agent knows its direct neighbours.

Codecs

Built-in JSON and protobuf serialisation; custom serialisers via add_serializer or the @json_serializable decorator.

FIPA ACL

Optional FIPA-compliant message wrapper for interoperability with other agent platforms.

Quick look¶

Agent

import asyncio
from mango import Agent, create_tcp_container, activate

class HelloAgent(Agent):
    def on_ready(self):
        self.schedule_instant_message("Hello, mango!", self.addr)

    def handle_message(self, content, meta):
        print(content)  # → Hello, mango!

async def main():
    c = create_tcp_container(addr=("127.0.0.1", 5555))
    c.register(HelloAgent())
    async with activate(c):
        await asyncio.sleep(0.05)

asyncio.run(main())

Roles

import asyncio
from mango import Role, agent_composed_of, run_with_tcp

class Ping:
    pass

class PingRole(Role):
    def setup(self):
        self.context.subscribe_message(
            self,
            self.handle_ping,
            lambda content, meta: isinstance(content, Ping),
        )

    def handle_ping(self, content, meta):
        print("Ping received!")

async def main():
    my_agent = agent_composed_of(PingRole())
    async with run_with_tcp(1, my_agent) as container:
        await container.send_message(Ping(), my_agent.addr)
        await asyncio.sleep(0.05)

asyncio.run(main())