Channel Plugins

Channel plugins add communication methods to TeamWeb AI. They are configured per assistant and handle both sending messages out and receiving inbound webhooks.

Directory Layout

my_channel/
    manifest.py
    channel.py

Example: A Simple SMS Channel

manifest.py:

from typing import Any

from app.plugins.base import PluginManifest, PluginType

manifest = PluginManifest(
    name="sms_provider",
    plugin_type=PluginType.CHANNEL,
    channel_type="messaging",
    version="1.0.0",
    description="Send and receive SMS messages",
    author="Your Name",
    config_schema={
        "api_key": {
            "type": "string",
            "required": True,
            "secret": True,
            "label": "API Key",
        },
        "phone_number": {
            "type": "string",
            "required": True,
            "label": "Phone Number",
            "placeholder": "+1234567890",
        },
    },
)


def get_setup_docs(app_config: dict[str, Any], **kwargs: Any) -> str:
    public_url = app_config.get("PUBLIC_URL", "https://yourdomain.com")
    assistant_id = kwargs.get("assistant_id", "<assistant_id>")
    webhook_url = (
        f"{public_url}/webhooks/channels/sms_provider/{assistant_id}/inbound"
    )
    return (
        "<h6>SMS Setup</h6>"
        "<ol>"
        "<li>Configure your SMS provider to send webhooks to:"
        f"<code>{webhook_url}</code></li>"
        "</ol>"
    )

channel.py:

import logging
from typing import Any

from app.plugins.base import Attachment, BaseChannel, InboundMessage

logger = logging.getLogger(__name__)


class SmsChannel(BaseChannel):
    """Send and receive SMS messages.

    Instantiated with config from AssistantPlugin.config_json.
    """

    def __init__(self, api_key: str, phone_number: str, **kwargs: Any) -> None:
        self.api_key = api_key
        self.phone_number = phone_number

    def send_message(
        self,
        to: str,
        subject: str,
        body_html: str,
        body_text: str,
        reply_to: str | None = None,
        headers: dict[str, str] | None = None,
        attachments: list[Attachment] | None = None,
    ) -> dict[str, Any]:
        """Send an SMS message.

        Args:
            to: Recipient phone number.
            subject: Ignored for SMS.
            body_html: Ignored for SMS.
            body_text: Message text content.
            reply_to: Ignored for SMS.
            headers: Optional channel-specific headers.
            attachments: Ignored for SMS.

        Returns:
            Dict with message_id from the SMS provider.
        """
        # Call your SMS provider API here
        # ...
        return {"message_id": "msg_123"}

    def parse_inbound(self, payload: dict[str, Any]) -> InboundMessage:
        """Parse an inbound SMS webhook into a normalised message.

        Args:
            payload: Raw webhook payload from the SMS provider.

        Returns:
            Normalised InboundMessage.
        """
        return InboundMessage(
            sender=payload.get("from", ""),
            recipient=payload.get("to", ""),
            body_text=payload.get("text", ""),
            message_id=payload.get("id"),
        )

    def build_reply_to(self, conversation_reply_token: str) -> str:
        """Build a reply-to identifier with the conversation token.

        For channels that support reply threading, embed the token so
        inbound messages can be routed to the correct conversation.

        Args:
            conversation_reply_token: UUID token for the conversation.

        Returns:
            Reply-to string (e.g. phone number with token suffix).
        """
        return ""

Channel Type Constraint

The channel_type in the manifest must be one of "email", "messaging", "voice", or "browser". Only one channel of each type may be active per assistant. Enabling a second messaging channel automatically disables the first.

Feedback Support

Channel plugins can declare support for TeamWeb AI’s feedback system by setting flags on the manifest:

manifest = PluginManifest(
    name="my_channel",
    plugin_type=PluginType.CHANNEL,
    channel_type="browser",
    # ...
    supports_message_feedback=True,   # Per-message thumbs up/down
    supports_session_feedback=True,   # Post-session rating form
)

Flag	Default	Description
`supports_message_feedback`	`False`	The channel UI can show thumbs up/down on assistant messages
`supports_session_feedback`	`False`	The channel UI can show a feedback form when a session ends

When these flags are True, the channel configuration page in the admin UI shows additional Feedback Settings where admins can choose which user types see feedback controls.

Only channels with a UI controlled by TeamWeb AI should set these flags (e.g., browser_internal and browser_external). External platforms like Slack, Telegram, and email don’t have a natural place for feedback controls.

Constructor

The channel class is instantiated with the plugin configuration unpacked as keyword arguments from AssistantPlugin.config_json. Define your __init__ parameters to match the keys in your config_schema. Always include **kwargs to handle any extra fields gracefully.

Webhook URL

Inbound messages are received at:

/webhooks/channels/<plugin_name>/<assistant_id>/inbound

TeamWeb AI automatically routes the payload to your parse_inbound() method. Use get_setup_docs() in your manifest to display the correct webhook URL in the configuration UI.

`InboundMessage` Fields

Field	Description
`sender`	Sender identifier (email, phone, user ID)
`recipient`	Recipient identifier
`subject`	Message subject (if applicable)
`body_text`	Plain text message body
`body_html`	HTML body (if available)
`message_id`	Channel-specific message ID
`reply_token`	Token for routing to an existing conversation
`attachments`	List of attachment dicts with `filename`, `content_type`, `data`

