telegram-toolkit-mcp by DronPascal - MCP Server

Telegram Toolkit MCP

Read-only MCP server for safe extraction of Telegram message history from public chats with proper pagination, error handling, and NDJSON resource streaming.

🚀 Key Features

Remote MCP Support: HTTP transport with Streamable endpoints for production deployment
Enterprise Ready: Docker deployment, health checks, monitoring, and SSL
Production Deployment: One-command setup with Nginx reverse proxy and Cloudflare integration
MCP 2025-06-18 Compliant: Full compliance with latest Model Context Protocol specification

🚀 Features

MCP 2025-06-18 Compliant: Full compliance with latest Model Context Protocol specification
Read-Only Access: Safe extraction from public Telegram channels only
Advanced Pagination: Cursor-based navigation with date filtering
NDJSON Streaming: Efficient large dataset exports
Error Resilience: Comprehensive FLOOD_WAIT handling with retry logic
Observability: Built-in Prometheus metrics and OpenTelemetry tracing
Security First: PII protection, secure session management
Production Ready: Async architecture with proper resource cleanup
Production Ready: Deploy with Docker + Nginx + SSL on any domain

📁 Scripts Directory

The scripts/ directory contains utility scripts for testing, validation, deployment, and maintenance of the Telegram Toolkit MCP server.

📖 For detailed information about script organization, usage, and workflows, see:

🚀 Quick Start

# 1. Clone the repository
git clone https://github.com/DronPascal/telegram-toolkit-mcp.git
cd telegram-toolkit-mcp

# 2. Configure (copy and edit)
cp env.example .env
# Edit .env with your Telegram API credentials

# 3. Run locally
python -m telegram_toolkit_mcp.server
# Or after installing the package:
# telegram-mcp-server
# Server will be available at http://localhost:8000/mcp

# 4. Test MCP endpoint
curl -X POST http://localhost:8000/mcp \
  -H "Content-Type: application/json" \
  -d '{"jsonrpc": "2.0", "id": 1, "method": "tools/list"}'

# 5. Connect your MCP client (see examples below)

🚀 Production Deployment

Enterprise Deployment Features

✅ Production Ready - Deploy on any domain with enterprise-grade infrastructure

Deployment Options:

Automated: One-command deployment script
Docker: Containerized with health checks
Nginx: Reverse proxy with SSL termination
SSL: Let's Encrypt certificates with auto-renewal
Monitoring: Built-in health checks and metrics

Automated Deployment

# Deploy to your domain
./deploy.sh your-domain.com admin@your-domain.com

# Example with a real domain
./deploy.sh mcp.example.com admin@example.com

Production Endpoints (After Deployment)

# Health check
curl https://your-domain.com/health

# MCP tools list
curl -X POST https://your-domain.com/mcp \
  -H "Content-Type: application/json" \
  -d '{"jsonrpc": "2.0", "id": 1, "method": "tools/list"}'

# Test chat resolution
curl -X POST https://your-domain.com/mcp \
  -H "Content-Type: application/json" \
  -d '{
    "jsonrpc": "2.0",
    "id": 2,
    "method": "tools/call",
    "params": {
      "name": "tg.resolve_chat",
      "arguments": {"input": "@telegram"}
    }
  }'

Infrastructure Requirements

Ubuntu 20.04+ VPS with root access
Domain name configured in DNS
Docker & Docker Compose installed
Telegram API credentials

📦 Installation

From Source (recommended)

# Clone the repository
git clone https://github.com/DronPascal/telegram-toolkit-mcp.git
cd telegram-toolkit-mcp

# Install dependencies
pip install -r requirements.txt

# For development with extra tools
pip install -r requirements.txt
pip install pytest black ruff mypy  # Development tools

Docker Deployment

# Build and run with Docker Compose
docker compose up -d

# Check logs
docker compose logs -f

⚙️ Configuration

Environment Variables

Copy env.example to .env and configure:

# Required: Telegram API credentials
TELEGRAM_API_ID=your_api_id
TELEGRAM_API_HASH=your_api_hash
TELEGRAM_STRING_SESSION=  # Will be auto-generated

