Inputs - Innate Docs

An input device sends data to the agent while it runs. Use one to connect an added sensor, a webhook, or any event the agent should react to.

Write an input

Save this as ~/innate-os/workspace/inputs/ticker.py. It sends a message every second:

import threading

from brain_client.inputs.types import InputDevice


class Ticker(InputDevice):
    def __init__(self):
        super().__init__()
        self._stop = threading.Event()

    @property
    def name(self) -> str:
        return "ticker"

    def on_open(self):  # the agent started
        self._stop.clear()

        def loop():
            while not self._stop.is_set():
                self.send_data({"text": "tick!"}, data_type="chat_in")
                self._stop.wait(1.0)

        threading.Thread(target=loop, daemon=True).start()

    def on_close(self):  # the agent stopped
        self._stop.set()

Then add it to an agent:

def get_inputs(self) -> list[InputRef]:
    return [Ticker]

Data types

`data_type`	Use for	Example
`chat_in`	Text, as if the user said it	`{"text": "Hello robot"}`
`custom`	Any other structured data	`{"air_quality": 42, "unit": "AQI"}`

The microphone

MARS ships with MicroInput, which streams speech from the built-in microphone. Add it to an agent to talk to it. Its source, workspace/inputs/micro_input.py, is the best example of a real input.