Self-Hosted Telegram SysAdmin Bot with Ollama
A Telegram sysadmin bot connects a chat interface to a local LLM and, optionally, to real system operations. The design challenge is not “how do I call the Telegram API,” it is “how do I let a chat message trigger AI reasoning and system actions without creating an unauthenticated remote-control channel into my infrastructure.” Treat this project as an API security exercise wrapped around a convenient chat frontend.
Telegram supports two mutually exclusive update mechanisms: long polling with getUpdates, and webhooks with setWebhook. getUpdates is a pull method where your bot repeatedly asks Telegram’s servers for new messages, and it cannot be used while a webhook is active. setWebhook is a push method: Telegram sends an HTTPS POST containing a JSON Update object to your specified URL whenever a message arrives.core.telegram+1
For a homelab or single-admin bot behind NAT, long polling is simpler because it needs no public inbound port. For a bot exposed through your own domain and reverse proxy, a webhook avoids constant polling and reacts faster. If you switch between the two, call deleteWebhook before starting getUpdates, since Telegram returns a conflict error if both are configured at once.
Long Polling Version
This is the easier starting point for a private admin bot with no public exposure.
import time
import requests
TELEGRAM_TOKEN = “YOUR_BOT_TOKEN”
API_BASE = f”https://api.telegram.org/bot{TELEGRAM_TOKEN}“
ALLOWED_CHAT_IDS = {123456789}
def get_updates(offset=None):
params = {“timeout”: 30, “allowed_updates”: [“message”]}
if offset:
params[“offset”] = offset
resp = requests.get(f”{API_BASE}/getUpdates”, params=params, timeout=35)
resp.raise_for_status()
return resp.json()[“result”]
def send_message(chat_id, text):
requests.post(f”{API_BASE}/sendMessage”, json={“chat_id”: chat_id, “text”: text})
def ask_ollama(prompt):
resp = requests.post(
“http://127.0.0.1:11434/api/chat”,
json={
“model”: “llama3.2”,
“messages”: [{“role”: “user”, “content”: prompt}],
“stream”: False,
},
timeout=60,
)
resp.raise_for_status()
return resp.json()[“message”][“content”]
def main():
offset = None
while True:
updates = get_updates(offset)
for update in updates:
offset = update[“update_id”] + 1
message = update.get(“message”)
if not message:
continue
chat_id = message[“chat”][“id”]
if chat_id not in ALLOWED_CHAT_IDS:
continue
text = message.get(“text”, “”)
reply = ask_ollama(text)
send_message(chat_id, reply)
time.sleep(1)
if __name__ == “__main__”:
main()
Ollama’s chat endpoint accepts a model name and a list of role-tagged messages, and returns the model’s reply; setting stream to false gives one complete response object instead of a token stream. The offset parameter in getUpdates must be one greater than the last processed, update_id or Telegram will resend already-handled messages.
If you expose the bot through a domain with a reverse proxy, register the webhook once:
curl -F "url=https://bot.example.internal/telegram/webhook" \
-F "secret_token=YOUR_RANDOM_SECRET" \
"https://api.telegram.org/bot${TELEGRAM_TOKEN}/setWebhook"Telegram lets you set a secret_token so incoming POST requests carry a header you can verify, confirming the request actually originated from your webhook configuration. Your Nginx layer should terminate TLS and forward only to your bot service, following the same reverse-proxy hardening pattern used for any self-hosted API: private-network binding for the backend, rate limiting, and access logging.
from fastapi import FastAPI, Header, HTTPException, Request
app = FastAPI()
WEBHOOK_SECRET = “YOUR_RANDOM_SECRET”
@app.post(“/telegram/webhook”)
async def telegram_webhook(request: Request, x_telegram_bot_api_secret_token: str = Header(None)):
if x_telegram_bot_api_secret_token != WEBHOOK_SECRET:
raise HTTPException(status_code=403, detail=“Invalid secret token”)
update = await request.json()
message = update.get(“message”)
if not message:
return {“ok”: True}
chat_id = message[“chat”][“id”]
if chat_id not in ALLOWED_CHAT_IDS:
return {“ok”: True}
reply = ask_ollama(message.get(“text”, “”))
send_message(chat_id, reply)
return {“ok”: True}
Never skip the secret-token check. Without it, anyone who discovers your webhook URL can submit forged Telegram-shaped JSON payloads directly to your bot.
A sysadmin bot is a privileged interface, so authorization must happen before any AI call or system action, not after. Maintain an explicit allowlist of Telegram chat IDs or user IDs, reject everything else silently or with a generic denial, and log rejected attempts. Do not rely on “security through obscurity” from an unlisted bot username; bot usernames and tokens can leak through screenshots, shared chats, or repository history.
Layer authorization by role if multiple admins use the bot:
{
"chat_id": 123456789,
"role": "read-only",
"allowed_commands": ["status", "ask"]
}
{
"chat_id": 987654321,
"role": "operator",
"allowed_commands": ["status", "ask", "restart-service", "run-backup"]
}Route every command through this permission table before execution. A “read-only” user should never reach a code path that can restart a service or touch a database.
Follow the same separation you’d apply to any AI-driven automation: the model can draft, explain, and suggest, but a structured, validated, allowlisted function should perform any real action. Do not let free-text bot input be interpolated directly into a shell command.
ALLOWED_ACTIONS = {
"status": lambda: run_readonly_check("systemctl status nginx"),
"disk": lambda: run_readonly_check("df -h"),
"restart_nginx": lambda: run_privileged_action("systemctl restart nginx"),
}
def handle_command(role, command):
if command not in ALLOWED_ACTIONS:
return “Unknown or unauthorized command.”
if command.startswith(“restart”) and role != “operator”:
return “You do not have permission for this action.”
return ALLOWED_ACTIONS[command]()
Use this pattern instead of asking the LLM to “generate and run whatever shell command answers the question.” Free-form command generation from a chat message is a direct path to remote code execution if the bot, token, or Telegram account is ever compromised.
Protect Secrets and the Runtime
Store the Telegram bot token, webhook secret, and any service credentials in environment variables or a secrets manager, never hardcoded or committed to a repository. Run the bot process under a dedicated, unprivileged Linux user, restrict its filesystem access, and grant it sudo rights only for the exact commands it must run, ideally via a tightly scoped sudoers entry rather than full root access.
Keep Ollama bound to 127.0.0.1 so only the bot process can reach it, matching the same loopback-binding pattern used for any self-hosted inference API. If the bot and Ollama run in separate containers, use an internal Docker network instead of exposing Ollama’s port to the host or LAN.
Log every command with timestamp, chat ID, role, command name, and outcome, but avoid logging full LLM prompts or responses by default if users might paste sensitive infrastructure details. Add a simple in-memory or Redis-backed rate limiter per chat ID to prevent one user or a compromised token from flooding Ollama with requests or triggering repeated privileged actions.
from collections import defaultdict
import time
_last_call = defaultdict(float)
MIN_INTERVAL_SECONDS = 3
def rate_limited(chat_id):
now = time.time()
if now – _last_call[chat_id] < MIN_INTERVAL_SECONDS:
return True
_last_call[chat_id] = now
return False