Developing Control Applications

This guide covers how to develop custom control applications using ai_nn_controller.

The framework is domain-agnostic and supports any type of network node — optical, wireless, RAN, core network, and more.

Application Structure

A control application consists of:

1. Class Definition: Inherits from AicApp

2. @aic_app Decorator: Registers the app with the framework

3. Configuration Attributes: Define measurements and controls

4. process() Method: Implement your control logic

Basic Template

from ai_nn_controller.decorators.aic_app import aic_app
from ai_nn_controller.AicApp import AicApp
from ai_nn_controller.AicController import AicController

@aic_app(name="MyControlApp")
class MyControlApp(AicApp):
    """Description of your control application."""

    # Unique identifier for this app
    aic_app_id = 100

    # Processing interval in seconds
    control_loop_update_time = 2

    # Measurements to subscribe to (node_id -> metric names)
    read_measurements = {
        3: ["gain", "power"],
        8: ["preamp_gain", "booster_gain"]
    }

    # Control functions available (node_id -> command names)
    control_functions = {
        8: ["SET_GAIN", "SET_VOA"]
    }

    # Note: cell_ids and send_commands are auto-generated by @aic_app decorator
    # cell_ids will be: [3, 8] (union of read_measurements and control_functions keys)

    @classmethod
    def process(cls, measurements):
        """
        Process incoming measurements.

        Args:
            measurements: Dict[int, List[Dict]] - node_id to measurement list
        """
        # Your control logic here
        pass

if __name__ == "__main__":
    AicController(with_api=True, verbose=True).run()

Template App Generator (First-Time Walkthrough)

If you are new to the controller, the easiest way to start is to generate a template app based on the nodes that have already registered. The generator builds a complete app folder and a focused test compose file that only includes the registered nodes plus the new template app.

What the generator creates:

• control_applications/template_app/ with aic_app.py, commands.py, aic_app.conf, Dockerfile, and requirements.txt

• docker-compose-test.yml in the project root with: - Redis + register + broker - Only the nodes that are currently registered - The generated template_app service

Step 1: Start the baseline stack

From the project root:

docker compose up -d

This starts Redis, the register, the broker, and the sample nodes. The nodes must register before the generator runs so the register has the PM/control lists.

Step 2: Wait for nodes to register

You should see registration logs in the register container:

docker logs -f aic_register

Look for lines like:

• REGISTER network_node node: 3

• Network Nodes sharing available PMs

• Network Nodes sharing available Control Functions

Step 3: Generate the template app + test compose

Run the generator inside the register container:

docker exec -it aic_register python3 /register/generate_template_app.py

Optional flags (shown with defaults):

docker exec -it aic_register python3 /register/generate_template_app.py \
  --app-name template_app \
  --app-id 100 \
  --update-time 2 \
  --output-root /control_applications \
  --compose-output /workspace/docker-compose-test.yml

The generator prints where it wrote the app folder and the compose file. It also prints the exact command to run the test compose.

Step 4: Run the generated test compose

Stop the original stack, then start the test stack:

docker compose down
docker compose -f docker-compose-test.yml up --build

This brings up only the registered nodes and the new template_app.

Configuration Attributes

control_loop_update_time

How often process() is called, in seconds:

control_loop_update_time = 2  # Process every 2 seconds

read_measurements

Dictionary mapping node IDs to lists of measurement names:

read_measurements = {
    3: ["amp1_target_gain", "amp1_gain_tilt", "amp1_target_power"],
    4: ["roadm1_preamp_gain", "roadm1_booster_gain"],
    8: ["roadm3_preamp_gain", "roadm3_booster_gain"]
}

Use MeasurementsHandler.ALL_MEASUREMENTS to receive all available metrics:

from ai_nn_controller.enums import MeasurementsHandler

read_measurements = {
    3: [MeasurementsHandler.ALL_MEASUREMENTS]
}

control_functions

Dictionary mapping node IDs to available command names:

control_functions = {
    3: ["SET_GAIN"],
    8: ["SET_GAIN", "SET_VOA", "SET_TILT"]
}

cell_ids (Auto-Generated)

List of node IDs to subscribe to. This is auto-generated by the @aic_app decorator from the union of read_measurements and control_functions keys. You do not need to define it in your app:

