OpenHome Ability SDK — Complete Reference

await self.capability_worker.speak("Hello! How can I help?")

await self.capability_worker.text_to_speech("Welcome aboard.", "pNInz6obpgDQGcFmaJgB")

user_input = await self.capability_worker.user_response()

full_input = await self.capability_worker.wait_for_complete_transcription()

import re

async def first_function(self):
    full_input = await self.capability_worker.wait_for_complete_transcription()
    reminder_text = re.sub(r"^\s*remind\b", "", full_input, flags=re.IGNORECASE).strip()
    await self.capability_worker.speak(f"Creating reminder: {reminder_text}")

answer = await self.capability_worker.run_io_loop("What's your favorite color?")

confirmed = await self.capability_worker.run_confirmation_loop("Should I send this email?")
if confirmed:
    # send it

response = self.capability_worker.text_to_text_response(
    "What's the capital of France?",
    history=[
        {"role": "user", "content": "Let's do geography trivia"},
        {"role": "assistant", "content": "Great, I'll ask you questions!"}
    ],
    system_prompt="You are a geography quiz host. Keep answers under 1 sentence."
)

import requests
audio = requests.get("https://example.com/song.mp3")
await self.capability_worker.play_audio(audio.content)

await self.capability_worker.play_from_audio_file("notification.mp3")

self.capability_worker.start_audio_recording()

self.capability_worker.stop_audio_recording()

wav_data = self.capability_worker.get_audio_recording()

length = self.capability_worker.get_audio_recording_length()

self.capability_worker.flush_audio_recording()

async def record_voice_note(self):
    await self.capability_worker.speak("I'll record a voice note. Start speaking.")
    self.capability_worker.start_audio_recording()

    await self.worker.session_tasks.sleep(10)  # Record for 10 seconds

    self.capability_worker.stop_audio_recording()

    duration = self.capability_worker.get_audio_recording_length()
    wav_file = self.capability_worker.get_audio_recording()

    await self.capability_worker.speak(f"Got it. Recorded {duration} of audio.")
    self.capability_worker.resume_normal_flow()

await self.capability_worker.stream_init()

await self.capability_worker.send_audio_data_in_stream(audio_bytes, chunk_size=4096)

await self.capability_worker.stream_end()

async def stream_long_audio(self):
    await self.capability_worker.stream_init()
    response = requests.get("https://example.com/long-audio.mp3")
    await self.capability_worker.send_audio_data_in_stream(response.content)
    await self.capability_worker.stream_end()

exists = await self.capability_worker.check_if_file_exists(
    "user_prefs.json",
    in_ability_directory=False
)

await self.capability_worker.write_file(
    "user_prefs.json",
    '{"theme": "dark"}',
    False
)

data = await self.capability_worker.read_file(
    "user_prefs.json",
    in_ability_directory=False
)

await self.capability_worker.delete_file(
    "user_prefs.json",
    in_ability_directory=False
)

files = await self.capability_worker.get_user_data_file_names()

# ✅ RECOMMENDED — delete + write (explicit, safe)
async def save_json(self, filename, data):
    if await self.capability_worker.check_if_file_exists(filename, False):
        await self.capability_worker.delete_file(filename, False)
    await self.capability_worker.write_file(filename, json.dumps(data), False)

# ✅ ALTERNATIVE — mode="w" (shorthand, but easy to forget)
await self.capability_worker.write_file("prefs.json", json.dumps(data), False, mode="w")

# ❌ WRONG — appending to JSON
await self.capability_worker.write_file("prefs.json", json.dumps(new_data), False)
# Result: {"old":"data"}{"new":"data"}  ← broken JSON, no error thrown

result = self.capability_worker.create_key(
    key="user_preferences",
    value={
        "language": "en",
        "theme": "dark",
        "notifications": True
    }
)

result = self.capability_worker.update_key(
    key="user_preferences",
    value={
        "language": "en",
        "theme": "light",
        "notifications": False
    }
)

result = self.capability_worker.delete_key("user_preferences")

all_context = self.capability_worker.get_all_keys()

preferences = self.capability_worker.get_single_key("user_preferences")

# 1) Create state
self.capability_worker.create_key(
    key="conversation_1234",
    value={
        "last_intent": "book_flight",
        "destination": "Dubai",
        "travel_date": "2026-04-01",
        "step": "awaiting_confirmation"
    }
)

# 2) Update state
self.capability_worker.update_key(
    key="conversation_1234",
    value={
        "last_intent": "book_flight",
        "destination": "Dubai",
        "travel_date": "2026-04-01",
        "step": "confirmed"
    }
)

# 3) Read state
context = self.capability_worker.get_single_key("conversation_1234")

existing = self.capability_worker.get_single_key("user_preferences")
if existing:
    self.capability_worker.update_key("user_preferences", updated_value)
else:
    self.capability_worker.create_key("user_preferences", updated_value)

await self.capability_worker.send_data_over_websocket("music-mode", {"mode": "on"})

await self.capability_worker.send_devkit_action("led_on")

self.capability_worker.resume_normal_flow()

interrupt_signal = await self.capability_worker.send_interrupt_signal()

