Skip to content

twitch_channel

twitch_channel

TwitchChannel — Twitch chat adapter via twitchio.

Classes

TwitchChannel 

TwitchChannel(
    access_token: str = "",
    *,
    client_id: str = "",
    nick: str = "",
    initial_channels: str = "",
    bus: Optional[EventBus] = None,
)

Bases: BaseChannel

Twitch chat messaging channel adapter.

Uses the Twitch IRC/EventSub API via twitchio.

PARAMETER DESCRIPTION
access_token

Twitch OAuth access token. Falls back to TWITCH_ACCESS_TOKEN env var.

TYPE: str DEFAULT: ''

client_id

Twitch application client ID. Falls back to TWITCH_CLIENT_ID env var.

TYPE: str DEFAULT: ''

nick

Bot nickname for IRC. Falls back to TWITCH_NICK env var.

TYPE: str DEFAULT: ''

initial_channels

Comma-separated list of channels to join. Falls back to TWITCH_CHANNELS env var.

TYPE: str DEFAULT: ''

bus

Optional event bus for publishing channel events.

TYPE: Optional[EventBus] DEFAULT: None

Source code in src/diapason/channels/twitch_channel.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
def __init__(
    self,
    access_token: str = "",
    *,
    client_id: str = "",
    nick: str = "",
    initial_channels: str = "",
    bus: Optional[EventBus] = None,
) -> None:
    self._access_token = access_token or os.environ.get("TWITCH_ACCESS_TOKEN", "")
    self._client_id = client_id or os.environ.get("TWITCH_CLIENT_ID", "")
    self._nick = nick or os.environ.get("TWITCH_NICK", "")
    self._initial_channels = initial_channels or os.environ.get(
        "TWITCH_CHANNELS", ""
    )
    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/twitch_channel.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
def connect(self) -> None:
    """Validate credentials and mark as connected."""
    if not self._access_token:
        logger.warning("No Twitch access_token configured")
        self._status = ChannelStatus.ERROR
        return
    try:
        import twitchio  # noqa: F401
    except ImportError:
        raise ImportError(
            "twitchio not installed. Install with: uv sync --extra channel-twitch"
        )
    self._status = ChannelStatus.CONNECTED
disconnect 
disconnect() -> None

Mark as disconnected.

Source code in src/diapason/channels/twitch_channel.py 
80
81
82
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 chat message to a Twitch channel.

PARAMETER DESCRIPTION
channel

Twitch channel name (without # prefix).

TYPE: str

content

Chat message content.

TYPE: str

Note

For send-only usage this uses the Twitch Helix API to send a chat message. A full interactive bot would use the twitchio event loop.

Source code in src/diapason/channels/twitch_channel.py 
 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
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
def send(
    self,
    channel: str,
    content: str,
    *,
    conversation_id: str = "",
    metadata: Dict[str, Any] | None = None,
) -> bool:
    """Send a chat message to a Twitch channel.

    Parameters
    ----------
    channel:
        Twitch channel name (without ``#`` prefix).
    content:
        Chat message content.

    Note
    ----
    For send-only usage this uses the Twitch Helix API to send a chat
    message.  A full interactive bot would use the twitchio event loop.
    """
    if not self._access_token:
        logger.warning("Cannot send: no Twitch credentials configured")
        return False

    try:
        import httpx

        url = "https://api.twitch.tv/helix/chat/messages"
        headers = {
            "Authorization": f"Bearer {self._access_token}",
            "Client-Id": self._client_id,
            "Content-Type": "application/json",
        }
        payload: Dict[str, Any] = {
            "broadcaster_id": channel,
            "sender_id": self._nick,
            "message": content,
        }

        resp = httpx.post(url, json=payload, headers=headers, timeout=10.0)
        if resp.status_code < 300:
            self._publish_sent(channel, content, conversation_id)
            return True
        logger.warning("Twitch API returned status %d", resp.status_code)
        return False
    except Exception:
        logger.debug("Twitch send failed", exc_info=True)
        return False
status 
status() -> ChannelStatus

Return the current connection status.

Source code in src/diapason/channels/twitch_channel.py 
137
138
139
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/twitch_channel.py 
141
142
143
def list_channels(self) -> List[str]:
    """Return available channel identifiers."""
    return ["twitch"]
on_message 
on_message(handler: ChannelHandler) -> None

Register a callback for incoming messages.

Source code in src/diapason/channels/twitch_channel.py 
145
146
147
def on_message(self, handler: ChannelHandler) -> None:
    """Register a callback for incoming messages."""
    self._handlers.append(handler)