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/openjarvis/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()
Functions
connect 
connect() -> None

Start listening for incoming messages via discord.py gateway.

Source code in src/openjarvis/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
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/openjarvis/channels/discord_channel.py 
74
75
76
77
78
79
80
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/openjarvis/channels/discord_channel.py 
 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
118
119
120
121
122
123
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

    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/openjarvis/channels/discord_channel.py 
125
126
127
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/openjarvis/channels/discord_channel.py 
129
130
131
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/openjarvis/channels/discord_channel.py 
133
134
135
def on_message(self, handler: ChannelHandler) -> None:
    """Register a callback for incoming messages."""
    self._handlers.append(handler)