`send_message` Parameters

Parameter	Description
`to`	Recipient address (email, channel ID, phone number)
`subject`	Message subject (ignore if not applicable to your channel)
`body_html`	HTML body content
`body_text`	Plain text body content
`reply_to`	Reply-to address for threading
`headers`	Channel-specific headers dict (e.g. `slack_thread_ts`)
`attachments`	List of `Attachment` objects with `filename`, `data` (bytes), and `content_type`

Return a dict with at least a message_id key.

Canonical Content Blocks

When the agent loop produces a message, it doesn’t write channel-flavoured text. It produces a list of structured content blocks, persisted on the Message row. Each channel turns those blocks into whatever its wire format needs at the moment of delivery.

The block kinds in v1 are text, media, card, data_table, actions, and the inbound-only action_click. Their schemas live in app/channels/canonical.py. Adding a new kind is a local change: define a Pydantic class, register it in app/channels/registry.py, and add a markdown fallback rendering — existing channel renderers pick up the fallback automatically.

The renderer contract

Each channel plugin pairs its BaseChannel subclass with a ChannelRenderer subclass (see app/channels/renderer_base.py). The renderer takes a list[ContentBlock] plus a RenderContext and returns a channel-specific payload object that the channel’s send_payload(payload, *, to, ctx) method delivers.

Override the per-kind methods you support natively; everything else falls through to a markdown rendering from app/channels/fallbacks.py, so every block always reaches the recipient in some form.

from app.channels.canonical import (
    Action,
    ActionsBlock,
    MediaBlock,
    TextBlock,
)
from app.channels.fallbacks import block_to_markdown
from app.channels.renderer_base import ChannelRenderer, Fragment, RenderContext


class SmsRenderer(ChannelRenderer):
    def render(self, blocks, ctx: RenderContext):
        # SMS is text-only, so collapse every block into one string.
        return "\n\n".join(block_to_markdown(b) for b in blocks)

Wire the renderer to the channel by overriding BaseChannel.get_renderer():

class SmsChannel(BaseChannel):
    def get_renderer(self):
        from my_channel.renderer import SmsRenderer
        return SmsRenderer()

    def send_payload(self, payload, *, to, ctx):
        # Send the rendered string via your provider's API.
        return {"message_id": self._sms_provider.send(to, payload)}

The dispatcher will use the canonical path automatically when a Message has content_blocks populated; otherwise it falls back to the legacy send_message(body_text, body_html, attachments) shape. New channels only need the renderer + send_payload.

Inbound blocks

InboundMessage carries a blocks: list[ContentBlock] field alongside the legacy text/HTML/attachment fields. Populate it in parse_inbound:

from app.channels.canonical import MediaBlock, TextBlock
from app.services.media_service import save_media_from_bytes


def parse_inbound(self, payload):
    blocks = []
    if text := payload.get("text"):
        blocks.append(TextBlock(body_md=text))
    for attachment in payload.get("attachments", []):
        media = save_media_from_bytes(
            data=attachment["bytes"],
            mime=attachment["content_type"],
            original_filename=attachment.get("filename"),
            source_channel="my_channel",
        )
        blocks.append(MediaBlock(
            media_id=media.id,
            role="inline" if media.mime.startswith("image/") else "attachment",
            alt=attachment.get("filename"),
        ))
    return InboundMessage(
        sender=...,
        recipient=...,
        body_text=text,
        attachments=...,
        blocks=blocks,
    )

The same blocks land on the resulting user Message.content_blocks, so when the conversation is reopened on a different channel the original photos and structured content are still there.

Action callbacks

Agents can ask for input as a small set of buttons via ask_human(response_type="callback", options=[...]). The framework signs each option with a callback token, emits an ActionsBlock, and parks the agent run in WAITING_ON_HUMAN.

If your channel supports interactive controls (Slack Block Kit, Telegram inline keyboards, etc.), render ActionsBlock natively and wire its native callback to the shared resolver:

from app.channels.router import resolve_action_click

# In your webhook handler for clicks:
resolve_action_click(token, channel="my_channel")

For text-only channels, render each action as a signed URL that points at /actions/click?t={callback_token} — the same resolver handles email-style click-throughs.

The resolver verifies the signature and expiry, validates the run is still waiting, inserts a system message recording the click, and enqueues a resume run. The agent loop then receives the chosen option’s payload as the return value of its original ask_human call.

Reply Threading

For channels that support reply threading, implement build_reply_to() to embed a conversation token into a reply address. TeamWeb AI calls this when sending the first message in a conversation so that subsequent inbound messages can be routed back to the same conversation.

For example, the Postmark email channel embeds the token in the local part of a reply-to email address: reply+{token}@yourdomain.com.

Request Verification

For channels that support request signing (e.g. Slack’s HMAC signatures), implement a verify_request class method that validates the webhook signature. TeamWeb AI calls this before processing the payload:

@classmethod
def verify_request(
    cls,
    signing_secret: str,
    timestamp: str,
    body: bytes,
    signature: str,
) -> bool:
    """Verify the webhook request signature.

    Args:
        signing_secret: The app's signing secret.
        timestamp: Request timestamp header value.
        body: Raw request body bytes.
        signature: Signature header value.

    Returns:
        True if the signature is valid.
    """
    # Implement your verification logic here
    return True

Tool Plugins Provider Plugins