# Optional: Server settings
MCP_SERVER_HOST=localhost
MCP_SERVER_PORT=8000
MAX_PAGE_SIZE=100

# Optional: Performance tuning
FLOOD_SLEEP_THRESHOLD=60
REQUEST_TIMEOUT=30

# Optional: Observability
ENABLE_PROMETHEUS_METRICS=true
ENABLE_OPENTELEMETRY_TRACING=true

Getting Telegram API Credentials

Go to https://my.telegram.org/auth
Log in with your phone number
Go to "API development tools"
Create an application to get api_id and api_hash

🎯 Usage

Basic Usage

import asyncio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

async def main():
    # Connect to MCP server
    async with stdio_client([
        "python", "-m", "telegram_toolkit_mcp.server"
    ]) as (read, write):
        async with ClientSession(read, write) as session:
            # Resolve chat
            result = await session.call_tool(
                "tg.resolve_chat",
                {"input": "@telegram"}
            )
            print(result)  # {"chat_id": "136817688", "kind": "channel"}

            # Fetch messages
            result = await session.call_tool(
                "tg.fetch_history",
                {
                    "chat": "@telegram",
                    "from": "2025-01-01T00:00:00Z",
                    "to": "2025-01-02T00:00:00Z",
                    "page_size": 50
                }
            )
            print(result)  # Structured message data with pagination

asyncio.run(main())

With Claude Desktop

Add to your claude_desktop_config.json:

{
  "mcpServers": {
    "telegram-toolkit": {
      "command": "npx",
      "args": ["@modelcontextprotocol/inspector", "--remote", "https://your-domain.com/mcp"]
    }
  }
}

Direct HTTP Connection

For direct HTTP connection without MCP Inspector:

{
  "mcpServers": {
    "telegram-toolkit": {
      "command": "curl",
      "args": ["-X", "POST", "https://your-domain.com/mcp", "-H", "Content-Type: application/json", "-d", "{\"jsonrpc\": \"2.0\", \"id\": 1, \"method\": \"tools/list\"}"]
    }
  }
}

🔧 MCP Tools

`tg.resolve_chat`

Resolves chat identifiers to internal format.

Input:

{
  "input": "@telegram"
}

Output:

{
  "chat_id": "136817688",
  "kind": "channel",
  "title": "Telegram News"
}

`tg.fetch_history`

Fetches message history with pagination and filtering.

Input:

{
  "chat": "@telegram",
  "from": "2025-01-01T00:00:00Z",
  "to": "2025-01-02T00:00:00Z",
  "page_size": 100,
  "cursor": "base64_encoded_cursor",
  "search": "bitcoin",
  "filter": "links"
}

Output:

{
  "messages": [
    {
      "id": 12345,
      "date": "2025-01-01T10:30:00Z",
      "from": {
        "peer_id": "channel:136817688",
        "kind": "channel",
        "display": "Telegram"
      },
      "text": "Breaking news...",
      "views": 10000,
      "forwards": 500
    }
  ],
  "page_info": {
    "cursor": "next_cursor_base64",
    "has_more": true,
    "count": 100,
    "fetched": 100
  },
  "export": {
    "uri": "mcp://resources/export/telegram.ndjson",
    "format": "ndjson"
  }
}

📚 API Reference

Response Format

All tools return MCP-compliant responses with:

content[]: Human-readable summary (for compatibility)
structuredContent: Machine-readable data with JSON Schema validation
Resources: NDJSON exports for large datasets (when applicable)

Error Handling

Standardized error codes:

FLOOD_WAIT: Rate limit exceeded, includes retry_after
CHANNEL_PRIVATE: Access denied to private channel
USERNAME_INVALID: Invalid chat identifier
INPUT_VALIDATION: Invalid request parameters

Pagination

Cursor-based: Opaque base64-encoded cursors
Date filtering: Server-side UTC filtering
Deduplication: Automatic message deduplication
Resource limits: Max 100 messages per page

🛠️ Development

🚀 Quick Development Setup

