MemosMemos

Telegram Bot Integration

Set up and configure a Telegram bot for creating memos directly from Telegram messages.

Telegram Bot Integration

Integrate Memos with Telegram to create memos directly from your mobile device or desktop Telegram client. This powerful integration allows you to quickly capture thoughts, notes, and ideas without opening the Memos web interface.

Overview

The Telegram bot integration enables you to:

  • Create memos by sending messages to your bot
  • Add media attachments (photos, documents, voice notes)
  • Use commands for advanced memo management
  • Set up scheduled reminders for important notes
  • Search existing memos directly from Telegram

Quick Setup: The bot integration uses Memos API tokens for authentication, making it secure and easy to set up without storing passwords.

Prerequisites

Before setting up the Telegram bot, ensure you have:

  • Memos instance running with API access
  • Telegram account to create and interact with the bot
  • API access token from your Memos instance
  • Basic understanding of Telegram bot interactions

Bot Creation

Step 1: Create Telegram Bot

  1. Open Telegram and search for @BotFather
  2. Start a chat with BotFather by clicking /start
  3. Create new bot by sending /newbot
  4. Choose bot name: Enter a display name (e.g., "My Memos Bot")
  5. Choose username: Enter a unique username ending in "bot" (e.g., "MyMemosBot")
  6. Save the token: BotFather will provide a token like 123456789:ABCdefGHIjklMNOpqrsTUVwxyZ

Security Note: Keep your bot token secure! Anyone with access to the token can control your bot.

Step 2: Configure Bot Settings

Customize your bot's behavior through BotFather:

/setdescription - Set a description for your bot
/setabouttext - Set an about text for your bot
/setuserpic - Set a profile picture for your bot
/setcommands - Set up command menu

Send /setcommands to BotFather and configure these commands:

start - Initialize bot and get setup instructions
help - Show available commands and usage
memo - Create a new memo
search - Search existing memos
recent - Show recent memos
settings - Configure bot preferences

Memos API Setup

Step 1: Create API Token

  1. Login to Memos web interface
  2. Go to SettingsAccess Tokens
  3. Create new token:
    • Name: "Telegram Bot"
    • Description: "Token for Telegram bot integration"
    • Expiration: Set appropriate expiry or leave blank for no expiry
  4. Copy the token - you'll need it for bot configuration

Step 2: Test API Access

Verify your API token works correctly:

# Test API access
curl -H "Authorization: Bearer YOUR_API_TOKEN" \
     "https://your-memos-instance.com/api/v1/auth/status"

# Create test memo
curl -X POST \
     -H "Authorization: Bearer YOUR_API_TOKEN" \
     -H "Content-Type: application/json" \
     -d '{"content": "Test memo from API"}' \
     "https://your-memos-instance.com/api/v1/memos"

Bot Implementation

Option 1: Using Existing Bot Scripts

Several community-developed Telegram bots are available for Memos integration.

Python Bot Example

import os
import requests
from telegram import Update
from telegram.ext import Application, CommandHandler, MessageHandler, filters, ContextTypes

# Configuration
TELEGRAM_BOT_TOKEN = os.getenv('TELEGRAM_BOT_TOKEN')
MEMOS_API_URL = os.getenv('MEMOS_API_URL', 'https://your-memos-instance.com')
MEMOS_API_TOKEN = os.getenv('MEMOS_API_TOKEN')

