Communicator Extension System

OpenMAS provides a robust system for discovering and using communicator extensions. This allows for extending the framework with custom communication methods, both as installable packages (via entry points) and as local project extensions.

Types of Extensions

The communicator extension system supports three types of extension sources:

1. Built-in Communicators: These are included with the OpenMAS package (HttpCommunicator, etc.)
2. Entry Point Extensions: These are installed packages that register communicators via the openmas.communicators entry point
3. Local Extensions: These are project-local communicator implementations found in directories specified in extension_paths

Using Communicator Extensions

To use a specific communicator in your agent, you can:

1. Set it in the configuration:

config = {
    "name": "my_agent",
    "communicator_type": "my_custom_communicator",
    "extension_paths": ["./my_extensions"]  # Optional: For local extensions
}
agent = MyAgent(config=config)

1. Pass it directly to the constructor:

from my_package import MyCustomCommunicator

agent = MyAgent(
    name="my_agent",
    communicator_class=MyCustomCommunicator
)

Creating Communicator Extensions

There are two ways to create and distribute communicator extensions:

1. As an Installable Package (Entry Points)

To create a communicator extension as an installable package:

1. Create a subclass of BaseCommunicator:

from openmas.communication import BaseCommunicator

class MyCustomCommunicator(BaseCommunicator):
    async def send_request(self, target_service, method, params=None, response_model=None, timeout=None):
        # Your implementation here
        ...

    async def send_notification(self, target_service, method, params=None):
        # Your implementation here
        ...

    async def register_handler(self, method, handler):
        # Your implementation here
        ...

    async def start(self):
        # Your implementation here
        ...

    async def stop(self):
        # Your implementation here
        ...

1. Register it as an entry point in your package's pyproject.toml:

[project.entry-points."openmas.communicators"]
my_communicator = "my_package.module:MyCustomCommunicator"

Or in setup.py:

setup(
    # ...
    entry_points={
        'openmas.communicators': [
            'my_communicator = my_package.module:MyCustomCommunicator',
        ],
    },
    # ...
)

2. As a Local Extension

To create a communicator as a local extension:

1. Create a Python file in your project directory (e.g., ./extensions/my_communicator.py):

from openmas.communication import BaseCommunicator

class MyLocalCommunicator(BaseCommunicator):
    # Implementation as above
    ...

1. Specify the extension path in your agent configuration:

config = {
    "name": "my_agent",
    "communicator_type": "my_communicator",  # Uses the filename without .py
    "extension_paths": ["./extensions"]
}
agent = MyAgent(config=config)

Discovery Order

When resolving a communicator type, OpenMAS uses the following order:

1. Check if a communicator class is directly provided to the agent constructor
2. Look for the type in the registry (built-ins and entry points)
3. Search local extension paths for matching communicator implementations

Environment Configuration

You can also configure the communicator type and extension paths through environment variables:

export COMMUNICATOR_TYPE=my_communicator
export EXTENSION_PATHS='["./extensions"]'  # JSON array

For agent-specific configuration with a prefix:

export MYAGENT_COMMUNICATOR_TYPE=my_communicator
export MYAGENT_EXTENSION_PATHS='["./extensions"]'

Then initialize the agent with:

agent = MyAgent(env_prefix="MYAGENT")