Skip to main content
Connect a device to Home Assistant once, and you can control it just by talking to your agent. No app, no buttons. Say what you want, and it happens.

Lights

“Turn on the kitchen light.” “Dim it to twenty percent.”

Climate

“Set the thermostat to seventy.”

Locks & doors

“Lock the front door.”

Media & scenes

“Start movie night.”

​
How it works

1

Connect the device in Home Assistant

Add it from the dashboard, as covered in Add integrations.
2

Build a Local Ability for it

Create a small Ability that reaches the device through Home Assistant on your DevKit.
3

Talk to your agent

Trigger the Ability by voice and run the device with natural commands.
This has to be a Local Ability. The action happens on your DevKit, where Home Assistant runs, so only the Local category can reach it.
New to building Abilities? Start with the Local Ability quickstart.

​
Example: voice-control Tasmota lights

This walkthrough builds a working voice integration end to end: a Local Ability that controls Tasmota smart lights through the Home Assistant instance on your device. Once it’s set up you can say “turn on the kitchen light”, “make it warm”, or “set it to red” and your agent does the rest. It comes in two files, which is the standard Local Ability split:
FileRuns onResponsibility
main.pyStandard Ability runtimeListens to the user, turns speech into an intent with the LLM, and calls the DevKit side.
devkit_functions.pyOpenHome DevKitTalks to Home Assistant’s REST API on the device and performs the actual light action.
This is a Local Ability, so it only runs on real OpenHome DevKit hardware, the device where Home Assistant is installed. See Local Abilities for the full reference on how the two files work together.

​
What you’ll need

  • Home Assistant installed on your DevKit (see Install & manage). The MQTT integration is set up automatically during install.
  • One or more Tasmota smart bulbs or strips on the same network as your DevKit.
  • The OpenHome Live Editor to create the Ability under the Local category.
There are three steps, and the first is automatic. The DevKit already connects to Home Assistant for you, so in practice you just pair your Tasmota device with Home Assistant and build the Ability.

​
Step 1: Connecting to Home Assistant (automatic)

You don’t set up authentication yourself. When you install Home Assistant through OpenHome, the credentials your Ability needs are provisioned on the device, and the DevKit side authenticates on its own. There’s no file to create, no token to paste, and nothing to configure. For this Tasmota example, it just works.

​
Reuse the connection in your own Ability

Building a different Ability that talks to Home Assistant? Reuse OpenHome’s connection logic directly. Drop this into your devkit_functions.py and call _ensure_auth() before any Home Assistant request. It finds the saved credentials, exchanges them for a short-lived access token at Home Assistant’s /auth/token endpoint, and caches the token until just before it expires: 
DEVKIT_AUTH_PATHS = [
    os.environ.get("OPENHOME_HA_AUTH_PATH", ""),
    os.path.expanduser("~/.ha_refresh_token"),
    "/home/openhome/.ha_refresh_token",
    "/root/.ha_refresh_token",
]

_AUTH = {
    "url": None,
    "client_id": None,
    "refresh_token": None,
    "access_token": None,
    "exp": 0,
}