# Clone and setup
git clone https://github.com/DronPascal/telegram-toolkit-mcp.git
cd telegram-toolkit-mcp

# Create environment and install
python -m venv .venv && source .venv/bin/activate
pip install -e ".[dev,observability]"
pre-commit install

# Run all tests (our 5-script testing system)
python scripts/testing/run_all_tests.py --verbose

🤝 Contributing

We welcome contributions! Please see our for details.

Development Guidelines

Code Quality: Follow PEP 8, use type hints, maintain test coverage >90%
Security: Never log PII, validate all inputs, handle secrets securely
Performance: Optimize for memory usage and API call efficiency
Compatibility: Maintain MCP specification compliance
Documentation: Update docs for any API changes

Testing Channels

For E2E testing, we use these public channels:

@telegram - Official Telegram news
@bloomberg - Financial news
@apod_telegram - NASA Astronomy Picture of the Day

📊 Observability

Metrics (Prometheus)

The server exposes metrics at /metrics:

mcp_tool_calls_total{tool,status} - Tool call counts
tg_fetch_history_messages_total - Messages processed
tg_fetch_history_duration_seconds - Response times
telegram_floodwait_seconds - FLOOD_WAIT durations

Tracing (OpenTelemetry)

Distributed tracing for request flows with:

Tool execution spans
Telegram API call spans
Resource generation spans

⚖️ License

This project is licensed under the Apache 2.0 License - see the file for details.

⚠️ Disclaimer

This is an unofficial project and is not affiliated with Telegram. Use responsibly and in accordance with Telegram's Terms of Service. Only access public channels and respect rate limits.

🔧 Troubleshooting

Common Issues

Rate Limiting

Error: Rate limit exceeded. Please wait 30 seconds before retrying.

Solution: The server automatically handles FLOOD_WAIT with exponential backoff. Reduce request frequency or implement client-side rate limiting.

Authentication Errors

Error: Invalid Telegram credentials

Solution: Verify your TELEGRAM_API_ID and TELEGRAM_API_HASH in environment variables.

Channel Not Found

Error: Chat not found

Solution: Ensure the channel exists and is public. Use the correct channel identifier format.

Large Dataset Handling

Warning: Dataset too large, using NDJSON resource

Solution: This is normal behavior for datasets >100 messages. The server automatically creates NDJSON resources.

Debug Mode

Enable debug logging:

export DEBUG=true
export LOG_LEVEL=DEBUG

Performance Tuning

Adjust performance settings:

# Connection pool size
export MAX_CONCURRENT_REQUESTS=20

# Request timeout
export REQUEST_TIMEOUT=60.0

# Page size limits
export MAX_PAGE_SIZE=200

📚 API Reference

MCP Tools

`tg.resolve_chat`

Resolves chat identifiers to standardized format.

Input Schema:

{
  "type": "object",
  "required": ["input"],
  "properties": {
    "input": {
      "type": "string",
      "description": "Chat identifier (@username, t.me URL, or numeric ID)",
      "minLength": 1,
      "maxLength": 100
    }
  }
}

Output Schema:

{
  "type": "object",
  "properties": {
    "chat_id": {"type": "string"},
    "kind": {"type": "string", "enum": ["user", "group", "channel"]},
    "title": {"type": "string"},
    "username": {"type": "string"},
    "description": {"type": "string"},
    "member_count": {"type": "integer"},
    "verified": {"type": "boolean"}
  }
}

`tg.fetch_history`

Fetches message history with advanced filtering.

Input Schema:

{
  "type": "object",
  "required": ["chat"],
  "properties": {
    "chat": {"type": "string"},
    "from_date": {"type": "string", "format": "date-time"},
    "to_date": {"type": "string", "format": "date-time"},
    "page_size": {"type": "integer", "minimum": 1, "maximum": 100, "default": 50},
    "cursor": {"type": "string"},
    "direction": {"type": "string", "enum": ["asc", "desc"], "default": "desc"},
    "search": {"type": "string"},
    "filter": {
      "type": "object",
      "properties": {
        "media_types": {
          "type": "array",
          "items": {"type": "string", "enum": ["text", "photo", "video", "document", "audio", "voice", "sticker", "link", "poll"]}
        },
        "has_media": {"type": "boolean"},
        "from_users": {"type": "array", "items": {"type": "integer"}},
        "min_views": {"type": "integer", "minimum": 0},
        "max_views": {"type": "integer", "minimum": 0}
      }
    }
  }
}

