Skip to content

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 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
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()
Functions
connect 
connect() -> None

Start listening for incoming messages via long polling.

Source code in src/openjarvis/channels/telegram.py 
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
def connect(self) -> None:
    """Start listening for incoming messages via long polling."""
    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

        self._listener_thread = threading.Thread(
            target=self._poll_loop, daemon=True,
        )
        self._listener_thread.start()
        self._status = ChannelStatus.CONNECTED
        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
disconnect 
disconnect() -> None

Stop the listener thread.

Source code in src/openjarvis/channels/telegram.py 
85
86
87
88
89
90
91
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 Telegram chat via the Bot API.

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

        url = f"https://api.telegram.org/bot{self._token}/sendMessage"
        payload: Dict[str, Any] = {
            "chat_id": channel,
            "text": content,
        }
        if self._parse_mode:
            payload["parse_mode"] = self._parse_mode
        if conversation_id:
            payload["reply_to_message_id"] = conversation_id

        resp = httpx.post(url, json=payload, timeout=10.0)
        if resp.status_code < 300:
            self._publish_sent(channel, content, conversation_id)
            return True
        logger.warning(
            "Telegram API returned status %d: %s",
            resp.status_code,
            resp.text,
        )
        return False
    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 
135
136
137
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 
139
140
141
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 
143
144
145
def on_message(self, handler: ChannelHandler) -> None:
    """Register a callback for incoming messages."""
    self._handlers.append(handler)