Agents API

Agent core

This module implements the base class for agents (Agent).

Every agent must live in a container. Containers are responsible for making

connections to other agents.

class mango.agent.core.Agent

Bases: ABC, AgentDelegates

Base class for all agents.

handle_message(content, meta: dict[str, Any])

Has to be implemented by the user. This method is called when a message is received at the agents inbox. :param content: The deserialized message object :param meta: Meta details of the message. In case of mqtt this dict includes at least the field ‘topic’

property observable_tasks

on_register()

Hook-in to define behavior of the agent directly after it got registered by a container

async on_stop()

Can be used as lifecycle callback when the agent is stopped

async shutdown()

Shutdown all tasks that are running and deregister from the container

property suspendable_tasks

class mango.agent.core.AgentContext(container)

Bases: object

property addr

property clock: Clock

property current_timestamp: float

Method that returns the current unix timestamp given the clock within the container

deregister(aid)

register(agent, suggested_aid)

async send_message(content, receiver_addr: AgentAddress, sender_id: None | str = None, **kwargs) → bool

See container.send_message(…)

service_of_type(type: type, default: Any = None) → Any

class mango.agent.core.AgentDelegates

Bases: object

add_forwarding_rule(from_addr: AgentAddress, to_addr: AgentAddress, forward_replies: bool = False) → None

Add an automatic message-forwarding rule.

After calling this, every message received from from_addr is automatically forwarded to to_addr. When forward_replies is True, replies originating from to_addr are forwarded back to from_addr.

Parameters:

• from_addr – source address to match

• to_addr – destination to forward to

• forward_replies – whether replies should be forwarded back

property addr

Return the address of the agent as AgentAddress

Returns:

_type_: AgentAddress

property aid

property category: str

Category tag for this agent.

property color: str

Visual color tag for this agent.

context: AgentContext

property current_timestamp: float

Method that returns the current unix timestamp given the clock within the container

delete_forwarding_rule(from_addr: AgentAddress, to_addr: AgentAddress | None = None) → None

Remove previously added forwarding rule(s).

Parameters:

• from_addr – source address of the rule to remove

• to_addr – if given, only remove rules that also match this destination; otherwise remove all rules matching from_addr

property description: AgentDescription

Return the agent’s AgentDescription.

property name: str

Human-readable name of this agent.

neighbors(state: State = State.NORMAL, *, tid: str = 'default', has_characteristic: str | None = None, include_connectors: tuple | list = (), match_func=None) → list[AgentAddress]

Return neighbor addresses from the topology.

Parameters:

• state – filter by edge state (default NORMAL)

• tid – topology identifier (default "default")

• has_characteristic – only include neighbors with this characteristic

• include_connectors – also include connector agents of these types

• match_func – optional (AgentDescription) -> bool predicate

Returns:

list of AgentAddress

on_agent_event(event: Any) → None

Called when a targeted agent event is emitted.

Override to react to events directed at this specific agent.

Parameters:

event – the event object

on_global_event(event: Any) → None

Called when a global event is emitted from the environment.

Override to react to environment-wide broadcasts.

Parameters:

event – the event object

on_ready()

Called when all container has been started using activate(…).

on_start()

Called when container started in which the agent is contained

on_step(env, clock, step_size_s: float) → None

Called on every simulation step (only in SimulationWorld).

Parameters:

• env – the simulation environment

• clock – the current simulation clock

• step_size_s – seconds advanced in this step

async reply_to(content: Any, received_meta: dict, **kwargs) → bool

Convenience helper to reply to a received message.

Extracts the sender address from received_meta and sends content back, preserving any tracking_id for transaction matching.

Parameters:

• content – reply content

• received_meta – the meta dict from the received message

Returns:

result of send_message()

schedule_conditional_process_task(coroutine_creator, condition_func, lookup_delay=0.1, on_stop=None, src=None)

Schedule a process task when a specified condition is met.

Parameters:

• coroutine_creator (coroutine_creator) – coroutine_creator creating coroutine to be scheduled

• condition_func (lambda () -> bool) – function for determining whether the confition is fullfiled

• lookup_delay (float) – delay between checking the condition

• src (Object) – creator of the task

schedule_conditional_task(coroutine, condition_func, lookup_delay=0.1, on_stop=None, src=None)

Schedule a task when a specified condition is met.

Parameters:

• coroutine (Coroutine) – coroutine to be scheduled

• condition_func (lambda () -> bool) – function for determining whether the confition is fullfiled

• lookup_delay (float) – delay between checking the condition