class MemosBot:
    def __init__(self):
        self.headers = {
            'Authorization': f'Bearer {MEMOS_API_TOKEN}',
            'Content-Type': 'application/json'
        }
    
    async def start(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
        """Send welcome message with setup instructions."""
        welcome_text = """
🎉 Welcome to Memos Bot!

I can help you create and manage memos directly from Telegram.

*Available Commands:*
• Send any message to create a memo
• /help - Show this help message
• /recent - Show your recent memos
• /search <query> - Search your memos

*Getting Started:*
Just send me a message and I'll create a memo for you!
        """
        await update.message.reply_text(welcome_text, parse_mode='Markdown')
    
    async def help_command(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
        """Show help information."""
        help_text = """
🤖 *Memos Bot Commands*

*Creating Memos:*
• Send any text message to create a memo
• Forward messages to save them as memos
• Send photos with captions to create visual memos
• Send documents to attach them to memos

*Managing Memos:*
• `/recent` - Show your 5 most recent memos
• `/search <query>` - Search memos by content
• `/settings` - Configure bot preferences

*Examples:*
📝 `Meeting notes: Discussed project timeline`
🏷️ `#important Remember to backup database`
📸 Send a photo with caption to create visual memo

Need help? Visit: https://your-memos-instance.com/docs
        """
        await update.message.reply_text(help_text, parse_mode='Markdown')
    
    async def create_memo(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
        """Create a memo from message content."""
        try:
            # Get message content
            content = update.message.text
            if not content:
                await update.message.reply_text("❌ Empty messages cannot be saved as memos.")
                return
            
            # Add metadata
            user_info = update.effective_user
            metadata = f"\n\n*Source:* Telegram Bot\n*User:* @{user_info.username or user_info.first_name}"
            full_content = content + metadata
            
            # Create memo via API
            response = requests.post(
                f'{MEMOS_API_URL}/api/v1/memos',
                headers=self.headers,
                json={'content': full_content}
            )
            
            if response.status_code == 200:
                memo = response.json()
                await update.message.reply_text(
                    f"✅ Memo created successfully!\n🆔 ID: {memo.get('id')}"
                )
            else:
                await update.message.reply_text(
                    f"❌ Failed to create memo. Status: {response.status_code}"
                )
                
        except Exception as e:
            await update.message.reply_text(f"❌ Error creating memo: {str(e)}")
    
    async def search_memos(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
        """Search existing memos."""
        try:
            query = ' '.join(context.args)
            if not query:
                await update.message.reply_text("❌ Please provide search query: `/search your query`", parse_mode='Markdown')
                return
            
            # Search via API
            response = requests.get(
                f'{MEMOS_API_URL}/api/v1/search',
                headers=self.headers,
                params={'q': query, 'limit': 5}
            )
            
            if response.status_code == 200:
                results = response.json()
                if results['data']:
                    message = f"🔍 *Search results for:* `{query}`\n\n"
                    for memo in results['data'][:5]:
                        preview = memo['content'][:100] + "..." if len(memo['content']) > 100 else memo['content']
                        message += f"📝 *ID {memo['id']}*\n{preview}\n\n"
                    await update.message.reply_text(message, parse_mode='Markdown')
                else:
                    await update.message.reply_text(f"❌ No memos found for query: `{query}`", parse_mode='Markdown')
            else:
                await update.message.reply_text("❌ Search failed. Please try again later.")
                
        except Exception as e:
            await update.message.reply_text(f"❌ Search error: {str(e)}")
    
    async def recent_memos(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
        """Show recent memos."""
        try:
            response = requests.get(
                f'{MEMOS_API_URL}/api/v1/memos',
                headers=self.headers,
                params={'limit': 5}
            )
            
            if response.status_code == 200:
                memos = response.json()['data']
                if memos:
                    message = "📋 *Your Recent Memos:*\n\n"
                    for memo in memos:
                        preview = memo['content'][:100] + "..." if len(memo['content']) > 100 else memo['content']
                        date = memo['createTime'][:10]  # Extract date
                        message += f"📝 *{date}* (ID: {memo['id']})\n{preview}\n\n"
                    await update.message.reply_text(message, parse_mode='Markdown')
                else:
                    await update.message.reply_text("📝 No memos found. Create your first memo by sending me a message!")
            else:
                await update.message.reply_text("❌ Failed to retrieve memos.")
                
        except Exception as e:
            await update.message.reply_text(f"❌ Error retrieving memos: {str(e)}")

def main():
    """Start the bot."""
    bot = MemosBot()
    application = Application.builder().token(TELEGRAM_BOT_TOKEN).build()
    
    # Command handlers
    application.add_handler(CommandHandler("start", bot.start))
    application.add_handler(CommandHandler("help", bot.help_command))
    application.add_handler(CommandHandler("search", bot.search_memos))
    application.add_handler(CommandHandler("recent", bot.recent_memos))
    
    # Message handler (for creating memos)
    application.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, bot.create_memo))
    
    # Start bot
    print("🤖 Memos Telegram Bot is starting...")
    application.run_polling()

if __name__ == '__main__':
    main()

Installation and Setup

# Create bot directory
mkdir memos-telegram-bot
cd memos-telegram-bot

# Create virtual environment
python -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate

# Install dependencies
pip install python-telegram-bot requests

# Create environment file
cat > .env << EOF
TELEGRAM_BOT_TOKEN=123456789:ABCdefGHIjklMNOpqrsTUVwxyZ
MEMOS_API_URL=https://your-memos-instance.com
MEMOS_API_TOKEN=your-memos-api-token
EOF

# Save the bot script as bot.py and run
python bot.py

Option 2: Docker Deployment

Create a containerized bot deployment:

# Dockerfile
FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install -r requirements.txt

COPY bot.py .

CMD ["python", "bot.py"]
# docker-compose.yml
version: '3.8'

services:
  memos-telegram-bot:
    build: .
    environment:
      - TELEGRAM_BOT_TOKEN=${TELEGRAM_BOT_TOKEN}
      - MEMOS_API_URL=${MEMOS_API_URL}
      - MEMOS_API_TOKEN=${MEMOS_API_TOKEN}
    restart: unless-stopped
    depends_on:
      - memos

  memos:
    image: neosmemo/memos:stable
    # ... your memos configuration
# requirements.txt
python-telegram-bot==20.6
requests==2.31.0
python-dotenv==1.0.0

Advanced Features

Media Support

Handle different types of media from Telegram:

async def handle_photo(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
    """Handle photo messages."""
    try:
        # Get largest photo
        photo = update.message.photo[-1]
        file = await context.bot.get_file(photo.file_id)
        
        # Download photo
        photo_data = await file.download_as_bytearray()
        
        # Upload to Memos
        files = {'file': ('photo.jpg', photo_data, 'image/jpeg')}
        response = requests.post(
            f'{MEMOS_API_URL}/api/v1/resources',
            headers={'Authorization': f'Bearer {MEMOS_API_TOKEN}'},
            files=files
        )
        
        if response.status_code == 200:
            resource = response.json()
            # Create memo with image
            caption = update.message.caption or "Photo from Telegram"
            content = f"{caption}\n\n![Photo](/api/v1/resources/{resource['id']})"
            
            memo_response = requests.post(
                f'{MEMOS_API_URL}/api/v1/memos',
                headers=self.headers,
                json={'content': content}
            )
            
            if memo_response.status_code == 200:
                await update.message.reply_text("✅ Photo memo created successfully!")
            else:
                await update.message.reply_text("❌ Failed to create photo memo.")
        else:
            await update.message.reply_text("❌ Failed to upload photo.")
            
    except Exception as e:
        await update.message.reply_text(f"❌ Error processing photo: {str(e)}")

# Add to application
application.add_handler(MessageHandler(filters.PHOTO, bot.handle_photo))

Voice Message Support

async def handle_voice(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
    """Handle voice messages."""
    try:
        voice = update.message.voice
        file = await context.bot.get_file(voice.file_id)
        voice_data = await file.download_as_bytearray()
        
        # Upload voice note
        files = {'file': ('voice.ogg', voice_data, 'audio/ogg')}
        response = requests.post(
            f'{MEMOS_API_URL}/api/v1/resources',
            headers={'Authorization': f'Bearer {MEMOS_API_TOKEN}'},
            files=files
        )
        
        if response.status_code == 200:
            resource = response.json()
            content = f"🎤 Voice message from Telegram\n\n[Voice Note](/api/v1/resources/{resource['id']})"
            
            memo_response = requests.post(
                f'{MEMOS_API_URL}/api/v1/memos',
                headers=self.headers,
                json={'content': content}
            )
            
            if memo_response.status_code == 200:
                await update.message.reply_text("✅ Voice memo created successfully!")
        
    except Exception as e:
        await update.message.reply_text(f"❌ Error processing voice: {str(e)}")

Scheduled Reminders

async def set_reminder(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
    """Set a reminder for a memo."""
    try:
        if len(context.args) < 2:
            await update.message.reply_text("Usage: `/remind <time> <message>`", parse_mode='Markdown')
            return
        
        time_str = context.args[0]
        message = ' '.join(context.args[1:])
        
        # Parse time (simplified - you can enhance this)
        if time_str.endswith('m'):
            minutes = int(time_str[:-1])
            reminder_time = datetime.now() + timedelta(minutes=minutes)
        elif time_str.endswith('h'):
            hours = int(time_str[:-1])
            reminder_time = datetime.now() + timedelta(hours=hours)
        else:
            await update.message.reply_text("❌ Time format: 30m (minutes) or 2h (hours)")
            return
        
        # Schedule reminder (you can use APScheduler or similar)
        context.job_queue.run_once(
            lambda ctx: ctx.bot.send_message(
                chat_id=update.effective_chat.id,
                text=f"⏰ Reminder: {message}"
            ),
            reminder_time
        )
        
        await update.message.reply_text(f"⏰ Reminder set for {reminder_time.strftime('%H:%M')}")
        
    except Exception as e:
        await update.message.reply_text(f"❌ Error setting reminder: {str(e)}")

Security Considerations

Bot Security

Security Best Practices: Implement proper authentication and access controls for production deployments.

# User authentication
AUTHORIZED_USERS = os.getenv('AUTHORIZED_USERS', '').split(',')

def authorized_only(func):
    """Decorator to restrict bot access to authorized users."""
    async def wrapper(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
        user_id = str(update.effective_user.id)
        username = update.effective_user.username
        
        if AUTHORIZED_USERS and user_id not in AUTHORIZED_USERS and username not in AUTHORIZED_USERS:
            await update.message.reply_text("❌ Unauthorized access. Contact administrator.")
            return
        
        return await func(self, update, context)
    return wrapper

# Apply to sensitive commands
@authorized_only
async def admin_command(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
    # Admin functionality here
    pass

Token Management

# Environment-based configuration
export TELEGRAM_BOT_TOKEN="your-bot-token"
export MEMOS_API_TOKEN="your-api-token"
export AUTHORIZED_USERS="user1,user2,123456789"

# Secure token storage
echo "TELEGRAM_BOT_TOKEN=your-token" > .env
chmod 600 .env

Monitoring and Logging

Bot Monitoring

import logging

# Configure logging
logging.basicConfig(
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    level=logging.INFO,
    handlers=[
        logging.FileHandler('/var/log/memos-bot.log'),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger(__name__)

# Add logging to functions
async def create_memo(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
    user_info = update.effective_user
    logger.info(f"User {user_info.username} ({user_info.id}) creating memo")
    
    try:
        # ... memo creation logic
        logger.info(f"Memo created successfully for user {user_info.id}")
    except Exception as e:
        logger.error(f"Failed to create memo for user {user_info.id}: {str(e)}")

Health Checks

async def health_check(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
    """Check bot and API health."""
    try:
        # Test Memos API
        response = requests.get(
            f'{MEMOS_API_URL}/api/v1/auth/status',
            headers=self.headers,
            timeout=5
        )
        
        if response.status_code == 200:
            await update.message.reply_text("✅ Bot and API are healthy!")
        else:
            await update.message.reply_text(f"⚠️ API health check failed: {response.status_code}")
            
    except Exception as e:
        await update.message.reply_text(f"❌ Health check failed: {str(e)}")

Troubleshooting

Common Issues

Bot Not Responding

# Check bot logs
tail -f /var/log/memos-bot.log

# Test bot token
curl -X GET "https://api.telegram.org/bot<YOUR_BOT_TOKEN>/getMe"

# Verify webhook (if using webhooks)
curl -X GET "https://api.telegram.org/bot<YOUR_BOT_TOKEN>/getWebhookInfo"

API Connection Issues

# Test API connectivity
def test_api_connection():
    try:
        response = requests.get(
            f'{MEMOS_API_URL}/api/v1/auth/status',
            headers={'Authorization': f'Bearer {MEMOS_API_TOKEN}'},
            timeout=10
        )
        print(f"API Status: {response.status_code}")
        print(f"Response: {response.text}")
    except Exception as e:
        print(f"API Connection Error: {e}")

# Run test
test_api_connection()

Debug Mode

# Enable debug logging
logging.getLogger("httpx").setLevel(logging.WARNING)
logging.getLogger("telegram").setLevel(logging.DEBUG)

# Add debug information
async def debug_info(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
    """Show debug information."""
    user = update.effective_user
    chat = update.effective_chat
    
    debug_text = f"""
🐛 *Debug Information*

*User:* {user.first_name} {user.last_name or ''}
*Username:* @{user.username or 'None'}
*User ID:* `{user.id}`
*Chat ID:* `{chat.id}`
*Chat Type:* {chat.type}

*Bot Status:* Running
*API URL:* `{MEMOS_API_URL}`
*Token Valid:* {bool(MEMOS_API_TOKEN)}
    """
    await update.message.reply_text(debug_text, parse_mode='Markdown')

Best Practices

Performance Optimization

  • Use async/await for API calls to avoid blocking
  • Implement rate limiting to prevent API abuse
  • Cache frequent API responses to reduce load
  • Use connection pooling for HTTP requests
  • Implement proper error handling and retries

User Experience

  • Provide clear commands and help documentation
  • Use emoji and formatting for better readability
  • Implement command shortcuts for frequent actions
  • Add confirmation messages for important operations
  • Support message editing and deletion

Maintenance

  • Monitor bot uptime and performance
  • Regularly update dependencies and tokens
  • Backup bot configuration and logs
  • Test bot functionality after Memos updates
  • Document custom modifications and configurations

Next Steps


Need help with Telegram bot integration? Check the troubleshooting guide or ask in GitHub Discussions.