agentlib.modules.communicator package

Package contains all modules to communicate messages with

Submodules

agentlib.modules.communicator.clonemap module

agentlib.modules.communicator.communicator module

Module contains basics communicator modules

class agentlib.modules.communicator.communicator.CommunicationDict

Bases: TypedDict

alias: str

source: str

timestamp: float

type: str

value: Any

class agentlib.modules.communicator.communicator.Communicator(*, config: dict, agent: Agent)

Bases: BaseModule

Base class for all communicators

process()

This abstract method must be implemented in order to sync the module with the other processes of the agent and the whole MAS.

register_callbacks()

Register all outputs to the callback function

short_dict(variable: AgentVariable) → CommunicationDict

Creates a short dict serialization of the Variable.

Only contains attributes of the AgentVariable, that are relevant for other modules or agents. For performance and privacy reasons, this function should be called for communicators.

to_json(payload: CommunicationDict) → bytes | str

Transforms the payload into json serialized form. Dynamically uses orjson if it is installed, and the builtin json otherwise.

Returns bytes or str depending on the library used, but this has not mattered with the communicators as of now.

pydantic model agentlib.modules.communicator.communicator.CommunicatorConfig

Bases: BaseModuleConfig

Config:

• arbitrary_types_allowed: bool = True

• validate_assignment: bool = True

• extra: str = forbid

• frozen: bool = True

Fields:

• log_level (Optional[str])

• module_id (str)

• shared_variable_fields (List[str])

• type (Union[str, Dict[str, str]])

• use_orjson (bool)

• validate_incoming_values (Optional[bool])

Validators:

• check_valid_fields » shared_variable_fields

• check_valid_level » log_level

field log_level: str | None = None

The log level for this Module. Default uses the root-loggers level.Options: DEBUG; INFO; WARNING; ERROR; CRITICAL

Validated by:

• check_valid_level

field module_id: str [Required]

The unqiue id of the module within an agent, used only to communicate withing the agent.

field shared_variable_fields: List[str] = []

A list of strings with each string being a field of the Modules configs. The field must be or contain an AgentVariable. If the field is added to this list, all shared attributes of the AgentVariables will be set to True.

Validated by:

• check_valid_fields

field type: str | Dict[str, str] [Required]

The type of the Module. Used to find the Python-Object from all agentlib-core and plugin Module options. If a dict is given,it must contain the keys ‘file’ and ‘class_name’. ‘file’ is the filepath of a python file containing the Module.’class_name’ is the name of the Module class within this file.

field use_orjson: bool = False

If true, the faster orjson library will be used for serialization deserialization. Requires the optional dependency.

field validate_incoming_values: bool | None = True

If true, the validator of the AgentVariable value is called when receiving a new value from the DataBroker.

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialise private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

• self – The BaseModel instance.

• context – The context.

class agentlib.modules.communicator.communicator.LocalCommunicator(config: dict, agent: Agent)

Bases: Communicator

Base class for local communicators.

property broker: Broker

Broker used by LocalCommunicator

abstract setup_broker()

Function to set up the broker object. Needs to return a valid broker option.

terminate()

Terminate all relevant processes of the module. This is necessary to correctly terminate an agent at runtime. Not all modules may need this, hence it is not an abstract method.

pydantic model agentlib.modules.communicator.communicator.LocalCommunicatorConfig

Bases: CommunicatorConfig

Config:

• arbitrary_types_allowed: bool = True

• validate_assignment: bool = True

• extra: str = forbid

• frozen: bool = True

Fields:

• log_level (Optional[str])

• module_id (str)

• parse_json (bool)

• shared_variable_fields (List[str])

• type (Union[str, Dict[str, str]])

• use_orjson (bool)

• validate_incoming_values (Optional[bool])

Validators:

• check_valid_fields » shared_variable_fields

• check_valid_level » log_level

field log_level: str | None = None