• src (Object) – creator of the task

schedule_instant_message(content, receiver_addr: AgentAddress, **kwargs)

Schedules sending a message without any delay. This is equivalent to using the schedulers ‘schedule_instant_task’ with the coroutine created by ‘container.send_message’.

Parameters:

• content – The content of the message

• receiver_addr – The address passed to the container

• kwargs – Additional parameters to provide protocol specific settings

Returns:

asyncio.Task for the scheduled coroutine

schedule_instant_process_task(coroutine_creator, on_stop=None, src=None)

Schedule an instantly executed task in another processes.

Parameters:

• coroutine_creator – coroutine_creator creating coroutine to be scheduled

• on_stop (Object) – coroutine to run on stop

• src (Object) – creator of the task

schedule_instant_task(coroutine, on_stop=None, src=None)

Schedule an instantly executed task.

Parameters:

• coroutine – coroutine to be scheduled

• on_stop (Object) – coroutine to run on stop

• src (Object) – creator of the task

schedule_periodic_process_task(coroutine_creator, delay, on_stop=None, src=None)

Schedule an open end periodically executed task in another process.

Parameters:

• coroutine_creator (Coroutine Function) – coroutine function creating coros to be scheduled

• delay (float) – delay in between the cycles

• on_stop (Object) – coroutine to run on stop

• src (Object) – creator of the task

schedule_periodic_task(coroutine_func, delay, on_stop=None, src=None)

Schedule an open end peridocally executed task.

Parameters:

• coroutine_func (Coroutine Function) – coroutine function creating coros to be scheduled

• delay (float) – delay in between the cycles

• on_stop (Object) – coroutine to run on stop

• src (Object) – creator of the task

schedule_process_task(task: ScheduledProcessTask, src=None)

Schedule a task with asyncio in another process. When the task is finished, if finite, its automatically removed afterwards. For scheduling options see the subclasses of ScheduledScheduledProcessTaskTask.

Parameters:

• task – task to be scheduled

• src – object, which represents the source of the task (for example the object in which the task got created)

schedule_recurrent_process_task(coroutine_creator, recurrency, on_stop=None, src=None)

Schedule a task using a fine-grained recurrency rule in another process.

Parameters:

• coroutine_creator (Coroutine Function) – coroutine function creating coros to be scheduled

• recurrency (dateutil.rrule.rrule) – recurrency rule to calculate next event

• on_stop (Object) – coroutine to run on stop

• src (Object) – creator of the task

schedule_recurrent_task(coroutine_func, recurrency, on_stop=None, src=None)

Schedule a task using a fine-grained recurrency rule in another process.

Parameters:

• coroutine_creator (Coroutine Function) – coroutine function creating coros to be scheduled

• recurrency (dateutil.rrule.rrule) – recurrency rule to calculate next event

• on_stop (Object) – coroutine to run on stop

• src (Object) – creator of the task

schedule_task(task: ScheduledTask, src=None)

Schedule a task with asyncio. When the task is finished, if finite, its automatically removed afterwards. For scheduling options see the subclasses of ScheduledTask.

Parameters:

• task – task to be scheduled

• src – object, which represents the source of the task (for example the object in which the task got created)

schedule_timestamp_process_task(coroutine_creator, timestamp: float, on_stop=None, src=None)

Schedule a task at specified unix timestamp dispatched to another process.

Parameters:

• coroutine_creator (coroutine_creator) – coroutine_creator creating coroutine to be scheduled

• timestamp (float) – unix timestamp defining when the task should start

• src (Object) – creator of the task

schedule_timestamp_task(coroutine, timestamp: float, on_stop=None, src=None)

Schedule a task at specified unix timestamp.

Parameters:

• coroutine (Coroutine) – coroutine to be scheduled

• timestamp (timestamp) – timestamp defining when the task should start

• src (Object) – creator of the task

scheduler: Scheduler

async send_message(content, receiver_addr: AgentAddress, **kwargs) → bool

See container.send_message(…)

async send_messages(content, receiver_addrs: list[AgentAddress], **kwargs) → list[bool]

Send the same content to multiple recipients.

Parameters:

• content – message content (sent to every recipient)

• receiver_addrs – list of target AgentAddress instances

Returns:

list of send results (one per recipient, in order)

async send_tracked_message(content: Any, receiver_addr: AgentAddress, response_handler=None, **kwargs)

Send a message and optionally register a response handler.

A tracking_id is attached to the message so that the reply can be matched. When response_handler is provided it will be called as response_handler(reply_content, reply_meta) when the matching reply arrives.

