amplifier-module-tool-telegram-input

Telegram Input Tool Module for Amplifier - polls Telegram for messages from authorized users and returns as tool input.

Features

• Message Polling: Long-polling to receive messages from Telegram
• Authorization: Whitelist-based user authorization via pairing.json
• Pairing Flow: Generate pairing codes for easy user authorization
• Rate Limiting: Blocks unauthorized users after repeated attempts
• Security: Only processes messages from authorized users

Installation

pip install git+https://github.com/ramparte/amplifier-module-tool-telegram-input@main

Setup

1. Create Telegram Bot

1. Message @BotFather on Telegram
2. Send /newbot and follow instructions
3. Save the bot token (format: 123456:ABC-DEF1234...)
4. Send /setprivacy → Disable (allows bot to see all messages)

2. Configure Module

Add to your Amplifier mount plan:

tools:
  - module: tool-telegram-input
    source: git+https://github.com/ramparte/amplifier-module-tool-telegram-input@main
    config:
      bot_token: "YOUR_BOT_TOKEN"
      pairing_file: ".amplifier/telegram_pairing.json"
      poll_interval: 30
      pairing_timeout: 5
      rate_limit:
        max_attempts: 5
        block_hours: 1

3. Authorize Users via Pairing Flow

Use the tool to generate pairing codes:

# Start pairing
result = await tool.execute({"action": "start_pairing"})
# Returns: {"pairing_code": "BEAR-7392", "expires_at": "...", "instructions": "..."}

# User sends code to bot on Telegram
# Complete pairing
result = await tool.execute({"action": "complete_pairing", "pairing_code": "BEAR-7392"})
# Returns: {"status": "paired", "user_id": 123456, "username": "alice"}

Or manually create .amplifier/telegram_pairing.json:

{
  "version": "1.0",
  "authorized_users": [
    {
      "user_id": 123456789,
      "chat_id": 123456789,
      "username": "alice",
      "paired_at": "2025-10-29T10:30:00Z"
    }
  ],
  "rate_limits": {}
}

Configuration Reference

Parameter	Type	Default	Description
bot_token	string	required	Telegram bot token from @BotFather
pairing_file	string	.amplifier/telegram_pairing.json	Path to authorization file
poll_interval	int	30	Long polling timeout (seconds)
pairing_timeout	int	5	Pairing code expiry (minutes)
rate_limit.max_attempts	int	5	Max unauthorized attempts before block
rate_limit.block_hours	int	1	Hours to block after max attempts

Tool Interface

Input Schema

{
  "type": "object",
  "properties": {
    "action": {
      "type": "string",
      "enum": ["poll", "start_pairing", "complete_pairing", "list_users"]
    },
    "pairing_code": {
      "type": "string",
      "description": "Required for complete_pairing"
    }
  },
  "required": ["action"]
}

Actions

1. poll

Poll for new messages from authorized users.

Input:

{"action": "poll"}

Output:

{
  "messages": [
    {
      "user_id": 123456,
      "username": "alice",
      "chat_id": 123456,
      "text": "Hello from Telegram",
      "timestamp": 1730203800
    }
  ],
  "count": 1,
  "unauthorized_attempts": 0
}

2. start_pairing

Generate pairing code for user authorization.

Input:

{"action": "start_pairing"}

Output:

{
  "pairing_code": "BEAR-7392",
  "expires_at": "2025-10-29T10:35:00Z",
  "instructions": "Send pairing code 'BEAR-7392' to your Telegram bot to authorize"
}

Pairing code format: WORD-NNNN (e.g., BEAR-7392, LION-4821)

• Word: Random from [BEAR, LION, WOLF, EAGLE, TIGER, SHARK, HAWK, PANDA]
• Digits: 4 random digits
• Expires: 5 minutes (configurable)

3. complete_pairing

Complete pairing when user sends code.

Input:

{
  "action": "complete_pairing",
  "pairing_code": "BEAR-7392"
}

Output (success):

{
  "status": "paired",
  "user_id": 123456,
  "username": "alice",
  "message": "Successfully paired with alice"
}

Output (waiting):

{
  "status": "waiting",
  "message": "No matching pairing code received yet"
}

4. list_users

List authorized users.

Input:

{"action": "list_users"}

Output:

{
  "authorized_users": [123456, 789012],
  "count": 2
}

Architecture

Components

1. TelegramInputTool (tool.py)

   • Implements Amplifier Tool interface
   • Manages actions (poll, pairing, list)
   • Filters by authorization
   • Emits security events

2. TelegramPoller (telegram_poller.py)

   • Long polling via getUpdates (timeout=30s)
   • Tracks last_update_id to avoid duplicates
   • Extracts text messages only

3. PairingManager (pairing_manager.py)

   • Generates pairing codes (WORD-NNNN format)
   • Manages expiration (5 min TTL)
   • Completes pairing on code match
   • Cleans up expired requests

4. AuthManager (auth_manager.py)

   • Reads/writes pairing.json
   • Checks authorization
   • Tracks rate limits (5 attempts → 1 hour block)
   • Shared with hook module

