Skip to content

viber_channel

viber_channel

ViberChannel — Viber adapter via viberbot.

Classes

ViberChannel 

ViberChannel(
    auth_token: str = "",
    *,
    name: str = "",
    avatar: str = "",
    bus: Optional[EventBus] = None,
)

Bases: BaseChannel

Viber messaging channel adapter.

Uses the Viber Bot API via viberbot.

PARAMETER DESCRIPTION
auth_token

Viber bot auth token. Falls back to VIBER_AUTH_TOKEN env var.

TYPE: str DEFAULT: ''

name

Bot display name. Falls back to VIBER_BOT_NAME env var.

TYPE: str DEFAULT: ''

avatar

Bot avatar URL. Falls back to VIBER_BOT_AVATAR env var.

TYPE: str DEFAULT: ''

bus

Optional event bus for publishing channel events.

TYPE: Optional[EventBus] DEFAULT: None

Source code in src/diapason/channels/viber_channel.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
def __init__(
    self,
    auth_token: str = "",
    *,
    name: str = "",
    avatar: str = "",
    bus: Optional[EventBus] = None,
) -> None:
    self._auth_token = auth_token or os.environ.get("VIBER_AUTH_TOKEN", "")
    self._name = name or os.environ.get("VIBER_BOT_NAME", "Diapason")
    self._avatar = avatar or os.environ.get("VIBER_BOT_AVATAR", "")
    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/viber_channel.py 
57
58
59
60
61
62
63
64
65
66
67
68
69
def connect(self) -> None:
    """Validate credentials and mark as connected."""
    if not self._auth_token:
        logger.warning("No Viber auth_token configured")
        self._status = ChannelStatus.ERROR
        return
    try:
        import viberbot  # noqa: F401
    except ImportError:
        raise ImportError(
            "viberbot not installed. Install with: uv sync --extra channel-viber"
        )
    self._status = ChannelStatus.CONNECTED
disconnect 
disconnect() -> None

Mark as disconnected.

Source code in src/diapason/channels/viber_channel.py 
71
72
73
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 a message to a Viber user.

PARAMETER DESCRIPTION
channel

Viber user ID to send to.

TYPE: str

content

Text message content.

TYPE: str

Source code in src/diapason/channels/viber_channel.py 
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
def send(
    self,
    channel: str,
    content: str,
    *,
    conversation_id: str = "",
    metadata: Dict[str, Any] | None = None,
) -> bool:
    """Send a message to a Viber user.

    Parameters
    ----------
    channel:
        Viber user ID to send to.
    content:
        Text message content.
    """
    if not self._auth_token:
        logger.warning("Cannot send: no Viber credentials configured")
        return False

    try:
        from viberbot import Api, BotConfiguration
        from viberbot.api.messages.text_message import TextMessage

        config = BotConfiguration(
            name=self._name,
            avatar=self._avatar,
            auth_token=self._auth_token,
        )
        api = Api(config)
        api.send_messages(channel, [TextMessage(text=content)])

        self._publish_sent(channel, content, conversation_id)
        return True
    except ImportError:
        logger.debug("viberbot not installed")
        return False
    except Exception:
        logger.debug("Viber send failed", exc_info=True)
        return False
status 
status() -> ChannelStatus

Return the current connection status.

Source code in src/diapason/channels/viber_channel.py 
119
120
121
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/viber_channel.py 
123
124
125
def list_channels(self) -> List[str]:
    """Return available channel identifiers."""
    return ["viber"]
on_message 
on_message(handler: ChannelHandler) -> None

Register a callback for incoming messages.

Source code in src/diapason/channels/viber_channel.py 
127
128
129
def on_message(self, handler: ChannelHandler) -> None:
    """Register a callback for incoming messages."""
    self._handlers.append(handler)