def _ensure_auth():
    if not _AUTH["refresh_token"]:
        path = None
        for p in DEVKIT_AUTH_PATHS:
            if p and os.path.exists(p) and os.access(p, os.R_OK):
                path = p
                break
        if not path:
            log.warning("[HA] no refresh token file found")
            return None, None
        try:
            values = {}
            with open(path, "r") as f:
                for line in f:
                    line = line.strip()
                    if not line or line.startswith("#") or "=" not in line:
                        continue
                    k, v = line.split("=", 1)
                    values[k.strip()] = v.strip().strip('"\'')
        except Exception as exc:
            log.error(f"[HA] read {path}: {exc}")
            return None, None

        url = (values.get("HA_URL") or "").rstrip("/")
        rt = values.get("HA_REFRESH_TOKEN")
        cid = values.get("HA_CLIENT_ID") or (url + "/" if url else None)
        if not url or not rt:
            log.warning("[HA] auth file missing HA_URL or HA_REFRESH_TOKEN")
            return None, None
        _AUTH["url"] = url
        _AUTH["refresh_token"] = rt
        _AUTH["client_id"] = cid

    if _AUTH["access_token"] and time.time() < _AUTH["exp"]:
        return _AUTH["url"], _AUTH["access_token"]

    data = urllib.parse.urlencode({
        "grant_type": "refresh_token",
        "refresh_token": _AUTH["refresh_token"],
        "client_id": _AUTH["client_id"],
    }).encode()
    req = urllib.request.Request(
        f"{_AUTH['url']}/auth/token", data=data,
        headers={"Content-Type": "application/x-www-form-urlencoded"},
        method="POST",
    )
    try:
        with urllib.request.urlopen(req, timeout=10) as resp:
            body = json.loads(resp.read().decode("utf-8", errors="replace"))
    except urllib.error.HTTPError as e:
        if e.code == 400:
            _AUTH["refresh_token"] = None
        log.error(f"[HA] /auth/token HTTP {e.code}")
        return _AUTH["url"], None
    except Exception as exc:
        log.error(f"[HA] /auth/token failed: {exc}")
        return _AUTH["url"], None

    access_token = body.get("access_token")
    expires_in = int(body.get("expires_in") or 1800)
    if not access_token:
        return _AUTH["url"], None
    _AUTH["access_token"] = access_token
    _AUTH["exp"] = time.time() + expires_in - 60
    return _AUTH["url"], access_token
Every Home Assistant request goes through _ensure_auth() first, so the access token is fetched and refreshed for you automatically. If a token expires mid-session, the next call refreshes it. (See _ha_request() in the full devkit_functions.py below for how each call uses it.)

​
Where the credentials live (reference)

You normally never touch this, but for advanced or custom setups it helps to know how the connection is stored. _ensure_auth() reads a small file containing: 
HA_URL=http://localhost:8123
HA_REFRESH_TOKEN=your_home_assistant_refresh_token
HA_CLIENT_ID=http://localhost:8123/
KeyValue
HA_URLThe Home Assistant address as seen from the DevKit. Because Home Assistant runs on the same device, http://localhost:8123 works.
HA_REFRESH_TOKENA refresh token issued by Home Assistant for user (openhome_devkit_user, created during install).
HA_CLIENT_IDThe OAuth client id the token was issued for. Defaults to HA_URL + / if you leave it out.
It checks these locations in order and uses the first one it finds:
  1. The path in the OPENHOME_HA_AUTH_PATH environment variable
  2. ~/.ha_refresh_token
  3. /home/openhome/.ha_refresh_token
OpenHome saves these credentials during install, so this file already exists at one of the paths above, and this is simply where the code looks. For a custom location, point OPENHOME_HA_AUTH_PATH at your own file.
Treat this file like a password, since anyone with the refresh token can control your Home Assistant. Keep it readable only by your user, and never commit it into an Ability or share it.

​
Step 2: Connect your Tasmota device to Home Assistant

Now pair the physical light with Home Assistant. Tasmota talks to the same MQTT broker your DevKit’s Home Assistant already uses, so once it connects, Home Assistant discovers it on its own.
1

Find your Tasmota device's IP

Check your router’s list of connected devices, or open tasmota-XXXXX.local in a browser.
2

Open the Tasmota web UI → Configuration → Configure MQTT

Set these values:
FieldValue
HostYour DevKit’s IP (for example 192.168.18.106)
Port1883
Useropenhome_devkit_user
Passwordadmin123
Topicleave default
Full Topicleave default (%prefix%/%topic%/)
The Host is your DevKit’s IP address, the same one you use to open the Home Assistant dashboard. You can find it in the DevKit section under MQTT, shown as Device IP.
3

Save

Tasmota restarts and connects to the broker.
4

Home Assistant auto-discovers it

The MQTT integration is already loaded in Home Assistant. Once Tasmota connects, it publishes to the discovery topic and Home Assistant’s Tasmota integration picks it up automatically, with no manual configuration needed. The light then shows up as a light.* entity, which is exactly what the Ability controls.

​
Step 3: Build the Ability

