Notify Telegram¶
NotifyTelegram sends task notifications to a Telegram chat via the Bot API. You can filter notifications by task status — for example, only receive alerts when a task fails.
Setup¶
1. Create a bot¶
- Open Telegram and search for BotFather
- Type
/newbotand follow the prompts to name your bot
2. Get your bot token¶
- Type
/tokenin the BotFather chat - Select your bot from the list
- Copy the token — you will need it to configure
NotifyTelegram
3. Get your chat ID¶
- Send a message to your bot in Telegram
- Run the following command:
curl 'https://api.telegram.org/bot<YOUR_BOT_TOKEN>/getUpdates'
- In the response, find
result[0].message.chat.id— that is your chat ID
Tip
Store your token and chat ID in environment variables or a .env file. Never hardcode secrets in your source code.
Example¶
import os
from dotenv import load_dotenv
from dotflow import Config, DotFlow, action
from dotflow.core.types.status import TypeStatus
from dotflow.providers import NotifyTelegram
@action
def step_one():
return {"data": "processed"}
@action
def step_two():
raise RuntimeError("Fail!")
def main():
load_dotenv()
notify = NotifyTelegram(
token=os.getenv("BOT_TOKEN", ""),
chat_id=int(os.getenv("CHAT_ID", "0")),
notification_type=TypeStatus.FAILED,
show_result=True,
)
workflow = DotFlow(config=Config(notify=notify))
workflow.task.add(step=step_one)
workflow.task.add(step=step_two)
workflow.start()
return workflow
if __name__ == "__main__":
main()
Notification types¶
notification_type |
When notified |
|---|---|
TypeStatus.FAILED |
Only on failure |
TypeStatus.COMPLETED |
Only on success |
None (default) |
Every task, regardless of status |
Show result¶
Set show_result=True to include the task result (as JSON) in the notification message. Disabled by default to keep messages compact.
show_result |
Behavior |
|---|---|
False (default) |
Only status, workflow ID, and task ID |
True |
Adds the task output as a JSON code block |
Message format¶
Notifications are sent as Telegram messages with:
- Task status with emoji indicator
- Workflow ID and Task ID
- Task result as JSON (when
show_result=True) - Error details (when failed)