Strands Telegram Listener

Real-time Telegram message processing with AI-powered responses for Strands Agents

A powerful background Telegram message processor that provides seamless real-time messaging capabilities with intelligent AI responses, event storage, and comprehensive message handling for Strands Agents.

🚀 Features

🎧 Real-Time Message Processing

• Long Polling: Efficient real-time message retrieval from Telegram
• Background Processing: Non-blocking message handling in separate threads
• Auto-Reply Mode: AI-powered automatic responses to incoming messages
• Message Filtering: Smart filtering to avoid processing own messages
• Event Storage: Persistent message history in JSONL format

🤖 AI-Powered Responses

• Strands Agent Integration: Seamless AI response generation
• Context Awareness: Maintains conversation context across messages
• Intelligent Processing: Advanced message understanding and response
• Multi-Modal Support: Text, media, and interactive message handling

📊 Event Management

• Persistent Storage: JSONL-based event logging for history
• Real-Time Monitoring: Live status updates and metrics
• Event Retrieval: Access to recent message history
• Thread-Safe Operations: Concurrent message processing

🔧 Advanced Configuration

• Environment-Based Setup: Easy configuration via environment variables
• Trigger Keywords: Optional keyword-based activation
• Auto-Reply Control: Toggle automatic responses on/off
• Listen-Only Mode: Process messages without responding
• Comprehensive Logging: Detailed operation logs

📦 Installation

pip install strands-telegram-listener

🤖 Bot Setup

Before using this tool, you need to create a Telegram bot and get an API token:

1. Create a Telegram Bot

1. Open Telegram and search for @BotFather
2. Start a chat with BotFather by clicking /start
3. Create new bot with /newbot command
4. Choose a name for your bot (e.g., "My Awesome Bot")
5. Choose a username for your bot (must end with "bot", e.g., "myawesomebot")
6. Save the token - BotFather will give you a token like 123456:ABC-DEF1234ghIkl-zyx57W2v1u123ew11

2. Configure Bot Settings (Optional)

You can customize your bot using BotFather commands:

• /setdescription - Set bot description
• /setcommands - Set bot commands menu
• /setprivacy - Enable/disable privacy mode for groups
• /setjoingroups - Allow bot to be added to groups

3. Get Your Bot Token

The token from BotFather is your TELEGRAM_BOT_TOKEN. Keep it secure!

4. Set Environment Variable

export TELEGRAM_BOT_TOKEN="your_bot_token_here"

📚 Official Documentation

• Telegram Bots Guide: https://core.telegram.org/bots
• Telegram Bot API: https://core.telegram.org/bots/api
• Full Telegram API: https://core.telegram.org/api

🛠️ Usage

Quick Start

from strands import Agent
from strands_telegram_listener import telegram_listener

# Create agent with Telegram listener
agent = Agent(tools=[telegram_listener])

# Start listening for messages
agent.tool.telegram_listener(action="start")

# Check listener status
agent.tool.telegram_listener(action="status")

# Get recent events
agent.tool.telegram_listener(action="get_recent_events", count=10)

# Stop listening
agent.tool.telegram_listener(action="stop")

Environment Setup

# Required: Set your Telegram Bot Token (get from @BotFather)
export TELEGRAM_BOT_TOKEN="your_bot_token_here"

# Optional: Enable automatic AI responses
export STRANDS_TELEGRAM_AUTO_REPLY="true"

# Optional: Only process messages containing this tag
export STRANDS_TELEGRAM_LISTEN_ONLY_TAG="@mybot"

# Optional: Number of events to retrieve by default
export TELEGRAM_DEFAULT_EVENT_COUNT="200"

Advanced Examples

Auto-Reply Bot

import os
from strands import Agent
from strands_telegram_listener import telegram_listener

# Enable auto-reply mode
os.environ["STRANDS_TELEGRAM_AUTO_REPLY"] = "true"

# Create intelligent bot agent
agent = Agent(
    system_prompt="""
    You are a helpful Telegram bot assistant. 
    Respond concisely and helpfully to user messages.
    Use appropriate emojis and Telegram formatting.
    """,
    tools=[telegram_listener]
)

# Start listening - bot will automatically respond to messages
agent.tool.telegram_listener(action="start")
print("🤖 AI Bot is now listening and responding!")

Monitoring Bot

from strands import Agent
from strands_telegram_listener import telegram_listener

# Create monitoring agent (no auto-reply)
agent = Agent(tools=[telegram_listener])

# Start listening in monitor-only mode
agent.tool.telegram_listener(action="start")

# Check status periodically
import time
while True:
    status = agent.tool.telegram_listener(action="status")
    print(f"Bot Status: {status}")
    
    # Get recent messages
    events = agent.tool.telegram_listener(action="get_recent_events", count=5)
    print(f"Recent Events: {events}")
    
    time.sleep(60)  # Check every minute

Selective Listening

import os
from strands import Agent
from strands_telegram_listener import telegram_listener

# Only process messages containing specific tag
os.environ["STRANDS_TELEGRAM_LISTEN_ONLY_TAG"] = "#support"

agent = Agent(
    system_prompt="You are a customer support bot. Handle support requests professionally.",
    tools=[telegram_listener]
)

# Bot will only respond to messages containing "#support"
agent.tool.telegram_listener(action="start")

📋 Available Actions

Core Operations