In the OpenHome Live Editor, create a new Ability and select the Local category (Local Abilities are the only type that can reach Home Assistant on the device; see Local Abilities). Give it trigger words so users can launch it by voice, such as “smart home” or “lights”. A Local Ability is two files. This example needs no third-party packages, so requirements.txt can stay empty, since everything uses the Python standard library.

​
main.py: voice and intent

main.py runs in the standard Ability runtime. It greets the user, runs a conversation loop, and uses the LLM to convert each spoken request into a structured intent (which integration, which action, which device). It then hands that intent to the DevKit side and speaks a short confirmation. 
import json
import re

from src.agent.capability import MatchingCapability
from src.main import AgentWorker
from src.agent.capability_worker import CapabilityWorker

def _tasmota_light_context():
    return (
        "=== TASMOTA LIGHTS (integration: tasmota_light) ===\n"
        "Smart bulbs/strips controlled via Tasmota over MQTT. The user may have "
        "one or more. Set `hint` to a short natural-language descriptor of which "
        "device (e.g. 'kitchen', 'bedroom lamp', 'the bulb'). Leave hint empty if "
        "only one is plausible from context — the DevKit will pick the only one.\n"
        "Actions:\n"
        "  on_off      params={\"state\":\"on\"|\"off\"|\"toggle\"}\n"
        "  brightness  params={\"percent\":0-100}  (\"bright\"=90, \"dim\"=20, \"half\"=50)\n"
        "  color       params={\"rgb\":[R,G,B]}    (0-255 each; convert any color name yourself)\n"
        "  color_temp  params={\"kelvin\":2000-6500}  (cozy=2700, reading=4000, daylight=6000)"
    )


def _tasmota_light_execute(action, params, hint):
    p = params or {}
    h = (hint or "").strip()
    if action == "on_off":
        state = (p.get("state") or "on").lower()
        if state == "on":
            service = "turn_on"
        elif state == "off":
            service = "turn_off"
        else:
            service = "toggle"
        return ("tasmota_light_action", [service, h, "{}"])
    if action == "brightness":
        try:
            pct = float(p.get("percent", 50))
        except Exception:
            return None
        b = max(0, min(255, int(round(pct * 2.55))))
        return ("tasmota_light_action", ["turn_on", h, json.dumps({"brightness": b})])
    if action == "color":
        rgb = p.get("rgb") or [255, 255, 255]
        try:
            rgb = [max(0, min(255, int(c))) for c in list(rgb)[:3]]
        except Exception:
            return None
        if len(rgb) != 3:
            return None
        return ("tasmota_light_action", ["turn_on", h, json.dumps({"rgb_color": rgb})])
    if action == "color_temp":
        try:
            k = max(2000, min(6500, int(p.get("kelvin", 4000))))
        except Exception:
            return None
        return ("tasmota_light_action", ["turn_on", h, json.dumps({"color_temp_kelvin": k})])
    return None


# Integration registry
INTEGRATION_REGISTRY = {
    "tasmota_light": {
        "context": _tasmota_light_context,
        "execute": _tasmota_light_execute,
    },
}


# Prompt builder
def _build_system_prompt():
    blocks = "\n\n".join(
        handler["context"]() for handler in INTEGRATION_REGISTRY.values()
    )
    return f"""You are a voice assistant for a smart home. Your output is read aloud by TTS to a native US English speaker.

Return ONLY valid JSON (no markdown, no code fences):
{{
  "integration": "<one of the loaded integration names, or 'none' for end_session/unrelated>",
  "action": "<an action listed under that integration, or 'end_session', or 'none'>",
  "params": {{ ... action-specific ... }},
  "hint": "<short natural-language descriptor of which device (e.g. 'kitchen', 'lamp'), or empty>",
  "spoken_response": "<short reply, see TTS RULES>"
}}

TTS RULES (apply to spoken_response):
- Plain spoken English. No markdown, asterisks, dashes, bullets, emojis, URLs, code, symbols.
- No abbreviations. Spell them out: "for example" not "e.g.", "by the way" not "FYI".
- Numbers spoken naturally: "thirty" not "30" when natural.
- No brand or technical jargon. Never say "Tasmota", "entity", "service", "RGB", "kelvin".
- Don't restate the user's command. Just confirm and stop.

LENGTH RULES:
- On/off: under 6 words. ("On it." / "Lights out.")
- Brightness/color/temp: under 10 words. ("Going warm." / "Setting it to red.")
- end_session: under 6 words. ("Catch you later." / "Anytime.")

INTENT RULES:
1. Return ONLY valid JSON.
2. integration MUST be one of the loaded names (or "none").
3. Only use actions listed for the chosen integration.
4. "it"/"that"/"the light" in follow-ups = previously hinted device. Reuse the same hint.
5. Goodbye, "I'm done", "that's all", "never mind", "thanks bye", silence-style closers -> action="end_session".
6. We CANNOT read the device's actual current state. NEVER claim it is on or off without a fresh action.
7. Unrelated request -> integration="none", action="none", short polite reply.

AVAILABLE INTEGRATIONS:
{blocks}
"""