# These definitions:
read_measurements = {3: ["gain"], 4: ["power"]}
control_functions = {8: ["SET_GAIN"]}

# Will auto-generate:
# cell_ids = [3, 4, 8]

Processing Measurements

The process() method receives measurements as a dictionary:

@classmethod
def process(cls, measurements):
    """
    Args:
        measurements: {
            node_id: [
                {"metric1": value, "metric2": value, ...},  # Oldest
                {"metric1": value, "metric2": value, ...},  # ...
                {"metric1": value, "metric2": value, ...}   # Newest
            ],
            ...
        }
    """
    # Check if we have measurements
    if not measurements:
        print("No measurements received yet")
        return

    # Get latest measurement from node 3
    if 3 in measurements and measurements[3]:
        latest = measurements[3][-1]  # Last item is newest
        gain = latest.get("amp1_target_gain", 0)
        print(f"Current gain: {gain}")

    # Process multiple measurements (batch processing)
    for node_id, measurement_list in measurements.items():
        for measurement in measurement_list:
            # Process each measurement
            pass

Sending Commands

Use add_command() to queue commands for sending:

@classmethod
def process(cls, measurements):
    # Get current gain
    latest = measurements.get(3, [{}])[-1]
    current_gain = latest.get("amp1_target_gain", 0)

    # Send command if gain is too high
    if current_gain > 25:
        cls.add_command((
            "SET_GAIN",  # Command name (must be registered)
            {
                "node_id": 3,
                "value": {
                    "amp_type": "line",
                    "target_gain": 20.0
                }
            }
        ))

Command Format

Commands are tuples of (command_name, payload):

(
    "SET_GAIN",  # Command name string
    {
        "node_id": 8,           # Target node
        "value": {              # Command parameters
            "amp_type": "preamp",
            "target_gain": 15.0
        }
    }
)

Command Validators (Optional)

Command validators provide guardrails for commands sent via the REST API or MCP tools. They are completely optional - if no validator is defined, commands pass through unchanged.

This is useful when you want to:

• Enforce safety limits (e.g., maximum gain values)

• Validate parameter combinations

• Reject commands based on current system state

from ai_nn_controller.decorators.aic_app import aic_app
from ai_nn_controller.decorators.command_validator import command_validator
from ai_nn_controller.AicApp import AicApp

@aic_app(name="NetworkApp2")
class SafeControlApp(AicApp):
    control_functions = {8: ["SET_GAIN", "SET_VOA"]}
    # cell_ids auto-generated: [8]

    # Safety limits
    MAX_GAIN = 25.0  # dB
    MIN_GAIN = 0.0   # dB

    # IMPORTANT: @classmethod must be ABOVE @command_validator
    @classmethod
    @command_validator("SET_GAIN")
    def validate_set_gain(cls, params: dict) -> tuple[bool, str | None]:
        """
        Validate SET_GAIN commands before execution.

        Args:
            params: Dict with node_id, target_gain, amp_type, etc.

        Returns:
            (True, None) if valid, (False, "error message") if invalid
        """
        target_gain = params.get("target_gain")

        if target_gain is None:
            return False, "target_gain parameter is required"

        if target_gain > cls.MAX_GAIN:
            return False, f"target_gain {target_gain} exceeds max {cls.MAX_GAIN} dB"

        if target_gain < cls.MIN_GAIN:
            return False, f"target_gain {target_gain} is below min {cls.MIN_GAIN} dB"

        return True, None

    # SET_VOA has no validator - passes through unchanged

    @classmethod
    def process(cls, measurements):
        pass

Validator Behavior

• No validator defined: Command passes through unchanged (backward compatible)

• Validator returns ``(True, None)``: Command is sent

• Validator returns ``(False, “reason”)``: Command is rejected

When a command is rejected via the REST API, the response includes:

{
  "status": "rejected",
  "reason": "target_gain 30.0 exceeds max 25.0 dB"
}

Note

Validators are primarily useful when the FastAPI REST API or MCP tools are being used to send manual control commands. Commands sent internally via add_command() in process() will also be validated if a validator is defined.

Agent-Controlled Operations (Optional)

Agent-controlled operations allow MCP/AI agents to execute logic inside the process loop, with access to live measurements and the app’s internal state. Unlike regular MCP control tools (which bypass process() via send_manual_control()), these handlers are synchronized with the process cycle.

This is useful when you want to:

