Tools

Tools: FastAPI + Telegram: Building a Real-Time Alert Bot in 30 Minutes - 2025 Update

2026-04-04 0 views admin

FastAPI + Telegram: Building a Real-Time Alert Bot in 30 Minutes

Why Telegram (Not Slack, Not Email)

Step 1: Create Your Bot with BotFather

Step 2: Project Setup

Step 3: The FastAPI Application

Step 4: Sending Alerts From Anywhere

Step 5: VPS Deployment

Real-World Performance Numbers

Want This Built for Your Business? At 2:47 AM, my server's CPU spiked to 98% and stayed there. I found out about it the next morning when users were already complaining. That incident cost me three hours of debugging and a genuinely embarrassing conversation with a client. Two days later, I had a Telegram bot pinging me within seconds of any anomaly. I haven't missed a critical alert since. If you're still relying on email alerts, cron-job log checks, or — God forbid — manually SSHing into servers to verify things are running, this tutorial will change how you operate. We're building a real-time alert system using FastAPI as the webhook receiver and Telegram as the delivery channel. The same architecture I use for trading signal alerts, CI/CD pipeline notifications, and server health monitoring. Slack is expensive at scale and requires your users to have accounts. Email has unpredictable delivery delays and gets buried. PagerDuty is overkill for solo developers and small teams. Telegram is free, instant, has a dead-simple bot API, and the mobile app actually wakes you up at 3 AM. The bot API rate limits are generous (30 messages/second to different chats), and there's no per-seat pricing. The Telegram Bot API also accepts both polling and webhooks. We're using webhooks because polling burns resources and introduces latency. With a webhook, Telegram pushes updates to your FastAPI endpoint the moment something happens. Open Telegram and search for @BotFather. Send /newbot, follow the prompts, and you'll receive a token that looks like this: Keep that token private. Anyone with it can send messages as your bot. Store it in an environment variable, not hardcoded in your source. Next, get your chat ID. Start a conversation with your new bot (just send /start), then hit this URL in your browser, replacing YOUR_TOKEN: The response will include your chat.id. For personal alerts, this is your personal ID. For team alerts, create a group, add the bot, send a message, and grab the group's chat ID from the same endpoint. Group IDs are negative numbers (e.g., -1001234567890). The WEBHOOK_SECRET is a string you'll attach to your webhook URL so random internet traffic can't trigger your alerts. Generate it with python -c "import secrets; print(secrets.token_hex(32))". Here's the full working application. I'll walk through the key decisions after the code: A few decisions worth explaining: I'm using httpx.AsyncClient instead of the requests library because FastAPI is async, and blocking the event loop with synchronous HTTP calls is a real problem under load. The retry logic with 429 handling isn't theoretical — Telegram will rate-limit you if you're sending burst alerts from multiple sources simultaneously. The secret in the URL path is crude but effective; a header-based approach works too, but this is easier to configure in most CI/CD systems. Run the server locally first: And on your phone, within about 300ms, you'll see: For a trading alert system, I send payloads like this from a separate signal-detection service: On your Ubuntu VPS (I use a $6/month Hetzner CX11 for this), set up a systemd service so it survives reboots: Put Nginx in front of it for SSL termination. Without HTTPS, your webhook secret is traveling in plaintext. Get a free cert with Certbot and point your Nginx location block to proxy_pass http://127.0.0.1:8000. For server monitoring, I run a simple cron job that hits the alert endpoint if disk or memory crosses a threshold: For CI/CD, add a step to your GitHub Actions workflow: In production, this setup handles about 2,000 alerts/day across three different projects without any issues. End-to-end latency from the triggering event to Telegram notification appearing on my phone averages 380ms on a good network day, occasionally spiking to 1.2 seconds if Telegram's servers are having a moment. The FastAPI endpoint itself responds in under 5ms — all the latency is in the outbound HTTP call to Telegram. The one error I hit repeatedly when first building this was {"ok":false,"error_code":400,"description":"Bad Request: chat not found"}. This almost always means your bot hasn't been started by the user, or for groups, the bot wasn't added as a member before you tried to message it. Send /start to your bot first, or verify the group ID is correct. Memory usage for the service sits at about 45MB RSS with two Uvicorn workers — completely negligible. This is the kind of infrastructure that runs forever without you thinking about it, which is exactly what you want from an alerting system. I build custom Python automation systems, trading bots, and AI-powered tools that run 24/7 in production. Currently available for consulting and contract work: DM me on dev.to or reach out on either platform. I respond within 24 hours. Templates let you quickly answer FAQs or store snippets for re-use. Hide child comments as well For further actions, you may consider blocking this person and/or reporting abuse