The log level for this Module. Default uses the root-loggers level.Options: DEBUG; INFO; WARNING; ERROR; CRITICAL

Validated by:

• check_valid_level

field module_id: str [Required]

The unqiue id of the module within an agent, used only to communicate withing the agent.

field parse_json: bool = False

field shared_variable_fields: List[str] = []

A list of strings with each string being a field of the Modules configs. The field must be or contain an AgentVariable. If the field is added to this list, all shared attributes of the AgentVariables will be set to True.

Validated by:

• check_valid_fields

field type: str | Dict[str, str] [Required]

The type of the Module. Used to find the Python-Object from all agentlib-core and plugin Module options. If a dict is given,it must contain the keys ‘file’ and ‘class_name’. ‘file’ is the filepath of a python file containing the Module.’class_name’ is the name of the Module class within this file.

field use_orjson: bool = False

If true, the faster orjson library will be used for serialization deserialization. Requires the optional dependency.

field validate_incoming_values: bool | None = True

If true, the validator of the AgentVariable value is called when receiving a new value from the DataBroker.

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialise private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

• self – The BaseModel instance.

• context – The context.

pydantic model agentlib.modules.communicator.communicator.SubscriptionCommunicatorConfig

Bases: CommunicatorConfig

Config:

• arbitrary_types_allowed: bool = True

• validate_assignment: bool = True

• extra: str = forbid

• frozen: bool = True

Fields:

• log_level (Optional[str])

• module_id (str)

• shared_variable_fields (List[str])

• subscriptions (List[str] | str)

• type (Union[str, Dict[str, str]])

• use_orjson (bool)

• validate_incoming_values (Optional[bool])

Validators:

• check_valid_fields » shared_variable_fields

• check_valid_level » log_level

• convert_to_list » subscriptions

field log_level: str | None = None

The log level for this Module. Default uses the root-loggers level.Options: DEBUG; INFO; WARNING; ERROR; CRITICAL

Validated by:

• check_valid_level

field module_id: str [Required]

The unqiue id of the module within an agent, used only to communicate withing the agent.

field shared_variable_fields: List[str] = []

A list of strings with each string being a field of the Modules configs. The field must be or contain an AgentVariable. If the field is added to this list, all shared attributes of the AgentVariables will be set to True.

Validated by:

• check_valid_fields

field subscriptions: List[str] | str = []

List of agent-id strings to subscribe to

Validated by:

• convert_to_list

field type: str | Dict[str, str] [Required]

The type of the Module. Used to find the Python-Object from all agentlib-core and plugin Module options. If a dict is given,it must contain the keys ‘file’ and ‘class_name’. ‘file’ is the filepath of a python file containing the Module.’class_name’ is the name of the Module class within this file.

field use_orjson: bool = False

If true, the faster orjson library will be used for serialization deserialization. Requires the optional dependency.

field validate_incoming_values: bool | None = True

If true, the validator of the AgentVariable value is called when receiving a new value from the DataBroker.

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialise private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

• self – The BaseModel instance.

• context – The context.

agentlib.modules.communicator.local module

class agentlib.modules.communicator.local.LocalClient(config: dict, agent: Agent)

Bases: LocalCommunicator

This communicator implements the communication between agents via a central process broker. Note: The broker is implemented as singleton this means that all agents must be in the same process!

setup_broker()

Use the LocalBroker

property subscriptions

pydantic model agentlib.modules.communicator.local.LocalSubscriptionCommunicatorConfig

Bases: LocalCommunicatorConfig, SubscriptionCommunicatorConfig

Config:

• arbitrary_types_allowed: bool = True

• validate_assignment: bool = True

• extra: str = forbid

• frozen: bool = True

Fields:

• log_level (Optional[str])

• module_id (str)

• parse_json (bool)

• shared_variable_fields (List[str])

• subscriptions (Union[List[str], str])

• type (Union[str, Dict[str, str]])

