> For the complete documentation index, see [llms.txt](https://docs.amesa.com/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://docs.amesa.com/build-multi-agent-systems/define-skills/configure-programmed-algorithms-as-skills/deep-dive-create-and-publish-controllers.md). # Deep Dive: Create and Publish Controllers This guide walks you through everything you need to build an Agent Controller from scratch and publish it to Amesa Orchestration Studio using the `amesa` CLI. By the end, you will have a working controller packaged as an artifact and registered in the AMESA registry. ## What Is a Controller? A **Controller** is a deterministic, hand-coded agent. Instead of learning a policy through reinforcement learning, a controller executes rule-based, algorithmic, or expert-system logic that you write directly. Every call to the controller runs your code — there is no training involved. Use a controller when: * The correct action can be computed directly from sensor values (PID, proportional, rule-based) * You have an existing algorithm or expert system you want to run as an agent * You want a reliable baseline to compare against a trained RL agent * The action logic is fully specified and does not need to adapt to new conditions Controllers and RL Teachers can coexist in the same orchestration. A common pattern is to use a controller as a stabilizing baseline agent alongside RL teachers that handle more complex sub-tasks. {% stepper %} {% step %} ## Understand the Controller Methods A Controller is a subclass of `AgentController` from `amesa_core`. You must implement four methods. ### Required Methods | Method | What it does | | -------------------------- | ------------------------------------------------------------------------------------------------------------ | | `compute_action` | Returns the action to send to the simulator. This is your core control logic — called once per timestep. | | `filtered_sensor_space` | Returns the list of sensor names this controller reads. Only listed sensors are visible in `compute_action`. | | `compute_success_criteria` | Returns `True` when the current episode should be considered a success. | | `compute_termination` | Returns `True` to end the episode early as a failure. | > **All four methods are async.** Declare every method with `async def`. > **`__init__` must accept zero required arguments.** The framework re-creates your controller at the start of every episode by calling `MyController()` with no arguments. Any `__init__` parameter without a default value causes a `TypeError` on every episode reset. ### Method Signatures ```python from typing import Dict, List from amesa_core import AgentController class MyController(AgentController): def __init__(self): # Must accept zero required arguments # All self.* state is wiped at every episode reset pass async def filtered_sensor_space(self, obs_spec) -> List[str]: # Declare which sensors this controller observes # Called once at startup; result is cached ... async def compute_action(self, obs_spec: Dict, action) -> any: # Core control logic — called every timestep # obs_spec: filtered sensor dict (keys match filtered_sensor_space) # action: previous action sent to the sim (None on first step) # Returns: the action to send to the simulator ... async def compute_success_criteria(self, obs_spec: Dict, action) -> bool: # Return True when the episode should end as a success ... async def compute_termination(self, obs_spec: Dict, action) -> bool: # Return True to end the episode early as a failure ... ``` ### `compute_action` Parameters | Parameter | Type | Description | | ---------- | ------ | ----------------------------------------------------------------------------------------------------- | | `obs_spec` | `Dict` | Filtered sensor observations for this step. Keys match the names returned by `filtered_sensor_space`. | | `action` | any | The action sent to the simulator on the **previous** step. `None` on the first step of each episode. | **Return value:** The action to send to the simulator. Must match the simulator's expected action format. {% endstep %} {% step %} ## Write Your Controller The example below implements a proportional controller that maintains dissolved oxygen in a fermentation reactor near a target setpoint by adjusting agitation speed. ```python # fermentation_controller/controller.py from typing import Dict, List from amesa_core import AgentController class FermentationController(AgentController): """ Proportional controller for dissolved oxygen in a fermentation reactor. Adjusts agitation speed based on deviation from target DO setpoint. """ TARGET_DO = 0.65 # target dissolved oxygen fraction TOLERANCE = 0.03 # success band: ±0.03 FAIL_DO_LOW = 0.10 # terminate if DO drops this low KP = 2.5 # proportional gain def __init__(self): # No required arguments — called with no args at every episode reset self.prev_error = 0.0 async def filtered_sensor_space(self, obs_spec) -> List[str]: # Expose only the sensors this controller actually reads return ["dissolved_oxygen", "agitation_rpm"] async def compute_action(self, obs_spec: Dict, action) -> List[float]: do_level = obs_spec["dissolved_oxygen"] error = self.TARGET_DO - do_level # Proportional control: positive error → increase agitation delta_rpm = self.KP * error self.prev_error = error return [float(delta_rpm)] async def compute_success_criteria(self, obs_spec: Dict, action) -> bool: do_level = obs_spec["dissolved_oxygen"] return abs(do_level - self.TARGET_DO) <= self.TOLERANCE async def compute_termination(self, obs_spec: Dict, action) -> bool: do_level = obs_spec["dissolved_oxygen"] return do_level < self.FAIL_DO_LOW ``` ### Accessing Full Sensor List To expose all orchestration sensors to the controller rather than a subset: ```python async def filtered_sensor_space(self, obs_spec) -> List[str]: return obs_spec # pass through — observe everything ``` {% endstep %} {% step %} ## Create the Artifact Directory ### Using the CLI Scaffold (Recommended) ```bash amesa agent new \ --name fermentation-controller \ --type controller \ --description "Proportional DO controller for fermentation reactor" \ --location ./ ``` This creates: ``` fermentation-controller/ fermentation_controller/ __init__.py ← empty; required controller.py ← your AgentController subclass goes here pyproject.toml ``` Replace the placeholder `controller.py` with your implementation from Step 2. ### Creating Manually ```bash mkdir -p fermentation-controller/fermentation_controller touch fermentation-controller/fermentation_controller/__init__.py touch fermentation-controller/fermentation_controller/controller.py touch fermentation-controller/pyproject.toml ``` ### Naming Rules | Layer | Convention | Example | | ------------------- | ------------- | ------------------------- | | Outer directory | kebab-case | `fermentation-controller` | | Inner Python module | snake\_case | `fermentation_controller` | | Source file | by convention | `controller.py` | | `__init__.py` | always empty | *(no content)* | The outer kebab-case directory is used by the CLI only. The inner snake\_case directory is the importable Python module referenced in `pyproject.toml`. Never use the kebab-case outer name in your entrypoint. {% endstep %} {% step %} ## Configure `pyproject.toml` ```toml [build-system] requires = ["setuptools"] build-backend = "setuptools.backends.legacy:build" [project] name = "fermentation-controller" version = "0.1.0" description = "Proportional DO controller for fermentation reactor" dependencies = [ "amesa-core", ] [amesa] type = "agent-controller" entrypoint = "fermentation_controller.controller:FermentationController" ``` ### Field Reference | Field | Where | Required | Description | | -------------- | ----------- | -------- | ------------------------------------------------------------ | | `name` | `[project]` | Yes | Artifact name as it will appear in the registry. Kebab-case. | | `version` | `[project]` | Yes | Semantic version string. | | `description` | `[project]` | Yes | Short human-readable description. | | `dependencies` | `[project]` | Yes | Must include `"amesa-core"`. | | `type` | `[amesa]` | Yes | Must be `"agent-controller"`. | | `entrypoint` | `[amesa]` | Yes | `inner_module.filename:ClassName` | > **Do not add a `variables` field.** The `variables` field is for perceptors only. Including it in a agent `pyproject.toml` causes a validation error on publish. ### Entrypoint Format | Part | Value in this example | | -------------- | ------------------------------------------------------- | | `inner_module` | `fermentation_controller` (inner snake\_case directory) | | `filename` | `controller` (the `.py` file, without extension) | | `ClassName` | `FermentationController` | **Full entrypoint:** `fermentation_controller.controller:FermentationController` {% endstep %} {% step %} ## Verify Your Directory ``` fermentation-controller/ ├── fermentation_controller/ │ ├── __init__.py ← must exist; must be empty │ └── controller.py ← contains FermentationController class └── pyproject.toml ← contains [amesa] section with type and entrypoint ``` Checklist: * [ ] `__init__.py` exists and is empty * [ ] `pyproject.toml` is in the outer directory (same level as `fermentation_controller/`) * [ ] `type` in `[amesa]` is `"agent-controller"` * [ ] `entrypoint` uses the inner snake\_case module name * [ ] `entrypoint` class name matches the class in `controller.py` * [ ] All four required methods are implemented with `async def` * [ ] `__init__` takes no required arguments * [ ] No `variables` field in `[amesa]` * [ ] All imports use `from amesa_core import ...` {% endstep %} {% step %} ## Publish ```bash amesa agent publish ./fermentation-controller/ ``` Or using the flag form: ```bash amesa agent publish --path ./fermentation-controller/ ``` The path must point to the **outer kebab-case directory** containing `pyproject.toml`. {% endstep %} {% step %} ## Confirm the Publish ```bash amesa agent list ``` | Name | Type | Version | Description | UUID | | ----------------------- | ---------- | ------- | ----------------------------- | ---- | | fermentation-controller | controller | 1 | Proportional DO controller... | ... | | {% endstep %} | | | | | | {% endstepper %} | | | | | ## Using a Controller in an Orchestration Controllers can be registered in-process the same way as teachers: ```python from amesa_core import Orchestration, Agent, Sensor from fermentation_controller.controller import FermentationController orchestration = Orchestration() orchestration.add_sensors([ Sensor("dissolved_oxygen", "Dissolved oxygen fraction [0..1]"), Sensor("agitation_rpm", "Agitator speed in RPM"), ]) # Pass the CLASS — not an instance agent = Agent("fermentation-ctrl", FermentationController) orchestration.add_agent(agent) ``` ## Updating a Published Controller Increment `version` in `pyproject.toml` and republish: ```bash amesa agent publish ./fermentation-controller/ ``` ## Deleting a Controller ```bash amesa agent delete ``` ## Troubleshooting ### `TypeError` during episode reset Your `__init__` has a required argument. The framework calls `MyController()` with no arguments at every episode boundary: ```python # WRONG — threshold has no default, raises TypeError on reset def __init__(self, threshold: float): self.threshold = threshold # CORRECT — provide a default value def __init__(self, threshold: float = 0.65): self.threshold = threshold ``` ### `ModuleNotFoundError` on publish The entrypoint is using the kebab-case outer directory name: ```toml # WRONG entrypoint = "fermentation-controller.controller:FermentationController" # CORRECT entrypoint = "fermentation_controller.controller:FermentationController" ``` ### Action type mismatch at runtime `compute_action` must return a value in the format the simulator expects. Returning a plain `float` when the simulator expects a list, or vice versa, causes a coercion error. Check your simulator's action space and match the return type accordingly. ### Legacy import error ```python # WRONG from composabl import AgentController # CORRECT from amesa_core import AgentController ``` --- # Agent Instructions This documentation is published with GitBook. GitBook is the documentation platform designed so that both humans and AI agents can read, navigate, and reason over technical content effectively. Learn more at gitbook.com. ## Querying This Documentation If you need additional information that is not directly available in this page, you can query the documentation dynamically by asking a question. Perform an HTTP GET request on the current page URL with the `ask` query parameter, and the optional `goal` query parameter: ``` GET https://docs.amesa.com/build-multi-agent-systems/define-skills/configure-programmed-algorithms-as-skills/deep-dive-create-and-publish-controllers.md?ask=&goal= ``` `ask` is the immediate question: it should be specific, self-contained, and written in natural language. `goal` is optional and describes the broader end goal you are ultimately trying to accomplish on behalf of the user. GitBook uses it to tailor the answer towards what is most useful for that goal. The response will contain a direct answer to the question and relevant excerpts and sources from the documentation. Use this mechanism when the answer is not explicitly present in the current page, you need clarification or additional context, or you want to retrieve related documentation sections.