Parameters:

• content – message content

• receiver_addr – target agent address

• response_handler – optional (content, meta) -> None callback

Returns:

the asyncio.Task for the sent message

service_of_type(type: type, default: Any = None) → Any

Return the service registered for type, creating it if absent.

If no service of type is registered yet, default is registered and returned; when default is None a new type() instance is created instead.

Parameters:

• type (type) – the type of the service

• default (Any (optional)) – the value to register if none exists; None creates a type() instance

Returns:

the service

Return type:

Any

async tasks_complete(timeout=1)

Wait for all scheduled tasks to complete using a timeout.

Parameters:

timeout – waiting timeout. Defaults to 1.

property uid: str

Unique identifier (UUID string) of this agent.

update_description(name: str | None = None, color: str | None = None, category: str | None = None) → None

Update one or more description fields.

Parameters:

• name – new human-readable name

• color – new color tag

• category – new category tag

class mango.agent.core.AgentDescription(name: str = '', category: str = 'agent', color: str = 'gray', uid: str = <factory>)

Bases: object

Metadata describing an agent (name, category, color, unique ID).

Mirrors the AgentDescription type in Mango.jl.

category: str = 'agent'

color: str = 'gray'

name: str = ''

uid: str

class mango.agent.core.ForwardingRule(from_addr: AgentAddress, to_addr: AgentAddress, forward_replies: bool = False)

Bases: object

Rule for automatic message forwarding.

When the agent receives a message from from_addr it is automatically forwarded to to_addr. If forward_replies is True, replies from to_addr are forwarded back to from_addr.

forward_replies: bool = False

from_addr: AgentAddress

to_addr: AgentAddress

class mango.agent.core.State(*values)

Bases: Enum

BROKEN = 2

EXT_CONNECTION = 4

INACTIVE = 1

NORMAL = 0

UNKNOWN = 3

class mango.agent.core.TopologyNeighbor(agent: Any, description: AgentDescription, characteristic: str = '')

Bases: object

A neighbor in the topology graph.

Addresses are resolved lazily so that topologies can be built before agents are registered in a container.

Parameters:

• agent – the neighbor agent (address resolved at access time)

• description – the agent’s AgentDescription

• characteristic – an optional label for the agent’s role in its node (e.g. "leader"); empty string means no characteristic

property address: AgentAddress

The neighbor’s current AgentAddress (resolved lazily).

class mango.agent.core.TopologyService

Bases: object

Stores neighborhood data injected by the topology system.

An instance is attached to each agent via service_of_type().

characteristic(tid: str = 'default') → str

Return this agent’s characteristic label in topology tid.

connection_types(tid: str = 'default') → list[str]

Return the connection type labels for connectors in topology tid.

connectors(tid: str = 'default', *, include_connectors: tuple[str, ...] | list[str] = (), match_func: Any = None) → list[AgentAddress]

Return addresses of connector agents for topology tid.

neighbors(state: State = State.NORMAL, *, tid: str = 'default', has_characteristic: str | None = None, include_connectors: tuple[str, ...] | list[str] = (), match_func: Any = None) → list[AgentAddress]

Return addresses of neighbors in topology tid with edge state.

Parameters:

• state – only return neighbors reachable via edges in this state

• tid – topology identifier

• has_characteristic – if given, only neighbors with this characteristic

• include_connectors – also include connectors of these connection types

• match_func – optional predicate (AgentDescription) -> bool

node_id(tid: str = 'default') → int

Return the node ID this agent occupies in topology tid.

Roles

API classes for using the role system. The role system is based on the idea, that everything an agent can do, is described as role/responsibility and is implemented in one separate class. For example participating in a coalition would be a separate role, monitoring grid voltage another one.

A role is part of a RoleAgent which inherits from Agent.

There are essentially two APIs for acting resp reacting:

• [Reacting] RoleContext.subscribe_message(), which allows you to subscribe to certain message types and lets you handle the message

• [Acting] RoleContext.schedule_task(), this allows you to schedule a task with delay/repeating/…

To interact with the environment an instance of the role context is provided. This context provides methods to share data with other roles and to communicate with other agents.

A message can be send using the method RoleContext.send_message().

There are often dependencies between different parts of an agent, there are options to interact with other roles: Roles have the possibility to use shared models and to act on changes of these models. A role can subscribe specific data that another role provides. To set this up, a model has to be created via RoleContext.get_or_create_model(). To notify other roles RoleContext.update() has to be called. In order to let a Role subscribe to a model you can use subscribe_model(). If you prefer a lightweight variant you can use RoleContext.data() to assign/access shared data.