Code Block

Copy

5839201847:AAHxyz_example_token_here_abc123def456 5839201847:AAHxyz_example_token_here_abc123def456 5839201847:AAHxyz_example_token_here_abc123def456 https://api.telegram.org/botYOUR_TOKEN/getUpdates https://api.telegram.org/botYOUR_TOKEN/getUpdates https://api.telegram.org/botYOUR_TOKEN/getUpdates mkdir fastapi-telegram-bot && cd fastapi-telegram-bot python -m venv venv && source venv/bin/activate pip install fastapi uvicorn httpx python-dotenv pydantic mkdir fastapi-telegram-bot && cd fastapi-telegram-bot python -m venv venv && source venv/bin/activate pip install fastapi uvicorn httpx python-dotenv pydantic mkdir fastapi-telegram-bot && cd fastapi-telegram-bot python -m venv venv && source venv/bin/activate pip install fastapi uvicorn httpx python-dotenv pydantic TELEGRAM_BOT_TOKEN=5839201847:AAHxyz_your_actual_token TELEGRAM_CHAT_ID=123456789 WEBHOOK_SECRET=your_random_secret_string_here TELEGRAM_BOT_TOKEN=5839201847:AAHxyz_your_actual_token TELEGRAM_CHAT_ID=123456789 WEBHOOK_SECRET=your_random_secret_string_here TELEGRAM_BOT_TOKEN=5839201847:AAHxyz_your_actual_token TELEGRAM_CHAT_ID=123456789 WEBHOOK_SECRET=your_random_secret_string_here import os import logging from contextlib import asynccontextmanager from typing import Any import httpx from fastapi import FastAPI, HTTPException, Header, Request, status from pydantic import BaseModel, Field from dotenv import load_dotenv load_dotenv() logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s" ) logger = logging.getLogger(__name__) TELEGRAM_BOT_TOKEN = os.getenv("TELEGRAM_BOT_TOKEN") TELEGRAM_CHAT_ID = os.getenv("TELEGRAM_CHAT_ID") WEBHOOK_SECRET = os.getenv("WEBHOOK_SECRET") TELEGRAM_API_URL = f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}" if not all([TELEGRAM_BOT_TOKEN, TELEGRAM_CHAT_ID, WEBHOOK_SECRET]): raise RuntimeError("Missing required environment variables. Check .env file.") class AlertPayload(BaseModel): title: str = Field(..., min_length=1, max_length=100) message: str = Field(..., min_length=1, max_length=3000) severity: str = Field(default="info", pattern="^(info|warning|critical)$") source: str = Field(default="system") chat_id: str | None = None # Override default chat if needed class TelegramSender: def __init__(self): self.client = httpx.AsyncClient(timeout=10.0) async def send_message( self, text: str, chat_id: str, parse_mode: str = "HTML", retries: int = 3 ) -> dict[str, Any]: url = f"{TELEGRAM_API_URL}/sendMessage" payload = { "chat_id": chat_id, "text": text, "parse_mode": parse_mode, "disable_web_page_preview": True, } for attempt in range(retries): try: response = await self.client.post(url, json=payload) response.raise_for_status() result = response.json() if not result.get("ok"): raise ValueError(f"Telegram API error: {result.get('description')}") logger.info(f"Message sent successfully. Message ID: {result['result']['message_id']}") return result except httpx.TimeoutException: logger.warning(f"Timeout on attempt {attempt + 1}/{retries}") if attempt == retries - 1: raise except httpx.HTTPStatusError as e: # 429 = rate limited, back off and retry if e.response.status_code == 429: retry_after = int(e.response.headers.get("Retry-After", 5)) logger.warning(f"Rate limited. Waiting {retry_after}s before retry.") import asyncio await asyncio.sleep(retry_after) else: raise raise RuntimeError("Failed to send message after all retries") async def close(self): await self.client.aclose() telegram = TelegramSender() def format_alert_message(alert: AlertPayload) -> str: severity_emoji = { "info": "ℹ️", "warning": "⚠️", "critical": "🚨" } emoji = severity_emoji.get(alert.severity, "ℹ️") return ( f"{emoji} [{alert.severity.upper()}] {alert.title}\n\n" f"{alert.message}\n\n" f"Source: {alert.source}" ) @asynccontextmanager async def lifespan(app: FastAPI): logger.info("Starting FastAPI Telegram Alert Bot") yield logger.info("Shutting down — closing HTTP client") await telegram.close() app = FastAPI( title="Telegram Alert Bot", description="Real-time alert delivery via Telegram", version="1.0.0", lifespan=lifespan ) @app.post("/alert/{secret}", status_code=status.HTTP_200_OK) async def receive_alert( secret: str, alert: AlertPayload, request: Request ): # Validate the secret before doing anything else if secret != WEBHOOK_SECRET: logger.warning(f"Invalid secret attempt from {request.client.host}") raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Invalid webhook secret" ) chat_id = alert.chat_id or TELEGRAM_CHAT_ID message_text = format_alert_message(alert) try: result = await telegram.send_message( text=message_text, chat_id=chat_id ) return { "status": "delivered", "message_id": result["result"]["message_id"], "chat_id": chat_id } except httpx.HTTPStatusError as e: logger.error(f"Telegram HTTP error: {e.response.status_code} - {e.response.text}") raise HTTPException( status_code=status.HTTP_502_BAD_GATEWAY, detail=f"Telegram API returned {e.response.status_code}" ) except Exception as e: logger.error(f"Unexpected error sending alert: {str(e)}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Failed to deliver alert" ) @app.get("/health") async def health_check(): return {"status": "ok", "bot_configured": bool(TELEGRAM_BOT_TOKEN)} import os import logging from contextlib import asynccontextmanager from typing import Any import httpx from fastapi import FastAPI, HTTPException, Header, Request, status from pydantic import BaseModel, Field from dotenv import load_dotenv load_dotenv() logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s" ) logger = logging.getLogger(__name__) TELEGRAM_BOT_TOKEN = os.getenv("TELEGRAM_BOT_TOKEN") TELEGRAM_CHAT_ID = os.getenv("TELEGRAM_CHAT_ID") WEBHOOK_SECRET = os.getenv("WEBHOOK_SECRET") TELEGRAM_API_URL = f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}" if not all([TELEGRAM_BOT_TOKEN, TELEGRAM_CHAT_ID, WEBHOOK_SECRET]): raise RuntimeError("Missing required environment variables. Check .env file.") class AlertPayload(BaseModel): title: str = Field(..., min_length=1, max_length=100) message: str = Field(..., min_length=1, max_length=3000) severity: str = Field(default="info", pattern="^(info|warning|critical)$") source: str = Field(default="system") chat_id: str | None = None # Override default chat if needed class TelegramSender: def __init__(self): self.client = httpx.AsyncClient(timeout=10.0) async def send_message( self, text: str, chat_id: str, parse_mode: str = "HTML", retries: int = 3 ) -> dict[str, Any]: url = f"{TELEGRAM_API_URL}/sendMessage" payload = { "chat_id": chat_id, "text": text, "parse_mode": parse_mode, "disable_web_page_preview": True, } for attempt in range(retries): try: response = await self.client.post(url, json=payload) response.raise_for_status() result = response.json() if not result.get("ok"): raise ValueError(f"Telegram API error: {result.get('description')}") logger.info(f"Message sent successfully. Message ID: {result['result']['message_id']}") return result except httpx.TimeoutException: logger.warning(f"Timeout on attempt {attempt + 1}/{retries}") if attempt == retries - 1: raise except httpx.HTTPStatusError as e: # 429 = rate limited, back off and retry if e.response.status_code == 429: retry_after = int(e.response.headers.get("Retry-After", 5)) logger.warning(f"Rate limited. Waiting {retry_after}s before retry.") import asyncio await asyncio.sleep(retry_after) else: raise raise RuntimeError("Failed to send message after all retries") async def close(self): await self.client.aclose() telegram = TelegramSender() def format_alert_message(alert: AlertPayload) -> str: severity_emoji = { "info": "ℹ️", "warning": "⚠️", "critical": "🚨" } emoji = severity_emoji.get(alert.severity, "ℹ️") return ( f"{emoji} [{alert.severity.upper()}] {alert.title}\n\n" f"{alert.message}\n\n" f"Source: {alert.source}" ) @asynccontextmanager async def lifespan(app: FastAPI): logger.info("Starting FastAPI Telegram Alert Bot") yield logger.info("Shutting down — closing HTTP client") await telegram.close() app = FastAPI( title="Telegram Alert Bot", description="Real-time alert delivery via Telegram", version="1.0.0", lifespan=lifespan ) @app.post("/alert/{secret}", status_code=status.HTTP_200_OK) async def receive_alert( secret: str, alert: AlertPayload, request: Request ): # Validate the secret before doing anything else if secret != WEBHOOK_SECRET: logger.warning(f"Invalid secret attempt from {request.client.host}") raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Invalid webhook secret" ) chat_id = alert.chat_id or TELEGRAM_CHAT_ID message_text = format_alert_message(alert) try: result = await telegram.send_message( text=message_text, chat_id=chat_id ) return { "status": "delivered", "message_id": result["result"]["message_id"], "chat_id": chat_id } except httpx.HTTPStatusError as e: logger.error(f"Telegram HTTP error: {e.response.status_code} - {e.response.text}") raise HTTPException( status_code=status.HTTP_502_BAD_GATEWAY, detail=f"Telegram API returned {e.response.status_code}" ) except Exception as e: logger.error(f"Unexpected error sending alert: {str(e)}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Failed to deliver alert" ) @app.get("/health") async def health_check(): return {"status": "ok", "bot_configured": bool(TELEGRAM_BOT_TOKEN)} import os import logging from contextlib import asynccontextmanager from typing import Any import httpx from fastapi import FastAPI, HTTPException, Header, Request, status from pydantic import BaseModel, Field from dotenv import load_dotenv load_dotenv() logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s" ) logger = logging.getLogger(__name__) TELEGRAM_BOT_TOKEN = os.getenv("TELEGRAM_BOT_TOKEN") TELEGRAM_CHAT_ID = os.getenv("TELEGRAM_CHAT_ID") WEBHOOK_SECRET = os.getenv("WEBHOOK_SECRET") TELEGRAM_API_URL = f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}" if not all([TELEGRAM_BOT_TOKEN, TELEGRAM_CHAT_ID, WEBHOOK_SECRET]): raise RuntimeError("Missing required environment variables. Check .env file.") class AlertPayload(BaseModel): title: str = Field(..., min_length=1, max_length=100) message: str = Field(..., min_length=1, max_length=3000) severity: str = Field(default="info", pattern="^(info|warning|critical)$") source: str = Field(default="system") chat_id: str | None = None # Override default chat if needed class TelegramSender: def __init__(self): self.client = httpx.AsyncClient(timeout=10.0) async def send_message( self, text: str, chat_id: str, parse_mode: str = "HTML", retries: int = 3 ) -> dict[str, Any]: url = f"{TELEGRAM_API_URL}/sendMessage" payload = { "chat_id": chat_id, "text": text, "parse_mode": parse_mode, "disable_web_page_preview": True, } for attempt in range(retries): try: response = await self.client.post(url, json=payload) response.raise_for_status() result = response.json() if not result.get("ok"): raise ValueError(f"Telegram API error: {result.get('description')}") logger.info(f"Message sent successfully. Message ID: {result['result']['message_id']}") return result except httpx.TimeoutException: logger.warning(f"Timeout on attempt {attempt + 1}/{retries}") if attempt == retries - 1: raise except httpx.HTTPStatusError as e: # 429 = rate limited, back off and retry if e.response.status_code == 429: retry_after = int(e.response.headers.get("Retry-After", 5)) logger.warning(f"Rate limited. Waiting {retry_after}s before retry.") import asyncio await asyncio.sleep(retry_after) else: raise raise RuntimeError("Failed to send message after all retries") async def close(self): await self.client.aclose() telegram = TelegramSender() def format_alert_message(alert: AlertPayload) -> str: severity_emoji = { "info": "ℹ️", "warning": "⚠️", "critical": "🚨" } emoji = severity_emoji.get(alert.severity, "ℹ️") return ( f"{emoji} [{alert.severity.upper()}] {alert.title}\n\n" f"{alert.message}\n\n" f"Source: {alert.source}" ) @asynccontextmanager async def lifespan(app: FastAPI): logger.info("Starting FastAPI Telegram Alert Bot") yield logger.info("Shutting down — closing HTTP client") await telegram.close() app = FastAPI( title="Telegram Alert Bot", description="Real-time alert delivery via Telegram", version="1.0.0", lifespan=lifespan ) @app.post("/alert/{secret}", status_code=status.HTTP_200_OK) async def receive_alert( secret: str, alert: AlertPayload, request: Request ): # Validate the secret before doing anything else if secret != WEBHOOK_SECRET: logger.warning(f"Invalid secret attempt from {request.client.host}") raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Invalid webhook secret" ) chat_id = alert.chat_id or TELEGRAM_CHAT_ID message_text = format_alert_message(alert) try: result = await telegram.send_message( text=message_text, chat_id=chat_id ) return { "status": "delivered", "message_id": result["result"]["message_id"], "chat_id": chat_id } except httpx.HTTPStatusError as e: logger.error(f"Telegram HTTP error: {e.response.status_code} - {e.response.text}") raise HTTPException( status_code=status.HTTP_502_BAD_GATEWAY, detail=f"Telegram API returned {e.response.status_code}" ) except Exception as e: logger.error(f"Unexpected error sending alert: {str(e)}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Failed to deliver alert" ) @app.get("/health") async def health_check(): return {"status": "ok", "bot_configured": bool(TELEGRAM_BOT_TOKEN)} uvicorn main:app --host 0.0.0.0 --port 8000 --reload uvicorn main:app --host 0.0.0.0 --port 8000 --reload uvicorn main:app --host 0.0.0.0 --port 8000 --reload curl -X POST "http://localhost:8000/alert/your_secret_here" \ -H "Content-Type: application/json" \ -d '{ "title": "CPU Spike Detected", "message": "Server cpu01 hit 97% CPU usage. Load average: 14.2, 13.8, 12.1. Top process: python3 (PID 18432).", "severity": "critical", "source": "prometheus-alertmanager" }' curl -X POST "http://localhost:8000/alert/your_secret_here" \ -H "Content-Type: application/json" \ -d '{ "title": "CPU Spike Detected", "message": "Server cpu01 hit 97% CPU usage. Load average: 14.2, 13.8, 12.1. Top process: python3 (PID 18432).", "severity": "critical", "source": "prometheus-alertmanager" }' curl -X POST "http://localhost:8000/alert/your_secret_here" \ -H "Content-Type: application/json" \ -d '{ "title": "CPU Spike Detected", "message": "Server cpu01 hit 97% CPU usage. Load average: 14.2, 13.8, 12.1. Top process: python3 (PID 18432).", "severity": "critical", "source": "prometheus-alertmanager" }' { "status": "delivered", "message_id": 47, "chat_id": "123456789" } { "status": "delivered", "message_id": 47, "chat_id": "123456789" } { "status": "delivered", "message_id": 47, "chat_id": "123456789" } 🚨 [CRITICAL] CPU Spike Detected Server cpu01 hit 97% CPU usage. Load average: 14.2, 13.8, 12.1. Top process: python3 (PID 18432). Source: prometheus-alertmanager 🚨 [CRITICAL] CPU Spike Detected Server cpu01 hit 97% CPU usage. Load average: 14.2, 13.8, 12.1. Top process: python3 (PID 18432). Source: prometheus-alertmanager 🚨 [CRITICAL] CPU Spike Detected Server cpu01 hit 97% CPU usage. Load average: 14.2, 13.8, 12.1. Top process: python3 (PID 18432). Source: prometheus-alertmanager import httpx import asyncio async def send_trading_alert(signal: dict): alert_url = "https://your-vps.com/alert/your_secret_here" message = ( f"Symbol: {signal['symbol']}\n" f"Action: {signal['action']}\n" f"Price: ${signal['price']:.4f}\n" f"RSI: {signal['rsi']:.1f} | MACD: {signal['macd']:.6f}\n" f"Confidence: {signal['confidence']}%" ) payload = { "title": f"{signal['action']} Signal — {signal['symbol']}", "message": message, "severity": "warning" if signal['confidence'] < 75 else "critical", "source": "trading-engine-v2" } async with httpx.AsyncClient(timeout=5.0) as client: response = await client.post(alert_url, json=payload) response.raise_for_status() return response.json() import httpx import asyncio async def send_trading_alert(signal: dict): alert_url = "https://your-vps.com/alert/your_secret_here" message = ( f"Symbol: {signal['symbol']}\n" f"Action: {signal['action']}\n" f"Price: ${signal['price']:.4f}\n" f"RSI: {signal['rsi']:.1f} | MACD: {signal['macd']:.6f}\n" f"Confidence: {signal['confidence']}%" ) payload = { "title": f"{signal['action']} Signal — {signal['symbol']}", "message": message, "severity": "warning" if signal['confidence'] < 75 else "critical", "source": "trading-engine-v2" } async with httpx.AsyncClient(timeout=5.0) as client: response = await client.post(alert_url, json=payload) response.raise_for_status() return response.json() import httpx import asyncio async def send_trading_alert(signal: dict): alert_url = "https://your-vps.com/alert/your_secret_here" message = ( f"Symbol: {signal['symbol']}\n" f"Action: {signal['action']}\n" f"Price: ${signal['price']:.4f}\n" f"RSI: {signal['rsi']:.1f} | MACD: {signal['macd']:.6f}\n" f"Confidence: {signal['confidence']}%" ) payload = { "title": f"{signal['action']} Signal — {signal['symbol']}", "message": message, "severity": "warning" if signal['confidence'] < 75 else "critical", "source": "trading-engine-v2" } async with httpx.AsyncClient(timeout=5.0) as client: response = await client.post(alert_url, json=payload) response.raise_for_status() return response.json() # /etc/systemd/system/telegram-alert-bot.service [Unit] Description=FastAPI Telegram Alert Bot After=network.target [Service] Type=simple User=ubuntu WorkingDirectory=/home/ubuntu/fastapi-telegram-bot EnvironmentFile=/home/ubuntu/fastapi-telegram-bot/.env ExecStart=/home/ubuntu/fastapi-telegram-bot/venv/bin/uvicorn main:app --host 0.0.0.0 --port 8000 --workers 2 Restart=always RestartSec=5 [Install] WantedBy=multi-user.target # /etc/systemd/system/telegram-alert-bot.service [Unit] Description=FastAPI Telegram Alert Bot After=network.target [Service] Type=simple User=ubuntu WorkingDirectory=/home/ubuntu/fastapi-telegram-bot EnvironmentFile=/home/ubuntu/fastapi-telegram-bot/.env ExecStart=/home/ubuntu/fastapi-telegram-bot/venv/bin/uvicorn main:app --host 0.0.0.0 --port 8000 --workers 2 Restart=always RestartSec=5 [Install] WantedBy=multi-user.target # /etc/systemd/system/telegram-alert-bot.service [Unit] Description=FastAPI Telegram Alert Bot After=network.target [Service] Type=simple User=ubuntu WorkingDirectory=/home/ubuntu/fastapi-telegram-bot EnvironmentFile=/home/ubuntu/fastapi-telegram-bot/.env ExecStart=/home/ubuntu/fastapi-telegram-bot/venv/bin/uvicorn main:app --host 0.0.0.0 --port 8000 --workers 2 Restart=always RestartSec=5 [Install] WantedBy=multi-user.target sudo systemctl daemon-reload sudo systemctl enable telegram-alert-bot sudo systemctl start telegram-alert-bot sudo systemctl status telegram-alert-bot sudo systemctl daemon-reload sudo systemctl enable telegram-alert-bot sudo systemctl start telegram-alert-bot sudo systemctl status telegram-alert-bot sudo systemctl daemon-reload sudo systemctl enable telegram-alert-bot sudo systemctl start telegram-alert-bot sudo systemctl status telegram-alert-bot # crontab -e */5 * * * * /home/ubuntu/scripts/check_health.sh # crontab -e */5 * * * * /home/ubuntu/scripts/check_health.sh # crontab -e */5 * * * * /home/ubuntu/scripts/check_health.sh - name: Send deployment alert run: | curl -s -X POST "${{ secrets.ALERT_WEBHOOK_URL }}/alert/${{ secrets.WEBHOOK_SECRET }}" \ -H "Content-Type: application/json" \ -d "{\"title\": \"Deployment Complete\", \"message\": \"${{ github.repository }} deployed commit ${{ github.sha }} to production.\", \"severity\": \"info\", \"source\": \"github-actions\"}" - name: Send deployment alert run: | curl -s -X POST "${{ secrets.ALERT_WEBHOOK_URL }}/alert/${{ secrets.WEBHOOK_SECRET }}" \ -H "Content-Type: application/json" \ -d "{\"title\": \"Deployment Complete\", \"message\": \"${{ github.repository }} deployed commit ${{ github.sha }} to production.\", \"severity\": \"info\", \"source\": \"github-actions\"}" - name: Send deployment alert run: | curl -s -X POST "${{ secrets.ALERT_WEBHOOK_URL }}/alert/${{ secrets.WEBHOOK_SECRET }}" \ -H "Content-Type: application/json" \ -d "{\"title\": \"Deployment Complete\", \"message\": \"${{ github.repository }} deployed commit ${{ github.sha }} to production.\", \"severity\": \"info\", \"source\": \"github-actions\"}" - Hire me on Upwork — Python automation, API integrations, trading systems - Check my Fiverr gigs — Bot development, web scraping, data pipelines

