Custom Interfaces

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

Method	Purpose
`_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:

Field	Type	Required	Description
`platform`	`str`	Yes	Platform name (e.g., `"slack"`)
`platform_user_id`	`str`	Yes	Sender’s user ID
`platform_chat_id`	`str`	Yes	Chat/channel ID
`platform_message_id`	`str`	Yes	Message ID
`text`	`str`	No	Text content
`username`	`str`	No	Display name
`images`	`List[Image]`	No	Attached images
`audio`	`List[Audio]`	No	Attached audio
`videos`	`List[Video]`	No	Attached videos
`files`	`List[File]`	No	Attached files
`reply_to_message_id`	`str`	No	ID of message being replied to
`metadata`	`dict`	No	Arbitrary 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:

Field	Type	Description
`content`	`str`	Text content
`images`	`List[Image]`	Images to send
`videos`	`List[Video]`	Videos to send
`audio`	`List[Audio]`	Audio to send
`files`	`List[File]`	Files to send
`metadata`	`dict`	Platform-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:

Subclass InterfaceConfig — Add platform-specific settings (tokens, URLs, modes). Use a frozen dataclass.
Implement _start_receiver — Set up your connection (WebSocket, HTTP server, polling loop). Store any client objects on self.
Implement _stop_receiver — Tear down connections. Must be idempotent (safe to call multiple times).
Implement _convert_inbound — Parse the platform’s message format. Extract text, media, user info. Return None for messages to skip.
Implement _send_response — Send text, images, files back to the platform. Handle message splitting if the platform has length limits.
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.

Get Started

Basics

Capabilities

Advanced

What You Implement

Skeleton Example

InterfaceMessage

InterfaceResponse

Error Types

InterfaceRateLimitError

Implementation Checklist

​What You Implement

​Skeleton Example

​InterfaceMessage

​InterfaceResponse

​Error Types

​InterfaceRateLimitError

​Implementation Checklist

What You Implement

Skeleton Example

InterfaceMessage

InterfaceResponse

Error Types

InterfaceRateLimitError

Implementation Checklist