mikl/telegram-mcp
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Compare commits

base: mikl:e96a6a8ca20b299051b7d3c48a75c6583d6bd465
Branches Tags
mikl:main
..
compare: mikl:2ee39ee069b438e505153d85384e00276bdf1299
Branches Tags
mikl:main

10 commits
e96a6a8ca2 ... 2ee39ee069

Author SHA1 Message Date
michaotic
2ee39ee069 Disable write operations for read-only mode
Some checks failed
Docker Build & Compose Validation / Build Docker Image (push) Has been cancelled
Details
Python Lint & Format Check / Check Linting & Formatting (push) Has been cancelled
Details
Docker Build & Compose Validation / Build & Validate Docker Compose (push) Has been cancelled
Details 
- Commented out all @mcp.tool decorators without readOnlyHint=True
- 47 write tools disabled (send_message, create_group, ban_user, etc.)
- 47 read-only tools remain active (get_chats, get_messages, search_messages, etc.)
- Created Python script to automate disabling process
- Now completely read-only - no accidental message sending possible
 2026-04-12 03:22:25 +03:00
Eugene Evstafev
ef01a39159
Merge pull request #97 from bunkerskyi/feat/send-contact 
feat(tools): add send_contact tool to send phone contacts via Telegram
 2026-04-09 09:33:26 +01:00
Eugene Evstafev
589fa53c7d
Merge pull request #95 from bunkerskyi/fix/inline-buttons-reply-markup-fallback 
fix(tools): fall back to reply_markup when buttons property is None for bot messages
 2026-04-09 09:31:03 +01:00
Eugene Evstafev
62d5be5ed5
Merge pull request #96 from bunkerskyi/fix/get-messages-json-output 
fix(tools): escape newlines in get_messages output to preserve line-based format
 2026-04-09 09:28:54 +01:00
Ivan Kolodrivskyi
e2ca3c78fe style: apply black formatting to send_contact 2026-04-09 09:23:14 +03:00
Ivan Kolodrivskyi
b2c3a865d8 style: apply black formatting to list_inline_buttons 2026-04-09 09:20:43 +03:00
Ivan Kolodrivskyi
cf6888e000 fix: fall back to reply_markup when buttons property is None for bot messages 
Telethon's client.get_messages(entity, ids=N) sometimes returns bot messages
with buttons=None and reply_markup=None, even when the message has inline
keyboard buttons. The same message fetched via history (limit=N) includes
the full ReplyInlineMarkup.

Changes:
- Add fallback: when message fetched by ID has no inline buttons, re-fetch
  via history and match by ID
- Check both msg.buttons and msg.reply_markup.rows when searching for
  messages with buttons
- Access .data/.url/.text directly on button objects (works for both
  MessageButton wrappers and raw KeyboardButtonCallback objects)
 2026-04-09 08:35:20 +03:00
Ivan Kolodrivskyi
b38051a9ee fix: escape newlines in get_messages instead of switching to JSON 
Replace the JSON output approach with a non-breaking fix: escape newlines
in message text (\n → \\n) to preserve the existing pipe-delimited format.
This fixes multi-line message parsing without breaking downstream consumers.
 2026-04-09 08:34:50 +03:00
Ivan Kolodrivskyi
ff37abdb01 fix: return JSON from get_messages to preserve multi-line message text 
The pipe-delimited text format (ID: N | ... | Message: <text>) breaks when
messages contain newlines, since \n serves as both record separator and can
appear inside message text. Downstream parsers that split on \n only capture
the first line

Switch to a JSON array output with fields: id, sender, date, text,
reply_to (optional), and engagement (optional, with views/forwards/reactions)

This is a breaking change for MCP clients that parse the old text format.
Since MCP tool results are typically consumed by LLMs which handle JSON
natively, this should be transparent for most consumers
 2026-04-09 08:34:38 +03:00
Ivan Kolodrivskyi
0185e2e36b feat: add send_contact tool to send phone contacts via MCP 2026-04-08 19:29:26 +03:00
1 changed files with 290 additions and 213 deletions
Show stats Expand all files Collapse all files

503
main.py
View file