• start - Begin listening for Telegram messages in background
• stop - Stop the message listener
• status - Get current listener status and configuration
• get_recent_events - Retrieve recent events from storage

Status Information

The status action returns comprehensive information:

{
  "running": true,
  "bot_info": {
    "id": 123456789,
    "is_bot": true,
    "first_name": "MyBot",
    "username": "my_bot"
  },
  "last_update_id": 123456,
  "events_file": "/path/to/telegram_events/events.jsonl",
  "auto_reply": "true",
  "listen_only_tag": "None"
}

🔒 Security & Best Practices

Token Security

• Never hardcode tokens in your code
• Use environment variables for bot tokens
• Rotate tokens regularly for production use
• Monitor bot permissions and access logs

Message Processing

• Own Message Filtering: Automatically ignores bot's own messages
• Duplicate Prevention: Handles message de-duplication
• Error Recovery: Robust error handling with automatic retries
• Rate Limiting: Respects Telegram's rate limits

Production Deployment

# Production configuration example
import os
import logging
from strands import Agent
from strands_telegram_listener import telegram_listener

# Configure logging
logging.basicConfig(level=logging.INFO)

# Production environment variables
os.environ["TELEGRAM_BOT_TOKEN"] = "your_production_token"
os.environ["STRANDS_TELEGRAM_AUTO_REPLY"] = "true"
os.environ["TELEGRAM_DEFAULT_EVENT_COUNT"] = "50"

# Create production agent
agent = Agent(
    system_prompt="""
    Production Telegram Bot Assistant.
    - Respond professionally and helpfully
    - Handle errors gracefully
    - Log important interactions
    - Maintain conversation context
    """,
    tools=[telegram_listener]
)

# Start with error handling
try:
    result = agent.tool.telegram_listener(action="start")
    print(f"✅ Production bot started: {result}")
except Exception as e:
    print(f"❌ Failed to start bot: {e}")

🎯 Integration Patterns

With strands-telegram Package

from strands import Agent
from strands_telegram import telegram
from strands_telegram_listener import telegram_listener

# Agent with both tools
agent = Agent(
    system_prompt="You are a comprehensive Telegram bot with full API access.",
    tools=[telegram, telegram_listener]
)

# Start listening
agent.tool.telegram_listener(action="start")

# Bot can now both listen and actively send messages
# The listener will automatically use the telegram tool for responses

Custom Message Processing

from strands import Agent
from strands_telegram_listener import telegram_listener

class CustomTelegramAgent(Agent):
    def process_telegram_message(self, message_data):
        """Custom message processing logic."""
        # Your custom processing here
        user = message_data.get("from", {})
        text = message_data.get("text", "")
        
        # Process with AI
        response = self(f"User {user.get('first_name', 'User')} says: {text}")
        
        # Return response for auto-reply
        return str(response)

agent = CustomTelegramAgent(tools=[telegram_listener])
agent.tool.telegram_listener(action="start")

📊 Event Storage Format

Events are stored in JSONL format at ./telegram_events/events.jsonl:

{
  "event_type": "telegram_update",
  "payload": {
    "update_id": 123456,
    "message": {
      "message_id": 789,
      "from": {
        "id": 987654321,
        "is_bot": false,
        "first_name": "John",
        "username": "johndoe"
      },
      "chat": {
        "id": 987654321,
        "first_name": "John",
        "username": "johndoe",
        "type": "private"
      },
      "date": 1697723456,
      "text": "Hello bot!"
    }
  },
  "timestamp": 1697723456.789,
  "update_id": 123456
}

🔧 Configuration Reference

Environment Variables

Bot Configuration

Get Bot Token

1. Message @BotFather on Telegram
2. Use /newbot command to create a new bot
3. Follow instructions to set bot name and username
4. Copy the provided bot token
5. Set as TELEGRAM_BOT_TOKEN environment variable

Bot Permissions

Ensure your bot has appropriate permissions:

• Send Messages: Required for auto-reply
• Read Messages: Required for message processing
• Manage Groups: If using in groups (optional)

🤝 Contributing

Contributions are welcome! Please feel free to submit issues, feature requests, or pull requests.

Development Setup

git clone https://github.com/eraykeskinmac/strands-telegram-listener
cd strands-telegram-listener
pip install -e ".[dev]"

Running Tests

# Run all tests
pytest

# Run with coverage
pytest --cov=strands_telegram_listener

# Run specific test
pytest tests/test_telegram_listener.py::TestTelegramListener::test_start_listener

📚 API Reference

telegram_listener(action, count=20, agent=None)

Parameters:

• action (str): The action to perform
  • "start": Begin listening for messages
  • "stop": Stop the listener
  • "status": Get listener status
  • "get_recent_events": Retrieve recent events
• count (int): Number of events to retrieve (for get_recent_events)
• agent (Agent): Strands agent instance (auto-provided)

Returns:

• str: Status message or event data based on the action

🔗 Related Packages

• strands-telegram: Complete Telegram Bot API tool
• strands: Core Strands Agents framework

📄 License

This project is licensed under the MIT License - see the LICENSE file for details.

🔗 Links

• Strands Agents SDK
• Telegram Bot API
• PyPI Package
• GitHub Repository

📞 Support

• Create an issue on GitHub for bug reports
• Join our community discussions for questions
• Check the Strands Agents SDK documentation for integration help

Made with ❤️ for the Strands Agents community