Output Schema:

{
  "type": "object",
  "properties": {
    "messages": {
      "type": "array",
      "items": {
        "type": "object",
        "properties": {
          "id": {"type": "integer"},
          "date": {"type": "number"},
          "text": {"type": "string"},
          "sender": {
            "type": "object",
            "properties": {
              "id": {"type": "integer"},
              "username": {"type": "string"},
              "first_name": {"type": "string"}
            }
          },
          "views": {"type": "integer"},
          "forwards": {"type": "integer"},
          "has_media": {"type": "boolean"}
        }
      }
    },
    "page_info": {
      "type": "object",
      "properties": {
        "has_more": {"type": "boolean"},
        "cursor": {"type": "string"},
        "total_fetched": {"type": "integer"}
      }
    },
    "export": {
      "type": "object",
      "properties": {
        "uri": {"type": "string"},
        "format": {"type": "string"}
      }
    }
  }
}

Error Responses

All errors follow MCP Problem Details format:

{
  "isError": true,
  "error": {
    "type": "ERROR_TYPE",
    "title": "Human readable title",
    "status": 400,
    "detail": "Detailed error description"
  },
  "content": [
    {
      "type": "text",
      "text": "Error message for user"
    }
  ]
}

Common Error Types:

VALIDATION_ERROR (400): Invalid input parameters
CHAT_NOT_FOUND (404): Chat does not exist or is private
CHANNEL_PRIVATE (403): Channel is private
FLOOD_WAIT (429): Rate limit exceeded
INTERNAL_ERROR (500): Server error

🚀 Deployment

Docker Deployment

FROM python:3.11-slim

WORKDIR /app
COPY . .

RUN pip install -e .
EXPOSE 8000

CMD ["python", "-m", "telegram_toolkit_mcp.server"]

Kubernetes Deployment

apiVersion: apps/v1
kind: Deployment
metadata:
  name: telegram-toolkit-mcp
spec:
  replicas: 1
  selector:
    matchLabels:
      app: telegram-toolkit-mcp
  template:
    metadata:
      labels:
        app: telegram-toolkit-mcp
    spec:
      containers:
      - name: telegram-toolkit-mcp
        image: DronPascal/telegram-toolkit-mcp:latest
        ports:
        - containerPort: 8000
        env:
        - name: TELEGRAM_API_ID
          valueFrom:
            secretKeyRef:
              name: telegram-secrets
              key: api-id
        - name: TELEGRAM_API_HASH
          valueFrom:
            secretKeyRef:
              name: telegram-secrets
              key: api-hash
        resources:
          limits:
            memory: "512Mi"
            cpu: "500m"

Environment Variables

# Required
TELEGRAM_API_ID=your_api_id
TELEGRAM_API_HASH=your_api_hash

# Optional
MCP_SERVER_HOST=0.0.0.0
MCP_SERVER_PORT=8000
LOG_LEVEL=INFO
DEBUG=false
MAX_PAGE_SIZE=100
REQUEST_TIMEOUT=30.0
MAX_CONCURRENT_REQUESTS=10
TEMP_DIR=/tmp/telegram_toolkit
RESOURCE_MAX_AGE_HOURS=24
NDJSON_CHUNK_SIZE=1024
ENABLE_PROMETHEUS_METRICS=true
ENABLE_OPENTELEMETRY_TRACING=false

📊 Monitoring

Health Checks

# Health endpoint (if implemented)
curl http://localhost:8000/health

# Metrics endpoint
curl http://localhost:8000/metrics

Key Metrics to Monitor