REPROMPTS = ["Sorry?", "One more time?", "Didn't catch that.", "Try saying turn it on or change the color."]


class AgentAbilityCapability(MatchingCapability):
    worker: AgentWorker = None
    capability_worker: CapabilityWorker = None

    #{{register capability}}

    def call(self, worker: AgentWorker):
        self.worker = worker
        self.capability_worker = CapabilityWorker(self.worker)
        self.worker.session_tasks.create(self.run())

    async def run(self):
        try:
            try:
                await self.capability_worker.send_devkit_capability_action(
                    function_name="ha_startup", args=[], timeout=15,
                )
            except Exception as exc:
                self.worker.editor_logging_handler.error(
                    f"[HA] ha_startup ping failed: {exc}"
                )

            system_prompt = _build_system_prompt()
            await self.capability_worker.speak("Smart home ready. What can I do?")

            history = []
            last_hint = ""
            miss_count = 0

            while True:
                msg = await self.capability_worker.user_response()
                if not msg or not msg.strip():
                    prompt = REPROMPTS[min(miss_count, len(REPROMPTS) - 1)]
                    await self.capability_worker.speak(prompt)
                    miss_count += 1
                    if miss_count >= 4:
                        await self.capability_worker.speak("Talk later.")
                        break
                    continue
                miss_count = 0

                hint_note = f" (last device hint: {last_hint})" if last_hint else ""
                user_prompt = f'User said: "{msg}"{hint_note}'

                raw = self.capability_worker.text_to_text_response(
                    user_prompt, history, system_prompt=system_prompt,
                )

                intent = self._parse_json(raw)
                if intent is None:
                    await self.capability_worker.speak("Try that again?")
                    continue

                history.append({"role": "user", "content": user_prompt})
                history.append({"role": "assistant", "content": raw or ""})
                if len(history) > 20:
                    history = history[-20:]

                hint_val = (intent.get("hint") or "").strip()
                if hint_val:
                    last_hint = hint_val

                if intent.get("action") == "end_session":
                    await self.capability_worker.speak(
                        intent.get("spoken_response") or "Bye."
                    )
                    break

                await self._dispatch(intent)

        except Exception as exc:
            self.worker.editor_logging_handler.error(f"[HA] run error: {exc}")
            await self.capability_worker.speak("Something didn't work right.")
        finally:
            self.capability_worker.resume_normal_flow()

    async def _dispatch(self, intent):
        integration = intent.get("integration", "")
        action = intent.get("action", "")
        params = intent.get("params") or {}
        hint = (intent.get("hint") or "").strip()
        spoken = intent.get("spoken_response") or "Done."

        if integration in ("none", "") or action in ("none", ""):
            await self.capability_worker.speak(spoken)
            return

        handler = INTEGRATION_REGISTRY.get(integration)
        if not handler:
            await self.capability_worker.speak("I can't do that one.")
            return

        translated = handler["execute"](action, params, hint)
        if not translated:
            await self.capability_worker.speak("Can't do that one.")
            return

        devkit_fn, devkit_args = translated
        try:
            await self.capability_worker.send_devkit_capability_action(
                function_name=devkit_fn, args=devkit_args, timeout=10,
            )
        except Exception as exc:
            self.worker.editor_logging_handler.error(
                f"[HA] devkit call {devkit_fn} failed: {exc}"
            )
        await self.capability_worker.speak(spoken)

    @staticmethod
    def _parse_json(text):
        if not text:
            return None
        try:
            cleaned = re.sub(r"^```[a-zA-Z]*\n|\n```$", "", text.strip())
            return json.loads(cleaned)
        except Exception:
            return None

