텔레그램 MCP 서버
🤖 MCP 작동 중
다음은 Claude 의 Telegram MCP 기능에 대한 데모입니다.
기본 사용 예:
예: Claude에게 채팅 기록을 분석하고 응답을 보내달라고 요청합니다.
그룹에 메시지를 성공적으로 보냈습니다:
보시다시피, AI는 귀하의 Telegram 계정과 원활하게 상호 작용하여 자연스러운 방식으로 귀하의 채팅, 메시지 및 기타 데이터를 검색하여 표시합니다.
Telethon 과 MCP(Model Context Protocol) 를 기반으로 Claude, Cursor 및 모든 MCP 호환 클라이언트를 위한 모든 기능을 갖춘 Telegram 통합 솔루션입니다. 이 프로젝트를 통해 Telegram 계정과 프로그래밍 방식으로 상호작용하여 메시징부터 그룹 관리까지 모든 것을 자동화할 수 있습니다.
Related MCP server: Telegram MCP Server
🚀 기능 및 도구
이 MCP 서버는 방대한 Telegram 도구 모음을 제공합니다. 모든 주요 Telegram/Telethon 기능을 도구로 사용할 수 있습니다!
채팅 및 그룹 관리
get_chats(page, page_size) : 페이지별로 정리된 채팅 목록
list_chats(chat_type, limit) : 메타데이터와 필터링을 사용하여 채팅을 나열합니다.
get_chat(chat_id) : 채팅에 대한 자세한 정보
create_group(title, user_ids) : 새로운 그룹을 생성합니다.
create_channel(제목, 정보, 메가그룹) : 채널 또는 슈퍼그룹을 생성합니다.
edit_chat_title(chat_id, title) : 채팅/그룹/채널 제목 변경
delete_chat_photo(chat_id) : 채팅/그룹/채널 사진 삭제
leave_chat(chat_id) : 그룹이나 채널에서 나가기
get_participants(chat_id) : 모든 참가자 나열
get_admins(chat_id) : 모든 관리자 나열
get_banned_users(chat_id) : 모든 차단된 사용자 나열
promote_admin(chat_id, user_id) : 사용자를 관리자로 승격합니다.
demote_admin(chat_id, user_id) : 관리자를 사용자로 강등합니다.
ban_user(chat_id, user_id) : 사용자 차단
unban_user(chat_id, user_id) : 사용자 차단 해제
get_invite_link(chat_id) : 초대 링크 받기
export_chat_invite(chat_id) : 초대 링크 내보내기
import_chat_invite(hash) : 초대 해시로 채팅에 참여
join_chat_by_link(링크) : 초대 링크로 채팅에 참여합니다.
메시징
get_messages(chat_id, page, page_size) : 페이지가 매겨진 메시지
list_messages(chat_id, limit, search_query, from_date, to_date) : 필터링된 메시지
send_message(chat_id, message) : 메시지를 보냅니다.
reply_to_message(chat_id, message_id, text) : 메시지에 답장
edit_message(chat_id, message_id, new_text) : 메시지를 편집합니다
delete_message(chat_id, message_id) : 메시지 삭제
forward_message(from_chat_id, message_id, to_chat_id) : 메시지 전달
pin_message(chat_id, message_id) : 메시지 고정
unpin_message(chat_id, message_id) : 메시지 고정 해제
mark_as_read(chat_id) : 모두 읽음으로 표시
get_message_context(chat_id, message_id, context_size) : 메시지 주변의 컨텍스트
get_history(chat_id, limit) : 전체 채팅 기록
get_pinned_messages(chat_id) : 고정된 메시지 목록
get_last_interaction(contact_id) : 연락처와의 가장 최근 메시지
연락처 관리
list_contacts() : 모든 연락처를 나열합니다
search_contacts(쿼리) : 연락처 검색
add_contact(전화번호, 이름, 성) : 연락처 추가
delete_contact(user_id) : 연락처 삭제
block_user(user_id) : 사용자 차단
unblock_user(user_id) : 사용자 차단 해제
import_contacts(연락처) : 대량 연락처 가져오기
export_contacts() : 모든 연락처를 JSON으로 내보냅니다.
get_blocked_users() : 차단된 사용자 목록
get_contact_ids() : 모든 연락처 ID를 나열합니다.
get_direct_chat_by_contact(contact_query) : 연락처와 직접 채팅을 찾습니다.
get_contact_chats(contact_id) : 연락처와의 모든 채팅을 나열합니다.
사용자 및 프로필
get_me() : 사용자 정보를 가져옵니다
update_profile(first_name, last_name, about) : 프로필을 업데이트하세요
delete_profile_photo() : 프로필 사진 삭제
get_user_photos(user_id, limit) : 사용자의 프로필 사진을 가져옵니다.
get_user_status(user_id) : 사용자의 온라인 상태를 가져옵니다.
메디아
검색 및 발견
search_public_chats(query) : 공개 채팅/채널/봇 검색
search_messages(chat_id, query, limit) : 채팅에서 메시지 검색
resolve_username(username) : 사용자 이름을 ID로 변환합니다.
스티커, GIF, 봇
get_sticker_sets() : 스티커 세트 나열
get_bot_info(bot_username) : 봇에 대한 정보를 가져옵니다.
set_bot_commands(bot_username, commands) : 봇 명령 설정(봇 계정만 해당)
개인정보 보호, 설정 및 기타
get_privacy_settings() : 개인정보 보호 설정 가져오기
set_privacy_settings(키, allow_users, disallow_users) : 개인정보 보호 설정을 지정합니다.
mute_chat(chat_id) : 알림 음소거
unmute_chat(chat_id) : 알림 음소거 해제
archive_chat(chat_id) : 채팅 보관
unarchive_chat(chat_id) : 채팅 보관 해제
get_recent_actions(chat_id) : 최근 관리자 작업 가져오기
제거된 기능
서버에서 직접 파일 경로에 접근해야 하는 도구( send_file , download_media , set_profile_photo , edit_chat_photo , send_voice , send_sticker , upload_file )가 main.py 에서 제거되었습니다. 이는 현재 MCP 환경에서 파일 첨부 및 로컬 파일 시스템 경로 처리에 대한 제한으로 인해 발생합니다.
또한, Telethon 라이브러리 또는 Telegram API 상호 작용의 안정성 문제로 인해 GIF 관련 도구( get_gif_search , get_saved_gifs , send_gif )가 제거되었습니다.
📋 요구 사항
🔧 설치 및 설정
1. 포크 및 복제
지엑스피1
2. 가상 환경 만들기
python3 -m venv .venv
source .venv/bin/activate # On Windows: .venv\Scripts\activate
pip install -r requirements.txt
3. 세션 문자열 생성
python3 session_string_generator.py
메시지에 따라 인증을 받고 .env 파일을 업데이트하세요.
4. .env 구성
.env.example .env 로 복사하고 값을 입력합니다.
TELEGRAM_API_ID=your_api_id_here
TELEGRAM_API_HASH=your_api_hash_here
TELEGRAM_SESSION_NAME=anon
TELEGRAM_SESSION_STRING=your_session_string_here
my.telegram.org/apps 에서 API 자격 증명을 받으세요.
🐳 Docker로 실행하기
Docker와 Docker Compose가 설치되어 있다면 컨테이너에서 서버를 빌드하고 실행하여 종속성 관리를 간소화할 수 있습니다.
1. 이미지 구축
프로젝트 루트 디렉토리에서 Docker 이미지를 빌드합니다.
docker build -t telegram-mcp:latest .
2. 컨테이너 실행
두 가지 옵션이 있습니다.
옵션 A: Docker Compose 사용(로컬 사용에 권장)
이 방법은 docker-compose.yml 파일을 사용하고 .env 파일에서 자동으로 자격 증명을 읽습니다.
.env 프로젝트 루트에 TELEGRAM_API_ID , TELEGRAM_API_HASH , TELEGRAM_SESSION_STRING (또는 TELEGRAM_SESSION_NAME )이 포함된 .env 파일이 있는지 확인하세요. .env.example 템플릿으로 사용하세요.
Compose 실행:
docker compose up --build
옵션 B:
자격 증명을 환경 변수로 전달하여 컨테이너를 직접 실행할 수 있습니다.
docker run -it --rm \
-e TELEGRAM_API_ID="YOUR_API_ID" \
-e TELEGRAM_API_HASH="YOUR_API_HASH" \
-e TELEGRAM_SESSION_STRING="YOUR_SESSION_STRING" \
telegram-mcp:latest
플레이스홀더를 실제 자격 증명으로 바꾸세요.
파일 기반 세션을 선호하는 경우 TELEGRAM_SESSION_STRING 대신 -e TELEGRAM_SESSION_NAME=your_session_file_name 사용하세요(볼륨 마운트가 필요함, 예시는 docker-compose.yml 참조).
-it 플래그는 서버와 상호 작용하는 데 중요합니다.
⚙️ Claude 및 커서 구성
MCP 구성
Claude 데스크톱 구성(예: ~/Library/Application Support/Claude/claude_desktop_config.json ) 또는 커서 구성( ~/.cursor/mcp.json )을 편집하세요.
{
"mcpServers": {
"telegram-mcp": {
"command": "uv",
"args": [
"--directory",
"/full/path/to/telegram-mcp",
"run",
"main.py"
]
}
}
}
📝 코드 및 출력을 포함한 도구 예제
다음은 가장 일반적으로 사용되는 도구와 그 구현 및 샘플 출력의 예입니다.
채팅 받기
@mcp.tool()
async def get_chats(page: int = 1, page_size: int = 20) -> str:
"""
Get a paginated list of chats.
Args:
page: Page number (1-indexed).
page_size: Number of chats per page.
"""
try:
dialogs = await client.get_dialogs()
start = (page - 1) * page_size
end = start + page_size
if start >= len(dialogs):
return "Page out of range."
chats = dialogs[start:end]
lines = []
for dialog in chats:
entity = dialog.entity
chat_id = entity.id
title = getattr(entity, "title", None) or getattr(entity, "first_name", "Unknown")
lines.append(f"Chat ID: {chat_id}, Title: {title}")
return "\n".join(lines)
except Exception as e:
logger.exception(f"get_chats failed (page={page}, page_size={page_size})")
return "An error occurred (code: GETCHATS-ERR-001). Check mcp_errors.log for details."
출력 예:
Chat ID: 123456789, Title: John Doe
Chat ID: -100987654321, Title: My Project Group
Chat ID: 111223344, Title: Jane Smith
Chat ID: -200123456789, Title: News Channel
메시지 보내기
@mcp.tool()
async def send_message(chat_id: int, message: str) -> str:
"""
Send a message to a specific chat.
Args:
chat_id: The ID of the chat.
message: The message content to send.
"""
try:
entity = await client.get_entity(chat_id)
await client.send_message(entity, message)
return "Message sent successfully."
except Exception as e:
logger.exception(f"send_message failed (chat_id={chat_id})")
return "An error occurred (code: SENDMSG-ERR-001). Check mcp_errors.log for details."
출력 예:
Message sent successfully.
채팅 초대 링크 받기
get_invite_link 함수는 특히 여러 대체 방법을 사용할 수 있는 강력한 함수입니다.
@mcp.tool()
async def get_invite_link(chat_id: int) -> str:
"""
Get the invite link for a group or channel.
"""
try:
entity = await client.get_entity(chat_id)
# Try using ExportChatInviteRequest first
try:
from telethon.tl import functions
result = await client(functions.messages.ExportChatInviteRequest(
peer=entity
))
return result.link
except AttributeError:
# If the function doesn't exist in the current Telethon version
logger.warning("ExportChatInviteRequest not available, using alternative method")
except Exception as e1:
# If that fails, log and try alternative approach
logger.warning(f"ExportChatInviteRequest failed: {e1}")
# Alternative approach using client.export_chat_invite_link
try:
invite_link = await client.export_chat_invite_link(entity)
return invite_link
except Exception as e2:
logger.warning(f"export_chat_invite_link failed: {e2}")
# Last resort: Try directly fetching chat info
try:
if isinstance(entity, (Chat, Channel)):
full_chat = await client(functions.messages.GetFullChatRequest(
chat_id=entity.id
))
if hasattr(full_chat, 'full_chat') and hasattr(full_chat.full_chat, 'invite_link'):
return full_chat.full_chat.invite_link or "No invite link available."
except Exception as e3:
logger.warning(f"GetFullChatRequest failed: {e3}")
return "Could not retrieve invite link for this chat."
except Exception as e:
logger.exception(f"get_invite_link failed (chat_id={chat_id})")
return f"Error getting invite link: {e}"
출력 예:
https://t.me/+AbCdEfGhIjKlMnOp
초대 링크를 통해 채팅에 참여하기
@mcp.tool()
async def join_chat_by_link(link: str) -> str:
"""
Join a chat by invite link.
"""
try:
# Extract the hash from the invite link
if '/' in link:
hash_part = link.split('/')[-1]
if hash_part.startswith('+'):
hash_part = hash_part[1:] # Remove the '+' if present
else:
hash_part = link
# Try checking the invite before joining
try:
# Try to check invite info first (will often fail if not a member)
invite_info = await client(functions.messages.CheckChatInviteRequest(hash=hash_part))
if hasattr(invite_info, 'chat') and invite_info.chat:
# If we got chat info, we're already a member
chat_title = getattr(invite_info.chat, 'title', 'Unknown Chat')
return f"You are already a member of this chat: {chat_title}"
except Exception:
# This often fails if not a member - just continue
pass
# Join the chat using the hash
result = await client(functions.messages.ImportChatInviteRequest(hash=hash_part))
if result and hasattr(result, 'chats') and result.chats:
chat_title = getattr(result.chats[0], 'title', 'Unknown Chat')
return f"Successfully joined chat: {chat_title}"
return f"Joined chat via invite hash."
except Exception as e:
err_str = str(e).lower()
if "expired" in err_str:
return "The invite hash has expired and is no longer valid."
elif "invalid" in err_str:
return "The invite hash is invalid or malformed."
elif "already" in err_str and "participant" in err_str:
return "You are already a member of this chat."
logger.exception(f"join_chat_by_link failed (link={link})")
return f"Error joining chat: {e}"
출력 예:
Successfully joined chat: Developer Community
공개 채팅 검색
@mcp.tool()
async def search_public_chats(query: str) -> str:
"""
Search for public chats, channels, or bots by username or title.
"""
try:
result = await client(functions.contacts.SearchRequest(q=query, limit=20))
return json.dumps([format_entity(u) for u in result.users], indent=2)
except Exception as e:
return f"Error searching public chats: {e}"
출력 예:
[
{
"id": 123456789,
"name": "TelegramBot",
"type": "user",
"username": "telegram_bot"
},
{
"id": 987654321,
"name": "Telegram News",
"type": "user",
"username": "telegram_news"
}
]
연락처와 직접 채팅하기
@mcp.tool()
async def get_direct_chat_by_contact(contact_query: str) -> str:
"""
Find a direct chat with a specific contact by name, username, or phone.
Args:
contact_query: Name, username, or phone number to search for.
"""
try:
# Fetch all contacts using the correct Telethon method
result = await client(functions.contacts.GetContactsRequest(hash=0))
contacts = result.users
found_contacts = []
for contact in contacts:
if not contact:
continue
name = f"{getattr(contact, 'first_name', '')} {getattr(contact, 'last_name', '')}".strip()
username = getattr(contact, 'username', '')
phone = getattr(contact, 'phone', '')
if (contact_query.lower() in name.lower() or
(username and contact_query.lower() in username.lower()) or
(phone and contact_query in phone)):
found_contacts.append(contact)
if not found_contacts:
return f"No contacts found matching '{contact_query}'."
# If we found contacts, look for direct chats with them
results = []
dialogs = await client.get_dialogs()
for contact in found_contacts:
contact_name = f"{getattr(contact, 'first_name', '')} {getattr(contact, 'last_name', '')}".strip()
for dialog in dialogs:
if isinstance(dialog.entity, User) and dialog.entity.id == contact.id:
chat_info = f"Chat ID: {dialog.entity.id}, Contact: {contact_name}"
if getattr(contact, 'username', ''):
chat_info += f", Username: @{contact.username}"
if dialog.unread_count:
chat_info += f", Unread: {dialog.unread_count}"
results.append(chat_info)
break
if not results:
return f"Found contacts matching '{contact_query}', but no direct chats with them."
return "\n".join(results)
except Exception as e:
return f"Error searching for direct chat: {e}"
출력 예:
Chat ID: 123456789, Contact: John Smith, Username: @johnsmith, Unread: 3
🎮 사용 예시
"최근 채팅 보기"
"채팅 123456789에 'Hello world'를 보내세요"
"전화번호 +1234567890, 이름 John Doe를 연락처에 추가하세요"
"사용자 111, 222, 333으로 '프로젝트 팀' 그룹을 만드세요"
"채팅 123456789의 메시지 42에서 미디어를 다운로드하세요"
"채팅 123456789에 대한 알림 음소거"
"사용자 111을 그룹 123456789의 관리자로 승격"
"뉴스"에 대한 공개 채널을 검색하세요
"초대 링크가 있는 Telegram 그룹에 가입하세요 https://t.me/+AbCdEfGhIjK "
"저장된 메시지에 스티커 보내기"
"내 스티커 세트를 모두 가져와"
Claude, Cursor 또는 MCP 호환 클라이언트에서 자연어로 이러한 도구를 사용할 수 있습니다.
🧠 오류 처리 및 견고성
이 구현에는 포괄적인 오류 처리가 포함됩니다.
세션 관리 : 파일 기반 세션과 문자열 기반 세션 모두에서 작동합니다.
오류 보고 : mcp_errors.log 에 자세한 오류가 기록됨
우아한 저하 : 중요 기능에 대한 다양한 대체 접근 방식
사용자 친화적인 메시지 : 기술적 오류 대신 명확하고 실행 가능한 오류 메시지
계정 유형 감지 : 봇 계정이 필요한 기능은 사용자 계정과 함께 사용될 때 감지하고 알림을 제공합니다.
초대 링크 처리 : 다양한 링크 형식 및 기존 회원 사례 처리
이 코드는 일반적인 Telegram API 문제와 제한 사항에도 견고하게 설계되었습니다.
🛠️ 기여 가이드
이 저장소를 포크하세요: chigwell/telegram-mcp
포크를 복제하세요:
git clone https://github.com/<your-github-username>/telegram-mcp.git
새로운 지점을 만드세요:
git checkout -b my-feature
변경 사항을 적용하고 필요한 경우 테스트/문서를 추가하세요.
chigwell/telegram-mcp 에 명확한 설명과 함께 풀 리퀘스트를 푸시하고 엽니다 .
검토를 위해 PR에 @chigwell 또는 @l1v0n1을 태그하세요 .
🔒 보안 고려 사항
.env
세션 문자열을 통해 Telegram 계정에 대한 전체 액세스 권한이 부여됩니다. 안전하게 보관하세요!
모든 처리는 로컬에서 이루어집니다. 즉, Telegram API 외에는 어디에도 데이터가 전송되지 않습니다.
.env.example 템플릿으로 사용하고 실제 .env 파일은 비공개로 유지하세요.
테스트 파일은 .gitignore 에서 자동으로 제외됩니다.
🛠️ 문제 해결
MCP 클라이언트(Claude/Cursor)와 터미널의 로그를 확인하여 오류를 확인하세요.
자세한 오류 로그는 mcp_errors.log 에서 확인할 수 있습니다.
인터프리터 오류가 있나요? .venv 생성되어 선택되었는지 확인하세요.
데이터베이스 잠금? 파일 기반 세션이 아닌 세션 문자열 인증을 사용하세요.
iCloud/Dropbox 문제? 이상한 오류가 표시되면 프로젝트를 공백 없는 로컬 경로로 옮기세요.
Telegram 비밀번호를 변경하거나 인증 오류가 발생하는 경우 세션 문자열을 다시 생성합니다 .
봇 전용 기능은 일반 사용자 계정에서 사용하면 명확한 메시지를 표시합니다.
테스트 스크립트에 오류가 발생했습니까? .env 의 테스트 구성을 확인하여 유효한 테스트 계정/그룹이 있는지 확인하세요.
📄 라이센스
이 프로젝트는 Apache 2.0 라이선스 에 따라 라이선스가 부여되었습니다.
🙏 감사의 말
@chigwell
스타 역사
