xmpp_channel

xmpp_channel

XMPPChannel — XMPP/Jabber adapter via slixmpp.

Classes

XMPPChannel

XMPPChannel(
    jid: str = "",
    *,
    password: str = "",
    server: str = "",
    port: int = 5222,
    bus: Optional[EventBus] = None,
)

Bases: BaseChannel

XMPP (Jabber) messaging channel adapter.

Uses the XMPP protocol via slixmpp.

PARAMETER	DESCRIPTION
jid	XMPP JID (e.g. bot@example.com). Falls back to XMPP_JID env var. TYPE: str DEFAULT: ''
password	XMPP account password. Falls back to XMPP_PASSWORD env var. TYPE: str DEFAULT: ''
server	Optional XMPP server hostname override. Falls back to XMPP_SERVER env var. TYPE: str DEFAULT: ''
port	XMPP server port (default 5222). Falls back to XMPP_PORT env var. TYPE: int DEFAULT: 5222
bus	Optional event bus for publishing channel events. TYPE: Optional[EventBus] DEFAULT: None

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

def __init__(
    self,
    jid: str = "",
    *,
    password: str = "",
    server: str = "",
    port: int = 5222,
    bus: Optional[EventBus] = None,
) -> None:
    self._jid = jid or os.environ.get("XMPP_JID", "")
    self._password = password or os.environ.get("XMPP_PASSWORD", "")
    self._server = server or os.environ.get("XMPP_SERVER", "")
    self._port = int(os.environ.get("XMPP_PORT", str(port)))
    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/xmpp_channel.py

def connect(self) -> None:
    """Validate credentials and mark as connected."""
    if not self._jid or not self._password:
        logger.warning("No XMPP jid or password configured")
        self._status = ChannelStatus.ERROR
        return
    try:
        import slixmpp  # noqa: F401
    except ImportError:
        raise ImportError(
            "slixmpp not installed. Install with: uv sync --extra channel-xmpp"
        )
    self._status = ChannelStatus.CONNECTED

disconnect

disconnect() -> None

Mark as disconnected.

Source code in src/diapason/channels/xmpp_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

Send an XMPP message.

PARAMETER	DESCRIPTION
channel	Recipient JID (user or MUC room). TYPE: str
content	Text message content. TYPE: str

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

def send(
    self,
    channel: str,
    content: str,
    *,
    conversation_id: str = "",
    metadata: Dict[str, Any] | None = None,
) -> bool:
    """Send an XMPP message.

    Parameters
    ----------
    channel:
        Recipient JID (user or MUC room).
    content:
        Text message content.
    """
    if not self._jid or not self._password:
        logger.warning("Cannot send: no XMPP credentials configured")
        return False

    try:
        import slixmpp

        xmpp = slixmpp.ClientXMPP(self._jid, self._password)
        msg_type = (metadata or {}).get("type", "chat")

        msg = xmpp.make_message(
            mto=channel,
            mbody=content,
            mtype=msg_type,
        )
        msg.send()

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

status

status() -> ChannelStatus

Return the current connection status.

Source code in src/diapason/channels/xmpp_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/xmpp_channel.py

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

on_message

on_message(handler: ChannelHandler) -> None

Register a callback for incoming messages.

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

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