@ -735,9 +735,10 @@ async def get_messages(chat_id: Union[int, str], page: int = 1, page_size: int =
reply_info = f" | reply to {msg.reply_to.reply_to_msg_id}" reply_info = f" | reply to {msg.reply_to.reply_to_msg_id}"
engagement_info = get_engagement_info(msg) engagement_info = get_engagement_info(msg)
safe_text = (msg.message or "").replace("\n", "\\n")
lines.append( lines.append(
f"ID: {msg.id} | {sender_name} | Date: {msg.date}{reply_info}{engagement_info} | Message: {msg.message}" f"ID: {msg.id} | {sender_name} | Date: {msg.date}{reply_info}{engagement_info} | Message: {safe_text}"
) )
return "\n".join(lines) return "\n".join(lines)
except Exception as e: except Exception as e:
@ -746,8 +747,8 @@ async def get_messages(chat_id: Union[int, str], page: int = 1, page_size: int =
) )
@mcp.tool( #@mcp.tool(
annotations=ToolAnnotations(title="Send Message", openWorldHint=True, destructiveHint=True) # annotations=ToolAnnotations(title="Send Message", openWorldHint=True, destructiveHint=True)
) )
@validate_id("chat_id") @validate_id("chat_id")
async def send_message( async def send_message(
@ -770,13 +771,13 @@ async def send_message(
return log_and_format_error("send_message", e, chat_id=chat_id) return log_and_format_error("send_message", e, chat_id=chat_id)
@mcp.tool( #@mcp.tool(
annotations=ToolAnnotations( # annotations=ToolAnnotations(
title="Subscribe Public Channel", # title="Subscribe Public Channel",
openWorldHint=True, # openWorldHint=True,
destructiveHint=True, # destructiveHint=True,
idempotentHint=True, # idempotentHint=True,
) # )
) )
@validate_id("channel") @validate_id("channel")
async def subscribe_public_channel(channel: Union[int, str]) -> str: async def subscribe_public_channel(channel: Union[int, str]) -> str:
@ -815,6 +816,22 @@ async def list_inline_buttons(
return "message_id must be an integer." return "message_id must be an integer."
entity = await resolve_entity(chat_id) entity = await resolve_entity(chat_id)
def _has_inline(msg):
if getattr(msg, "buttons", None):
return True
rm = getattr(msg, "reply_markup", None)
return bool(rm and hasattr(rm, "rows"))
def _flat_buttons(msg):
btns = getattr(msg, "buttons", None)
if btns:
return [btn for row in btns for btn in row]
rm = getattr(msg, "reply_markup", None)
if rm and hasattr(rm, "rows"):
return [btn for row in rm.rows for btn in row.buttons]
return []
target_message = None target_message = None
if message_id is not None: if message_id is not None:
@ -823,18 +840,12 @@ async def list_inline_buttons(
target_message = target_message[0] if target_message else None target_message = target_message[0] if target_message else None
else: else:
recent_messages = await client.get_messages(entity, limit=limit) recent_messages = await client.get_messages(entity, limit=limit)
target_message = next( target_message = next((msg for msg in recent_messages if _has_inline(msg)), None)
(msg for msg in recent_messages if getattr(msg, "buttons", None)), None
)
if not target_message: if not target_message:
return "No message with inline buttons found." return "No message with inline buttons found."
buttons_attr = getattr(target_message, "buttons", None) buttons = _flat_buttons(target_message)
if not buttons_attr:
return f"Message {target_message.id} does not contain inline buttons."
buttons = [btn for row in buttons_attr for btn in row]
if not buttons: if not buttons:
return f"Message {target_message.id} does not contain inline buttons." return f"Message {target_message.id} does not contain inline buttons."
@ -842,9 +853,8 @@ async def list_inline_buttons(
f"Buttons for message {target_message.id} (date {target_message.date}):", f"Buttons for message {target_message.id} (date {target_message.date}):",
] ]
for idx, btn in enumerate(buttons): for idx, btn in enumerate(buttons):
raw_button = getattr(btn, "button", None)
text = getattr(btn, "text", "") or "<no text>" text = getattr(btn, "text", "") or "<no text>"
url = getattr(raw_button, "url", None) if raw_button else None url = getattr(btn, "url", None)
has_callback = bool(getattr(btn, "data", None)) has_callback = bool(getattr(btn, "data", None))
parts = [f"[{idx}] text='{text}'"] parts = [f"[{idx}] text='{text}'"]
parts.append("callback=yes" if has_callback else "callback=no") parts.append("callback=yes" if has_callback else "callback=no")
@ -863,10 +873,10 @@ async def list_inline_buttons(
) )
@mcp.tool( #@mcp.tool(
annotations=ToolAnnotations( # annotations=ToolAnnotations(
title="Press Inline Button", openWorldHint=True, destructiveHint=True # title="Press Inline Button", openWorldHint=True, destructiveHint=True
) # )
) )
@validate_id("chat_id") @validate_id("chat_id")
async def press_inline_button( async def press_inline_button(
@ -903,25 +913,49 @@ async def press_inline_button(
entity = await resolve_entity(chat_id) entity = await resolve_entity(chat_id)
def _has_inline_buttons(msg):
"""Check if a message has inline buttons via buttons property or reply_markup."""
if getattr(msg, "buttons", None):
return True
rm = getattr(msg, "reply_markup", None)
return bool(rm and hasattr(rm, "rows"))
def _extract_buttons(msg):
"""Extract flat list of buttons from buttons property or reply_markup fallback."""
btns = getattr(msg, "buttons", None)
if btns:
return [btn for row in btns for btn in row]
rm = getattr(msg, "reply_markup", None)
if rm and hasattr(rm, "rows"):
return [btn for row in rm.rows for btn in row.buttons]
return []
target_message = None target_message = None
if message_id is not None: if message_id is not None:
# Fetch by ID first, then fall back to recent-message search if
# reply_markup is missing (Telethon sometimes omits it for ID fetches).
target_message = await client.get_messages(entity, ids=message_id) target_message = await client.get_messages(entity, ids=message_id)
if isinstance(target_message, list): if isinstance(target_message, list):
target_message = target_message[0] if target_message else None target_message = target_message[0] if target_message else None
if target_message and not _has_inline_buttons(target_message):
# Fallback: search recent messages for the same ID with markup
recent = await client.get_messages(entity, limit=30)
fallback = next(
(m for m in recent if m.id == target_message.id and _has_inline_buttons(m)),
None,
)
if fallback:
target_message = fallback
else: else:
recent_messages = await client.get_messages(entity, limit=20) recent_messages = await client.get_messages(entity, limit=20)
target_message = next( target_message = next(
(msg for msg in recent_messages if getattr(msg, "buttons", None)), None (msg for msg in recent_messages if _has_inline_buttons(msg)), None
) )
if not target_message: if not target_message:
return "No message with inline buttons found. Specify message_id to target a specific message." return "No message with inline buttons found. Specify message_id to target a specific message."
buttons_attr = getattr(target_message, "buttons", None) buttons = _extract_buttons(target_message)
if not buttons_attr:
return f"Message {target_message.id} does not contain inline buttons."
buttons = [btn for row in buttons_attr for btn in row]
if not buttons: if not buttons:
return f"Message {target_message.id} does not contain inline buttons." return f"Message {target_message.id} does not contain inline buttons."
@ -949,16 +983,16 @@ async def press_inline_button(
) )
return f"Button not found. Available buttons: {available}" return f"Button not found. Available buttons: {available}"
if not getattr(target_button, "data", None): btn_data = getattr(target_button, "data", None)
raw_button = getattr(target_button, "button", None) if not btn_data:
url = getattr(raw_button, "url", None) if raw_button else None url = getattr(target_button, "url", None)
if url: if url:
return f"Selected button opens a URL instead of sending a callback: {url}" return f"Selected button opens a URL instead of sending a callback: {url}"
return "Selected button does not provide callback data to press." return "Selected button does not provide callback data to press."
callback_result = await client( callback_result = await client(
functions.messages.GetBotCallbackAnswerRequest( functions.messages.GetBotCallbackAnswerRequest(
peer=entity, msg_id=target_message.id, data=target_button.data peer=entity, msg_id=target_message.id, data=btn_data
) )
) )
@ -1665,10 +1699,10 @@ async def get_message_context(
) )
@mcp.tool( #@mcp.tool(
annotations=ToolAnnotations( # annotations=ToolAnnotations(
title="Add Contact", openWorldHint=True, destructiveHint=True, idempotentHint=True # title="Add Contact", openWorldHint=True, destructiveHint=True, idempotentHint=True
) # )
) )
async def add_contact( async def add_contact(
phone: Optional[str] = None, phone: Optional[str] = None,
@ -1798,10 +1832,10 @@ async def add_contact(
return log_and_format_error("add_contact", e, phone=phone, username=username) return log_and_format_error("add_contact", e, phone=phone, username=username)
@mcp.tool( #@mcp.tool(
annotations=ToolAnnotations( # annotations=ToolAnnotations(
title="Delete Contact", openWorldHint=True, destructiveHint=True, idempotentHint=True # title="Delete Contact", openWorldHint=True, destructiveHint=True, idempotentHint=True
) # )
) )
@validate_id("user_id") @validate_id("user_id")
async def delete_contact(user_id: Union[int, str]) -> str: async def delete_contact(user_id: Union[int, str]) -> str:
@ -1818,10 +1852,10 @@ async def delete_contact(user_id: Union[int, str]) -> str:
return log_and_format_error("delete_contact", e, user_id=user_id) return log_and_format_error("delete_contact", e, user_id=user_id)
@mcp.tool( #@mcp.tool(
annotations=ToolAnnotations( # annotations=ToolAnnotations(
title="Block User", openWorldHint=True, destructiveHint=True, idempotentHint=True # title="Block User", openWorldHint=True, destructiveHint=True, idempotentHint=True
) # )
) )
@validate_id("user_id") @validate_id("user_id")
async def block_user(user_id: Union[int, str]) -> str: async def block_user(user_id: Union[int, str]) -> str:
@ -1838,10 +1872,10 @@ async def block_user(user_id: Union[int, str]) -> str:
return log_and_format_error("block_user", e, user_id=user_id) return log_and_format_error("block_user", e, user_id=user_id)
@mcp.tool( #@mcp.tool(
annotations=ToolAnnotations( # annotations=ToolAnnotations(
title="Unblock User", openWorldHint=True, destructiveHint=True, idempotentHint=True # title="Unblock User", openWorldHint=True, destructiveHint=True, idempotentHint=True
) # )
) )
@validate_id("user_id") @validate_id("user_id")
async def unblock_user(user_id: Union[int, str]) -> str: async def unblock_user(user_id: Union[int, str]) -> str:
@ -1870,8 +1904,8 @@ async def get_me() -> str:
return log_and_format_error("get_me", e) return log_and_format_error("get_me", e)
@mcp.tool( #@mcp.tool(
annotations=ToolAnnotations(title="Create Group", openWorldHint=True, destructiveHint=True) # annotations=ToolAnnotations(title="Create Group", openWorldHint=True, destructiveHint=True)
) )
@validate_id("user_ids") @validate_id("user_ids")
async def create_group(title: str, user_ids: List[Union[int, str]]) -> str: async def create_group(title: str, user_ids: List[Union[int, str]]) -> str:
@ -1931,10 +1965,10 @@ async def create_group(title: str, user_ids: List[Union[int, str]]) -> str:
return log_and_format_error("create_group", e, title=title, user_ids=user_ids) return log_and_format_error("create_group", e, title=title, user_ids=user_ids)
@mcp.tool( #@mcp.tool(
annotations=ToolAnnotations( # annotations=ToolAnnotations(
title="Invite To Group", openWorldHint=True, destructiveHint=True, idempotentHint=True # title="Invite To Group", openWorldHint=True, destructiveHint=True, idempotentHint=True
) # )
) )
@validate_id("group_id", "user_ids") @validate_id("group_id", "user_ids")
async def invite_to_group(group_id: Union[int, str], user_ids: List[Union[int, str]]) -> str: async def invite_to_group(group_id: Union[int, str], user_ids: List[Union[int, str]]) -> str:
@ -1985,10 +2019,10 @@ async def invite_to_group(group_id: Union[int, str], user_ids: List[Union[int, s
return log_and_format_error("invite_to_group", e, group_id=group_id, user_ids=user_ids) return log_and_format_error("invite_to_group", e, group_id=group_id, user_ids=user_ids)
@mcp.tool( #@mcp.tool(
annotations=ToolAnnotations( # annotations=ToolAnnotations(
title="Leave Chat", openWorldHint=True, destructiveHint=True, idempotentHint=True # title="Leave Chat", openWorldHint=True, destructiveHint=True, idempotentHint=True
) # )
) )
@validate_id("chat_id") @validate_id("chat_id")
async def leave_chat(chat_id: Union[int, str]) -> str: async def leave_chat(chat_id: Union[int, str]) -> str:
@ -2091,8 +2125,8 @@ async def get_participants(chat_id: Union[int, str]) -> str:
return log_and_format_error("get_participants", e, chat_id=chat_id) return log_and_format_error("get_participants", e, chat_id=chat_id)
@mcp.tool(annotations=ToolAnnotations(title="Send File", openWorldHint=True, destructiveHint=True)) #@mcp.tool(annotations=ToolAnnotations(title="Send File", openWorldHint=True, destructiveHint=True))
@validate_id("chat_id") #@validate_id("chat_id")
async def send_file( async def send_file(
chat_id: Union[int, str], chat_id: Union[int, str],
file_path: str, file_path: str,
@ -2123,8 +2157,8 @@ async def send_file(
) )
@mcp.tool( #@mcp.tool(
annotations=ToolAnnotations(title="Download Media", openWorldHint=True, destructiveHint=True) # annotations=ToolAnnotations(title="Download Media", openWorldHint=True, destructiveHint=True)
) )
@validate_id("chat_id") @validate_id("chat_id")
async def download_media( async def download_media(
@ -2184,10 +2218,10 @@ async def download_media(
) )
@mcp.tool( #@mcp.tool(
annotations=ToolAnnotations( # annotations=ToolAnnotations(
title="Update Profile", openWorldHint=True, destructiveHint=True, idempotentHint=True # title="Update Profile", openWorldHint=True, destructiveHint=True, idempotentHint=True
) # )
) )
async def update_profile(first_name: str = None, last_name: str = None, about: str = None) -> str: async def update_profile(first_name: str = None, last_name: str = None, about: str = None) -> str:
""" """
@ -2206,10 +2240,10 @@ async def update_profile(first_name: str = None, last_name: str = None, about: s
) )
@mcp.tool( #@mcp.tool(
annotations=ToolAnnotations( # annotations=ToolAnnotations(
title="Set Profile Photo", openWorldHint=True, destructiveHint=True, idempotentHint=True # title="Set Profile Photo", openWorldHint=True, destructiveHint=True, idempotentHint=True
) # )
) )
async def set_profile_photo(file_path: str, ctx: Optional[Context] = None) -> str: async def set_profile_photo(file_path: str, ctx: Optional[Context] = None) -> str:
""" """
@ -2233,10 +2267,10 @@ async def set_profile_photo(file_path: str, ctx: Optional[Context] = None) -> st
return log_and_format_error("set_profile_photo", e, file_path=file_path) return log_and_format_error("set_profile_photo", e, file_path=file_path)
@mcp.tool( #@mcp.tool(
annotations=ToolAnnotations( # annotations=ToolAnnotations(
title="Delete Profile Photo", openWorldHint=True, destructiveHint=True, idempotentHint=True # title="Delete Profile Photo", openWorldHint=True, destructiveHint=True, idempotentHint=True
) # )
) )
async def delete_profile_photo() -> str: async def delete_profile_photo() -> str:
""" """
@ -2282,10 +2316,10 @@ async def get_privacy_settings() -> str:
return log_and_format_error("get_privacy_settings", e) return log_and_format_error("get_privacy_settings", e)
@mcp.tool( #@mcp.tool(
annotations=ToolAnnotations( # annotations=ToolAnnotations(
title="Set Privacy Settings", openWorldHint=True, destructiveHint=True, idempotentHint=True # title="Set Privacy Settings", openWorldHint=True, destructiveHint=True, idempotentHint=True
) # )
) )
@validate_id("allow_users", "disallow_users") @validate_id("allow_users", "disallow_users")
async def set_privacy_settings( async def set_privacy_settings(
@ -2383,8 +2417,8 @@ async def set_privacy_settings(
return log_and_format_error("set_privacy_settings", e, key=key) return log_and_format_error("set_privacy_settings", e, key=key)
@mcp.tool( #@mcp.tool(
annotations=ToolAnnotations(title="Import Contacts", openWorldHint=True, destructiveHint=True) # annotations=ToolAnnotations(title="Import Contacts", openWorldHint=True, destructiveHint=True)
) )
async def import_contacts(contacts: list) -> str: async def import_contacts(contacts: list) -> str:
""" """
@ -2435,8 +2469,8 @@ async def get_blocked_users() -> str:
return log_and_format_error("get_blocked_users", e) return log_and_format_error("get_blocked_users", e)
@mcp.tool( #@mcp.tool(
annotations=ToolAnnotations(title="Create Channel", openWorldHint=True, destructiveHint=True) # annotations=ToolAnnotations(title="Create Channel", openWorldHint=True, destructiveHint=True)
) )
async def create_channel(title: str, about: str = "", megagroup: bool = False) -> str: async def create_channel(title: str, about: str = "", megagroup: bool = False) -> str:
""" """
@ -2453,10 +2487,10 @@ async def create_channel(title: str, about: str = "", megagroup: bool = False) -
) )
@mcp.tool( #@mcp.tool(
annotations=ToolAnnotations( # annotations=ToolAnnotations(
title="Edit Chat Title", openWorldHint=True, destructiveHint=True, idempotentHint=True # title="Edit Chat Title", openWorldHint=True, destructiveHint=True, idempotentHint=True
) # )
) )
@validate_id("chat_id") @validate_id("chat_id")
async def edit_chat_title(chat_id: Union[int, str], title: str) -> str: async def edit_chat_title(chat_id: Union[int, str], title: str) -> str:
@ -2477,10 +2511,10 @@ async def edit_chat_title(chat_id: Union[int, str], title: str) -> str:
return log_and_format_error("edit_chat_title", e, chat_id=chat_id, title=title) return log_and_format_error("edit_chat_title", e, chat_id=chat_id, title=title)
@mcp.tool( #@mcp.tool(
annotations=ToolAnnotations( # annotations=ToolAnnotations(
title="Edit Chat Photo", openWorldHint=True, destructiveHint=True, idempotentHint=True # title="Edit Chat Photo", openWorldHint=True, destructiveHint=True, idempotentHint=True
) # )
) )
@validate_id("chat_id") @validate_id("chat_id")
async def edit_chat_photo( async def edit_chat_photo(
@ -2522,10 +2556,10 @@ async def edit_chat_photo(
return log_and_format_error("edit_chat_photo", e, chat_id=chat_id, file_path=file_path) return log_and_format_error("edit_chat_photo", e, chat_id=chat_id, file_path=file_path)
@mcp.tool( #@mcp.tool(
annotations=ToolAnnotations( # annotations=ToolAnnotations(
title="Delete Chat Photo", openWorldHint=True, destructiveHint=True, idempotentHint=True # title="Delete Chat Photo", openWorldHint=True, destructiveHint=True, idempotentHint=True
) # )
) )
@validate_id("chat_id") @validate_id("chat_id")
async def delete_chat_photo(chat_id: Union[int, str]) -> str: async def delete_chat_photo(chat_id: Union[int, str]) -> str:
@ -2555,10 +2589,10 @@ async def delete_chat_photo(chat_id: Union[int, str]) -> str:
return log_and_format_error("delete_chat_photo", e, chat_id=chat_id) return log_and_format_error("delete_chat_photo", e, chat_id=chat_id)
@mcp.tool( #@mcp.tool(
annotations=ToolAnnotations( # annotations=ToolAnnotations(
title="Promote Admin", openWorldHint=True, destructiveHint=True, idempotentHint=True # title="Promote Admin", openWorldHint=True, destructiveHint=True, idempotentHint=True
) # )
) )
@validate_id("group_id", "user_id") @validate_id("group_id", "user_id")
async def promote_admin( async def promote_admin(
@ -2626,10 +2660,10 @@ async def promote_admin(
return log_and_format_error("promote_admin", e, group_id=group_id, user_id=user_id) return log_and_format_error("promote_admin", e, group_id=group_id, user_id=user_id)
@mcp.tool( #@mcp.tool(
annotations=ToolAnnotations( # annotations=ToolAnnotations(
title="Demote Admin", openWorldHint=True, destructiveHint=True, idempotentHint=True # title="Demote Admin", openWorldHint=True, destructiveHint=True, idempotentHint=True
) # )
) )
@validate_id("group_id", "user_id") @validate_id("group_id", "user_id")
async def demote_admin(group_id: Union[int, str], user_id: Union[int, str]) -> str: async def demote_admin(group_id: Union[int, str], user_id: Union[int, str]) -> str:
@ -2679,10 +2713,10 @@ async def demote_admin(group_id: Union[int, str], user_id: Union[int, str]) -> s
return log_and_format_error("demote_admin", e, group_id=group_id, user_id=user_id) return log_and_format_error("demote_admin", e, group_id=group_id, user_id=user_id)
@mcp.tool( #@mcp.tool(
annotations=ToolAnnotations( # annotations=ToolAnnotations(
title="Ban User", openWorldHint=True, destructiveHint=True, idempotentHint=True # title="Ban User", openWorldHint=True, destructiveHint=True, idempotentHint=True
) # )
) )
@validate_id("chat_id", "user_id") @validate_id("chat_id", "user_id")
async def ban_user(chat_id: Union[int, str], user_id: Union[int, str]) -> str: async def ban_user(chat_id: Union[int, str], user_id: Union[int, str]) -> str:
@ -2730,10 +2764,10 @@ async def ban_user(chat_id: Union[int, str], user_id: Union[int, str]) -> str:
return log_and_format_error("ban_user", e, chat_id=chat_id, user_id=user_id) return log_and_format_error("ban_user", e, chat_id=chat_id, user_id=user_id)
@mcp.tool( #@mcp.tool(
annotations=ToolAnnotations( # annotations=ToolAnnotations(
title="Unban User", openWorldHint=True, destructiveHint=True, idempotentHint=True # title="Unban User", openWorldHint=True, destructiveHint=True, idempotentHint=True
) # )
) )
@validate_id("chat_id", "user_id") @validate_id("chat_id", "user_id")
async def unban_user(chat_id: Union[int, str], user_id: Union[int, str]) -> str: async def unban_user(chat_id: Union[int, str], user_id: Union[int, str]) -> str:
@ -2869,10 +2903,10 @@ async def get_invite_link(chat_id: Union[int, str]) -> str:
return log_and_format_error("get_invite_link", e, chat_id=chat_id) return log_and_format_error("get_invite_link", e, chat_id=chat_id)
@mcp.tool( #@mcp.tool(
annotations=ToolAnnotations( # annotations=ToolAnnotations(
title="Join Chat By Link", openWorldHint=True, destructiveHint=True, idempotentHint=True # title="Join Chat By Link", openWorldHint=True, destructiveHint=True, idempotentHint=True
) # )
) )
async def join_chat_by_link(link: str) -> str: async def join_chat_by_link(link: str) -> str:
""" """
@ -2954,10 +2988,10 @@ async def export_chat_invite(chat_id: Union[int, str]) -> str:
return log_and_format_error("export_chat_invite", e, chat_id=chat_id) return log_and_format_error("export_chat_invite", e, chat_id=chat_id)
@mcp.tool( #@mcp.tool(
annotations=ToolAnnotations( # annotations=ToolAnnotations(
title="Import Chat Invite", openWorldHint=True, destructiveHint=True, idempotentHint=True # title="Import Chat Invite", openWorldHint=True, destructiveHint=True, idempotentHint=True
) # )
) )
async def import_chat_invite(hash: str) -> str: async def import_chat_invite(hash: str) -> str:
""" """
@ -3015,8 +3049,8 @@ async def import_chat_invite(hash: str) -> str:
return log_and_format_error("import_chat_invite", e, hash=hash) return log_and_format_error("import_chat_invite", e, hash=hash)
@mcp.tool( #@mcp.tool(
annotations=ToolAnnotations(title="Send Voice", openWorldHint=True, destructiveHint=True) # annotations=ToolAnnotations(title="Send Voice", openWorldHint=True, destructiveHint=True)
) )
@validate_id("chat_id") @validate_id("chat_id")
async def send_voice( async def send_voice(
@ -3058,8 +3092,8 @@ async def send_voice(
return log_and_format_error("send_voice", e, chat_id=chat_id, file_path=file_path) return log_and_format_error("send_voice", e, chat_id=chat_id, file_path=file_path)
@mcp.tool( #@mcp.tool(
annotations=ToolAnnotations(title="Upload File", openWorldHint=True, destructiveHint=True) # annotations=ToolAnnotations(title="Upload File", openWorldHint=True, destructiveHint=True)
) )
async def upload_file(file_path: str, ctx: Optional[Context] = None) -> str: async def upload_file(file_path: str, ctx: Optional[Context] = None) -> str:
""" """
@ -3089,8 +3123,8 @@ async def upload_file(file_path: str, ctx: Optional[Context] = None) -> str:
return log_and_format_error("upload_file", e, file_path=file_path) return log_and_format_error("upload_file", e, file_path=file_path)
@mcp.tool( #@mcp.tool(
annotations=ToolAnnotations(title="Forward Message", openWorldHint=True, destructiveHint=True) # annotations=ToolAnnotations(title="Forward Message", openWorldHint=True, destructiveHint=True)
) )
@validate_id("from_chat_id", "to_chat_id") @validate_id("from_chat_id", "to_chat_id")
async def forward_message( async def forward_message(
@ -3114,10 +3148,10 @@ async def forward_message(
) )
@mcp.tool( #@mcp.tool(
annotations=ToolAnnotations( # annotations=ToolAnnotations(
title="Edit Message", openWorldHint=True, destructiveHint=True, idempotentHint=True # title="Edit Message", openWorldHint=True, destructiveHint=True, idempotentHint=True
) # )
) )
@validate_id("chat_id") @validate_id("chat_id")
async def edit_message(chat_id: Union[int, str], message_id: int, new_text: str) -> str: async def edit_message(chat_id: Union[int, str], message_id: int, new_text: str) -> str:
@ -3134,10 +3168,10 @@ async def edit_message(chat_id: Union[int, str], message_id: int, new_text: str)
) )
@mcp.tool( #@mcp.tool(
annotations=ToolAnnotations( # annotations=ToolAnnotations(
title="Delete Message", openWorldHint=True, destructiveHint=True, idempotentHint=True # title="Delete Message", openWorldHint=True, destructiveHint=True, idempotentHint=True
) # )
) )
@validate_id("chat_id") @validate_id("chat_id")
async def delete_message(chat_id: Union[int, str], message_id: int) -> str: async def delete_message(chat_id: Union[int, str], message_id: int) -> str:
@ -3152,10 +3186,10 @@ async def delete_message(chat_id: Union[int, str], message_id: int) -> str:
return log_and_format_error("delete_message", e, chat_id=chat_id, message_id=message_id) return log_and_format_error("delete_message", e, chat_id=chat_id, message_id=message_id)
@mcp.tool( #@mcp.tool(
annotations=ToolAnnotations( # annotations=ToolAnnotations(
title="Pin Message", openWorldHint=True, destructiveHint=True, idempotentHint=True # title="Pin Message", openWorldHint=True, destructiveHint=True, idempotentHint=True
) # )
) )
@validate_id("chat_id") @validate_id("chat_id")
async def pin_message(chat_id: Union[int, str], message_id: int) -> str: async def pin_message(chat_id: Union[int, str], message_id: int) -> str:
@ -3170,10 +3204,10 @@ async def pin_message(chat_id: Union[int, str], message_id: int) -> str:
return log_and_format_error("pin_message", e, chat_id=chat_id, message_id=message_id) return log_and_format_error("pin_message", e, chat_id=chat_id, message_id=message_id)
@mcp.tool( #@mcp.tool(
annotations=ToolAnnotations( # annotations=ToolAnnotations(
title="Unpin Message", openWorldHint=True, destructiveHint=True, idempotentHint=True # title="Unpin Message", openWorldHint=True, destructiveHint=True, idempotentHint=True
) # )
) )
@validate_id("chat_id") @validate_id("chat_id")
async def unpin_message(chat_id: Union[int, str], message_id: int) -> str: async def unpin_message(chat_id: Union[int, str], message_id: int) -> str:
@ -3188,10 +3222,10 @@ async def unpin_message(chat_id: Union[int, str], message_id: int) -> str:
return log_and_format_error("unpin_message", e, chat_id=chat_id, message_id=message_id) return log_and_format_error("unpin_message", e, chat_id=chat_id, message_id=message_id)
@mcp.tool( #@mcp.tool(
annotations=ToolAnnotations( # annotations=ToolAnnotations(
title="Mark As Read", openWorldHint=True, destructiveHint=True, idempotentHint=True # title="Mark As Read", openWorldHint=True, destructiveHint=True, idempotentHint=True
) # )
) )
@validate_id("chat_id") @validate_id("chat_id")
async def mark_as_read(chat_id: Union[int, str]) -> str: async def mark_as_read(chat_id: Union[int, str]) -> str:
@ -3206,8 +3240,8 @@ async def mark_as_read(chat_id: Union[int, str]) -> str:
return log_and_format_error("mark_as_read", e, chat_id=chat_id) return log_and_format_error("mark_as_read", e, chat_id=chat_id)
@mcp.tool( #@mcp.tool(
annotations=ToolAnnotations(title="Reply To Message", openWorldHint=True, destructiveHint=True) # annotations=ToolAnnotations(title="Reply To Message", openWorldHint=True, destructiveHint=True)
) )
@validate_id("chat_id") @validate_id("chat_id")
async def reply_to_message( async def reply_to_message(
@ -3353,10 +3387,10 @@ async def resolve_username(username: str) -> str:
return log_and_format_error("resolve_username", e, username=username) return log_and_format_error("resolve_username", e, username=username)
@mcp.tool( #@mcp.tool(
annotations=ToolAnnotations( # annotations=ToolAnnotations(
title="Mute Chat", openWorldHint=True, destructiveHint=True, idempotentHint=True # title="Mute Chat", openWorldHint=True, destructiveHint=True, idempotentHint=True
) # )
) )
@validate_id("chat_id") @validate_id("chat_id")
async def mute_chat(chat_id: Union[int, str]) -> str: async def mute_chat(chat_id: Union[int, str]) -> str:
@ -3396,10 +3430,10 @@ async def mute_chat(chat_id: Union[int, str]) -> str:
return log_and_format_error("mute_chat", e, chat_id=chat_id) return log_and_format_error("mute_chat", e, chat_id=chat_id)
@mcp.tool( #@mcp.tool(
annotations=ToolAnnotations( # annotations=ToolAnnotations(
title="Unmute Chat", openWorldHint=True, destructiveHint=True, idempotentHint=True # title="Unmute Chat", openWorldHint=True, destructiveHint=True, idempotentHint=True
) # )
) )
@validate_id("chat_id") @validate_id("chat_id")
async def unmute_chat(chat_id: Union[int, str]) -> str: async def unmute_chat(chat_id: Union[int, str]) -> str:
@ -3439,10 +3473,10 @@ async def unmute_chat(chat_id: Union[int, str]) -> str:
return log_and_format_error("unmute_chat", e, chat_id=chat_id) return log_and_format_error("unmute_chat", e, chat_id=chat_id)
@mcp.tool( #@mcp.tool(
annotations=ToolAnnotations( # annotations=ToolAnnotations(
title="Archive Chat", openWorldHint=True, destructiveHint=True, idempotentHint=True # title="Archive Chat", openWorldHint=True, destructiveHint=True, idempotentHint=True
) # )
) )
@validate_id("chat_id") @validate_id("chat_id")
async def archive_chat(chat_id: Union[int, str]) -> str: async def archive_chat(chat_id: Union[int, str]) -> str:
@ -3462,10 +3496,10 @@ async def archive_chat(chat_id: Union[int, str]) -> str:
return log_and_format_error("archive_chat", e, chat_id=chat_id) return log_and_format_error("archive_chat", e, chat_id=chat_id)
@mcp.tool( #@mcp.tool(
annotations=ToolAnnotations( # annotations=ToolAnnotations(
title="Unarchive Chat", openWorldHint=True, destructiveHint=True, idempotentHint=True # title="Unarchive Chat", openWorldHint=True, destructiveHint=True, idempotentHint=True
) # )
) )
@validate_id("chat_id") @validate_id("chat_id")
async def unarchive_chat(chat_id: Union[int, str]) -> str: async def unarchive_chat(chat_id: Union[int, str]) -> str:
@ -3499,8 +3533,8 @@ async def get_sticker_sets() -> str:
return log_and_format_error("get_sticker_sets", e) return log_and_format_error("get_sticker_sets", e)
@mcp.tool( #@mcp.tool(
annotations=ToolAnnotations(title="Send Sticker", openWorldHint=True, destructiveHint=True) # annotations=ToolAnnotations(title="Send Sticker", openWorldHint=True, destructiveHint=True)
) )
@validate_id("chat_id") @validate_id("chat_id")
async def send_sticker( async def send_sticker(
@ -3589,8 +3623,8 @@ async def get_gif_search(query: str, limit: int = 10) -> str:
return log_and_format_error("get_gif_search", e, query=query, limit=limit) return log_and_format_error("get_gif_search", e, query=query, limit=limit)
@mcp.tool(annotations=ToolAnnotations(title="Send Gif", openWorldHint=True, destructiveHint=True)) #@mcp.tool(annotations=ToolAnnotations(title="Send Gif", openWorldHint=True, destructiveHint=True))
@validate_id("chat_id") #@validate_id("chat_id")
async def send_gif(chat_id: Union[int, str], gif_id: int) -> str: async def send_gif(chat_id: Union[int, str], gif_id: int) -> str:
""" """
Send a GIF to a chat by Telegram GIF document ID (not a file path). Send a GIF to a chat by Telegram GIF document ID (not a file path).
@ -3609,6 +3643,49 @@ async def send_gif(chat_id: Union[int, str], gif_id: int) -> str:
return log_and_format_error("send_gif", e, chat_id=chat_id, gif_id=gif_id) return log_and_format_error("send_gif", e, chat_id=chat_id, gif_id=gif_id)
#@mcp.tool(
# annotations=ToolAnnotations(title="Send Contact", openWorldHint=True, destructiveHint=True)
)
@validate_id("chat_id")
async def send_contact(
chat_id: Union[int, str],
phone_number: str,
first_name: str,
last_name: str = "",
vcard: str = "",
) -> str:
"""
Send a contact to a chat.
Args:
chat_id: The chat ID or username.
phone_number: Contact's phone number.
first_name: Contact's first name.
last_name: Contact's last name (optional).
vcard: Additional vCard data (optional).
"""
try:
entity = await resolve_entity(chat_id)
from telethon.tl.types import InputMediaContact
import random
await client(
functions.messages.SendMediaRequest(
peer=entity,
media=InputMediaContact(
phone_number=phone_number,
first_name=first_name,
last_name=last_name,
vcard=vcard,
),
message="",
random_id=random.randint(0, 2**63 - 1),
)
)
return f"Contact sent to chat {chat_id}."
except Exception as e:
return log_and_format_error("send_contact", e, chat_id=chat_id, phone_number=phone_number)
@mcp.tool(annotations=ToolAnnotations(title="Get Bot Info", openWorldHint=True, readOnlyHint=True)) @mcp.tool(annotations=ToolAnnotations(title="Get Bot Info", openWorldHint=True, readOnlyHint=True))
async def get_bot_info(bot_username: str) -> str: async def get_bot_info(bot_username: str) -> str:
""" """
@ -3645,10 +3722,10 @@ async def get_bot_info(bot_username: str) -> str:
return log_and_format_error("get_bot_info", e, bot_username=bot_username) return log_and_format_error("get_bot_info", e, bot_username=bot_username)
@mcp.tool( #@mcp.tool(
annotations=ToolAnnotations( # annotations=ToolAnnotations(
title="Set Bot Commands", openWorldHint=True, destructiveHint=True, idempotentHint=True # title="Set Bot Commands", openWorldHint=True, destructiveHint=True, idempotentHint=True
) # )
) )
async def set_bot_commands(bot_username: str, commands: list) -> str: async def set_bot_commands(bot_username: str, commands: list) -> str:
""" """
@ -3825,8 +3902,8 @@ async def get_pinned_messages(chat_id: Union[int, str]) -> str:
return log_and_format_error("get_pinned_messages", e, chat_id=chat_id) return log_and_format_error("get_pinned_messages", e, chat_id=chat_id)
@mcp.tool( #@mcp.tool(
annotations=ToolAnnotations(title="Create Poll", openWorldHint=True, destructiveHint=True) # annotations=ToolAnnotations(title="Create Poll", openWorldHint=True, destructiveHint=True)
) )
async def create_poll( async def create_poll(
chat_id: int, chat_id: int,
@ -3900,10 +3977,10 @@ async def create_poll(
) )
@mcp.tool( #@mcp.tool(
annotations=ToolAnnotations( # annotations=ToolAnnotations(
title="Send Reaction", openWorldHint=True, destructiveHint=False, idempotentHint=True # title="Send Reaction", openWorldHint=True, destructiveHint=False, idempotentHint=True
) # )
) )
@validate_id("chat_id") @validate_id("chat_id")
async def send_reaction( async def send_reaction(
@ -3943,10 +4020,10 @@ async def send_reaction(
) )
@mcp.tool( #@mcp.tool(
annotations=ToolAnnotations( # annotations=ToolAnnotations(
title="Remove Reaction", openWorldHint=True, destructiveHint=True, idempotentHint=True # title="Remove Reaction", openWorldHint=True, destructiveHint=True, idempotentHint=True
) # )
) )
@validate_id("chat_id") @validate_id("chat_id")
async def remove_reaction( async def remove_reaction(
@ -4051,10 +4128,10 @@ async def get_message_reactions(
# ============================================================================ # ============================================================================
@mcp.tool( #@mcp.tool(
annotations=ToolAnnotations( # annotations=ToolAnnotations(
title="Save Draft", openWorldHint=True, destructiveHint=False, idempotentHint=True # title="Save Draft", openWorldHint=True, destructiveHint=False, idempotentHint=True
) # )
) )
@validate_id("chat_id") @validate_id("chat_id")
async def save_draft( async def save_draft(
@ -4155,10 +4232,10 @@ async def get_drafts() -> str:
return log_and_format_error("get_drafts", e) return log_and_format_error("get_drafts", e)
@mcp.tool( #@mcp.tool(
annotations=ToolAnnotations( # annotations=ToolAnnotations(
title="Clear Draft", openWorldHint=True, destructiveHint=True, idempotentHint=True # title="Clear Draft", openWorldHint=True, destructiveHint=True, idempotentHint=True
) # )
) )
@validate_id("chat_id") @validate_id("chat_id")
async def clear_draft(chat_id: Union[int, str]) -> str: async def clear_draft(chat_id: Union[int, str]) -> str:
@ -4357,10 +4434,10 @@ async def get_folder(folder_id: int) -> str:
return log_and_format_error("get_folder", e, ErrorCategory.FOLDER, folder_id=folder_id) return log_and_format_error("get_folder", e, ErrorCategory.FOLDER, folder_id=folder_id)
@mcp.tool( #@mcp.tool(
annotations=ToolAnnotations( # annotations=ToolAnnotations(
title="Create Folder", openWorldHint=True, destructiveHint=True, idempotentHint=False # title="Create Folder", openWorldHint=True, destructiveHint=True, idempotentHint=False
) # )
) )
async def create_folder( async def create_folder(
title: str, title: str,
@ -4457,10 +4534,10 @@ async def create_folder(
return log_and_format_error("create_folder", e, ErrorCategory.FOLDER, title=title) return log_and_format_error("create_folder", e, ErrorCategory.FOLDER, title=title)
@mcp.tool( #@mcp.tool(
annotations=ToolAnnotations( # annotations=ToolAnnotations(
title="Add Chat to Folder", openWorldHint=True, destructiveHint=True, idempotentHint=True # title="Add Chat to Folder", openWorldHint=True, destructiveHint=True, idempotentHint=True
) # )
) )
@validate_id("chat_id") @validate_id("chat_id")
async def add_chat_to_folder( async def add_chat_to_folder(
@ -4558,13 +4635,13 @@ async def add_chat_to_folder(
) )
@mcp.tool( #@mcp.tool(
annotations=ToolAnnotations( # annotations=ToolAnnotations(
title="Remove Chat from Folder", # title="Remove Chat from Folder",
openWorldHint=True, # openWorldHint=True,
destructiveHint=True, # destructiveHint=True,
idempotentHint=True, # idempotentHint=True,
) # )
) )
@validate_id("chat_id") @validate_id("chat_id")
async def remove_chat_from_folder(folder_id: int, chat_id: Union[int, str]) -> str: async def remove_chat_from_folder(folder_id: int, chat_id: Union[int, str]) -> str:
@ -4668,10 +4745,10 @@ async def remove_chat_from_folder(folder_id: int, chat_id: Union[int, str]) -> s
) )
@mcp.tool( #@mcp.tool(
annotations=ToolAnnotations( # annotations=ToolAnnotations(
title="Delete Folder", openWorldHint=True, destructiveHint=True, idempotentHint=True # title="Delete Folder", openWorldHint=True, destructiveHint=True, idempotentHint=True
) # )
) )
async def delete_folder(folder_id: int) -> str: async def delete_folder(folder_id: int) -> str:
""" """
@ -4712,10 +4789,10 @@ async def delete_folder(folder_id: int) -> str:
return log_and_format_error("delete_folder", e, ErrorCategory.FOLDER, folder_id=folder_id) return log_and_format_error("delete_folder", e, ErrorCategory.FOLDER, folder_id=folder_id)
@mcp.tool( #@mcp.tool(
annotations=ToolAnnotations( # annotations=ToolAnnotations(
title="Reorder Folders", openWorldHint=True, destructiveHint=True, idempotentHint=True # title="Reorder Folders", openWorldHint=True, destructiveHint=True, idempotentHint=True
) # )
) )
async def reorder_folders(folder_ids: List[int]) -> str: async def reorder_folders(folder_ids: List[int]) -> str:
""" """