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

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()
    # Serialize listener registration and shutdown so a concurrent
    # connect() cannot clear a stop request after disconnect() returns.
    self._lifecycle_lock = threading.RLock()
    # Published by _gateway_loop while it's running so disconnect()
    # can wake the async lifecycle owner (#784).
    self._client: Optional[Any] = None
    self._loop: Optional[Any] = None
    self._async_stop_event: Optional[asyncio.Event] = None
    self._runtime_lock = threading.Lock()

Functions

connect

connect() -> None

Start listening for incoming messages via discord.py gateway.

Source code in src/openjarvis/channels/discord_channel.py

def connect(self) -> None:
    """Start listening for incoming messages via discord.py gateway."""
    with self._lifecycle_lock:
        if self._listener_thread is not None and self._listener_thread.is_alive():
            # A gateway connection is already running; starting another
            # one duplicates message handling (#784).
            logger.warning(
                "Discord channel already connected; ignoring duplicate connect()"
            )
            return

        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

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

disconnect

disconnect() -> None

Stop the listener thread.

The listener owns login, gateway connection, and client cleanup. disconnect publishes a durable stop request onto that loop; the listener races each async startup phase against it and performs the close itself. Status is only reported DISCONNECTED after the thread has actually terminated.

Source code in src/openjarvis/channels/discord_channel.py

def disconnect(self) -> None:
    """Stop the listener thread.

    The listener owns login, gateway connection, and client cleanup.
    ``disconnect`` publishes a durable stop request onto that loop; the
    listener races each async startup phase against it and performs the
    close itself. Status is only reported DISCONNECTED after the thread
    has actually terminated.
    """
    with self._lifecycle_lock:
        self._stop_event.set()

        with self._runtime_lock:
            loop = self._loop
            async_stop_event = self._async_stop_event
        if loop is not None and async_stop_event is not None:
            try:
                loop.call_soon_threadsafe(async_stop_event.set)
            except RuntimeError:
                logger.debug("Discord gateway event loop already closed")

        if self._listener_thread is not None:
            self._listener_thread.join(timeout=5.0)
            if self._listener_thread.is_alive():
                logger.warning(
                    "Discord listener thread did not stop within timeout;"
                    " leaving status unchanged to avoid a duplicate"
                    " gateway connection"
                )
                return
            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

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/openjarvis/channels/discord_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/openjarvis/channels/discord_channel.py

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

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