Best Practices

await self.capability_worker.speak(
    "Okay, I've successfully processed your request and sent the email "
    "to [email protected] with the subject line 'Meeting Tomorrow' and "
    "the body text you provided. The email was sent at 3:45 PM Eastern "
    "Time from your configured email account. Would you like me to do "
    "anything else for you today?"
)

confirmed = await self.capability_worker.run_confirmation_loop(
    f"Send email to {recipient} with subject '{subject}'?"
)

if confirmed:
    # Perform action
    await self.capability_worker.speak("Email sent.")
else:
    await self.capability_worker.speak("Cancelled.")

await self.capability_worker.speak(
    "I'm ready to help. Say 'stop' when you're done."
)

while True:
    user_input = await self.capability_worker.user_response()
    
    if any(w in user_input.lower() for w in EXIT_WORDS):
        await self.capability_worker.speak("Goodbye!")
        break
    
    # Process input

try:
    user_input = await asyncio.wait_for(
        self.capability_worker.user_response(),
        timeout=10.0
    )
except asyncio.TimeoutError:
    await self.capability_worker.speak(
        "I didn't catch that. Let's try again later."
    )
    self.capability_worker.resume_normal_flow()
    return

async def run(self):
    result = await self.fetch_data()  # Might throw exception
    await self.capability_worker.speak(result)
    self.capability_worker.resume_normal_flow()

self.worker.editor_logging_handler.info(f"User request: {user_input}")
self.worker.editor_logging_handler.info(f"API URL: {api_url}")

try:
    response = requests.get(api_url)
    self.worker.editor_logging_handler.info(
        f"API response: {response.status_code}"
    )
except Exception as e:
    self.worker.editor_logging_handler.error(f"API error: {str(e)}")

user_input = await self.capability_worker.user_response()

if not user_input or len(user_input.strip()) == 0:
    await self.capability_worker.speak("I didn't catch that. Can you repeat?")
    return

# Sanitize input for API calls
import re
email = re.match(r"[^@]+@[^@]+\.[^@]+", user_input)
if not email:
    await self.capability_worker.speak("That doesn't look like a valid email.")
    return

async def run(self):
    await self.capability_worker.speak("Task complete.")
    self.capability_worker.resume_normal_flow()  # Required!

async def first_function(self):
    while True:
        # Background work
        await self.worker.session_tasks.sleep(5.0)
    # No resume_normal_flow() here!

async def run(self):
    try:
        # Your logic
        await self.capability_worker.speak("Done!")
    except Exception as e:
        self.worker.editor_logging_handler.error(f"Error: {e}")
        await self.capability_worker.speak("Something went wrong.")
    finally:
        self.capability_worker.resume_normal_flow()  # Always called

EXIT_WORDS = {"stop", "exit", "quit", "done", "cancel", "bye", "goodbye", "leave"}

while True:
    user_input = await self.capability_worker.user_response()
    
    if any(word in user_input.lower() for word in EXIT_WORDS):
        await self.capability_worker.speak("Goodbye!")
        break
    
    # Continue processing

await self.capability_worker.speak(
    "Let's start. Say 'stop' anytime to exit."
)

self.worker.editor_logging_handler.info(f"API key: {API_KEY}")

API_KEY = "sk-1234567890abcdef"
DB_PASSWORD = "mypassword123"

command = await self.capability_worker.user_response()
await self.capability_worker.exec_local_command(command)  # User could say "rm -rf /"

import os

def is_safe_path(path: str) -> bool:
    # Prevent directory traversal
    if ".." in path:
        return False
    
    # Only allow specific directories
    allowed_dirs = ["/home/user/documents", "/tmp"]
    abs_path = os.path.abspath(path)
    
    return any(abs_path.startswith(d) for d in allowed_dirs)

if not is_safe_path(user_provided_path):
    await self.capability_worker.speak("Invalid file path.")
    return

import time

time.sleep(5)  # Blocks entire event loop
await self.capability_worker.speak("Done")

data = requests.get(large_file_url).content  # Could be GBs

async def run(self):
    file_handle = None
    
    try:
        file_handle = open("data.txt", "r")
        data = file_handle.read()
        # Process data
    except Exception as e:
        self.worker.editor_logging_handler.error(f"Error: {e}")
    finally:
        if file_handle:
            file_handle.close()
        self.capability_worker.resume_normal_flow()

self.worker.editor_logging_handler.info(f"Starting ability with input: {user_input}")
self.worker.editor_logging_handler.info(f"Step 1 complete")
self.worker.editor_logging_handler.info(f"API returned: {response}")
self.worker.editor_logging_handler.info(f"Exiting ability")

await self.capability_worker.write_file("data.json", json.dumps(data))
# write_file() appends - calling twice creates: [old][new]

if not await self.capability_worker.check_if_file_exists("config.json", False):
    # File doesn't exist - use defaults
    config = {"default": "value"}
else:
    raw = await self.capability_worker.read_file("config.json", False)
    try:
        config = json.loads(raw)
    except json.JSONDecodeError:
        # Corrupted file - use defaults
        config = {"default": "value"}