• use_orjson (bool)

• validate_incoming_values (Optional[bool])

Validators:

• check_valid_fields » shared_variable_fields

• check_valid_level » log_level

• convert_to_list » subscriptions

field log_level: str | None = None

The log level for this Module. Default uses the root-loggers level.Options: DEBUG; INFO; WARNING; ERROR; CRITICAL

Validated by:

• check_valid_level

field module_id: str [Required]

The unqiue id of the module within an agent, used only to communicate withing the agent.

field parse_json: bool = False

field shared_variable_fields: List[str] = []

A list of strings with each string being a field of the Modules configs. The field must be or contain an AgentVariable. If the field is added to this list, all shared attributes of the AgentVariables will be set to True.

Validated by:

• check_valid_fields

field subscriptions: List[str] | str = []

List of agent-id strings to subscribe to

Validated by:

• convert_to_list

field type: str | Dict[str, str] [Required]

The type of the Module. Used to find the Python-Object from all agentlib-core and plugin Module options. If a dict is given,it must contain the keys ‘file’ and ‘class_name’. ‘file’ is the filepath of a python file containing the Module.’class_name’ is the name of the Module class within this file.

field use_orjson: bool = False

If true, the faster orjson library will be used for serialization deserialization. Requires the optional dependency.

field validate_incoming_values: bool | None = True

If true, the validator of the AgentVariable value is called when receiving a new value from the DataBroker.

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialise private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

• self – The BaseModel instance.

• context – The context.

agentlib.modules.communicator.local_broadcast module

class agentlib.modules.communicator.local_broadcast.LocalBroadcastClient(config: dict, agent: Agent)

Bases: LocalCommunicator

This communicator implements the communication between agents via a broadcast broker central process. Note: The broker is implemented as singleton. This means that all agents must be in the same process!

setup_broker()

Use the LocalBroadCastBroker

agentlib.modules.communicator.local_multiprocessing module

class agentlib.modules.communicator.local_multiprocessing.MultiProcessingBroadcastClient(config: dict, agent: Agent)

Bases: Communicator

This communicator implements the communication between agents via a central process broker.

process()

Only start the loop once the env is running.

terminate()

Closes all of the connections.

pydantic model agentlib.modules.communicator.local_multiprocessing.MultiProcessingBroadcastClientConfig

Bases: CommunicatorConfig

Config:

• arbitrary_types_allowed: bool = True

• validate_assignment: bool = True

• extra: str = forbid

• frozen: bool = True

Fields:

• authkey (bytes)

• ipv4 (ipaddress.IPv4Address)

• log_level (Optional[str])

• module_id (str)

• port (int)

• shared_variable_fields (List[str])

• type (Union[str, Dict[str, str]])

• use_orjson (bool)

• validate_incoming_values (Optional[bool])

Validators:

• check_valid_fields » shared_variable_fields

• check_valid_level » log_level

field authkey: bytes = b'useTheAgentlib'

Authorization key for the connection with the broker.

field ipv4: IPv4Address = '127.0.0.1'

IP Address for the communication server. Defaults to localhost.

field log_level: str | None = None

The log level for this Module. Default uses the root-loggers level.Options: DEBUG; INFO; WARNING; ERROR; CRITICAL

Validated by:

• check_valid_level

field module_id: str [Required]

The unqiue id of the module within an agent, used only to communicate withing the agent.

field port: int = 50000

Port for setting up the connection with the broker.

field shared_variable_fields: List[str] = []

A list of strings with each string being a field of the Modules configs. The field must be or contain an AgentVariable. If the field is added to this list, all shared attributes of the AgentVariables will be set to True.

Validated by:

• check_valid_fields

field type: str | Dict[str, str] [Required]

The type of the Module. Used to find the Python-Object from all agentlib-core and plugin Module options. If a dict is given,it must contain the keys ‘file’ and ‘class_name’. ‘file’ is the filepath of a python file containing the Module.’class_name’ is the name of the Module class within this file.