self.worker.editor_logging_handler.info("Something happened")
self.worker.editor_logging_handler.error("Something broke")
self.worker.editor_logging_handler.warning("Something suspicious")
self.worker.editor_logging_handler.debug("Debugging")

self.worker.session_tasks.create(self.my_async_method())

await self.worker.session_tasks.sleep(5.0)

timezone = self.capability_worker.get_timezone()

token = self.capability_worker.get_token("google")
self.worker.editor_logging_handler.info(token)

user_ip = self.worker.user_socket.client.host
self.worker.editor_logging_handler.info(f"User connected from: {user_ip}")

import requests

def get_user_location(self):
    """Get user's city and timezone from their IP address."""
    try:
        ip = self.worker.user_socket.client.host
        resp = requests.get(f"http://ip-api.com/json/{ip}", timeout=5)
        if resp.status_code == 200:
            data = resp.json()
            if data.get("status") == "success":
                isp = data.get("isp", "").lower()
                cloud_indicators = ["amazon", "aws", "google", "microsoft", "azure", "digitalocean"]
                if any(c in isp for c in cloud_indicators):
                    self.worker.editor_logging_handler.warning("Cloud IP detected, location may be inaccurate")
                    return None
                return {
                    "city": data.get("city"),
                    "region": data.get("regionName"),
                    "country": data.get("country"),
                    "timezone": data.get("timezone"),
                    "lat": data.get("lat"),
                    "lon": data.get("lon"),
                }
    except Exception as e:
        self.worker.editor_logging_handler.error(f"Geolocation error: {e}")
    return None

keys = self.capability_worker.get_api_keys()
gmail_password = keys.get("gmail_app_password")
self.worker.editor_logging_handler.info(keys)

async def call(self, worker):
    self.worker = worker
    self.capability_worker = CapabilityWorker(self)

    keys = self.capability_worker.get_api_keys()
    required_keys = ["openai_api_key", "sendgrid_api_key"]
    missing = [k for k in required_keys if not keys.get(k)]

    if missing:
        missing_list = ", ".join(missing)
        await self.capability_worker.speak(
            f"I'm missing required keys: {missing_list}. "
            "Please set them in Settings under API Keys."
        )
        self.capability_worker.resume_normal_flow()
        return

    openai_key = keys.get("openai_api_key")
    sendgrid_key = keys.get("sendgrid_api_key")

    self.worker.editor_logging_handler.info("Required API keys found. Calling service...")
    # Use openai_key and sendgrid_key here

history = self.capability_worker.get_full_message_history()
self.worker.editor_logging_handler.info(f"Messages so far: {len(history)}")

self.capability_worker.delete_conversation_history()

self.capability_worker.update_personality_agent_prompt(
    "The user prefers concise answers and metric units."
)

self.history = []

async def main_loop(self):
    system = "You are a helpful cooking assistant. Keep answers under 2 sentences."
    while True:
        user_input = await self.capability_worker.user_response()
        if "exit" in user_input.lower():
            break
        self.history.append({"role": "user", "content": user_input})
        response = self.capability_worker.text_to_text_response(  # No await!
            user_input,
            history=self.history,
            system_prompt=system
        )
        self.history.append({"role": "assistant", "content": response})
        await self.capability_worker.speak(response)
    self.capability_worker.resume_normal_flow()

async def play_track(self, audio_bytes):
    # 1. Enter music mode (system stops listening, won't interrupt)
    self.worker.music_mode_event.set()
    await self.capability_worker.send_data_over_websocket("music-mode", {"mode": "on"})

    # 2. Play the audio
    await self.capability_worker.play_audio(audio_bytes)

    # 3. Exit music mode (system resumes listening)
    await self.capability_worker.send_data_over_websocket("music-mode", {"mode": "off"})
    self.worker.music_mode_event.clear()

def classify_intent(self, user_input: str) -> dict:
    prompt = (
        "Classify this user input. Return ONLY valid JSON.\n"
        '{"intent": "weather|timer|music|chat", "confidence": 0.0-1.0}\n\n'
        f"User: {user_input}"
    )
    raw = self.capability_worker.text_to_text_response(prompt)  # No await!
    clean = raw.replace("```json", "").replace("```", "").strip()
    try:
        return json.loads(clean)
    except json.JSONDecodeError:
        return {"intent": "chat", "confidence": 0.0}

async def do_something(self):
    try:
        response = requests.get("https://api.example.com/data", timeout=10)
        if response.status_code == 200:
            data = response.json()
            await self.capability_worker.speak(f"Here's what I found: {data['result']}")
        else:
            await self.capability_worker.speak("Sorry, I couldn't get that information right now.")
    except Exception as e:
        self.worker.editor_logging_handler.error(f"API error: {e}")
        await self.capability_worker.speak("Something went wrong. Let me hand you back.")
    self.capability_worker.resume_normal_flow()  # ALWAYS called

ABILITY_VOICE_ID = "pNInz6obpgDQGcFmaJgB"  # Deep, American, male narration voice

async def speak(self, text: str):
    await self.capability_worker.text_to_speech(text, ABILITY_VOICE_ID)