텔레그램 MCP 서버
🤖 MCP 작동 중
다음은 Claude에서 Telegram MCP 기능을 보여주는 데모입니다.
기본 사용 예:
- 예: Claude에게 채팅 기록을 분석하고 응답을 보내달라고 요청합니다.
- 그룹에 메시지를 성공적으로 보냈습니다:
보시다시피, AI는 귀하의 Telegram 계정과 원활하게 상호 작용하여 자연스러운 방식으로 귀하의 채팅, 메시지 및 기타 데이터를 검색하여 표시합니다.
Telethon 과 MCP(Model Context Protocol) 를 기반으로 Claude, Cursor 및 모든 MCP 호환 클라이언트를 위한 모든 기능을 갖춘 Telegram 통합 솔루션입니다. 이 프로젝트를 통해 Telegram 계정과 프로그래밍 방식으로 상호작용하여 메시징부터 그룹 관리까지 모든 것을 자동화할 수 있습니다.
🚀 기능 및 도구
이 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) : 채팅/그룹/채널 제목 변경
- edit_chat_photo(chat_id, file_path) : 채팅/그룹/채널 사진 설정
- 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) : 고정된 메시지 목록
연락처 관리
- 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_last_interaction(contact_id) : 연락처와의 가장 최근 메시지
사용자 및 프로필
- get_me() : 사용자 정보를 가져옵니다
- update_profile(first_name, last_name, about) : 프로필을 업데이트하세요
- set_profile_photo(file_path) : 프로필 사진을 설정합니다.
- delete_profile_photo() : 프로필 사진 삭제
- get_user_photos(user_id, limit) : 사용자의 프로필 사진을 가져옵니다.
- get_user_status(user_id) : 사용자의 온라인 상태를 가져옵니다.
- send_file(chat_id, file_path, caption) : 파일 전송
- send_voice(chat_id, file_path) : 음성 메시지 보내기
- download_media(chat_id, message_id, file_path) : 미디어 다운로드
- upload_file(file_path) : Telegram 서버에 파일을 업로드합니다.
- get_media_info(chat_id, message_id) : 메시지의 미디어에 대한 정보를 가져옵니다.
검색 및 발견
- search_public_chats(query) : 공개 채팅/채널/봇 검색
- search_messages(chat_id, query, limit) : 채팅에서 메시지 검색
- resolve_username(username) : 사용자 이름을 ID로 변환합니다.
스티커, GIF, 봇
- get_sticker_sets() : 스티커 세트 나열
- send_sticker(chat_id, file_path) : 스티커 보내기
- get_gif_search(query, limit) : GIF 검색
- send_gif(chat_id, gif_id) : GIF 보내기
- 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) : 최근 관리자 작업 가져오기
📋 요구 사항
🔧 설치 및 설정
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 자격 증명을 받으세요.
⚙️ Claude 및 Cursor 구성
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-server",
"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 과 @l1v0n1 이 관리합니다. PR 환영!
스타 역사