Overview
The JARVIS Telegram bot provides an alternative capture channel for sending photos to the identification pipeline. This is particularly useful for:
Testing without Meta glasses hardware
Remote capture from any device with Telegram
Multi-user deployments where multiple people can submit photos
Fallback when webhook endpoints are unavailable
The Telegram bot is automatically disabled when TELEGRAM_BOT_TOKEN is not configured, allowing graceful degradation.
Architecture
┌─────────────────────────────────────────────────────┐
│ Telegram Flow │
│ │
│ User → Telegram App → Bot API → JARVIS Backend │
│ │ │ │ │
│ │ │ ▼ │
│ │ │ TelegramCaptureBot │
│ │ │ │ │
│ ▼ ▼ ▼ │
│ Send Photo Webhook CapturePipeline │
│ Polling │ │
│ ▼ │
│ Face Detection → │
│ Identification → │
│ Person Dossier │
└─────────────────────────────────────────────────────┘
Prerequisites
Create a Telegram Bot
Message @BotFather on Telegram: Follow the prompts:
Bot name : JARVIS Capture Bot
Username : your_jarvis_bot (must end in ‘bot’)
BotFather will respond with your bot token: Use this token to access the HTTP API:
1234567890:ABCdefGHIjklMNOpqrsTUVwxyz
Keep your bot token secret! Anyone with this token can control your bot.
Install Dependencies
Install the Python Telegram Bot library: pip install python-telegram-bot
Or add to requirements.txt: python-telegram-bot>=20.0
Configure Environment
Add your bot token to .env: TELEGRAM_BOT_TOKEN = 1234567890:ABCdefGHIjklMNOpqrsTUVwxyz
Implementation
The TelegramCaptureBot class handles photo reception and pipeline integration.
Bot Initialization
from telegram import Update
from telegram.ext import Application, CommandHandler, MessageHandler, filters
from loguru import logger
class TelegramCaptureBot :
"""Telegram bot that receives photos and runs them through the capture pipeline."""
def __init__ ( self , token : str , pipeline : CapturePipeline) -> None :
self ._token = token
self ._pipeline = pipeline
self ._app: Application | None = None
async def start ( self ) -> None :
"""Initialize and start polling in the background."""
self ._app = (
Application.builder()
.token( self ._token)
.build()
)
self ._app.add_handler(CommandHandler( "start" , self ._handle_start))
self ._app.add_handler(
MessageHandler(filters. PHOTO | filters.Document. IMAGE , self ._handle_photo)
)
await self ._app.initialize()
await self ._app.start()
await self ._app.updater.start_polling()
logger.info( "Telegram capture bot started polling" )
View the complete implementation in source/backend/capture/telegram_bot.py:24
Command Handlers
/start Command
Photo Handler
Welcome message when users first interact with the bot: @ staticmethod
async def _handle_start ( update : Update, _context : object ) -> None :
if update.effective_message:
await update.effective_message.reply_text(
"SPECTER capture bot ready. Send a photo to analyze."
)
Users will see: SPECTER capture bot ready. Send a photo to analyze.
Process incoming photos and send to pipeline: async def _handle_photo ( self , update : Update, _context : object ) -> None :
message = update.effective_message
if message is None :
return
# Prefer highest resolution photo, fall back to document
if message.photo:
file_obj = await message.photo[ - 1 ].get_file()
elif message.document:
file_obj = await message.document.get_file()
else :
await message.reply_text( "No image found in message." )
return
data = await file_obj.download_as_bytearray()
logger.info( "Telegram photo received, {} bytes" , len (data))
from uuid import uuid4
capture_id = f "cap_ { uuid4().hex[: 12 ] } "
try :
result = await self ._pipeline.process(
capture_id = capture_id,
data = bytes (data),
content_type = "image/jpeg" ,
source = "telegram" ,
)
await message.reply_text(
f "Processed { capture_id } : { result.faces_detected } face(s) detected, "
f " { len (result.persons_created) } person(s) created."
)
except Exception as exc:
logger.error( "Telegram pipeline error: {} " , exc)
await message.reply_text( f "Processing failed: { exc } " )
View implementation in source/backend/capture/telegram_bot.py:67
Factory Function
Create bot instances with graceful fallback:
def create_telegram_bot (
token : str | None ,
pipeline : CapturePipeline
) -> TelegramCaptureBot | None :
"""Factory that returns None when token is missing or library unavailable."""
if not token:
logger.info( "TELEGRAM_BOT_TOKEN not set, Telegram bot disabled" )
return None
if not _TELEGRAM_AVAILABLE :
logger.warning( "python-telegram-bot not installed, Telegram bot disabled" )
return None
return TelegramCaptureBot( token = token, pipeline = pipeline)
View implementation in source/backend/capture/telegram_bot.py:103
Integration with Backend
Add Telegram bot to your FastAPI application:
# main.py
from backend.capture.telegram_bot import create_telegram_bot
from pipeline import CapturePipeline
import os
app = FastAPI()
pipeline = CapturePipeline()
@app.on_event ( "startup" )
async def startup ():
# Initialize Telegram bot if configured
telegram_token = os.getenv( "TELEGRAM_BOT_TOKEN" )
telegram_bot = create_telegram_bot(telegram_token, pipeline)
if telegram_bot:
await telegram_bot.start()
logger.info( "Telegram bot started successfully" )
else :
logger.info( "Telegram bot disabled" )
@app.on_event ( "shutdown" )
async def shutdown ():
if telegram_bot:
await telegram_bot.stop()
logger.info( "Telegram bot stopped" )
Usage Examples
Basic Photo Capture
Open Telegram and find your bot: @your_jarvis_bot
Send /start to initialize
Send any photo
Receive processing result:
Processed cap_a1b2c3d4e5f6: 2 face(s) detected, 2 person(s) created.
Document Upload
For higher quality images, send as document:
Tap attachment icon (📎)
Select File instead of Photo
Choose your image file
Bot downloads full resolution image
Photos sent as regular messages are compressed by Telegram. Send as documents for original quality.
Batch Processing
Send multiple photos in quick succession:
User: [Sends photo 1]
Bot: Processed cap_abc123: 1 face(s) detected...
User: [Sends photo 2]
Bot: Processed cap_def456: 3 face(s) detected...
User: [Sends photo 3]
Bot: Processed cap_ghi789: 0 face(s) detected...
The bot replies with processing details:
f "Processed { capture_id } : { faces_detected } face(s) detected, { persons_created } person(s) created."
Success Example
Processed cap_a1b2c3d4e5f6: 2 face(s) detected, 2 person(s) created.
No Faces Example
Processed cap_xyz789abc123: 0 face(s) detected, 0 person(s) created.
Error Example
Processing failed: Face detection model not loaded
Configuration Options
Environment Variables
# .env
TELEGRAM_BOT_TOKEN = 1234567890:ABCdefGHIjklMNOpqrsTUVwxyz
# Optional: Restrict to specific users
TELEGRAM_ALLOWED_USERS = 123456789,987654321
# Optional: Enable debug logging
TELEGRAM_DEBUG = true
Advanced Configuration
Customize bot behavior:
class TelegramCaptureBot :
def __init__ ( self , token : str , pipeline : CapturePipeline,
allowed_users : list[ int ] | None = None ) -> None :
self ._token = token
self ._pipeline = pipeline
self ._allowed_users = allowed_users or []
async def _handle_photo ( self , update : Update, _context : object ) -> None :
# Check user whitelist
if self ._allowed_users:
user_id = update.effective_user.id
if user_id not in self ._allowed_users:
await update.effective_message.reply_text(
"Unauthorized. This bot is private."
)
return
# Continue with photo processing...
Monitoring and Logging
The bot logs all activities using loguru:
logger.info( "Telegram capture bot started polling" )
logger.info( "Telegram photo received, {} bytes" , len (data))
logger.error( "Telegram pipeline error: {} " , exc)
logger.info( "Telegram capture bot stopped" )
Log Output Example
2026-03-05 10:23:45 | INFO | Telegram capture bot started polling
2026-03-05 10:24:12 | INFO | Telegram photo received, 245678 bytes
2026-03-05 10:24:15 | INFO | Processed cap_a1b2c3: 2 faces detected
2026-03-05 10:25:03 | ERROR | Telegram pipeline error: Database connection failed
Testing
Manual Testing
Start the backend:
Check logs for bot initialization:
Telegram capture bot started polling
Send test photo to bot on Telegram
Verify response message
Check backend logs for processing
Unit Testing
import pytest
from telegram_bot import TelegramCaptureBot
from pipeline import CapturePipeline
@pytest.mark.asyncio
async def test_bot_creation ():
pipeline = CapturePipeline()
bot = TelegramCaptureBot( token = "test_token" , pipeline = pipeline)
assert bot is not None
@pytest.mark.asyncio
async def test_bot_graceful_disable ():
from telegram_bot import create_telegram_bot
pipeline = CapturePipeline()
# Should return None when token is missing
bot = create_telegram_bot( token = None , pipeline = pipeline)
assert bot is None
Security Considerations
Important Security Notes :
Never commit bot tokens to version control
Use environment variables for token storage
Implement user whitelisting for production
Rate limit photo uploads to prevent abuse
Validate image file sizes and formats
User Whitelisting
ALLOWED_USERS = [
123456789 , # Your Telegram user ID
987654321 , # Team member user ID
]
if update.effective_user.id not in ALLOWED_USERS :
await update.effective_message.reply_text( "Unauthorized" )
return
Find your user ID:
Message @userinfobot
Copy the Id value
Rate Limiting
from collections import defaultdict
from time import time
class TelegramCaptureBot :
def __init__ ( self , * args , ** kwargs ):
super (). __init__ ( * args, ** kwargs)
self ._rate_limits = defaultdict( list )
self ._max_photos_per_minute = 5
async def _check_rate_limit ( self , user_id : int ) -> bool :
now = time()
cutoff = now - 60 # 1 minute window
# Remove old timestamps
self ._rate_limits[user_id] = [
ts for ts in self ._rate_limits[user_id] if ts > cutoff
]
if len ( self ._rate_limits[user_id]) >= self ._max_photos_per_minute:
return False
self ._rate_limits[user_id].append(now)
return True
Troubleshooting
Symptoms : Messages sent but no replySolutions :
Verify bot token is correct in .env
Check backend logs for startup errors
Ensure python-telegram-bot is installed
Test token with curl:
curl https://api.telegram.org/bot < TOKE N > /getMe
Restart the backend service
Symptoms : Bot replies but no faces detectedSolutions :
Check if pipeline is initialized correctly
Verify face detection model is loaded
Test with different photos
Check backend logs: docker logs jarvis-backend
Ensure image format is supported (JPEG, PNG)
Try sending as document for full resolution
ImportError: No module named 'telegram'
Symptoms : Bot fails to start with import errorSolutions :
Install the library:
pip install python-telegram-bot
Verify installation:
pip show python-telegram-bot
Check Python version compatibility (3.8+)
Symptoms : 401 Unauthorized errorSolutions :
Verify token is copied correctly (no spaces)
Check if token was revoked in BotFather
Generate new token:
/mybots → Select bot → API Token → Revoke → Generate new
Update .env with new token
Next Steps
Meta Glasses Upgrade to hands-free capture with smart glasses
Camera Setup Configure iPhone camera fallback
Reference