​
devkit_functions.py: Home Assistant on the device

devkit_functions.py runs on the DevKit. It authenticates with the credentials file from Step 1, finds the matching Tasmota light entity, and calls Home Assistant’s light services to perform the action. ha_startup is a lightweight ping that also reports which integrations and how many entities are available. 
import os
import sys
import json
import time
import urllib.parse
import urllib.request
import urllib.error

try:
    from devkit_utils.devkit_logging import web_logger as log
except Exception:
    class _StubLogger:
        def info(self, *a, **k): pass
        def warning(self, *a, **k): pass
        def error(self, *a, **k): pass
        def debug(self, *a, **k): pass
    log = _StubLogger()


log.info("[HA devkit] module loaded")

REQUEST_TIMEOUT = 8

DEVKIT_AUTH_PATHS = [
    os.environ.get("OPENHOME_HA_AUTH_PATH", ""),
    os.path.expanduser("~/.ha_refresh_token"),
    "/home/openhome/.ha_refresh_token",
    "/root/.ha_refresh_token",
]

_AUTH = {
    "url": None,
    "client_id": None,
    "refresh_token": None,
    "access_token": None,
    "exp": 0,
}

_NON_TASMOTA_LIGHT_PLATFORMS = ("hue", "lifx", "shelly", "wled", "yeelight", "tplink")


def _ensure_auth():
    if not _AUTH["refresh_token"]:
        path = None
        for p in DEVKIT_AUTH_PATHS:
            if p and os.path.exists(p) and os.access(p, os.R_OK):
                path = p
                break
        if not path:
            log.warning("[HA] no refresh token file found")
            return None, None
        try:
            values = {}
            with open(path, "r") as f:
                for line in f:
                    line = line.strip()
                    if not line or line.startswith("#") or "=" not in line:
                        continue
                    k, v = line.split("=", 1)
                    values[k.strip()] = v.strip().strip('"\'')
        except Exception as exc:
            log.error(f"[HA] read {path}: {exc}")
            return None, None

        url = (values.get("HA_URL") or "").rstrip("/")
        rt = values.get("HA_REFRESH_TOKEN")
        cid = values.get("HA_CLIENT_ID") or (url + "/" if url else None)
        if not url or not rt:
            log.warning("[HA] auth file missing HA_URL or HA_REFRESH_TOKEN")
            return None, None
        _AUTH["url"] = url
        _AUTH["refresh_token"] = rt
        _AUTH["client_id"] = cid

    if _AUTH["access_token"] and time.time() < _AUTH["exp"]:
        return _AUTH["url"], _AUTH["access_token"]

    data = urllib.parse.urlencode({
        "grant_type": "refresh_token",
        "refresh_token": _AUTH["refresh_token"],
        "client_id": _AUTH["client_id"],
    }).encode()
    req = urllib.request.Request(
        f"{_AUTH['url']}/auth/token", data=data,
        headers={"Content-Type": "application/x-www-form-urlencoded"},
        method="POST",
    )
    try:
        with urllib.request.urlopen(req, timeout=10) as resp:
            body = json.loads(resp.read().decode("utf-8", errors="replace"))
    except urllib.error.HTTPError as e:
        if e.code == 400:
            _AUTH["refresh_token"] = None
        log.error(f"[HA] /auth/token HTTP {e.code}")
        return _AUTH["url"], None
    except Exception as exc:
        log.error(f"[HA] /auth/token failed: {exc}")
        return _AUTH["url"], None

    access_token = body.get("access_token")
    expires_in = int(body.get("expires_in") or 1800)
    if not access_token:
        return _AUTH["url"], None
    _AUTH["access_token"] = access_token
    _AUTH["exp"] = time.time() + expires_in - 60
    return _AUTH["url"], access_token