Flow: Pairing

1. Start Pairing
   Tool generates code: "BEAR-7392"
   ↓
2. User sends code to bot
   User: "BEAR-7392"
   ↓
3. Complete Pairing
   Tool polls, sees matching code
   Adds user to pairing.json
   ↓
4. Authorized
   Future messages from user accepted

Flow: Polling

Poll Action
    ↓
TelegramPoller.poll() (long polling)
    ↓
Filter messages by authorization
    ↓
Authorized → Return messages
Unauthorized → Record attempt, emit event, block if threshold

Security

Authorization

• Whitelist-only: Only authorized users in pairing.json
• Rate limiting: Blocks users after 5 unauthorized attempts for 1 hour
• Pairing expiry: Codes expire after 5 minutes
• No ambient authority: Bot token required explicitly

Rate Limiting

When unauthorized user sends message:

1. Record failed attempt
2. If attempts ≥ 5: Block for 1 hour
3. Emit bridge:auth_attempt event
4. Emit bridge:rate_limit event on block

Rate limit stored in pairing.json:

{
  "rate_limits": {
    "999999": {
      "failed_attempts": 5,
      "blocked_until": "2025-10-29T11:30:00Z"
    }
  }
}

Events Emitted

Tool emits these events for observability:

• bridge:message_received - Authorized message received

  • Data: {user_id, username, message}

• bridge:auth_attempt - Authorization attempted

  • Data: {user_id, username, authorized: bool}

• bridge:rate_limit - User rate limited

  • Data: {user_id, failed_attempts, blocked_until}

• bridge:pairing_complete - Pairing completed

  • Data: {user_id, username, chat_id}

Usage Examples

Example 1: Poll for Messages

from amplifier_core import AmplifierSession

async with AmplifierSession(config=mount_plan) as session:
    # Poll for messages
    result = await session.call_tool("telegram_input", {"action": "poll"})

    if result.success:
        for msg in result.output["messages"]:
            print(f"{msg['username']}: {msg['text']}")

Example 2: Pairing Flow

# Start pairing
pairing_result = await session.call_tool("telegram_input", {"action": "start_pairing"})
pairing_code = pairing_result.output["pairing_code"]

print(f"Send '{pairing_code}' to your Telegram bot")

# Wait for user to send code, then complete
complete_result = await session.call_tool("telegram_input", {
    "action": "complete_pairing",
    "pairing_code": pairing_code
})

if complete_result.output["status"] == "paired":
    print(f"Paired with {complete_result.output['username']}")

Example 3: Combined with Hook Module

Mount both modules for bidirectional communication:

# Hook: Amplifier → Telegram (notifications)
hooks:
  - module: hooks-telegram-bridge
    source: git+https://github.com/ramparte/amplifier-module-hooks-telegram-bridge@main
    config:
      bot_token: "YOUR_BOT_TOKEN"
      pairing_file: ".amplifier/telegram_pairing.json"

# Tool: Telegram → Amplifier (user input)
tools:
  - module: tool-telegram-input
    source: git+https://github.com/ramparte/amplifier-module-tool-telegram-input@main
    config:
      bot_token: "YOUR_BOT_TOKEN"
      pairing_file: ".amplifier/telegram_pairing.json"

Now you can:

• Send session events to Telegram (via hook)
• Receive user messages from Telegram (via tool)

Testing

# Install dev dependencies
uv sync --dev

# Run tests
uv run pytest tests/ -v

# Run specific test
uv run pytest tests/test_tool.py::test_tool_poll_authorized_messages -v

Development

Structure

amplifier-module-tool-telegram-input/
├── amplifier_module_tool_telegram_input/
│   ├── __init__.py          # Module mount point
│   ├── tool.py              # TelegramInputTool
│   ├── telegram_poller.py   # Long polling
│   ├── pairing_manager.py   # Pairing flow
│   └── auth_manager.py      # Authorization (shared)
├── tests/
│   ├── test_tool.py
│   └── ...
├── pyproject.toml
└── README.md

Philosophy

Follows Amplifier design principles:

• Ruthless simplicity: Direct API calls, minimal abstractions
• Security by default: Whitelist-only, rate limiting
• Mechanism not policy: Tool provides capability, orchestrator decides usage
• Event-first observability: All security events emitted

Troubleshooting

No messages received

1. Check bot token: curl https://api.telegram.org/bot<TOKEN>/getMe
2. Verify user is in pairing.json with correct user_id/chat_id
3. Check user not rate limited (blocked_until in pairing.json)
4. Send test message to bot and check logs

Pairing code not working

1. Check code hasn't expired (5 min default)
2. Verify exact code sent (case-insensitive match)
3. Check logs for "Pairing complete" message
4. Try generating new code

Rate limiting too aggressive

Adjust in config:

rate_limit:
  max_attempts: 10  # Increase threshold
  block_hours: 2    # Increase block duration

License

MIT

Contributing

See CONTRIBUTING.md

Related Modules

• amplifier-module-hooks-telegram-bridge - Complementary hook for sending events to Telegram