Share this article

Twitter Facebook LinkedIn Reddit

🏷️ Tags

toolsutilitiessecurity toolsfastapitelegrambuildingalertminutesupdate

More from Tools

Tools: Mistakes to Avoid When Self-Hosting OpenClaw (Learned the Hard Way) 5

2026-04-04 0

Tools: Docker Build Output: 50 Lines You Don't Need (2026)

2026-04-04 0

Tools: How to Self-Host Your Entire Dev Stack for Under $20/Month in 2026

2026-04-04 0

Tools: Ollama + Open WebUI Self-Hosting Guide 2026 — Run Your Own AI for $0

2026-04-04 0

Trending

1

CVE-2025-61481: Critical Remote Code Execution Vulnerability in MikroTik RouterOS & SwitchOS

2025-10-27 • 189 views

2

CVE-2025-43939: Dell Unity OS Command Injection (High)

2025-10-30 • 148 views

3

Google disputes false claims of massive Gmail data breach

2025-10-30 • 130 views

4

Microsoft: DNS outage impacts Azure and Microsoft 365 services

2025-10-30 • 88 views

5

3.5B Accounts, 1 Critical Flaw: Meta Closes WhatsApp Data-Harvesting

2025-11-25 • 81 views

InfinitSec - Latest Cybersecurity, Technology & Gaming News

