telegram

telegram

TelegramChannel — native Telegram Bot API adapter.

Classes

TelegramChannel

TelegramChannel(bot_token: str = '', *, allowed_chat_ids: str = '', parse_mode: str = 'Markdown', bus: Optional[EventBus] = None)

Bases: BaseChannel

Native Telegram channel adapter using the Bot API.

PARAMETER	DESCRIPTION
bot_token	Telegram Bot API token. Falls back to TELEGRAM_BOT_TOKEN env var. TYPE: str DEFAULT: ''
allowed_chat_ids	Comma-separated list of chat IDs allowed to interact. TYPE: str DEFAULT: ''
parse_mode	Message parse mode (Markdown, HTML, etc.). TYPE: str DEFAULT: 'Markdown'
bus	Optional event bus for publishing channel events. TYPE: Optional[EventBus] DEFAULT: None

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

def __init__(
    self,
    bot_token: str = "",
    *,
    allowed_chat_ids: str = "",
    parse_mode: str = "Markdown",
    bus: Optional[EventBus] = None,
) -> None:
    self._token = bot_token or os.environ.get("TELEGRAM_BOT_TOKEN", "")
    self._allowed_chat_ids = allowed_chat_ids
    self._parse_mode = parse_mode
    self._bus = bus
    self._handlers: List[ChannelHandler] = []
    self._status = ChannelStatus.DISCONNECTED
    self._listener_thread: Optional[threading.Thread] = None
    self._stop_event = threading.Event()
    # Serialize connect/disconnect all the way through listener
    # registration and shutdown.  Without this lock, disconnect() can
    # complete just before a concurrent connect() clears the shared stop
    # event, losing the stop request and leaving a fresh poller behind.
    self._lifecycle_lock = threading.RLock()
    # Published by _poll_loop while it's running so disconnect() can
    # wake the async lifecycle owner (#784).
    self._app: 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 long polling.

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

def connect(self) -> None:
    """Start listening for incoming messages via long polling."""
    with self._lifecycle_lock:
        if self._listener_thread is not None and self._listener_thread.is_alive():
            # A poller is already running against this bot token; starting
            # another one causes Telegram's getUpdates API to return 409
            # Conflict for one or both pollers (#784).
            logger.warning(
                "Telegram channel already connected; ignoring duplicate connect()"
            )
            return

        if not self._token:
            logger.warning("No Telegram bot token configured")
            self._status = ChannelStatus.ERROR
            return

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

        try:
            from telegram.ext import ApplicationBuilder  # noqa: F401

            listener = threading.Thread(
                target=self._poll_loop,
                daemon=True,
            )
            self._listener_thread = listener
            # Publish CONNECTED before starting.  A listener that fails
            # immediately can then reliably overwrite it with ERROR;
            # connect() must not race afterward and hide that failure.
            self._status = ChannelStatus.CONNECTED
            listener.start()
            logger.info("Telegram channel connected (long polling)")
        except ImportError:
            # python-telegram-bot not installed — send-only mode
            logger.info(
                "python-telegram-bot not installed; send-only mode",
            )
            self._status = ChannelStatus.CONNECTED
        except Exception:
            self._listener_thread = None
            self._status = ChannelStatus.ERROR
            logger.exception("Failed to start Telegram listener")

disconnect

disconnect() -> None

Stop the listener thread.

The listener owns every async startup/shutdown phase. disconnect only publishes a stop request onto that owned loop, which is observed between initialize, polling startup, and application startup as well as during steady-state polling. Status is only reported DISCONNECTED once the thread has actually terminated.

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

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

    The listener owns every async startup/shutdown phase. ``disconnect``
    only publishes a stop request onto that owned loop, which is observed
    between initialize, polling startup, and application startup as well
    as during steady-state polling. Status is only reported DISCONNECTED
    once 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("Telegram poll loop's 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(
                    "Telegram listener thread did not stop within timeout;"
                    " leaving status unchanged to avoid a duplicate poller"
                )
                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 Telegram chat via the Bot API.

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

def send(
    self,
    channel: str,
    content: str,
    *,
    conversation_id: str = "",
    metadata: Dict[str, Any] | None = None,
) -> bool:
    """Send a message to a Telegram chat via the Bot API."""
    if not self._token:
        logger.warning("Cannot send: no Telegram bot token")
        return False

    try:
        import httpx

        _TELEGRAM_MAX_LEN = 4096
        url = f"https://api.telegram.org/bot{self._token}/sendMessage"
        # Canonical channel send contract (see BaseChannel.send): the first
        # positional ``channel`` arg is the DESTINATION (the Telegram chat
        # id).  ``conversation_id`` is the inbound message id used as a
        # reply/thread reference (``reply_to_message_id``).  We fall back to
        # ``conversation_id`` as the chat id only when ``channel`` is empty,
        # for backwards compatibility with legacy callers that passed the
        # chat id via ``conversation_id``.
        chat_id = channel or conversation_id
        reply_to = conversation_id if (channel and conversation_id) else ""
        chunks = textwrap.wrap(
            content,
            width=_TELEGRAM_MAX_LEN,
            break_long_words=True,
            replace_whitespace=False,
        )
        if not chunks:
            # Empty content wraps to an empty chunk list -- there is
            # nothing to send, so this must not fall through to the
            # success path below and report a message that was never
            # transmitted (#783).
            return False
        for chunk in chunks:
            payload: Dict[str, Any] = {
                "chat_id": chat_id,
                "text": chunk,
            }
            if self._parse_mode:
                payload["parse_mode"] = self._parse_mode
            if reply_to:
                payload["reply_to_message_id"] = reply_to

            resp = httpx.post(url, json=payload, timeout=10.0)
            if resp.status_code >= 300:
                # Telegram rejects unparseable Markdown (lone asterisks,
                # unclosed code fences, unescaped snake_case
                # identifiers, etc.) with a 400 naming the cause in the
                # response body. Retry once as plain text instead of
                # dropping the message outright (#783).
                if self._parse_mode and "can't parse entities" in resp.text.lower():
                    logger.warning(
                        "Telegram rejected Markdown formatting, "
                        "retrying as plain text: %s",
                        resp.text,
                    )
                    plain_payload = {
                        k: v for k, v in payload.items() if k != "parse_mode"
                    }
                    resp = httpx.post(url, json=plain_payload, timeout=10.0)
                if resp.status_code >= 300:
                    logger.warning(
                        "Telegram API returned status %d: %s",
                        resp.status_code,
                        resp.text,
                    )
                    return False
        self._publish_sent(channel, content, conversation_id)
        return True
    except Exception:
        logger.debug("Telegram send failed", exc_info=True)
        return False

status

status() -> ChannelStatus

Return the current connection status.

Source code in src/openjarvis/channels/telegram.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/telegram.py

def list_channels(self) -> List[str]:
    """Return available channel identifiers."""
    return ["telegram"]

on_message

on_message(handler: ChannelHandler) -> None

Register a callback for incoming messages.

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

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