• Give agents access to live measurements for decision-making

• Let agents trigger complex logic that uses app-internal state

• Keep agent actions synchronized with the control loop

• Return computed results back to the agent

from ai_nn_controller.decorators.aic_app import aic_app
from ai_nn_controller.decorators.agent_controlled import agent_controlled
from ai_nn_controller.AicApp import AicApp

@aic_app(name="SmartApp")
class SmartApp(AicApp):
    read_measurements = {8: ["preamp_gain", "signal_power"]}
    control_functions = {8: ["SET_GAIN"]}
    MAX_GAIN = 25.0

    # IMPORTANT: @classmethod must be ABOVE @agent_controlled
    @classmethod
    @agent_controlled(
        name="optimize_gain",
        description="Optimize gain based on a strategy",
        schema={
            "properties": {
                "node_id": {"type": "integer"},
                "strategy": {"type": "string", "enum": ["max_snr", "min_power"]}
            },
            "required": ["node_id", "strategy"]
        }
    )
    def handle_optimize_gain(cls, request, measurements):
        """Runs inside the process loop with access to live measurements."""
        node_id = request["node_id"]
        latest = measurements.get(node_id, [{}])[-1] or {}
        current_gain = latest.get("preamp_gain", 15.0)
        new_gain = min(current_gain + 2.0, cls.MAX_GAIN)

        cls.add_command(("SET_GAIN", {
            "node_id": node_id,
            "value": {"target_gain": new_gain}
        }))

        # Return value goes back to the MCP caller
        return {"status": "applied", "new_gain": new_gain}

    @classmethod
    def process(cls, measurements):
        # Normal control logic — agent requests are handled separately
        pass

Handler Signature

The handler receives two arguments:

• request (dict): The MCP tool arguments (matching the schema you defined)

• measurements (dict): Current measurements from the process loop, same format as what process() receives

The handler should return a dict that will be sent back to the MCP caller.

How It Works

Agent (MCP tool call) → push AgentRequest to queue → process() cycle picks it up
  → handler runs with (request, measurements) → response set → MCP returns result

1. The MCP handler creates an AgentRequest with a threading.Event

2. The request is pushed to cls.agent_requests

3. On the next process cycle, the controller drains the queue and calls the handler

4. The handler’s return value is stored in the request and the event is set

5. The MCP handler wakes up and returns the result to the caller

Edge Cases

• App not running: MCP handler returns an error immediately

• Timeout: Defaults to control_loop_update_time * 3 + 5 seconds; returns a timeout error

• Handler exception: Caught by the controller, error returned to MCP caller

• App stopped while request pending: All pending requests are drained with an error

Declaring Plugin Dependencies (Optional)

Control applications can depend on plugins — reusable, independently installable packages that provide typed access to external services (a time-series database, an ML model registry, a monitoring backend, etc.). Declare a dependency with required_plugins and access the loaded plugin through cls.plugins:

from ai_nn_controller.decorators.aic_app import aic_app
from ai_nn_controller.AicApp import AicApp

@aic_app(name="MonitoredApp")
class MonitoredApp(AicApp):
    required_plugins = ["ConsolePlugin"]
    read_measurements = {3: ["gain", "power"]}
    control_functions = {}

    @classmethod
    def process(cls, measurements):
        console = cls.plugins["ConsolePlugin"]
        latest = measurements.get(3, [{}])[-1]
        console.log_measurement(3, latest)

cls.plugins is populated by AicController at startup, after all plugin and app entry points have loaded. If a plugin listed in required_plugins isn’t installed (or wasn’t registered via the ai_nn_controller.plugin_init entry-point group), the controller refuses to start the app and raises RuntimeError naming the missing plugin.

See Developing Plugins for details on writing your own plugin.

State Variables

Store state between process() calls using class attributes:

@aic_app(name="StatefulApp")
class StatefulApp(AicApp):
    # State variables
    measurement_history = []
    last_command_time = 0
    command_counter = 0

    @classmethod
    def process(cls, measurements):
        import time

        # Track measurement history
        if 3 in measurements:
            cls.measurement_history.extend(measurements[3])
            # Keep only last 100
            cls.measurement_history = cls.measurement_history[-100:]

        # Rate-limit commands (once per 10 seconds)
        current_time = time.time()
        if current_time - cls.last_command_time > 10:
            cls.add_command(...)
            cls.last_command_time = current_time
            cls.command_counter += 1

