Impulses¶

Overview¶

Normally, the user prompts UNI. Impulses let your plugin do the prompting: an event fires, and UNI has something to say about it. Connected devices show an indicator so the user knows a message is waiting. UNI will bring it up naturally in the next voice conversation, or once it becomes due, deliver it through a message channel.

Basic usage¶

Call send_impulse() from event handlers, callbacks, or background tasks:

Basic impulse

uni.send_impulse(
    type="download_complete",
    instruction="Tell the user their file finished downloading",
    due_after="15m",
    data={
        "file": file_path,
        "size_mb": round(Path(file_path).stat().st_size / 1024 / 1024, 2)
    }
)

The due_after controls how long UNI waits for a natural conversation before initiating contact (e.g., via a message channel). While waiting, connected devices show an indicator the user can tap for immediate delivery.

Without due_after, the impulse stays in the queue until a conversation picks it up or the user taps the indicator.

Expiration¶

Use expires_after to silently drop the impulse after a given time window elapses.

Impulse with expiration

uni.send_impulse(
    type="evening_reflection",
    instruction="Ask the user how their day went",
    expires_after="2h",
)

Without expires_after, the impulse is never dropped.

Quiet hours¶

Quiet hours prevent UNI from initiating contact. However, the impulse indicator still shows on connected devices — if the user taps it or starts a conversation, impulses are delivered normally.

Time spent in quiet hours does not count toward expires_after, but do count toward due_after.

Checking quiet hours from your plugin¶

Use is_quiet_hours() to check the current state, and get_quiet_hours_window() to read the configured window:

Quiet hours API

if uni.is_quiet_hours():
    # Ssh.
    ...

window = uni.get_quiet_hours_window()
if window is not None:
    start_hour, end_hour = window  # e.g. (22, 8)

Examples¶

Background jobs¶

Polling with state tracking:

Reminder polling

mentioned_reminders = set()

@uni.job(interval="5m")
def check_reminders():
    for reminder in get_pending_reminders():
        if reminder.id not in mentioned_reminders:
            mentioned_reminders.add(reminder.id)
            uni.send_impulse(
                type="reminder",
                instruction="Remind the user",
                due_after="15m"
                data={
                    "text": reminder.text,
                    "created": reminder.created_at
                }
            )

@uni.on_nightly_reset
def reset_mentions():
    mentioned_reminders.clear()

Async task results¶

Start a long-running task and deliver results when ready:

Background research

@uni.tool
def research_topic(topic: str):
    """
    Research a topic using multiple sources.

    Args:
        topic: Topic to research
    """
    def run_research():
        findings = my_research_implementation(topic)
        uni.send_impulse(
            type="research_complete",
            instruction="Share the research findings",
            due_after="5m",
            data={
                "topic": topic,
                "key_points": findings["summary"],
                "sources": findings["sources"]
            }
        )

    uni.run_task(run_research)
    return f"Researching {topic}..."