Furthermore there are two lifecycle methods to know about:

• Role.setup() is called when the Role is added to the agent, so its the perfect place

  for initialization and scheduling of tasks

• Role.on_stop() is called when the container the agent lives in, is shut down

class mango.agent.role.DataContainer

Bases: object

get(key, default=None)

update(data: dict)

class mango.agent.role.MessagePreprocessor

Bases: ABC

Abstract base for message preprocessors in the role system.

A preprocessor intercepts messages before they reach a role’s handler, allowing transformation or rate-limiting. Pass an instance to RoleContext.subscribe_message() via the preprocessor keyword.

Subclasses must implement handle(). Override process() to transform the message content/meta before delivery.

Example:

class LoggingPreprocessor(MessagePreprocessor):
    def handle(self, role, handler, content, meta):
        print(f"[{role}] received: {content}")
        handler(content, meta)

class MyRole(Role):
    def setup(self):
        self.context.subscribe_message(
            self, self.on_msg, lambda c, m: True,
            preprocessor=LoggingPreprocessor(),
        )

    def on_msg(self, content, meta):
        ...

abstractmethod handle(role_or_agent: Any, handler: Callable, content: Any, meta: dict) → None

Intercept a message. Must call handler(content, meta) to deliver.

Parameters:

• role_or_agent – the subscribing role or agent

• handler – the original message handler

• content – message content

• meta – message metadata

init(role_or_agent: Any) → None

Called once when the preprocessor is registered.

Parameters:

role_or_agent – the role (or agent) that owns the subscription

process(content: Any, meta: dict) → tuple[Any, dict]

Transform message before delivery. Default: identity.

Parameters:

• content – message content

• meta – message metadata

Returns:

transformed (content, meta) tuple

class mango.agent.role.Role

Bases: ABC

General role class, defining the API every role can use. A role implements one responsibility of an agent.

Every role must be added to a RoleAgent and is defined by some lifecycle methods:

• Role.setup() is called when the Role is added to the agent, so its the perfect place for

  initialization and scheduling of tasks

• Role.on_stop() is called when the container the agent lives in, is shut down

To interact with the environment you have to use the context, accessible via :func:Role.context.

property context: RoleContext

Return the context of the role. This context can be send as bridge to the agent.

Returns:

the context of the role

handle_message(content: Any, meta: dict)

on_agent_event(event: Any) → None

Called when a targeted agent event is emitted.

Parameters:

event – the event object

on_change_model(model) → None

Will be invoked when a subscribed model changes via RoleContext.update().

Parameters:

model – the model

on_deactivation(src) → None

Hook in, which will be called when another role deactivates this instance (temporarily)

on_global_event(event: Any) → None

Called when a global event is emitted from the environment.

Parameters:

event – the event object

on_ready()

Called after the start of all container using activate

on_start() → None

Called when container started in which the agent is contained

on_step(env, clock, step_size_s: float) → None

Called on every simulation step (only in SimulationWorld).

Parameters:

• env – the simulation environment

• clock – the current simulation clock

• step_size_s – seconds advanced in this step

async on_stop() → None

Lifecycle hook in, which will be called when the container is shut down or if the role got removed.

setup() → None

Lifecycle hook in, which will be called on adding the role to agent. The role context is known from hereon.

class mango.agent.role.RoleAgent

Bases: Agent

Agent, which support the role API-system. When you want to use the role-api you always need a RoleAgent as base for your agents. A role can be added with RoleAgent.add_role().

add_role(role: Role)

Add a role to the agent. This will lead to the call of Role.setup().

Parameters:

role – the role to add

handle_message(content, meta: dict[str, Any])

Has to be implemented by the user. This method is called when a message is received at the agents inbox. :param content: The deserialized message object :param meta: Meta details of the message. In case of mqtt this dict includes at least the field ‘topic’

on_ready()

Called when all container has been started using activate(…).

on_register()

Hook-in to define behavior of the agent directly after it got registered by a container

on_start()

Called when container started in which the agent is contained

remove_role(role: Role)

Remove a role permanently from the agent.

Parameters:

role (Role) – [description]

property roles: list[Role]

Returns list of roles

Returns:

list of roles

async shutdown()

Shutdown all tasks that are running and deregister from the container

class mango.agent.role.RoleContext(role_handler: RoleHandler, aid: str, inbox)

Bases: AgentDelegates

