Skip to main content

Documentation Index

Fetch the complete documentation index at: https://docs.definable.ai/llms.txt

Use this file to discover all available pages before exploring further.

You can connect an agent to any messaging platform by subclassing BaseInterface and implementing four abstract methods. The base class handles sessions, hooks, concurrency, and error handling — you only write the platform-specific code.

​
What You Implement

MethodPurpose
_start_receiver()Start listening for messages (polling, webhook, WebSocket)
_stop_receiver()Stop listening and clean up resources
_convert_inbound(raw)Convert a platform message into an InterfaceMessage
_send_response(msg, resp, raw)Send an InterfaceResponse back to the platform
Everything else — sessions, hooks, agent execution, error handling, concurrency — is handled by BaseInterface.

​
Skeleton Example

Here is a minimal custom interface for a WebSocket-based chat platform:
Looking for Slack? There’s a built-in Slack interface with full Block Kit, slash commands, and interactive component support.
from typing import Any, Optional
from definable.agent.interface import (
    BaseInterface,
    InterfaceConfig,
    InterfaceMessage,
    InterfaceResponse,
)

class MyChatConfig(InterfaceConfig):
    platform: str = "mychat"
    api_token: str = ""
    ws_url: str = "wss://chat.example.com/ws"

class MyChatInterface(BaseInterface):
    def __init__(self, agent, **kwargs):
        super().__init__(agent=agent, **kwargs)
        self._ws = None

    async def _start_receiver(self) -> None:
        """Connect to the chat platform via WebSocket."""
        import websockets
        self._ws = await websockets.connect(self.config.ws_url)

        # Start receiving messages in background
        import asyncio
        asyncio.create_task(self._listen())

    async def _listen(self) -> None:
        """Read messages from the WebSocket and dispatch them."""
        import json
        async for raw in self._ws:
            event = json.loads(raw)
            await self.handle_platform_message(event)

    async def _stop_receiver(self) -> None:
        """Disconnect from the chat platform."""
        if self._ws:
            await self._ws.close()
            self._ws = None

    async def _convert_inbound(self, raw_message: Any) -> Optional[InterfaceMessage]:
        """Convert a platform event to InterfaceMessage."""
        # Skip bot messages
        if raw_message.get("is_bot"):
            return None

        return InterfaceMessage(
            platform="mychat",
            platform_user_id=raw_message.get("user_id", ""),
            platform_chat_id=raw_message.get("room_id", ""),
            platform_message_id=raw_message.get("msg_id", ""),
            text=raw_message.get("text", ""),
        )

    async def _send_response(
        self,
        original_msg: InterfaceMessage,
        response: InterfaceResponse,
        raw_message: Any,
    ) -> None:
        """Send the response back to the platform."""
        import json
        if response.content and self._ws:
            await self._ws.send(json.dumps({
                "type": "message",
                "room_id": original_msg.platform_chat_id,
                "text": response.content,
            }))

​
InterfaceMessage

The platform-agnostic inbound message your _convert_inbound method must produce:
FieldTypeRequiredDescription
platformstrYesPlatform name (e.g., "slack")
platform_user_idstrYesSender’s user ID
platform_chat_idstrYesChat/channel ID
platform_message_idstrYesMessage ID
textstrNoText content
usernamestrNoDisplay name
imagesList[Image]NoAttached images
audioList[Audio]NoAttached audio
videosList[Video]NoAttached videos
filesList[File]NoAttached files
reply_to_message_idstrNoID of message being replied to
metadatadictNoArbitrary platform-specific data
Return None from _convert_inbound to silently skip a message (e.g., bot’s own messages, unsupported message types).

​
InterfaceResponse

The platform-agnostic response produced by the agent:
FieldTypeDescription
contentstrText content
imagesList[Image]Images to send
videosList[Video]Videos to send
audioList[Audio]Audio to send
filesList[File]Files to send
metadatadictPlatform-specific response data
Your _send_response method receives this and translates it into platform API calls.

​
Error Types

Use the built-in error hierarchy for consistent error handling across interfaces: 
from definable.agent.interface import (
    InterfaceError,                 # Base — general interface failure
    InterfaceConnectionError,       # 503 — cannot connect to platform
    InterfaceAuthenticationError,   # 401 — invalid credentials
    InterfaceRateLimitError,        # 429 — rate limited by platform
    InterfaceMessageError,          # 400 — bad message (invalid chat, etc.)
)
Raise these from your platform methods, and the base class handles them:
  • Runs all on_error hooks
  • Sends the configured error_message to the user
  • Logs the error

​
InterfaceRateLimitError

Includes an optional retry_after field for backoff: 
raise InterfaceRateLimitError(
    "Rate limited by Slack",
    platform="slack",
    retry_after=30.0,  # seconds
)

​
Implementation Checklist

When building a custom interface:
  1. Subclass InterfaceConfig — Add platform-specific settings (tokens, URLs, modes). Use a frozen dataclass.
  2. Implement _start_receiver — Set up your connection (WebSocket, HTTP server, polling loop). Store any client objects on self.
  3. Implement _stop_receiver — Tear down connections. Must be idempotent (safe to call multiple times).
  4. Implement _convert_inbound — Parse the platform’s message format. Extract text, media, user info. Return None for messages to skip.
  5. Implement _send_response — Send text, images, files back to the platform. Handle message splitting if the platform has length limits.
  6. Call handle_platform_message — From your receiver (event handler, webhook route, poll loop), call await self.handle_platform_message(raw_message). This triggers the full pipeline. 
# In your polling loop or event handler:
async def _on_platform_event(self, event):
    await self.handle_platform_message(event)
The base class takes it from there — converting the message, running hooks, managing sessions, calling the agent, and invoking your _send_response.
Look at the Telegram or Slack implementations as references. Telegram demonstrates polling vs webhooks, media handling, and error mapping. Slack demonstrates Socket Mode, Block Kit interactions, slash commands, and modals.