def _ha_request(method, path, body=None, timeout=REQUEST_TIMEOUT):
    url, token = _ensure_auth()
    if not url or not token:
        return 0, "auth unavailable"

    data = json.dumps(body).encode("utf-8") if body is not None else None

    for attempt in (1, 2):
        req = urllib.request.Request(
            url + path, data=data, method=method,
            headers={
                "Content-Type": "application/json",
                "Authorization": f"Bearer {token}",
            },
        )
        try:
            with urllib.request.urlopen(req, timeout=timeout) as resp:
                text = resp.read().decode("utf-8", errors="replace")
                return resp.status, text
        except urllib.error.HTTPError as e:
            if e.code == 401 and attempt == 1:
                log.warning("[HA] <- 401, refreshing token")
                _AUTH["access_token"] = None
                _AUTH["exp"] = 0
                _, token = _ensure_auth()
                if not token:
                    return 401, "auth refresh failed"
                continue
            text = ""
            try:
                text = e.read().decode("utf-8", errors="replace")
            except Exception:
                pass
            log.warning(f"[HA] <- HTTP {e.code}")
            return e.code, text
        except Exception as e:
            log.error(f"[HA] {type(e).__name__}: {e}")
            return 0, str(e)

    return 0, "exhausted retries"


def _emit(data):
    payload = json.dumps(data, separators=(",", ":"))
    sys.stdout.write(payload + "\n")
    try:
        sys.stdout.flush()
    except Exception:
        pass
    return data


def _pick_entity(candidates, name_hint):
    if not candidates:
        return None
    needle = (name_hint or "").strip().lower()
    if not needle:
        return candidates[0].get("entity_id")
    for s in candidates:
        attrs = s.get("attributes") or {}
        fn = (attrs.get("friendly_name") or "").lower()
        eid = s.get("entity_id", "").lower()
        if needle in fn or needle in eid:
            return s.get("entity_id")
    for s in candidates:
        attrs = s.get("attributes") or {}
        fn = (attrs.get("friendly_name") or "").lower()
        eid = s.get("entity_id", "").lower()
        for word in needle.split():
            if word and (word in fn or word in eid):
                return s.get("entity_id")
    return candidates[0].get("entity_id")


# DevKit functions
def ha_startup():
    url, token = _ensure_auth()
    if not url or not token:
        return _emit({"ok": False, "err": "auth unavailable"})

    status, _ = _ha_request("GET", "/api/")
    if status != 200:
        return _emit({"ok": False, "err": f"HA ping HTTP {status}"})

    integrations = []
    status, body = _ha_request("GET", "/api/config/config_entries/entry", timeout=10)
    if status == 200:
        try:
            entries = json.loads(body)
            integrations = sorted({e.get("domain") for e in entries if e.get("domain")})
        except Exception as exc:
            log.warning(f"[HA] config_entries parse: {exc}")

    entities = 0
    status, body = _ha_request("GET", "/api/states", timeout=15)
    if status == 200:
        try:
            entities = len(json.loads(body))
        except Exception as exc:
            log.warning(f"[HA] states parse: {exc}")

    return _emit({
        "ok": True,
        "url": url,
        "integrations": integrations,
        "entities": entities,
    })


def tasmota_light_action(service=None, name_hint="", extra_json="{}"):
    if not service:
        return _emit({"ok": False, "err": "service required"})

    status, body = _ha_request("GET", "/api/states", timeout=15)
    if status != 200:
        return _emit({"ok": False, "err": f"states HTTP {status}"})
    try:
        states = json.loads(body)
    except Exception as exc:
        return _emit({"ok": False, "err": f"parse: {exc}"})

    candidates = []
    for s in states:
        if not isinstance(s, dict):
            continue
        eid = s.get("entity_id", "")
        if not eid.startswith("light."):
            continue
        if s.get("state") in ("unavailable", "unknown"):
            continue
        attrs = s.get("attributes") or {}
        if attrs.get("platform", "") in _NON_TASMOTA_LIGHT_PLATFORMS:
            continue
        candidates.append(s)

    entity_id = _pick_entity(candidates, name_hint)
    if not entity_id:
        log.warning(f"[HA] no tasmota lights to match hint={name_hint!r}")
        return _emit({"ok": False, "err": "no tasmota lights", "hint": name_hint})

    try:
        extra = json.loads(extra_json) if extra_json and extra_json != "{}" else {}
    except Exception:
        extra = {}
    payload = {"entity_id": entity_id}
    payload.update(extra)

    status, _ = _ha_request("POST", f"/api/services/light/{service}", body=payload)
    return _emit({"ok": status in (200, 201), "status": status, "entity_id": entity_id})