field use_orjson: bool = False

If true, the faster orjson library will be used for serialization deserialization. Requires the optional dependency.

field validate_incoming_values: bool | None = True

If true, the validator of the AgentVariable value is called when receiving a new value from the DataBroker.

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialise private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

• self – The BaseModel instance.

• context – The context.

agentlib.modules.communicator.mqtt module

pydantic model agentlib.modules.communicator.mqtt.BaseMQTTClientConfig

Bases: SubscriptionCommunicatorConfig

Config:

• arbitrary_types_allowed: bool = True

• validate_assignment: bool = True

• extra: str = forbid

• frozen: bool = True

Fields:

• clean_start (bool)

• client_id (str | None)

• connection_timeout (float)

• keepalive (int)

• log_level (Optional[str])

• module_id (str)

• password (str | None)

• prefix (str)

• qos (int)

• shared_variable_fields (List[str])

• subscriptions (Union[List[str], str])

• subtopics (List[str] | str)

• tls_ca_certs (str | None)

• type (Union[str, Dict[str, str]])

• use_orjson (bool)

• use_tls (bool | None)

• username (str | None)

• validate_incoming_values (Optional[bool])

Validators:

• check_valid_fields » shared_variable_fields

• check_valid_level » log_level

• convert_to_list » subscriptions

• convert_to_list » subtopics

field clean_start: bool = True

True, False or MQTT_CLEAN_START_FIRST_ONLY.Sets the MQTT v5.0 clean_start flag always, never or on the first successful connect only, respectively. MQTT session data (such as outstanding messages and subscriptions) is cleared on successful connect when the clean_start flag is set.

field client_id: str | None = None

field connection_timeout: float = 10

Number of seconds to wait for the initial connection until throwing an Error.

field keepalive: int = 60

Maximum period in seconds between communications with the broker. If no other messages are being exchanged, this controls the rate at which the client will send ping messages to the broker.

field log_level: str | None = None

The log level for this Module. Default uses the root-loggers level.Options: DEBUG; INFO; WARNING; ERROR; CRITICAL

Validated by:

• check_valid_level

field module_id: str [Required]

The unqiue id of the module within an agent, used only to communicate withing the agent.

field password: str | None = None

field prefix: str = '/agentlib'

Prefix for MQTT-Topic

field qos: int = 0

Quality of Service

Constraints:

• ge = 0

• le = 2

field shared_variable_fields: List[str] = []

A list of strings with each string being a field of the Modules configs. The field must be or contain an AgentVariable. If the field is added to this list, all shared attributes of the AgentVariables will be set to True.

Validated by:

• check_valid_fields

field subscriptions: List[str] | str = []

List of agent-id strings to subscribe to

Validated by:

• convert_to_list

field subtopics: List[str] | str = []

Topics to that the agent subscribes

Validated by:

• convert_to_list

field tls_ca_certs: str | None = None

Path to the Certificate Authority certificate files. If None, windows certificate will be used.

field type: str | Dict[str, str] [Required]

The type of the Module. Used to find the Python-Object from all agentlib-core and plugin Module options. If a dict is given,it must contain the keys ‘file’ and ‘class_name’. ‘file’ is the filepath of a python file containing the Module.’class_name’ is the name of the Module class within this file.

field use_orjson: bool = False

If true, the faster orjson library will be used for serialization deserialization. Requires the optional dependency.

field use_tls: bool | None = None

Option to use TLS certificates

field username: str | None = None

field validate_incoming_values: bool | None = True

If true, the validator of the AgentVariable value is called when receiving a new value from the DataBroker.

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialise private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

• self – The BaseModel instance.

• context – The context.

class agentlib.modules.communicator.mqtt.BaseMqttClient(config: dict, agent: Agent)

Bases: Communicator

abstract connect()

disconnect(reasoncode=None, properties=None)

