# Directory Structure
```
├── .DS_Store
├── .gitattributes
├── .gitignore
├── LICENSE.txt
├── pyproject.toml
├── README.md
├── src
│ ├── .DS_Store
│ ├── mcp_telegram
│ │ ├── __init__.py
│ │ ├── .env.example
│ │ ├── convostyle.txt
│ │ ├── main.py
│ │ └── telethon_auth.py
│ └── telegram_mcp.egg-info
│ ├── dependency_links.txt
│ ├── entry_points.txt
│ ├── PKG-INFO
│ ├── requires.txt
│ ├── SOURCES.txt
│ └── top_level.txt
└── uv.lock
```
# Files
--------------------------------------------------------------------------------
/.gitattributes:
--------------------------------------------------------------------------------
```
# Auto detect text files and perform LF normalization
* text=auto
```
--------------------------------------------------------------------------------
/src/mcp_telegram/.env.example:
--------------------------------------------------------------------------------
```
# Telegram MCP Configuration
# Copy this file to .env and fill in your values
# Required: Get these from https://my.telegram.org/apps
TELEGRAM_API_ID=
TELEGRAM_API_HASH=
# Optional: Will be filled automatically during authentication
TELEGRAM_PHONE=
TELEGRAM_2FA_PASSWORD=
TELEGRAM_SESSION_STRING=
```
--------------------------------------------------------------------------------
/.gitignore:
--------------------------------------------------------------------------------
```
# Byte-compiled / optimized / DLL files
__pycache__/
*.py[cod]
*$py.class
# C extensions
*.so
# Distribution / packaging
.Python
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
share/python-wheels/
*.egg-info/
.installed.cfg
*.egg
MANIFEST
# PyInstaller
# Usually these files are written by a python script from a template
# before PyInstaller builds the exe, so as to inject date/other infos into it.
*.manifest
*.spec
# Installer logs
pip-log.txt
pip-delete-this-directory.txt
# Unit test / coverage reports
htmlcov/
.tox/
.nox/
.coverage
.coverage.*
.cache
nosetests.xml
coverage.xml
*.cover
*.py,cover
.hypothesis/
.pytest_cache/
cover/
# Translations
*.mo
*.pot
# Django stuff:
*.log
local_settings.py
db.sqlite3
db.sqlite3-journal
# Flask stuff:
instance/
.webassets-cache
# Scrapy stuff:
.scrapy
# Sphinx documentation
docs/_build/
# PyBuilder
.pybuilder/
target/
# Jupyter Notebook
.ipynb_checkpoints
# IPython
profile_default/
ipython_config.py
# pyenv
# For a library or package, you might want to ignore these files since the code is
# intended to run in multiple environments; otherwise, check them in:
# .python-version
# pipenv
# According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
# However, in case of collaboration, if having platform-specific dependencies or dependencies
# having no cross-platform support, pipenv may install dependencies that don't work, or not
# install all needed dependencies.
#Pipfile.lock
# UV
# Similar to Pipfile.lock, it is generally recommended to include uv.lock in version control.
# This is especially recommended for binary packages to ensure reproducibility, and is more
# commonly ignored for libraries.
#uv.lock
# poetry
# Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control.
# This is especially recommended for binary packages to ensure reproducibility, and is more
# commonly ignored for libraries.
# https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control
#poetry.lock
# pdm
# Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control.
#pdm.lock
# pdm stores project-wide configurations in .pdm.toml, but it is recommended to not include it
# in version control.
# https://pdm.fming.dev/latest/usage/project/#working-with-version-control
.pdm.toml
.pdm-python
.pdm-build/
# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm
__pypackages__/
# Celery stuff
celerybeat-schedule
celerybeat.pid
# SageMath parsed files
*.sage.py
# Environments
.env
.venv
env/
venv/
ENV/
env.bak/
venv.bak/
# Spyder project settings
.spyderproject
.spyproject
# Rope project settings
.ropeproject
# mkdocs documentation
/site
# mypy
.mypy_cache/
.dmypy.json
dmypy.json
# Pyre type checker
.pyre/
# pytype static type analyzer
.pytype/
# Cython debug symbols
cython_debug/
# Ruff stuff:
.ruff_cache/
# PyPI configuration file
.pypirc
# Environment variables
.env
```
--------------------------------------------------------------------------------
/README.md:
--------------------------------------------------------------------------------
```markdown
# Telegram MCP Server
Connect Claude to your Telegram account to read and send messages.
## Features
### Available Tools
1. **get_chats** - List your Telegram chats
- Returns paginated list with chat names, IDs, and unread counts
- For page 1, just provide page number
- For subsequent pages, use the pagination parameters from the previous response
2. **get_messages** - Read messages from a specific chat
- Fetches paginated message history
- Automatically marks messages as read
3. **mark_messages_read** - Mark all unread messages in a chat as read
4. **send_message** - Send messages to any chat
- Supports replying to specific messages
5. **get_conversation_context** - Analyze chat style for natural responses
- Reads your conversation style guide from `convostyle.txt`
- Helps Claude match your texting patterns
## Setup Guide
### Step 1: Get Telegram API Credentials
1. Go to [https://my.telegram.org/apps](https://my.telegram.org/apps)
2. Log in and create an application
3. Save your **API ID** and **API Hash**
### Step 2: Install
```bash
# Clone the repository
git clone https://github.com/alexandertsai/mcp-telegram
cd mcp-telegram
# Set up Python environment
pip install uv
uv venv
source .venv/bin/activate # On Windows: .venv\Scripts\activate
uv sync
```
### Step 3: Configure
```bash
# Copy the example file
cp .env.example .env
# Edit .env and add your API credentials:
# TELEGRAM_API_ID=your_api_id_here
# TELEGRAM_API_HASH=your_api_hash_here
```
### Step 4: Authenticate
```bash
cd src/mcp_telegram
python telethon_auth.py
```
Follow the prompts:
- Enter your phone number (with country code, e.g., +1234567890)
- Enter the code sent to your Telegram
- Enter your 2FA password if you have one
### Step 5: Add to Claude Desktop
Find your Claude Desktop config file:
- **macOS**: `~/Library/Application Support/Claude/claude_desktop_config.json`
- **Windows**: `%APPDATA%\Claude\claude_desktop_config.json`
Add this configuration:
```json
{
"mcpServers": {
"telegram": {
"command": "/path/to/python",
"args": ["/path/to/mcp-telegram/src/mcp_telegram/main.py"]
}
}
}
```
**To find paths:**
- Python: Run `which python` (Mac) or `where.exe python` (Windows)
- main.py: Right-click the file and select "Copy Path"
Restart Claude Desktop.
## Usage
After setup, you can ask Claude to:
- "Check my Telegram messages"
- "Send a message to [contact name]"
- "What are my unread chats?"
- "Reply to the last message from [contact name]"
## Style Guide (Optional)
Create `src/mcp_telegram/convostyle.txt` to help Claude match your texting style:
```
I text casually with friends, formally with work contacts.
I use emojis sparingly and prefer short messages.
```
## Troubleshooting
### Authentication Issues
If authentication fails:
1. Check your API credentials in `.env`
2. Remove the TELEGRAM_SESSION_STRING line from `.env`
3. Run `python telethon_auth.py` again
### Common Errors
- **"Please set TELEGRAM_API_ID and TELEGRAM_API_HASH"**: Missing `.env` file or credentials
- **"Session string is invalid or expired"**: Re-run authentication
- **2FA password not showing**: This is normal - keep typing
## Requirements
- Python 3.10+
- Claude Desktop
- Telegram account
## License
Apache 2.0
```
--------------------------------------------------------------------------------
/src/telegram_mcp.egg-info/dependency_links.txt:
--------------------------------------------------------------------------------
```
```
--------------------------------------------------------------------------------
/src/telegram_mcp.egg-info/top_level.txt:
--------------------------------------------------------------------------------
```
mcp_telegram
```
--------------------------------------------------------------------------------
/src/mcp_telegram/__init__.py:
--------------------------------------------------------------------------------
```python
"""Telegram MCP server for reading and writing messages."""
__version__ = "0.1.0"
```
--------------------------------------------------------------------------------
/src/telegram_mcp.egg-info/requires.txt:
--------------------------------------------------------------------------------
```
httpx>=0.24.0
mcp>=1.4.1
nest-asyncio>=1.5.6
python-dotenv>=1.0.0
telethon>=1.28.0
```
--------------------------------------------------------------------------------
/src/telegram_mcp.egg-info/entry_points.txt:
--------------------------------------------------------------------------------
```
[console_scripts]
telegram-auth = mcp_telegram.telethon_auth:main
telegram-mcp = mcp_telegram.main:main
```
--------------------------------------------------------------------------------
/src/telegram_mcp.egg-info/SOURCES.txt:
--------------------------------------------------------------------------------
```
LICENSE.txt
README.md
pyproject.toml
src/mcp_telegram/__init__.py
src/mcp_telegram/main.py
src/mcp_telegram/telethon_auth.py
src/telegram_mcp.egg-info/PKG-INFO
src/telegram_mcp.egg-info/SOURCES.txt
src/telegram_mcp.egg-info/dependency_links.txt
src/telegram_mcp.egg-info/entry_points.txt
src/telegram_mcp.egg-info/requires.txt
src/telegram_mcp.egg-info/top_level.txt
```
--------------------------------------------------------------------------------
/src/mcp_telegram/convostyle.txt:
--------------------------------------------------------------------------------
```
EXAMPLE: I type concisely in proper english. I do not use many emojis (except 😭 occasionally) but I do sometimes use text emotes. However, I do keep my tone friendly.
Here are some examples of how I'd respond to messages:
SENDER: "How was your day?"
ME: "Good! What about yours :)"
SENDER: "Can you get this done?"
ME: "Okays"
SENDER: "What do you think of this idea?"
ME: "Honestly it sounds good, but I'm a bit worried it might be too static."
SENDER: "I'm really tired and my test went so badly..."
ME: "Oh dear what happened?"
```
--------------------------------------------------------------------------------
/pyproject.toml:
--------------------------------------------------------------------------------
```toml
[project]
name = "telegram-mcp"
version = "0.1.0"
description = "A Telegram MCP server for reading and writing messages."
readme = "README.md"
requires-python = ">=3.10"
license = "Apache-2.0"
authors = [
{name = "Alexander Tsai"}
]
keywords = ["telegram", "mcp", "claude", "ai", "llm"]
dependencies = [
"httpx>=0.24.0",
"mcp>=1.4.1",
"python-dotenv>=1.0.0",
"telethon>=1.28.0",
]
[project.scripts]
telegram-mcp = "mcp_telegram.main:main"
telegram-auth = "mcp_telegram.telethon_auth:main"
[project.urls]
"Homepage" = "https://github.com/alexandertsai/telegram-mcp"
"Bug Tracker" = "https://github.com/alexandertsai/telegram-mcp/issues"
[build-system]
requires = ["setuptools>=61.0"]
build-backend = "setuptools.build_meta"
[tool.setuptools]
package-dir = {"" = "src"}
packages = ["mcp_telegram"]
```
--------------------------------------------------------------------------------
/src/mcp_telegram/telethon_auth.py:
--------------------------------------------------------------------------------
```python
#!/usr/bin/env python3
# telethon_auth.py
import asyncio
import os
import sys
import logging
import getpass
from telethon import TelegramClient
from telethon.sessions import StringSession
from telethon.errors import SessionPasswordNeededError
from dotenv import load_dotenv, set_key
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='[%(asctime)s] %(levelname)-8s %(message)s',
datefmt='%m/%d/%y %H:%M:%S',
)
logger = logging.getLogger(__name__)
async def authenticate():
# Load existing .env file
load_dotenv()
# Get API credentials from .env
api_id = os.getenv('TELEGRAM_API_ID')
api_hash = os.getenv('TELEGRAM_API_HASH')
phone = os.getenv('TELEGRAM_PHONE')
password = os.getenv('TELEGRAM_2FA_PASSWORD') # Optional, for 2FA
session_string = os.getenv('TELEGRAM_SESSION_STRING')
# Check if we already have a valid session
if session_string:
logger.info("Found existing session string in .env file. Verifying...")
client = TelegramClient(StringSession(session_string), api_id, api_hash)
try:
await client.connect()
if await client.is_user_authorized():
logger.info("✓ Existing session is valid! You're already authenticated.")
await client.disconnect()
return True
else:
logger.warning("Existing session is invalid or expired. Need to re-authenticate.")
except Exception as e:
logger.warning(f"Error with existing session: {str(e)}")
finally:
await client.disconnect()
# Check required credentials
if not api_id or not api_hash:
logger.error("Missing required credentials in .env file!")
logger.error("Please add the following to your .env file:")
logger.error("TELEGRAM_API_ID=your_api_id")
logger.error("TELEGRAM_API_HASH=your_api_hash")
logger.error("Get these from https://my.telegram.org/apps")
return False
# Get phone number if not in .env
if not phone:
phone = input("Enter your phone number (with country code, e.g. +12345678900): ")
# Save phone to .env for future use
env_path = os.path.join(os.getcwd(), '.env')
set_key(env_path, 'TELEGRAM_PHONE', phone)
logger.info(f"Phone number saved to .env file")
else:
logger.info(f"Using phone number from .env: {phone}")
use_saved = input(f"Use saved phone number {phone}? (y/n): ").lower()
if use_saved != 'y':
phone = input("Enter your phone number (with country code, e.g. +12345678900): ")
env_path = os.path.join(os.getcwd(), '.env')
set_key(env_path, 'TELEGRAM_PHONE', phone)
# Create new session
client = TelegramClient(StringSession(), api_id, api_hash)
try:
await client.connect()
# Send code request
logger.info(f"Sending authentication code to {phone}...")
await client.send_code_request(phone)
# Get the code from the user
code = input("Enter the code you received: ")
# Initialize save_pwd to avoid reference error
save_pwd = 'n'
try:
# Try to sign in with the code
await client.sign_in(phone, code)
except SessionPasswordNeededError:
# 2FA is enabled
logger.info("Two-factor authentication is enabled.")
# Try to use saved password first
if password:
logger.info("Using 2FA password from .env file...")
try:
await client.sign_in(password=password)
logger.info("✓ Successfully authenticated with saved password!")
except Exception as e:
logger.warning("Saved password didn't work, please enter manually.")
password = None
# If no saved password or it didn't work, ask user
if not password:
password = getpass.getpass("Enter your 2FA password: ")
await client.sign_in(password=password)
# Ask if user wants to save password
save_pwd = input("Save 2FA password to .env file for future use? (y/n): ").lower()
if save_pwd == 'y':
env_path = os.path.join(os.getcwd(), '.env')
set_key(env_path, 'TELEGRAM_2FA_PASSWORD', password)
logger.info("2FA password saved to .env file")
if await client.is_user_authorized():
# Save the string session to .env
session_string = client.session.save()
env_path = os.path.join(os.getcwd(), '.env')
set_key(env_path, 'TELEGRAM_SESSION_STRING', session_string)
logger.info("✓ Authentication successful!")
logger.info("✓ Session string saved to .env file")
logger.info("")
logger.info("You can now start the MCP server with: python main.py")
logger.info("")
logger.info("Your .env file has been updated with:")
logger.info("- TELEGRAM_SESSION_STRING (required)")
logger.info("- TELEGRAM_PHONE (for convenience)")
if save_pwd == 'y':
logger.info("- TELEGRAM_2FA_PASSWORD (optional, for automatic re-auth)")
return True
else:
logger.error("Authentication failed.")
return False
except Exception as e:
logger.error(f"Authentication error: {str(e)}")
return False
finally:
await client.disconnect()
def main():
logger.info("Telegram MCP Authentication Setup")
logger.info("-" * 40)
# Check if .env file exists
env_path = os.path.join(os.getcwd(), '.env')
if not os.path.exists(env_path):
logger.info("No .env file found. Creating one...")
with open(env_path, 'w') as f:
f.write("# Telegram MCP Configuration\n")
f.write("# Get API credentials from https://my.telegram.org/apps\n")
f.write("TELEGRAM_API_ID=\n")
f.write("TELEGRAM_API_HASH=\n")
f.write("TELEGRAM_PHONE=\n")
f.write("TELEGRAM_2FA_PASSWORD=\n")
f.write("TELEGRAM_SESSION_STRING=\n")
logger.info(f"Created .env file at: {env_path}")
logger.info("Please edit it and add your TELEGRAM_API_ID and TELEGRAM_API_HASH")
logger.info("Then run this script again.")
sys.exit(1)
success = asyncio.run(authenticate())
sys.exit(0 if success else 1)
if __name__ == "__main__":
main()
```
--------------------------------------------------------------------------------
/LICENSE.txt:
--------------------------------------------------------------------------------
```
Apache License
Version 2.0, January 2004
http://www.apache.org/licenses/
TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
1. Definitions.
"License" shall mean the terms and conditions for use, reproduction,
and distribution as defined by Sections 1 through 9 of this document.
"Licensor" shall mean the copyright owner or entity authorized by
the copyright owner that is granting the License.
"Legal Entity" shall mean the union of the acting entity and all
other entities that control, are controlled by, or are under common
control with that entity. For the purposes of this definition,
"control" means (i) the power, direct or indirect, to cause the
direction or management of such entity, whether by contract or
otherwise, or (ii) ownership of fifty percent (50%) or more of the
outstanding shares, or (iii) beneficial ownership of such entity.
"You" (or "Your") shall mean an individual or Legal Entity
exercising permissions granted by this License.
"Source" form shall mean the preferred form for making modifications,
including but not limited to software source code, documentation
source, and configuration files.
"Object" form shall mean any form resulting from mechanical
transformation or translation of a Source form, including but
not limited to compiled object code, generated documentation,
and conversions to other media types.
"Work" shall mean the work of authorship, whether in Source or
Object form, made available under the License, as indicated by a
copyright notice that is included in or attached to the work
(an example is provided in the Appendix below).
"Derivative Works" shall mean any work, whether in Source or Object
form, that is based on (or derived from) the Work and for which the
editorial revisions, annotations, elaborations, or other modifications
represent, as a whole, an original work of authorship. For the purposes
of this License, Derivative Works shall not include works that remain
separable from, or merely link (or bind by name) to the interfaces of,
the Work and Derivative Works thereof.
"Contribution" shall mean any work of authorship, including
the original version of the Work and any modifications or additions
to that Work or Derivative Works thereof, that is intentionally
submitted to Licensor for inclusion in the Work by the copyright owner
or by an individual or Legal Entity authorized to submit on behalf of
the copyright owner. For the purposes of this definition, "submitted"
means any form of electronic, verbal, or written communication sent
to the Licensor or its representatives, including but not limited to
communication on electronic mailing lists, source code control systems,
and issue tracking systems that are managed by, or on behalf of, the
Licensor for the purpose of discussing and improving the Work, but
excluding communication that is conspicuously marked or otherwise
designated in writing by the copyright owner as "Not a Contribution."
"Contributor" shall mean Licensor and any individual or Legal Entity
on behalf of whom a Contribution has been received by Licensor and
subsequently incorporated within the Work.
2. Grant of Copyright License. Subject to the terms and conditions of
this License, each Contributor hereby grants to You a perpetual,
worldwide, non-exclusive, no-charge, royalty-free, irrevocable
copyright license to reproduce, prepare Derivative Works of,
publicly display, publicly perform, sublicense, and distribute the
Work and such Derivative Works in Source or Object form.
3. Grant of Patent License. Subject to the terms and conditions of
this License, each Contributor hereby grants to You a perpetual,
worldwide, non-exclusive, no-charge, royalty-free, irrevocable
(except as stated in this section) patent license to make, have made,
use, offer to sell, sell, import, and otherwise transfer the Work,
where such license applies only to those patent claims licensable
by such Contributor that are necessarily infringed by their
Contribution(s) alone or by combination of their Contribution(s)
with the Work to which such Contribution(s) was submitted. If You
institute patent litigation against any entity (including a
cross-claim or counterclaim in a lawsuit) alleging that the Work
or a Contribution incorporated within the Work constitutes direct
or contributory patent infringement, then any patent licenses
granted to You under this License for that Work shall terminate
as of the date such litigation is filed.
4. Redistribution. You may reproduce and distribute copies of the
Work or Derivative Works thereof in any medium, with or without
modifications, and in Source or Object form, provided that You
meet the following conditions:
(a) You must give any other recipients of the Work or
Derivative Works a copy of this License; and
(b) You must cause any modified files to carry prominent notices
stating that You changed the files; and
(c) You must retain, in the Source form of any Derivative Works
that You distribute, all copyright, patent, trademark, and
attribution notices from the Source form of the Work,
excluding those notices that do not pertain to any part of
the Derivative Works; and
(d) If the Work includes a "NOTICE" text file as part of its
distribution, then any Derivative Works that You distribute must
include a readable copy of the attribution notices contained
within such NOTICE file, excluding those notices that do not
pertain to any part of the Derivative Works, in at least one
of the following places: within a NOTICE text file distributed
as part of the Derivative Works; within the Source form or
documentation, if provided along with the Derivative Works; or,
within a display generated by the Derivative Works, if and
wherever such third-party notices normally appear. The contents
of the NOTICE file are for informational purposes only and
do not modify the License. You may add Your own attribution
notices within Derivative Works that You distribute, alongside
or as an addendum to the NOTICE text from the Work, provided
that such additional attribution notices cannot be construed
as modifying the License.
You may add Your own copyright statement to Your modifications and
may provide additional or different license terms and conditions
for use, reproduction, or distribution of Your modifications, or
for any such Derivative Works as a whole, provided Your use,
reproduction, and distribution of the Work otherwise complies with
the conditions stated in this License.
5. Submission of Contributions. Unless You explicitly state otherwise,
any Contribution intentionally submitted for inclusion in the Work
by You to the Licensor shall be under the terms and conditions of
this License, without any additional terms or conditions.
Notwithstanding the above, nothing herein shall supersede or modify
the terms of any separate license agreement you may have executed
with Licensor regarding such Contributions.
6. Trademarks. This License does not grant permission to use the trade
names, trademarks, service marks, or product names of the Licensor,
except as required for reasonable and customary use in describing the
origin of the Work and reproducing the content of the NOTICE file.
7. Disclaimer of Warranty. Unless required by applicable law or
agreed to in writing, Licensor provides the Work (and each
Contributor provides its Contributions) on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
implied, including, without limitation, any warranties or conditions
of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
PARTICULAR PURPOSE. You are solely responsible for determining the
appropriateness of using or redistributing the Work and assume any
risks associated with Your exercise of permissions under this License.
8. Limitation of Liability. In no event and under no legal theory,
whether in tort (including negligence), contract, or otherwise,
unless required by applicable law (such as deliberate and grossly
negligent acts) or agreed to in writing, shall any Contributor be
liable to You for damages, including any direct, indirect, special,
incidental, or consequential damages of any character arising as a
result of this License or out of the use or inability to use the
Work (including but not limited to damages for loss of goodwill,
work stoppage, computer failure or malfunction, or any and all
other commercial damages or losses), even if such Contributor
has been advised of the possibility of such damages.
9. Accepting Warranty or Additional Liability. While redistributing
the Work or Derivative Works thereof, You may choose to offer,
and charge a fee for, acceptance of support, warranty, indemnity,
or other liability obligations and/or rights consistent with this
License. However, in accepting such obligations, You may act only
on Your own behalf and on Your sole responsibility, not on behalf
of any other Contributor, and only if You agree to indemnify,
defend, and hold each Contributor harmless for any liability
incurred by, or claims asserted against, such Contributor by reason
of your accepting any such warranty or additional liability.
END OF TERMS AND CONDITIONS
Copyright [2025] [Alexander Tsai]
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
```
--------------------------------------------------------------------------------
/src/mcp_telegram/main.py:
--------------------------------------------------------------------------------
```python
#!/usr/bin/env python3
import os
import sys
import json
from telethon import TelegramClient
from telethon.sessions import StringSession
from mcp.server.fastmcp import FastMCP
from typing import Optional
from dotenv import load_dotenv
# Client will be initialized when needed
client: Optional[TelegramClient] = None
class TelegramServer:
def __init__(self):
self.app = FastMCP("telegram")
self.client = None
self.register_tools()
async def initialize_client(self):
"""Initialize Telegram client if not already initialized"""
if self.client and self.client.is_connected():
return
api_id = os.getenv('TELEGRAM_API_ID')
api_hash = os.getenv('TELEGRAM_API_HASH')
session_string = os.getenv('TELEGRAM_SESSION_STRING')
if not api_id or not api_hash or not session_string:
raise ValueError("Missing Telegram credentials in .env file")
self.client = TelegramClient(StringSession(session_string), api_id, api_hash)
await self.client.connect()
if not await self.client.is_user_authorized():
raise ValueError("Session string is invalid or expired")
def register_tools(self):
"""Register Telegram-related tools with the MCP server"""
@self.app.tool()
async def get_chats(page: int, page_size: int = 20, offset_id: int = 0, offset_date: str = None, offset_peer_id: int = None) -> str:
"""
Used when checking messages. Gets a paginated list of chats from Telegram.
Args:
page: Page number (1-indexed).
page_size: Number of chats per page.
offset_id: Message ID to use as offset for pagination (from previous page's last_message_id).
offset_date: Date to use as offset for pagination (from previous page's last_message_date).
offset_peer_id: Peer ID to use as offset for pagination (from previous page's last_peer_id).
After using this tool, use get_messages to read the actual messages from chats
with unread_count > 0
"""
try:
await self.initialize_client()
# For page 1, start from the beginning
if page == 1 or (offset_id == 0 and offset_date is None and offset_peer_id is None):
dialogs = await self.client.get_dialogs(limit=page_size, archived=False)
else:
# For subsequent pages, use the provided offset parameters
offset_peer = None
if offset_peer_id:
# Get the peer entity from the ID
offset_peer = await self.client.get_entity(offset_peer_id)
# Parse the offset date if provided
from datetime import datetime
offset_date_obj = None
if offset_date:
try:
offset_date_obj = datetime.fromisoformat(offset_date.replace('Z', '+00:00'))
except:
offset_date_obj = None
dialogs = await self.client.get_dialogs(
limit=page_size,
offset_date=offset_date_obj,
offset_id=offset_id,
offset_peer=offset_peer,
archived=False
)
result = []
pagination_info = None
for dialog in dialogs:
chat = {
"id": dialog.id,
"name": dialog.name,
"unread_count": dialog.unread_count,
"type": str(dialog.entity.__class__.__name__)
}
result.append(chat)
# Store pagination info from the last dialog
if dialog == dialogs[-1] and dialog.message:
pagination_info = {
"last_message_id": dialog.message.id,
"last_message_date": dialog.date.isoformat() if dialog.date else None,
"last_peer_id": dialog.id
}
# Return both results and pagination info
response = {
"chats": result,
"page": page,
"page_size": page_size,
"has_more": len(result) == page_size
}
if pagination_info:
response["next_page_params"] = {
"page": page + 1,
"page_size": page_size,
"offset_id": pagination_info["last_message_id"],
"offset_date": pagination_info["last_message_date"],
"offset_peer_id": pagination_info["last_peer_id"]
}
return json.dumps(response, indent=2)
except Exception as e:
return json.dumps({"error": str(e)})
@self.app.tool()
async def get_messages(chat_id: int, page: int, page_size: int = 20) -> str:
"""
Get paginated messages from a specific chat from Telegram.
Args:
chat_id: The ID of the chat.
page: Page number (1-indexed).
page_size: Number of messages per page.
"""
offset = (page - 1) * page_size
limit = page_size
try:
await self.initialize_client()
messages = await self.client.get_messages(chat_id, limit=limit, offset_id=0, offset_date=None, add_offset=offset)
await self.client.send_read_acknowledge(entity=chat_id)
result = []
for message in messages:
msg = {
"id": message.id,
"date": message.date.isoformat(),
"text": message.text,
"sender_id": message.sender_id,
"reply_to_msg_id": message.reply_to_msg_id
}
result.append(msg)
return json.dumps(result, indent=2)
except Exception as e:
return json.dumps({"error": str(e)})
@self.app.tool()
async def mark_messages_read(chat_id: int) -> str:
"""
Mark all unread messages in a specific Telegram chat as read.
Args:
chat_id: The ID of the chat whose messages should be marked as read.
"""
try:
await self.initialize_client()
# The read_history method marks messages as read
result = await self.client.send_read_acknowledge(entity=chat_id)
return json.dumps({
"success": True,
"message": f"Successfully marked messages as read in chat {chat_id}"
})
except Exception as e:
return json.dumps({"success": False, "error": str(e)})
@self.app.tool()
async def send_message(chat_id: int, message: str, reply_to_msg_id: int = None) -> str:
"""
Send a message to a specific chat in Telegram.
Args:
chat_id: The ID of the chat.
message: The message content to send.
reply_to_msg_id: Optional ID of a message to reply to. If provided, this message will be a reply to that specific message.
"""
try:
await self.initialize_client()
result = await self.client.send_message(
entity=chat_id,
message=message,
reply_to=reply_to_msg_id
)
return json.dumps({
"success": True,
"message_id": result.id,
"is_reply": reply_to_msg_id is not None,
"replied_to_message_id": reply_to_msg_id
})
except Exception as e:
return json.dumps({
"success": False,
"error": str(e),
"is_reply": reply_to_msg_id is not None,
"replied_to_message_id": reply_to_msg_id
})
@self.app.tool()
async def get_conversation_context(chat_id: int, message_count: int = 30) -> str:
"""
This function retrieves recent messages from a specific chat to help
understand the conversational style and tone, allowing it to
generate responses that match the existing conversation pattern.
The function also reads a user-defined style guide from convostyle.txt
to further refine the response style.
Args:
chat_id: The ID of the chat to analyze.
message_count: Number of recent messages to retrieve (default: 30).
"""
try:
await self.initialize_client()
# Get messages from the chat
messages = await self.client.get_messages(chat_id, limit=message_count)
# Process and organize the conversation
conversation = []
sender_info = {}
# First pass: collect unique senders and their information
for msg in messages:
if msg.sender_id and msg.sender_id not in sender_info:
try:
# Get sender information
entity = await self.client.get_entity(msg.sender_id)
sender_name = getattr(entity, 'first_name', '') or getattr(entity, 'title', '') or str(msg.sender_id)
sender_info[msg.sender_id] = {
'id': msg.sender_id,
'name': sender_name,
'is_self': msg.out
}
except Exception as e:
# If we can't get entity info, use minimal information
sender_info[msg.sender_id] = {
'id': msg.sender_id,
'name': f"User {msg.sender_id}",
'is_self': msg.out
}
# Second pass: organize messages into conversation format
# Start with newest messages first in the API response, so we reverse to get chronological order
for msg in reversed(messages):
if not msg.text: # Skip non-text messages
continue
sender = sender_info.get(msg.sender_id, {'name': 'Unknown', 'is_self': False})
conversation.append({
'timestamp': msg.date.isoformat(),
'sender_name': sender['name'],
'is_self': sender['is_self'],
'text': msg.text,
'message_id': msg.id
})
# Read the user's conversation style guide
style_guide = ""
style_guide_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "convostyle.txt")
try:
with open(style_guide_path, 'r', encoding='utf-8') as file:
style_guide = file.read().strip()
except Exception as e:
style_guide = "Style guide file not available. Focus only on conversation history."
# Add analysis information to help Claude
result = {
'conversation': conversation,
'user_style_guide': style_guide,
'analysis_instructions': """
You're helping generate messages that match the user's texting style.
To do this effectively:
1. FIRST, carefully read the user's own description of their texting style in the 'user_style_guide' field.
This is the primary source of information about how they want to come across in messages.
2. SECOND, analyze the conversation history to understand both:
- The overall conversation context (topic, relationship between participants)
- Specific examples of the user's actual writing style in practice, paying attention to:
* Tone (formal, casual, friendly, professional)
* Typical message length
* Use of emoji, slang, abbreviations, or special formatting
* Common greeting/closing patterns
* Sentence structure and vocabulary level
3. SYNTHESIS: Blend the explicit style guide with observed patterns in their messages,
but when there's any conflict, the explicit style guide takes precedence.
Generate a response that feels authentic to both how they say they write
and how they actually write, while being appropriate to the current conversation.
"""
}
return json.dumps(result, indent=2, ensure_ascii=False)
except Exception as e:
return json.dumps({
"error": str(e),
"message": "Failed to retrieve conversation context"
})
def run(self):
"""Run the MCP server"""
self.app.run(transport='stdio')
def main():
# Load .env file
load_dotenv()
# Get Telegram API credentials from .env file
api_id = os.getenv('TELEGRAM_API_ID')
api_hash = os.getenv('TELEGRAM_API_HASH')
session_string = os.getenv('TELEGRAM_SESSION_STRING')
if not api_id or not api_hash:
print("Please set TELEGRAM_API_ID and TELEGRAM_API_HASH in your .env file", file=sys.stderr)
print("Get these from https://my.telegram.org/apps", file=sys.stderr)
sys.exit(1)
if not session_string:
print("Please set TELEGRAM_SESSION_STRING in your .env file", file=sys.stderr)
print("Run: python telethon_auth.py to generate a session string", file=sys.stderr)
sys.exit(1)
try:
# Create and run server
server = TelegramServer()
server.run()
except Exception as e:
print(f"Error: {str(e)}", file=sys.stderr)
sys.exit(1)
if __name__ == "__main__":
main()
```