mastodon_channel

mastodon_channel

MastodonChannel — Mastodon adapter via Mastodon.py.

Classes

MastodonChannel

MastodonChannel(
    api_base_url: str = "",
    *,
    access_token: str = "",
    bus: Optional[EventBus] = None,
)

Bases: BaseChannel

Mastodon messaging channel adapter.

Uses the Mastodon API via Mastodon.py.

PARAMETER	DESCRIPTION
api_base_url	Mastodon instance URL (e.g. https://mastodon.social). Falls back to MASTODON_API_BASE_URL env var. TYPE: str DEFAULT: ''
access_token	Mastodon access token. Falls back to MASTODON_ACCESS_TOKEN env var. TYPE: str DEFAULT: ''
bus	Optional event bus for publishing channel events. TYPE: Optional[EventBus] DEFAULT: None

Source code in src/diapason/channels/mastodon_channel.py

def __init__(
    self,
    api_base_url: str = "",
    *,
    access_token: str = "",
    bus: Optional[EventBus] = None,
) -> None:
    self._api_base_url = api_base_url or os.environ.get("MASTODON_API_BASE_URL", "")
    self._access_token = access_token or os.environ.get("MASTODON_ACCESS_TOKEN", "")
    self._bus = bus
    self._handlers: List[ChannelHandler] = []
    self._status = ChannelStatus.DISCONNECTED

Methods:

connect

connect() -> None

Validate credentials and mark as connected.

Source code in src/diapason/channels/mastodon_channel.py

def connect(self) -> None:
    """Validate credentials and mark as connected."""
    if not self._api_base_url or not self._access_token:
        logger.warning("No Mastodon api_base_url or access_token configured")
        self._status = ChannelStatus.ERROR
        return
    try:
        import mastodon  # noqa: F401
    except ImportError:
        raise ImportError(
            "Mastodon.py not installed. Install with: "
            "uv sync --extra channel-mastodon"
        )
    self._status = ChannelStatus.CONNECTED

disconnect

disconnect() -> None

Mark as disconnected.

Source code in src/diapason/channels/mastodon_channel.py

def disconnect(self) -> None:
    """Mark as disconnected."""
    self._status = ChannelStatus.DISCONNECTED

send

send(
    channel: str,
    content: str,
    *,
    conversation_id: str = "",
    metadata: Dict[str, Any] | None = None,
) -> bool

Post a status or send a direct message on Mastodon.

PARAMETER	DESCRIPTION
channel	Visibility level (public, unlisted, private, direct) or a username to DM (prefix with @). TYPE: str
content	Toot / message content. TYPE: str

Source code in src/diapason/channels/mastodon_channel.py

def send(
    self,
    channel: str,
    content: str,
    *,
    conversation_id: str = "",
    metadata: Dict[str, Any] | None = None,
) -> bool:
    """Post a status or send a direct message on Mastodon.

    Parameters
    ----------
    channel:
        Visibility level (``public``, ``unlisted``, ``private``,
        ``direct``) or a username to DM (prefix with ``@``).
    content:
        Toot / message content.
    """
    if not self._api_base_url or not self._access_token:
        logger.warning("Cannot send: no Mastodon credentials configured")
        return False

    try:
        from mastodon import Mastodon

        client = Mastodon(
            access_token=self._access_token,
            api_base_url=self._api_base_url,
        )

        visibility = "public"
        if channel in ("public", "unlisted", "private", "direct"):
            visibility = channel
        elif channel.startswith("@"):
            # Direct message — prepend mention
            visibility = "direct"
            if not content.startswith(channel):
                content = f"{channel} {content}"

        in_reply_to_id = conversation_id or None
        client.status_post(
            content,
            visibility=visibility,
            in_reply_to_id=in_reply_to_id,
        )

        self._publish_sent(channel, content, conversation_id)
        return True
    except ImportError:
        logger.debug("Mastodon.py not installed")
        return False
    except Exception:
        logger.debug("Mastodon send failed", exc_info=True)
        return False

status

status() -> ChannelStatus

Return the current connection status.

Source code in src/diapason/channels/mastodon_channel.py

def status(self) -> ChannelStatus:
    """Return the current connection status."""
    return self._status

list_channels

list_channels() -> List[str]

Return available channel identifiers.

Source code in src/diapason/channels/mastodon_channel.py

def list_channels(self) -> List[str]:
    """Return available channel identifiers."""
    return ["mastodon"]

on_message

on_message(handler: ChannelHandler) -> None

Register a callback for incoming messages.

Source code in src/diapason/channels/mastodon_channel.py

def on_message(self, handler: ChannelHandler) -> None:
    """Register a callback for incoming messages."""
    self._handlers.append(handler)