Official starter templates for building OpenHome abilities
OpenHome provides official templates for every common ability pattern. Each template is a minimal, working boilerplate that demonstrates a core architectural approach.
Skill is the workhorse—triggered by a hotword, runs its task, and hands control back via resume_normal_flow().Background Daemon starts automatically when a user connects and runs in a while True loop for the entire session.Local abilities run directly on Raspberry Pi hardware, bypassing the cloud sandbox entirely.
Type: Skill · Complexity: MinimalThe absolute minimum scaffolding for a working Skill. Use this as your starting point for any new Skill.Key SDK methods:speak(), user_response(), resume_normal_flow()
from src.agent.capability import MatchingCapabilityfrom src.main import AgentWorkerfrom src.agent.capability_worker import CapabilityWorkerclass BasicTemplateCapability(MatchingCapability): worker: AgentWorker = None capability_worker: CapabilityWorker = None #{{register capability}} async def run(self): # Greet the user await self.capability_worker.speak("Hi! How can I help you today?") # Listen for input user_input = await self.capability_worker.user_response() # Generate a response response = self.capability_worker.text_to_text_response( f"Give a short, helpful response to: {user_input}" ) # Speak the response await self.capability_worker.speak(response) # IMPORTANT: Always call this when done self.capability_worker.resume_normal_flow() def call(self, worker: AgentWorker): self.worker = worker self.capability_worker = CapabilityWorker(self) self.worker.session_tasks.create(self.run())
The register_capability comment tag is required boilerplate—copy it exactly. The platform handles config.json automatically at runtime.
Type: Skill · Complexity: MinimalExtends basic-template with an outbound HTTP request to an external REST API. Demonstrates the request/response cycle within the ability lifecycle.Key SDK methods:text_to_text_response(), speak(), resume_normal_flow()
import requestsfrom src.agent.capability import MatchingCapabilityfrom src.main import AgentWorkerfrom src.agent.capability_worker import CapabilityWorkerAPI_URL = "https://api.example.com/data"API_HEADERS = { "Authorization": "Bearer YOUR_API_KEY_HERE", "Content-Type": "application/json",}class ApiTemplateCapability(MatchingCapability): worker: AgentWorker = None capability_worker: CapabilityWorker = None #{{register capability}} async def fetch_data(self, query: str) -> str | None: try: response = requests.get( API_URL, headers=API_HEADERS, params={"q": query}, ) if response.status_code == 200: data = response.json() return str(data) else: self.worker.editor_logging_handler.error( f"API returned {response.status_code}" ) return None except Exception as e: self.worker.editor_logging_handler.error(f"Error: {e}") return None async def run(self): await self.capability_worker.speak("What would you like me to look up?") user_input = await self.capability_worker.user_response() await self.capability_worker.speak("Let me check on that.") result = await self.fetch_data(user_input) if result: response = self.capability_worker.text_to_text_response( f"Summarize this data in one short sentence: {result}" ) await self.capability_worker.speak(response) else: await self.capability_worker.speak( "Sorry, I couldn't get that information right now." ) self.capability_worker.resume_normal_flow() def call(self, worker: AgentWorker): self.worker = worker self.capability_worker = CapabilityWorker(self) self.worker.session_tasks.create(self.run())
Type: Skill (long-running) · Complexity: MediumInteractive back-and-forth conversation until the user says an exit word.Key SDK methods:speak(), user_response(), resume_normal_flow()
from src.agent.capability import MatchingCapabilityfrom src.main import AgentWorkerfrom src.agent.capability_worker import CapabilityWorkerEXIT_WORDS = {"stop", "exit", "quit", "done", "cancel", "bye", "goodbye"}class LoopTemplateCapability(MatchingCapability): worker: AgentWorker = None capability_worker: CapabilityWorker = None #{{register capability}} async def run(self): await self.capability_worker.speak( "I'm ready to help. Ask me anything, or say stop when you're done." ) while True: user_input = await self.capability_worker.user_response() if not user_input: continue # Check for exit commands if any(word in user_input.lower() for word in EXIT_WORDS): await self.capability_worker.speak("Goodbye!") break # Process input response = self.capability_worker.text_to_text_response( f"Respond in one short sentence: {user_input}" ) await self.capability_worker.speak(response) # ALWAYS resume normal flow when the loop ends self.capability_worker.resume_normal_flow() def call(self, worker: AgentWorker): self.worker = worker self.capability_worker = CapabilityWorker(self) self.worker.session_tasks.create(self.run())
Type: Skill · Pattern: Fire-and-forget · Complexity: MinimalWhen triggered by a hotword, sends an email via SMTP. No conversation, no user input—just trigger → send → speak result → exit.Key SDK methods:send_email(), speak(), resume_normal_flow()
from src.agent.capability import MatchingCapabilityfrom src.main import AgentWorkerfrom src.agent.capability_worker import CapabilityWorkerclass SendemailhCapability(MatchingCapability): worker: AgentWorker = None capability_worker: CapabilityWorker = None #{{register capability}} async def email_sender(self): # Hardcoded config HOST = "smtp.gmail.com" PORT = 465 SENDER_EMAIL = "[email protected]" SENDER_PASSWORD = "test 1234 5678 9121" # Gmail app password RECEIVER_EMAIL = "[email protected]" SUBJECT = "Test Email" BODY = "Hello! This is a test email." ATTACHMENTS = ["testfile.txt"] status = self.capability_worker.send_email( host=HOST, port=PORT, sender_email=SENDER_EMAIL, sender_password=SENDER_PASSWORD, receiver_email=RECEIVER_EMAIL, cc_emails=[], subject=SUBJECT, body=BODY, attachment_paths=ATTACHMENTS, ) if status: await self.capability_worker.speak("Email has been sent successfully.") else: await self.capability_worker.speak("Failed to send email") self.capability_worker.resume_normal_flow() def call(self, worker: AgentWorker): self.worker = worker self.capability_worker = CapabilityWorker(self.worker) self.worker.session_tasks.create(self.email_sender())
Critical pattern — resume_normal_flow():Every Skill must call this when done. It hands control back to the agent’s normal conversation. Forget it, and the speaker goes permanently silent.
Type: Skill · Pattern: LLM-as-translator · Complexity: MediumYou say “list all my Python files” and the speaker translates that into a real terminal command, runs it on your local machine, and reads back the result.Key SDK methods:wait_for_complete_transcription(), text_to_text_response(), exec_local_command()
Abilities run in OpenHome’s cloud sandbox, not on your local machine. exec_local_command() bridges that gap via WebSocket to whatever device is connected.
Type: Skill · Pattern: Sandbox escape · Complexity: MinimalForwards the user’s raw speech directly to OpenClaw—a desktop AI agent with 2,800+ community skills.Key SDK methods:wait_for_complete_transcription(), exec_local_command(), speak()
Type: Background Daemon · Pattern: Poll loop · Complexity: MediumStarts automatically when a user connects and runs in an infinite loop. This example reads conversation history and logs it every 20 seconds.Key SDK methods:get_full_message_history(), session_tasks.sleep()
Type: Skill + Daemon · Pattern: Coordinated · Complexity: Advancedmain.py (Skill) parses “set an alarm for 3pm” and writes to alarms.json. watcher.py (Daemon) polls that file every 5 seconds and fires when the target time hits.Key SDK methods:send_interrupt_signal(), play_from_audio_file(), session_tasks.sleep()
from datetime import datetimefrom zoneinfo import ZoneInfoimport jsonclass AlarmCapabilityWatcher(MatchingCapability): async def first_function(self): while True: try: alarms = await self._read_alarms_safe() if not alarms: await self.worker.session_tasks.sleep(5.0) continue tz_name = self.capability_worker.get_timezone() tz = ZoneInfo(tz_name) now = datetime.now(tz=tz) # Find due alarms due = [] for a in alarms: if a.get("status") != "scheduled": continue target_dt = datetime.fromisoformat(a.get("target_iso")) if now >= target_dt: due.append(a) # Fire alarms for a in due: await self.capability_worker.play_from_audio_file("alarm.mp3") await self._mark_alarm_triggered(alarms, a.get("id")) await self.worker.session_tasks.sleep(5.0) except Exception as e: self.worker.editor_logging_handler.error(f"Watcher error: {e}") await self.worker.session_tasks.sleep(2.0)
