Skill Cheat Sheet

from huxley_sdk import (
    # Core protocols (you implement Skill)
    Skill,
    SkillContext,

    # Tool definitions
    ToolDefinition,
    ToolResult,

    # Side effects
    AudioStream,
    PlaySound,
    CancelMedia,
    SetVolume,
    InputClaim,
    ClaimEndReason,

    # Focus / proactive
    InjectPriority,

    # Content type for AudioStream
    ContentType,

    # Errors
    ClaimBusyError,
)
from huxley_sdk.audio import load_pcm_palette

# load_pcm_palette is sync and takes the role names you want to load.
# It returns a {role: pcm_bytes} dict; missing roles are silently skipped.
palette = load_pcm_palette(
    ctx.persona_data_dir / "sounds",
    roles=["news_start", "book_start", "book_end"],
)

class MySkill:
    name = "my_skill"  # required, must be unique

    @property
    def tools(self) -> list[ToolDefinition]:
        return [...]

    async def setup(self, ctx: SkillContext) -> None: ...
    async def handle(self, tool_name: str, args: dict) -> ToolResult: ...
    async def teardown(self) -> None: ...

ToolDefinition(
    name="search_audiobooks",
    description="Search the user's library by title or author.",
    parameters={
        "type": "object",
        "properties": {
            "query": {"type": "string", "description": "..."},
        },
        "required": ["query"],
    },
)

return ToolResult(
    output=json.dumps({"matches": [...]}),  # text the model narrates
    side_effect=PlaySound(pcm=...),          # optional
)

AudioStream(
    factory: Callable[[], AsyncIterator[bytes]],   # required
    on_complete_prompt: str | None = None,         # narrate on natural end
    completion_silence_ms: int = 0,                # silence before prompt
    content_type: ContentType = ContentType.NONMIXABLE,
    label: str | None = None,                      # shown in dev UI
    preroll_ms: int = 0,
    on_patience_expired: Callable[[], Awaitable[None]] | None = None,
    patience: timedelta | None = None,
)

InputClaim(
    on_mic_frame: Callable[[bytes], Awaitable[None]],  # mic now goes here
    speaker_source: AsyncIterator[bytes] | None = None, # speaker reads from here
    on_claim_end: Callable[[ClaimEndReason], Awaitable[None]] | None = None,
    title: str | None = None,
)

await ctx.inject_turn(
    "Tell the user: time for medication.",       # prompt — LLM instruction
    dedup_key="med_2026-04-29",                  # prevents duplicate queue
    priority=InjectPriority.BLOCK_BEHIND_COMMS,  # see priorities
)

await ctx.storage.set_setting(key: str, value: str) -> None
await ctx.storage.get_setting(key: str, default: str | None = None) -> str | None
await ctx.storage.list_settings(prefix: str = "") -> list[tuple[str, str]]
await ctx.storage.delete_setting(key: str) -> None

await ctx.logger.adebug(event: str, **kwargs)
await ctx.logger.ainfo(event: str, **kwargs)
await ctx.logger.awarning(event: str, **kwargs)
await ctx.logger.aerror(event: str, **kwargs)
await ctx.logger.aexception(event: str, **kwargs)  # captures traceback

handle = ctx.background_task(
    name="my_listener",
    coro_factory=self._listen,            # callable returning a coroutine
    restart_on_crash=True,
    max_restarts_per_hour=10,
    on_permanent_failure=self._on_dead,
)

catalog = ctx.catalog()
await catalog.upsert(
    id="el_principito",
    fields={"title": "El Principito", "author": "Saint-Exupéry"},
    payload={"path": "/data/audiobooks/el_principito.m4b"},
)
hits = await catalog.search("principito", limit=5)
prompt_text = catalog.as_prompt_lines(
    limit=20,
    line=lambda h: f"- {h.fields['title']} ({h.fields['author']})",
)

from huxley_sdk.audio import load_pcm_palette

# Load all WAVs in a directory as PCM blobs.
sounds = load_pcm_palette(ctx.persona_data_dir / "sounds")
chime = sounds["news_start"]  # bytes, ready for PlaySound

from huxley_sdk import ClaimBusyError

try:
    handle = await ctx.start_input_claim(claim)
except ClaimBusyError:
    # Another skill has an active claim on COMMS.
    ...