Multiple Applications

You can define multiple apps in the same file:

@aic_app(name="MonitoringApp")
class MonitoringApp(AicApp):
    aic_app_id = 1
    control_loop_update_time = 2
    read_measurements = {3: ["gain"]}
    control_functions = {}
    # cell_ids auto-generated: [3]

    @classmethod
    def process(cls, measurements):
        # Monitoring logic
        pass

@aic_app(name="ControlApp")
class ControlApp(AicApp):
    aic_app_id = 2
    control_loop_update_time = 2
    read_measurements = {8: ["preamp_gain"]}
    control_functions = {8: ["SET_GAIN"]}
    # cell_ids auto-generated: [8]

    @classmethod
    def process(cls, measurements):
        # Control logic
        pass

if __name__ == "__main__":
    AicController(with_api=True).run()

Both apps run in separate threads and share the same API server.

Auto-Generated Endpoints

The @aic_app decorator automatically creates REST endpoints:

Auto-Generated Endpoints

Endpoint	Method	Description
/apps/{name}/info	GET	App configuration
/apps/{name}/state	GET	Current state
/apps/{name}/state	PUT	Update state (running/paused/stopped)
/apps/{name}/measurements	GET	Latest measurements
/apps/{name}/control	POST	Send manual command

Auto-Generated MCP Tools

For each app, MCP tools are automatically created:

• {AppName}_get_state - Get current state

• {AppName}_set_state - Set state

• {AppName}_get_measurements - Get measurements

• {AppName}_{command_name} - For each control function

• {AppName}_{operation_name} - For each @agent_controlled operation

Best Practices

1. Keep process() Fast: Avoid blocking operations

2. Handle Missing Data: Check if measurements exist before accessing

3. Use Rate Limiting: Don’t send commands every cycle

4. Log Important Events: Use print or logging for debugging

5. Validate Inputs: Check measurement values before acting

Example: Complete Application

import time
from ai_nn_controller.decorators.aic_app import aic_app
from ai_nn_controller.AicApp import AicApp
from ai_nn_controller.AicController import AicController

@aic_app(name="GainController")
class GainController(AicApp):
    """
    Maintains amplifier gain within target range.
    Monitors node 3 and sends corrections to node 8.
    """

    aic_app_id = 100
    control_loop_update_time = 2

    read_measurements = {
        3: ["amp1_target_gain", "amp1_gain_tilt"],
        8: ["roadm3_preamp_gain"]
    }

    control_functions = {
        8: ["SET_GAIN"]
    }

    # Note: cell_ids and send_commands are auto-generated
    # cell_ids will be: [3, 8]

    # Configuration
    TARGET_GAIN = 20.0
    GAIN_TOLERANCE = 2.0
    MIN_COMMAND_INTERVAL = 10  # seconds

    # State
    last_correction_time = 0

    @classmethod
    def process(cls, measurements):
        if not measurements:
            return

        # Get current gain from node 3
        amp_data = measurements.get(3, [{}])[-1]
        current_gain = amp_data.get("amp1_target_gain")

        if current_gain is None:
            return

        print(f"[GainController] Current gain: {current_gain:.2f}")

        # Check if correction needed
        gain_error = current_gain - cls.TARGET_GAIN

        if abs(gain_error) > cls.GAIN_TOLERANCE:
            # Rate limit corrections
            current_time = time.time()
            if current_time - cls.last_correction_time > cls.MIN_COMMAND_INTERVAL:
                # Calculate correction
                new_gain = cls.TARGET_GAIN

                print(f"[GainController] Correcting gain: {current_gain:.2f} -> {new_gain:.2f}")

                cls.add_command((
                    "SET_GAIN",
                    {
                        "node_id": 8,
                        "value": {
                            "amp_type": "preamp",
                            "target_gain": new_gain
                        }
                    }
                ))

                cls.last_correction_time = current_time

if __name__ == "__main__":
    AicController(with_api=True, api_port=8000, verbose=True).run()

Next Steps

• Developing Network Nodes - Build network nodes with the controlled_entity framework

• Developing Plugins - Write reusable plugins for external services

• Defining Commands - Define custom commands

• Core Framework API (ai_nn_controller) - Full API reference

• Basic Application Example - More examples