Tools: FastAPI + Telegram: Building a Real-Time Alert Bot in 30 Minutes - 2025 Update

FastAPI + Telegram: Building a Real-Time Alert Bot in 30 Minutes

Why Telegram (Not Slack, Not Email)

Step 1: Create Your Bot with BotFather

Step 2: Project Setup

Step 3: The FastAPI Application

Step 4: Sending Alerts From Anywhere

Step 5: VPS Deployment

Real-World Performance Numbers

🏷️ Tags

More from Tools

Tools: Mistakes to Avoid When Self-Hosting OpenClaw (Learned the Hard Way) 5

Tools: Docker Build Output: 50 Lines You Don't Need (2026)

Tools: How to Self-Host Your Entire Dev Stack for Under $20/Month in 2026

Tools: Ollama + Open WebUI Self-Hosting Guide 2026 — Run Your Own AI for $0

Trending

CVE-2025-61481: Critical Remote Code Execution Vulnerability in MikroTik RouterOS & SwitchOS

CVE-2025-43939: Dell Unity OS Command Injection (High)

Google disputes false claims of massive Gmail data breach

Microsoft: DNS outage impacts Azure and Microsoft 365 services

3.5B Accounts, 1 Critical Flaw: Meta Closes WhatsApp Data-Harvesting