FUNCTION_REGISTRY = {
    "ha_startup": ha_startup,
    "tasmota_light_action": tasmota_light_action,
}

log.info(f"[HA devkit] registry: {list(FUNCTION_REGISTRY.keys())}")


def list_functions():
    print("=" * 60)
    print("  HA Devkit Functions (Tasmota Lights)")
    print("=" * 60)
    for name, func in FUNCTION_REGISTRY.items():
        doc = (func.__doc__ or "").strip().split("\n")[0]
        print(f"  {name}: {doc}")


FUNCTION_REGISTRY["list_functions"] = list_functions


def main():
    if len(sys.argv) < 2:
        print("Usage: python3 devkit_functions.py <function_name> [args...]")
        sys.exit(1)
    func_name = sys.argv[1]
    func_args = sys.argv[2:]
    if func_name in ("--help", "-h"):
        list_functions()
        sys.exit(0)
    if func_name not in FUNCTION_REGISTRY:
        print(f"[error] unknown function '{func_name}'")
        sys.exit(1)
    try:
        FUNCTION_REGISTRY[func_name](*func_args)
        sys.exit(0)
    except TypeError as e:
        print(f"[error] wrong arguments for '{func_name}': {e}")
        sys.exit(1)
    except Exception as e:
        print(f"[error] {e}")
        sys.exit(1)


if __name__ == "__main__":
    main()
When you save in the Live Editor with the DevKit connected, the files sync to the device automatically. See Sync Local Abilities with the DevKit if your DevKit was offline while editing.

​
Step 4: Talk to it

Launch the Ability with one of your trigger words, then speak naturally. The conversation loop keeps listening until you say you’re done.
  • “Turn on the kitchen light.”
  • “Make it warm.” (reuses the last light you mentioned)
  • “Set the lamp to red.”
  • “Dim the bedroom to twenty percent.”
  • “That’s all, thanks.” (ends the session)
If you have more than one light, include a short descriptor like “kitchen” or “bedroom lamp”, and the DevKit side matches it against your Home Assistant light names and picks the right one. With a single light, no descriptor is needed.

​
Extending it

This template is built around an integration registry. To support more than Tasmota lights, add another entry to INTEGRATION_REGISTRY in main.py (a context function describing its actions, and an execute function that maps an intent to a DevKit call) and a matching handler in FUNCTION_REGISTRY in devkit_functions.py. The voice loop, prompt builder, and dispatch logic stay the same.

​
Edge cases

The Ability checks for a running Home Assistant on the device when it starts. If Home Assistant isn’t installed, or is still starting up, the Ability can’t control anything. Make sure it’s installed and running (see Install & manage), then relaunch.
If the Ability says it can’t find a light, the Tasmota device probably isn’t connected to Home Assistant yet. Recheck Step 2. Once Tasmota connects to the broker, Home Assistant discovers it as a light and the Ability can control it.
This example deliberately ignores lights from other ecosystems (such as Hue, LIFX, Shelly, WLED, Yeelight, and TP-Link) so it only acts on your Tasmota bulbs. To control other brands, extend the Ability as described above.
Add a short descriptor to your command, like “kitchen” or “bedroom lamp”. The DevKit side matches it against your Home Assistant light names. If it can’t match your words to a specific light, it falls back to the first one it finds, so naming your lights clearly in Home Assistant helps.
A bulb that’s cut off from power shows as unavailable in Home Assistant and is skipped. Restore power so it reconnects, then try your command again.
The Ability sends actions; it doesn’t read a light’s current state. It will never claim a light is on or off without performing an action, so ask it to turn things on, off, or toggle rather than asking for status.
If your speech isn’t recognized, the Ability asks you to repeat. After a few unclear tries in a row it ends the session politely. Just relaunch it with your trigger word.

​
See also