Implementation of the RoleContext.

activate(role) → None

add_role(role: Role)

Add a role to the context.

Parameters:

role – the Role

property current_timestamp: float

Method that returns the current unix timestamp given the clock within the container

property data

Return data container of the agent

Returns:

the data container

Return type:

DataContainer

deactivate(role) → None

emit_event(event: Any, event_source: Any = None)

Emit an custom event to other roles.

Parameters:

• event (Any) – the event

• event_source (Any, optional) – emitter of the event (mostly the emitting role), defaults to None

get_or_create_model(cls)

get_role(cls: type) → Role | None

returns the first role of a given class returns None if no role of this type exists in the current context

handle_message(content, meta: dict[str, Any])

Handle an incoming message, delegating it to all applicable subscribers

for role, message_condition, method, _ in self._message_subs:
    if self._is_role_active(role) and message_condition(content, meta):
        method(content, meta)

Parameters:

• content – content

• meta – meta

inbox_length()

on_ready()

Called when all container has been started using activate(…).

on_start()

Called when container started in which the agent is contained

remove_role(role: Role)

Remove a role and call on_stop for clean up

Parameters:

role (Role) – the role to remove

async send_message(content, receiver_addr: AgentAddress, **kwargs) → bool

See container.send_message(…)

subscribe_event(role: Role, event_type: Any, handler_method: Callable)

Subscribe to specific event types. The listener will be evaluated based on their order of subscription

Parameters:

• role (Role) – the role in which you want to handle the event

• event_type (Any) – the event type you want to handle

subscribe_message(role, method, message_condition, priority=0, preprocessor: MessagePreprocessor | None = None)

subscribe_model(role, role_model_type)

subscribe_send(role, method)

update(role_model)

class mango.agent.role.RoleHandler(scheduler)

Bases: object

Contains all roles and their models. Implements the communication between roles.

activate(role) → None

Activates the given role.

Parameters:

role (Role) – the role to activate

add_role(role: Role) → None

Add a new role

Args:

role ([type]): the role

deactivate(role) → None

Deactivates the role. This includes all tasks (soft suspending)

Parameters:

role (Role) – the role to deactivate

emit_event(event: Any, event_source: Any = None)

get_or_create_model(cls)

Creates or return (when already created) a central role model.

Returns:

[type]: the model

handle_message(content, meta: dict[str, Any])

Handle an incoming message, delegating it to all applicable subscribers

for role, message_condition, method, _, preprocessor in self._message_subs:
    if self._is_role_active(role) and message_condition(content, meta):
        if preprocessor:
            preprocessor.handle(role, method, content, meta)
        else:
            method(content, meta)

Parameters:

• content – content

• meta – meta

on_ready()

on_start()

async on_stop()

Notify all roles when the container is shutdown

remove_role(role: Role) → None

Remove a given role

Args:

role ([type]): the role

property roles: list[Role]

Returns all roles

Returns:

List[Role]: the roles hold by this handler

subscribe(role: Role, role_model_type) → None

Subscribe a role to change events of a specific role model type

Args:

role ([type]): the role role_model_type ([type]): the type of the role model

subscribe_event(role: Role, event_type: type, method: Callable)

subscribe_message(role, method, message_condition, priority=0, preprocessor: MessagePreprocessor | None = None)

subscribe_send(role: Role, method: Callable)

update(role_model) → None

Notifies all subscribers of an update of the given role_model.

Args:

role_model ([type]): the role model to notify about

class mango.agent.role.WaitingMessagePreprocessor

Bases: MessagePreprocessor

Prevents concurrent message handling for a role.

Messages are queued and dispatched one at a time. The next message is only delivered after the handler for the current one has returned (or its coroutine completed). This avoids race conditions when a role’s handler is async or triggers further messages.

Example:

class MyRole(Role):
    def setup(self):
        self.context.subscribe_message(
            self, self.on_data, lambda c, m: True,
            preprocessor=WaitingMessagePreprocessor(),
        )

    async def on_data(self, content, meta):
        await asyncio.sleep(0.1)   # safe – next msg waits
        ...

handle(role_or_agent: Any, handler: Callable, content: Any, meta: dict) → None

Intercept a message. Must call handler(content, meta) to deliver.

Parameters:

• role_or_agent – the subscribing role or agent

• handler – the original message handler

• content – message content

• meta – message metadata

init(role_or_agent: Any) → None

Called once when the preprocessor is registered.

Parameters:

role_or_agent – the role (or agent) that owns the subscription