mcp_tool_calls_total{status="success"} - Successful tool calls
telegram_api_calls_total{status="error"} - API errors
telegram_flood_wait_events_total - Rate limiting events
telegram_messages_fetched_total - Data volume
telegram_ndjson_exports_total - Large export usage

Alerting Rules

# Prometheus alerting rules
groups:
  - name: telegram_toolkit_alerts
    rules:
      - alert: HighErrorRate
        expr: rate(mcp_tool_calls_total{status="error"}[5m]) > 0.1
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "High error rate detected"

      - alert: FloodWaitSpike
        expr: increase(telegram_flood_wait_events_total[5m]) > 10
        for: 2m
        labels:
          severity: warning
        annotations:
          summary: "High FLOOD_WAIT rate detected"

🧪 Testing

Running Tests

# Run all tests
pytest

# Run specific test categories
pytest -m unit
pytest -m integration
pytest -m e2e

# Run with coverage
pytest --cov=src --cov-report=html

# Run security tests
pytest -m security

Test Configuration

# Test environment variables
export TELEGRAM_API_ID=12345
export TELEGRAM_API_HASH=test_hash_12345678901234567890123456789012
export DEBUG=true
export LOG_LEVEL=DEBUG

Mock Data for Testing

The test suite includes comprehensive mocks for:

Telegram API responses
MCP protocol interactions
File system operations
Network requests
Rate limiting scenarios

🔒 Security

Data Protection

PII Masking: All sensitive data is automatically masked in logs
No Data Persistence: Messages are never stored on disk
Secure Sessions: In-memory session management with integrity checks
Rate Limiting: Built-in protection against abuse

Authentication

Telegram API: Uses official Telegram API with proper credentials
No User Data: Only accesses public channel information
Session Security: StringSession encryption for persistent sessions

Compliance

GDPR: PII masking for European data protection
Telegram ToS: Respects rate limits and access policies
Open Source: Transparent security implementation

📈 Performance

Benchmarks

Operation	p50	p95	p99
Chat Resolution	50ms	200ms	500ms
Message Fetch (50)	100ms	500ms	1s
Large Export	2s	5s	10s

Optimization Tips

Use Appropriate Page Sizes: Balance between latency and throughput
Implement Client-Side Caching: Cache frequently accessed chat info
Monitor Rate Limits: Respect Telegram's rate limiting
Use Filtering: Reduce data transfer with server-side filtering

🤝 Contributing

Development Setup

# Clone repository
git clone https://github.com/DronPascal/telegram-toolkit-mcp.git
cd telegram-toolkit-mcp

# Install dependencies
pip install -e ".[dev]"

# Run tests
pytest

# Run linting
ruff check .
mypy src/

Code Standards

Python 3.11+ with type hints
Black for code formatting
Ruff for linting
MyPy for type checking
Pytest for testing

Pull Request Process

Fork the repository
Create a feature branch
Add tests for new functionality
Ensure all tests pass
Update documentation
Submit pull request

🌐 Remote MCP Usage

For detailed instructions on how to use the MCP server from any programming language, see:

📖 - Complete guide for remote MCP integration

This guide covers:

Language-agnostic HTTP communication
JSON-RPC 2.0 protocol usage
Examples in Python, JavaScript, Go, and cURL
Error handling and best practices
Pagination and advanced features

💡 Examples

Simple CLI Client

A ready-to-use command-line client is available in the examples/ directory:

# Navigate to examples
cd examples/simple-client

# Configure your domain
cp env.example .env
# Edit .env with your MCP_DOMAIN

# Install dependencies
pip install requests python-dotenv

# Use the client
python simple_client.py health
python simple_client.py tools
python simple_client.py resolve "@telegram"
python simple_client.py fetch "@telegram" --from "2025-01-01T00:00:00Z"
python simple_client.py interactive

The client demonstrates:

Environment-based configuration
JSON-RPC 2.0 communication
Error handling and retry logic
Interactive exploration mode
Command-line interface patterns

See: examples/simple-client/README.md for detailed usage instructions.

🙏 Acknowledgments

Telethon for Telegram API client
MCP for the protocol specification
FastMCP for Python SDK
Prometheus for monitoring capabilities