Message channels¶

Overview¶

Message channels let UNI send and receive messages through external services. They serve two purposes:

You can message UNI from your phone and receive replies
When impulses expire without a voice conversation, UNI can forward them as messages

All installed channels can receive messages in parallel. The "preferred" channel is used for UNI-initiated outbound messages.

Creating a channel¶

Extend uni.MessageChannel and register it with the @uni.message_channel decorator:

Minimal message channel

import uni_plugin_sdk as uni

@uni.message_channel
class MyChannel(uni.MessageChannel):

    def __init__(self):
        self.client = None
        self.client.on_message(self._on_message)

    @property
    def online(self) -> bool:
        """Whether the channel is ready to send/receive."""
        return self.client is not None and self.client.is_connected()

    def send(self, message: str) -> None:
        """Send a message to the user."""
        self.client.send_dm(message)

    async def approve(self, tool_name: str, arguments: dict[str, Any]) -> bool:
        """Approve or deny tool execution remotely."""
        return False

    def _on_message(self, sender_id, text):
        """Vendor callback — validate and route to UNI."""
        if sender_id != self.authorized_user_id:
            return
        self.receive(text)

The channel handles both directions: UNI calls send() to deliver messages, and your vendor callback calls self.receive() to route incoming replies back to UNI.

Sender validation

Always validate incoming messages are from the authorized user. Without validation, anyone who discovers UNI's bot could send commands.

Tool approval¶

When UNI wants to use a tool that isn't marked as safe, it calls approve. The default returns False (deny all). If your channel supports interactive prompts, override it — for example, using emoji reactions:

Reaction-based approval

async def approve(self, tool_name: str, arguments: dict[str, Any]) -> bool:
    message = await self.channel.send(
        f"**{tool_name}**\n```json\n{json.dumps(arguments, indent=2)}\n```"
    )
    await message.add_reaction("✅")
    await message.add_reaction("❌")

    try:
        reaction, _ = await self.bot.wait_for(
            "reaction_add",
            check=lambda r, u: str(r.emoji) in ("✅", "❌") and u == self.user,
            timeout=60,
        )
        return str(reaction.emoji) == "✅"
    except asyncio.TimeoutError:
        return False

Quiet hours¶

UNI-initiated messages (expired impulses) respect quiet hours. User-initiated messages always get replies regardless.

Settings page¶

Most channels need a settings page for credentials. See Routes & sockets for the full routing guide.

For impulse routing options (voice, message, or auto), see Impulses.