Skip to content

discord_channel

discord_channel

DiscordChannel — native Discord Bot API adapter.

Classes

DiscordChannel 

DiscordChannel(
    bot_token: str = "", *, bus: Optional[EventBus] = None
)

Bases: BaseChannel

Native Discord channel adapter using the Discord REST API.

PARAMETER DESCRIPTION
bot_token

Discord bot token. Falls back to DISCORD_BOT_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/discord_channel.py 
36
37
38
39
40
41
42
43
44
45
46
47
def __init__(
    self,
    bot_token: str = "",
    *,
    bus: Optional[EventBus] = None,
) -> None:
    self._token = bot_token or os.environ.get("DISCORD_BOT_TOKEN", "")
    self._bus = bus
    self._handlers: List[ChannelHandler] = []
    self._status = ChannelStatus.DISCONNECTED
    self._listener_thread: Optional[threading.Thread] = None
    self._stop_event = threading.Event()
Methods:
connect 
connect() -> None

Start listening for incoming messages via discord.py gateway.

Source code in src/diapason/channels/discord_channel.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
def connect(self) -> None:
    """Start listening for incoming messages via discord.py gateway."""
    if not self._token:
        logger.warning("No Discord bot token configured")
        self._status = ChannelStatus.ERROR
        return

    self._stop_event.clear()
    self._status = ChannelStatus.CONNECTING

    try:
        import discord  # noqa: F401

        self._listener_thread = threading.Thread(
            target=self._gateway_loop,
            daemon=True,
        )
        self._listener_thread.start()
        self._status = ChannelStatus.CONNECTED
        logger.info("Discord channel connected (gateway)")
    except ImportError:
        logger.info("discord.py not installed; send-only mode")
        self._status = ChannelStatus.CONNECTED
disconnect 
disconnect() -> None

Stop the listener thread.

Source code in src/diapason/channels/discord_channel.py 
75
76
77
78
79
80
81
def disconnect(self) -> None:
    """Stop the listener thread."""
    self._stop_event.set()
    if self._listener_thread is not None:
        self._listener_thread.join(timeout=5.0)
        self._listener_thread = None
    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 Discord channel via REST API.

Source code in src/diapason/channels/discord_channel.py 
 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
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
def send(
    self,
    channel: str,
    content: str,
    *,
    conversation_id: str = "",
    metadata: Dict[str, Any] | None = None,
) -> bool:
    """Send a message to a Discord channel via REST API."""
    if not self._token:
        logger.warning("Cannot send: no Discord bot token")
        return False
    # Defensive guard for #459 follow-up: an empty `channel` arg would
    # produce /channels//messages and silently 404. Better to fail
    # explicitly so the upstream bug (wrong field passed in) surfaces
    # in the warning log instead of silently blackholing the reply.
    if not channel:
        logger.warning(
            "Cannot send: no Discord channel destination "
            "(caller passed empty channel id)"
        )
        return False

    try:
        import httpx

        url = f"https://discord.com/api/v10/channels/{channel}/messages"
        headers = {
            "Authorization": f"Bot {self._token}",
            "Content-Type": "application/json",
        }
        payload: Dict[str, Any] = {"content": content}
        if conversation_id:
            payload["message_reference"] = {"message_id": conversation_id}

        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(
            "Discord API returned status %d: %s",
            resp.status_code,
            resp.text,
        )
        return False
    except Exception:
        logger.debug("Discord send failed", exc_info=True)
        return False
status 
status() -> ChannelStatus

Return the current connection status.

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

Register a callback for incoming messages.

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