Trigger the disconnect

mqttc_type

alias of Client

terminate()

Disconnect from client and join loop

property topics_size

pydantic model agentlib.modules.communicator.mqtt.MQTTClientConfig

Bases: BaseMQTTClientConfig

Config:

• arbitrary_types_allowed: bool = True

• validate_assignment: bool = True

• extra: str = forbid

• frozen: bool = True

Fields:

• clean_start (bool)

• client_id (Optional[str])

• connection_timeout (float)

• keepalive (int)

• log_level (Optional[str])

• module_id (str)

• password (Optional[str])

• prefix (str)

• qos (int)

• shared_variable_fields (List[str])

• subscriptions (Union[List[str], str])

• subtopics (Union[List[str], str])

• tls_ca_certs (Optional[str])

• type (Union[str, Dict[str, str]])

• url (pydantic.networks.AnyUrl)

• use_orjson (bool)

• use_tls (Optional[bool])

• username (Optional[str])

• validate_incoming_values (Optional[bool])

Validators:

• check_url » url

• check_valid_fields » shared_variable_fields

• check_valid_level » log_level

• convert_to_list » subscriptions

• convert_to_list » subtopics

field clean_start: bool = True

True, False or MQTT_CLEAN_START_FIRST_ONLY.Sets the MQTT v5.0 clean_start flag always, never or on the first successful connect only, respectively. MQTT session data (such as outstanding messages and subscriptions) is cleared on successful connect when the clean_start flag is set.

field client_id: str | None = None

field connection_timeout: float = 10

Number of seconds to wait for the initial connection until throwing an Error.

field keepalive: int = 60

Maximum period in seconds between communications with the broker. If no other messages are being exchanged, this controls the rate at which the client will send ping messages to the broker.

field log_level: str | None = None

The log level for this Module. Default uses the root-loggers level.Options: DEBUG; INFO; WARNING; ERROR; CRITICAL

Validated by:

• check_valid_level

field module_id: str [Required]

The unqiue id of the module within an agent, used only to communicate withing the agent.

field password: str | None = None

field prefix: str = '/agentlib'

Prefix for MQTT-Topic

field qos: int = 0

Quality of Service

Constraints:

• ge = 0

• le = 2

field shared_variable_fields: List[str] = []

A list of strings with each string being a field of the Modules configs. The field must be or contain an AgentVariable. If the field is added to this list, all shared attributes of the AgentVariables will be set to True.

Validated by:

• check_valid_fields

field subscriptions: List[str] | str = []

List of agent-id strings to subscribe to

Validated by:

• convert_to_list

field subtopics: List[str] | str = []

Topics to that the agent subscribes

Validated by:

• convert_to_list

field tls_ca_certs: str | None = None

Path to the Certificate Authority certificate files. If None, windows certificate will be used.

field type: str | Dict[str, str] [Required]

The type of the Module. Used to find the Python-Object from all agentlib-core and plugin Module options. If a dict is given,it must contain the keys ‘file’ and ‘class_name’. ‘file’ is the filepath of a python file containing the Module.’class_name’ is the name of the Module class within this file.

field url: AnyUrl [Required]

Host is the hostname or IP address of the remote broker.

Validated by:

• check_url

field use_orjson: bool = False

If true, the faster orjson library will be used for serialization deserialization. Requires the optional dependency.

field use_tls: bool | None = None

Option to use TLS certificates

field username: str | None = None

field validate_incoming_values: bool | None = True

If true, the validator of the AgentVariable value is called when receiving a new value from the DataBroker.

validator check_url  »  url

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialise private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

• self – The BaseModel instance.

• context – The context.

class agentlib.modules.communicator.mqtt.MqttClient(config: dict, agent: Agent)

Bases: BaseMqttClient

connect()

generate_topic(agent_id: str, subscription: bool = True)

Generate the topic with the given agent_id and configs prefix

property pubtopic

property topics_size