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

feat: add comprehensive ToolAnnotations to all MCP tools

Browse source 
- Add openWorldHint=True to all 63+ @mcp.tool decorators
- Add readOnlyHint=True to read-only operations (get/list/search/export functions)
- Add destructiveHint=True to state-changing operations (send/create/delete/edit functions)
- Add idempotentHint=True to safely repeatable operations (block/unblock/pin/unpin/ban/unban/etc.)
- Fix syntax errors: openWorldHint(True) → openWorldHint=True
- Remove incorrect idempotentHint from non-idempotent operations (send_file, press_inline_button, import_contacts)

These annotations help MCP clients understand tool behavior:
- readOnlyHint: tool doesn't modify environment
- destructiveHint: tool performs destructive updates
- idempotentHint: repeated calls with same args have no additional effect
- openWorldHint: tool interacts with external entities (Telegram API)

Improves tool discoverability and helps AI agents make better decisions about tool usage.
This commit is contained in:
Valeriy_Pavlovich 2025-11-14 00:26:35 +03:00
parent 57b6ef44f8
commit 1c71806082
1 changed files with 79 additions and 78 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

157
main.py
View file

@ -14,6 +14,7 @@ from typing import List, Dict, Optional, Union, Any
import nest_asyncio import nest_asyncio
from dotenv import load_dotenv from dotenv import load_dotenv
from mcp.server.fastmcp import FastMCP from mcp.server.fastmcp import FastMCP
from mcp.types import ToolAnnotations
from pythonjsonlogger import jsonlogger from pythonjsonlogger import jsonlogger
from telethon import TelegramClient, functions, utils from telethon import TelegramClient, functions, utils
from telethon.sessions import StringSession from telethon.sessions import StringSession
@ -316,7 +317,7 @@ def get_sender_name(message) -> str:
return "Unknown" return "Unknown"
@mcp.tool() @mcp.tool(annotations=ToolAnnotations(openWorldHint=True, readOnlyHint=True))
async def get_chats(page: int = 1, page_size: int = 20) -> str: async def get_chats(page: int = 1, page_size: int = 20) -> str:
""" """
Get a paginated list of chats. Get a paginated list of chats.
@ -342,7 +343,7 @@ async def get_chats(page: int = 1, page_size: int = 20) -> str:
return log_and_format_error("get_chats", e) return log_and_format_error("get_chats", e)
@mcp.tool() @mcp.tool(annotations=ToolAnnotations(openWorldHint=True, readOnlyHint=True))
@validate_id("chat_id") @validate_id("chat_id")
async def get_messages(chat_id: Union[int, str], page: int = 1, page_size: int = 20) -> str: async def get_messages(chat_id: Union[int, str], page: int = 1, page_size: int = 20) -> str:
""" """
@ -374,7 +375,7 @@ async def get_messages(chat_id: Union[int, str], page: int = 1, page_size: int =
) )
@mcp.tool() @mcp.tool(annotations=ToolAnnotations(openWorldHint=True, destructiveHint=True))
@validate_id("chat_id") @validate_id("chat_id")
async def send_message(chat_id: Union[int, str], message: str) -> str: async def send_message(chat_id: Union[int, str], message: str) -> str:
""" """
@ -391,7 +392,7 @@ async def send_message(chat_id: Union[int, str], message: str) -> str:
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(openWorldHint=True, destructiveHint=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:
""" """
@ -411,7 +412,7 @@ async def subscribe_public_channel(channel: Union[int, str]) -> str:
return log_and_format_error("subscribe_public_channel", e, channel=channel) return log_and_format_error("subscribe_public_channel", e, channel=channel)
@mcp.tool() @mcp.tool(annotations=ToolAnnotations(openWorldHint=True, readOnlyHint=True))
@validate_id("chat_id") @validate_id("chat_id")
async def list_inline_buttons( async def list_inline_buttons(
chat_id: Union[int, str], message_id: Optional[Union[int, str]] = None, limit: int = 20 chat_id: Union[int, str], message_id: Optional[Union[int, str]] = None, limit: int = 20
@ -475,7 +476,7 @@ async def list_inline_buttons(
) )
@mcp.tool() @mcp.tool(annotations=ToolAnnotations(openWorldHint=True, destructiveHint=True))
@validate_id("chat_id") @validate_id("chat_id")
async def press_inline_button( async def press_inline_button(
chat_id: Union[int, str], chat_id: Union[int, str],
@ -590,7 +591,7 @@ async def press_inline_button(
) )
@mcp.tool() @mcp.tool(annotations=ToolAnnotations(openWorldHint=True, readOnlyHint=True))
async def list_contacts() -> str: async def list_contacts() -> str:
""" """
List all contacts in your Telegram account. List all contacts in your Telegram account.
@ -616,7 +617,7 @@ async def list_contacts() -> str:
return log_and_format_error("list_contacts", e) return log_and_format_error("list_contacts", e)
@mcp.tool() @mcp.tool(annotations=ToolAnnotations(openWorldHint=True, readOnlyHint=True))
async def search_contacts(query: str) -> str: async def search_contacts(query: str) -> str:
""" """
Search for contacts by name, username, or phone number using Telethon's SearchRequest. Search for contacts by name, username, or phone number using Telethon's SearchRequest.
@ -644,7 +645,7 @@ async def search_contacts(query: str) -> str:
return log_and_format_error("search_contacts", e, query=query) return log_and_format_error("search_contacts", e, query=query)
@mcp.tool() @mcp.tool(annotations=ToolAnnotations(openWorldHint=True, readOnlyHint=True))
async def get_contact_ids() -> str: async def get_contact_ids() -> str:
""" """
Get all contact IDs in your Telegram account. Get all contact IDs in your Telegram account.
@ -658,7 +659,7 @@ async def get_contact_ids() -> str:
return log_and_format_error("get_contact_ids", e) return log_and_format_error("get_contact_ids", e)
@mcp.tool() @mcp.tool(annotations=ToolAnnotations(openWorldHint=True, readOnlyHint=True))
@validate_id("chat_id") @validate_id("chat_id")
async def list_messages( async def list_messages(
chat_id: Union[int, str], chat_id: Union[int, str],
@ -781,7 +782,7 @@ async def list_messages(
return log_and_format_error("list_messages", e, chat_id=chat_id) return log_and_format_error("list_messages", e, chat_id=chat_id)
@mcp.tool() @mcp.tool(annotations=ToolAnnotations(openWorldHint=True, readOnlyHint=True))
async def list_topics( async def list_topics(
chat_id: int, chat_id: int,
limit: int = 200, limit: int = 200,
@ -868,7 +869,7 @@ async def list_topics(
) )
@mcp.tool() @mcp.tool(annotations=ToolAnnotations(openWorldHint=True, readOnlyHint=True))
async def list_chats(chat_type: str = None, limit: int = 20) -> str: async def list_chats(chat_type: str = None, limit: int = 20) -> str:
""" """
List available chats with metadata. List available chats with metadata.
@ -929,7 +930,7 @@ async def list_chats(chat_type: str = None, limit: int = 20) -> str:
return log_and_format_error("list_chats", e, chat_type=chat_type, limit=limit) return log_and_format_error("list_chats", e, chat_type=chat_type, limit=limit)
@mcp.tool() @mcp.tool(annotations=ToolAnnotations(openWorldHint=True, readOnlyHint=True))
@validate_id("chat_id") @validate_id("chat_id")
async def get_chat(chat_id: Union[int, str]) -> str: async def get_chat(chat_id: Union[int, str]) -> str:
""" """
@ -1010,7 +1011,7 @@ async def get_chat(chat_id: Union[int, str]) -> str:
return log_and_format_error("get_chat", e, chat_id=chat_id) return log_and_format_error("get_chat", e, chat_id=chat_id)
@mcp.tool() @mcp.tool(annotations=ToolAnnotations(openWorldHint=True, readOnlyHint=True))
async def get_direct_chat_by_contact(contact_query: str) -> str: async def get_direct_chat_by_contact(contact_query: str) -> str:
""" """
Find a direct chat with a specific contact by name, username, or phone. Find a direct chat with a specific contact by name, username, or phone.
@ -1065,7 +1066,7 @@ async def get_direct_chat_by_contact(contact_query: str) -> str:
return log_and_format_error("get_direct_chat_by_contact", e, contact_query=contact_query) return log_and_format_error("get_direct_chat_by_contact", e, contact_query=contact_query)
@mcp.tool() @mcp.tool(annotations=ToolAnnotations(openWorldHint=True, readOnlyHint=True))
@validate_id("contact_id") @validate_id("contact_id")
async def get_contact_chats(contact_id: Union[int, str]) -> str: async def get_contact_chats(contact_id: Union[int, str]) -> str:
""" """
@ -1118,7 +1119,7 @@ async def get_contact_chats(contact_id: Union[int, str]) -> str:
return log_and_format_error("get_contact_chats", e, contact_id=contact_id) return log_and_format_error("get_contact_chats", e, contact_id=contact_id)
@mcp.tool() @mcp.tool(annotations=ToolAnnotations(openWorldHint=True, readOnlyHint=True))
@validate_id("contact_id") @validate_id("contact_id")
async def get_last_interaction(contact_id: Union[int, str]) -> str: async def get_last_interaction(contact_id: Union[int, str]) -> str:
""" """
@ -1155,7 +1156,7 @@ async def get_last_interaction(contact_id: Union[int, str]) -> str:
return log_and_format_error("get_last_interaction", e, contact_id=contact_id) return log_and_format_error("get_last_interaction", e, contact_id=contact_id)
@mcp.tool() @mcp.tool(annotations=ToolAnnotations(openWorldHint=True, readOnlyHint=True))
@validate_id("chat_id") @validate_id("chat_id")
async def get_message_context( async def get_message_context(
chat_id: Union[int, str], message_id: int, context_size: int = 3 chat_id: Union[int, str], message_id: int, context_size: int = 3
@ -1222,7 +1223,7 @@ async def get_message_context(
) )
@mcp.tool() @mcp.tool(annotations=ToolAnnotations(openWorldHint=True, destructiveHint=True, idempotentHint=True))
async def add_contact(phone: str, first_name: str, last_name: str = "") -> str: async def add_contact(phone: str, first_name: str, last_name: str = "") -> str:
""" """
Add a new contact to your Telegram account. Add a new contact to your Telegram account.
@ -1278,7 +1279,7 @@ async def add_contact(phone: str, first_name: str, last_name: str = "") -> str:
return log_and_format_error("add_contact", e, phone=phone) return log_and_format_error("add_contact", e, phone=phone)
@mcp.tool() @mcp.tool(annotations=ToolAnnotations(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:
""" """
@ -1294,7 +1295,7 @@ 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(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:
""" """
@ -1310,7 +1311,7 @@ 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(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:
""" """
@ -1326,7 +1327,7 @@ async def unblock_user(user_id: Union[int, str]) -> str:
return log_and_format_error("unblock_user", e, user_id=user_id) return log_and_format_error("unblock_user", e, user_id=user_id)
@mcp.tool() @mcp.tool(annotations=ToolAnnotations(openWorldHint=True, readOnlyHint=True))
async def get_me() -> str: async def get_me() -> str:
""" """
Get your own user information. Get your own user information.
@ -1338,7 +1339,7 @@ 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(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:
""" """
@ -1397,7 +1398,7 @@ 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(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:
""" """
@ -1447,7 +1448,7 @@ 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(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:
""" """
@ -1528,7 +1529,7 @@ async def leave_chat(chat_id: Union[int, str]) -> str:
return log_and_format_error("leave_chat", e, chat_id=chat_id) return log_and_format_error("leave_chat", e, chat_id=chat_id)
@mcp.tool() @mcp.tool(annotations=ToolAnnotations(openWorldHint=True, readOnlyHint=True))
@validate_id("chat_id") @validate_id("chat_id")
async def get_participants(chat_id: Union[int, str]) -> str: async def get_participants(chat_id: Union[int, str]) -> str:
""" """
@ -1547,7 +1548,7 @@ 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() @mcp.tool(annotations=ToolAnnotations(openWorldHint=True, destructiveHint=True))
@validate_id("chat_id") @validate_id("chat_id")
async def send_file(chat_id: Union[int, str], file_path: str, caption: str = None) -> str: async def send_file(chat_id: Union[int, str], file_path: str, caption: str = None) -> str:
""" """
@ -1571,7 +1572,7 @@ async def send_file(chat_id: Union[int, str], file_path: str, caption: str = Non
) )
@mcp.tool() @mcp.tool(annotations=ToolAnnotations(openWorldHint=True, readOnlyHint=True))
@validate_id("chat_id") @validate_id("chat_id")
async def download_media(chat_id: Union[int, str], message_id: int, file_path: str) -> str: async def download_media(chat_id: Union[int, str], message_id: int, file_path: str) -> str:
""" """
@ -1604,7 +1605,7 @@ async def download_media(chat_id: Union[int, str], message_id: int, file_path: s
) )
@mcp.tool() @mcp.tool(annotations=ToolAnnotations(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:
""" """
Update your profile information (name, bio). Update your profile information (name, bio).
@ -1622,7 +1623,7 @@ async def update_profile(first_name: str = None, last_name: str = None, about: s
) )
@mcp.tool() @mcp.tool(annotations=ToolAnnotations(openWorldHint=True, destructiveHint=True, idempotentHint=True))
async def set_profile_photo(file_path: str) -> str: async def set_profile_photo(file_path: str) -> str:
""" """
Set a new profile photo. Set a new profile photo.
@ -1636,7 +1637,7 @@ async def set_profile_photo(file_path: str) -> str:
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(openWorldHint=True, destructiveHint=True, idempotentHint=True))
async def delete_profile_photo() -> str: async def delete_profile_photo() -> str:
""" """
Delete your current profile photo. Delete your current profile photo.
@ -1653,7 +1654,7 @@ async def delete_profile_photo() -> str:
return log_and_format_error("delete_profile_photo", e) return log_and_format_error("delete_profile_photo", e)
@mcp.tool() @mcp.tool(annotations=ToolAnnotations(openWorldHint=True, readOnlyHint=True))
async def get_privacy_settings() -> str: async def get_privacy_settings() -> str:
""" """
Get your privacy settings for last seen status. Get your privacy settings for last seen status.
@ -1677,7 +1678,7 @@ 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(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(
key: str, key: str,
@ -1774,7 +1775,7 @@ 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(openWorldHint=True, destructiveHint=True))
async def import_contacts(contacts: list) -> str: async def import_contacts(contacts: list) -> str:
""" """
Import a list of contacts. Each contact should be a dict with phone, first_name, last_name. Import a list of contacts. Each contact should be a dict with phone, first_name, last_name.
@ -1795,7 +1796,7 @@ async def import_contacts(contacts: list) -> str:
return log_and_format_error("import_contacts", e, contacts=contacts) return log_and_format_error("import_contacts", e, contacts=contacts)
@mcp.tool() @mcp.tool(annotations=ToolAnnotations(openWorldHint=True, readOnlyHint=True))
async def export_contacts() -> str: async def export_contacts() -> str:
""" """
Export all contacts as a JSON string. Export all contacts as a JSON string.
@ -1808,7 +1809,7 @@ async def export_contacts() -> str:
return log_and_format_error("export_contacts", e) return log_and_format_error("export_contacts", e)
@mcp.tool() @mcp.tool(annotations=ToolAnnotations(openWorldHint=True, readOnlyHint=True))
async def get_blocked_users() -> str: async def get_blocked_users() -> str:
""" """
Get a list of blocked users. Get a list of blocked users.
@ -1820,7 +1821,7 @@ 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(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:
""" """
Create a new channel or supergroup. Create a new channel or supergroup.
@ -1836,7 +1837,7 @@ async def create_channel(title: str, about: str = "", megagroup: bool = False) -
) )
@mcp.tool() @mcp.tool(annotations=ToolAnnotations(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:
""" """
@ -1856,7 +1857,7 @@ 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(openWorldHint=True, destructiveHint=True, idempotentHint=True))
@validate_id("chat_id") @validate_id("chat_id")
async def edit_chat_photo(chat_id: Union[int, str], file_path: str) -> str: async def edit_chat_photo(chat_id: Union[int, str], file_path: str) -> str:
""" """
@ -1890,7 +1891,7 @@ async def edit_chat_photo(chat_id: Union[int, str], file_path: str) -> str:
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(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:
""" """
@ -1919,7 +1920,7 @@ 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(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(
group_id: Union[int, str], user_id: Union[int, str], rights: dict = None group_id: Union[int, str], user_id: Union[int, str], rights: dict = None
@ -1986,7 +1987,7 @@ 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(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:
""" """
@ -2035,7 +2036,7 @@ 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(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:
""" """
@ -2082,7 +2083,7 @@ 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(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:
""" """
@ -2129,7 +2130,7 @@ async def unban_user(chat_id: Union[int, str], user_id: Union[int, str]) -> str:
return log_and_format_error("unban_user", e, chat_id=chat_id, user_id=user_id) return log_and_format_error("unban_user", e, chat_id=chat_id, user_id=user_id)
@mcp.tool() @mcp.tool(annotations=ToolAnnotations(openWorldHint=True, readOnlyHint=True))
@validate_id("chat_id") @validate_id("chat_id")
async def get_admins(chat_id: Union[int, str]) -> str: async def get_admins(chat_id: Union[int, str]) -> str:
""" """
@ -2148,7 +2149,7 @@ async def get_admins(chat_id: Union[int, str]) -> str:
return log_and_format_error("get_admins", e, chat_id=chat_id) return log_and_format_error("get_admins", e, chat_id=chat_id)
@mcp.tool() @mcp.tool(annotations=ToolAnnotations(openWorldHint=True, readOnlyHint=True))
@validate_id("chat_id") @validate_id("chat_id")
async def get_banned_users(chat_id: Union[int, str]) -> str: async def get_banned_users(chat_id: Union[int, str]) -> str:
""" """
@ -2169,7 +2170,7 @@ async def get_banned_users(chat_id: Union[int, str]) -> str:
return log_and_format_error("get_banned_users", e, chat_id=chat_id) return log_and_format_error("get_banned_users", e, chat_id=chat_id)
@mcp.tool() @mcp.tool(annotations=ToolAnnotations(openWorldHint=True, readOnlyHint=True))
@validate_id("chat_id") @validate_id("chat_id")
async def get_invite_link(chat_id: Union[int, str]) -> str: async def get_invite_link(chat_id: Union[int, str]) -> str:
""" """
@ -2213,7 +2214,7 @@ 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(openWorldHint=True, destructiveHint=True, idempotentHint=True))
async def join_chat_by_link(link: str) -> str: async def join_chat_by_link(link: str) -> str:
""" """
Join a chat by invite link. Join a chat by invite link.
@ -2257,7 +2258,7 @@ async def join_chat_by_link(link: str) -> str:
return f"Error joining chat: {e}" return f"Error joining chat: {e}"
@mcp.tool() @mcp.tool(annotations=ToolAnnotations(openWorldHint=True, readOnlyHint=True))
@validate_id("chat_id") @validate_id("chat_id")
async def export_chat_invite(chat_id: Union[int, str]) -> str: async def export_chat_invite(chat_id: Union[int, str]) -> str:
""" """
@ -2292,7 +2293,7 @@ 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(openWorldHint=True, destructiveHint=True, idempotentHint=True))
async def import_chat_invite(hash: str) -> str: async def import_chat_invite(hash: str) -> str:
""" """
Import a chat invite by hash. Import a chat invite by hash.
@ -2349,7 +2350,7 @@ 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(openWorldHint=True, destructiveHint=True))
@validate_id("chat_id") @validate_id("chat_id")
async def send_voice(chat_id: Union[int, str], file_path: str) -> str: async def send_voice(chat_id: Union[int, str], file_path: str) -> str:
""" """
@ -2383,7 +2384,7 @@ async def send_voice(chat_id: Union[int, str], file_path: str) -> str:
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(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(
from_chat_id: Union[int, str], message_id: int, to_chat_id: Union[int, str] from_chat_id: Union[int, str], message_id: int, to_chat_id: Union[int, str]
@ -2406,7 +2407,7 @@ async def forward_message(
) )
@mcp.tool() @mcp.tool(annotations=ToolAnnotations(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:
""" """
@ -2422,7 +2423,7 @@ async def edit_message(chat_id: Union[int, str], message_id: int, new_text: str)
) )
@mcp.tool() @mcp.tool(annotations=ToolAnnotations(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:
""" """
@ -2436,7 +2437,7 @@ 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(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:
""" """
@ -2450,7 +2451,7 @@ 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(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:
""" """
@ -2464,7 +2465,7 @@ 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(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:
""" """
@ -2478,7 +2479,7 @@ 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(openWorldHint=True, destructiveHint=True))
@validate_id("chat_id") @validate_id("chat_id")
async def reply_to_message(chat_id: Union[int, str], message_id: int, text: str) -> str: async def reply_to_message(chat_id: Union[int, str], message_id: int, text: str) -> str:
""" """
@ -2494,7 +2495,7 @@ async def reply_to_message(chat_id: Union[int, str], message_id: int, text: str)
) )
@mcp.tool() @mcp.tool(annotations=ToolAnnotations(openWorldHint=True, readOnlyHint=True))
@validate_id("chat_id") @validate_id("chat_id")
async def get_media_info(chat_id: Union[int, str], message_id: int) -> str: async def get_media_info(chat_id: Union[int, str], message_id: int) -> str:
""" """
@ -2516,7 +2517,7 @@ async def get_media_info(chat_id: Union[int, str], message_id: int) -> str:
return log_and_format_error("get_media_info", e, chat_id=chat_id, message_id=message_id) return log_and_format_error("get_media_info", e, chat_id=chat_id, message_id=message_id)
@mcp.tool() @mcp.tool(annotations=ToolAnnotations(openWorldHint=True, readOnlyHint=True))
async def search_public_chats(query: str) -> str: async def search_public_chats(query: str) -> str:
""" """
Search for public chats, channels, or bots by username or title. Search for public chats, channels, or bots by username or title.
@ -2528,7 +2529,7 @@ async def search_public_chats(query: str) -> str:
return log_and_format_error("search_public_chats", e, query=query) return log_and_format_error("search_public_chats", e, query=query)
@mcp.tool() @mcp.tool(annotations=ToolAnnotations(openWorldHint=True, readOnlyHint=True))
@validate_id("chat_id") @validate_id("chat_id")
async def search_messages(chat_id: Union[int, str], query: str, limit: int = 20) -> str: async def search_messages(chat_id: Union[int, str], query: str, limit: int = 20) -> str:
""" """
@ -2554,7 +2555,7 @@ async def search_messages(chat_id: Union[int, str], query: str, limit: int = 20)
) )
@mcp.tool() @mcp.tool(annotations=ToolAnnotations(openWorldHint=True, readOnlyHint=True))
async def resolve_username(username: str) -> str: async def resolve_username(username: str) -> str:
""" """
Resolve a username to a user or chat ID. Resolve a username to a user or chat ID.
@ -2566,7 +2567,7 @@ 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(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:
""" """
@ -2605,7 +2606,7 @@ 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(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:
""" """
@ -2644,7 +2645,7 @@ 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(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:
""" """
@ -2661,7 +2662,7 @@ 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(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:
""" """
@ -2678,7 +2679,7 @@ async def unarchive_chat(chat_id: Union[int, str]) -> str:
return log_and_format_error("unarchive_chat", e, chat_id=chat_id) return log_and_format_error("unarchive_chat", e, chat_id=chat_id)
@mcp.tool() @mcp.tool(annotations=ToolAnnotations(openWorldHint=True, readOnlyHint=True))
async def get_sticker_sets() -> str: async def get_sticker_sets() -> str:
""" """
Get all sticker sets. Get all sticker sets.
@ -2690,7 +2691,7 @@ 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(openWorldHint=True, destructiveHint=True))
@validate_id("chat_id") @validate_id("chat_id")
async def send_sticker(chat_id: Union[int, str], file_path: str) -> str: async def send_sticker(chat_id: Union[int, str], file_path: str) -> str:
""" """
@ -2715,7 +2716,7 @@ async def send_sticker(chat_id: Union[int, str], file_path: str) -> str:
return log_and_format_error("send_sticker", e, chat_id=chat_id, file_path=file_path) return log_and_format_error("send_sticker", e, chat_id=chat_id, file_path=file_path)
@mcp.tool() @mcp.tool(annotations=ToolAnnotations(openWorldHint=True, readOnlyHint=True))
async def get_gif_search(query: str, limit: int = 10) -> str: async def get_gif_search(query: str, limit: int = 10) -> str:
""" """
Search for GIFs by query. Returns a list of Telegram document IDs (not file paths). Search for GIFs by query. Returns a list of Telegram document IDs (not file paths).
@ -2771,7 +2772,7 @@ 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() @mcp.tool(annotations=ToolAnnotations(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:
""" """
@ -2791,7 +2792,7 @@ 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() @mcp.tool(annotations=ToolAnnotations(openWorldHint=True, readOnlyHint=True))
async def get_bot_info(bot_username: str) -> str: async def get_bot_info(bot_username: str) -> str:
""" """
Get information about a bot by username. Get information about a bot by username.
@ -2827,7 +2828,7 @@ 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(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:
""" """
Set bot commands for a bot you own. Set bot commands for a bot you own.
@ -2874,7 +2875,7 @@ async def set_bot_commands(bot_username: str, commands: list) -> str:
return log_and_format_error("set_bot_commands", e, bot_username=bot_username) return log_and_format_error("set_bot_commands", e, bot_username=bot_username)
@mcp.tool() @mcp.tool(annotations=ToolAnnotations(openWorldHint=True, readOnlyHint=True))
@validate_id("chat_id") @validate_id("chat_id")
async def get_history(chat_id: Union[int, str], limit: int = 100) -> str: async def get_history(chat_id: Union[int, str], limit: int = 100) -> str:
""" """
@ -2898,7 +2899,7 @@ async def get_history(chat_id: Union[int, str], limit: int = 100) -> str:
return log_and_format_error("get_history", e, chat_id=chat_id, limit=limit) return log_and_format_error("get_history", e, chat_id=chat_id, limit=limit)
@mcp.tool() @mcp.tool(annotations=ToolAnnotations(openWorldHint=True, readOnlyHint=True))
@validate_id("user_id") @validate_id("user_id")
async def get_user_photos(user_id: Union[int, str], limit: int = 10) -> str: async def get_user_photos(user_id: Union[int, str], limit: int = 10) -> str:
""" """
@ -2914,7 +2915,7 @@ async def get_user_photos(user_id: Union[int, str], limit: int = 10) -> str:
return log_and_format_error("get_user_photos", e, user_id=user_id, limit=limit) return log_and_format_error("get_user_photos", e, user_id=user_id, limit=limit)
@mcp.tool() @mcp.tool(annotations=ToolAnnotations(openWorldHint=True, readOnlyHint=True))
@validate_id("user_id") @validate_id("user_id")
async def get_user_status(user_id: Union[int, str]) -> str: async def get_user_status(user_id: Union[int, str]) -> str:
""" """
@ -2927,7 +2928,7 @@ async def get_user_status(user_id: Union[int, str]) -> str:
return log_and_format_error("get_user_status", e, user_id=user_id) return log_and_format_error("get_user_status", e, user_id=user_id)
@mcp.tool() @mcp.tool(annotations=ToolAnnotations(openWorldHint=True, readOnlyHint=True))
@validate_id("chat_id") @validate_id("chat_id")
async def get_recent_actions(chat_id: Union[int, str]) -> str: async def get_recent_actions(chat_id: Union[int, str]) -> str:
""" """
@ -2956,7 +2957,7 @@ async def get_recent_actions(chat_id: Union[int, str]) -> str:
return log_and_format_error("get_recent_actions", e, chat_id=chat_id) return log_and_format_error("get_recent_actions", e, chat_id=chat_id)
@mcp.tool() @mcp.tool(annotations=ToolAnnotations(openWorldHint=True, readOnlyHint=True))
@validate_id("chat_id") @validate_id("chat_id")
async def get_pinned_messages(chat_id: Union[int, str]) -> str: async def get_pinned_messages(chat_id: Union[int, str]) -> str:
""" """
@ -2995,7 +2996,7 @@ 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(openWorldHint=True, destructiveHint=True))
async def create_poll( async def create_poll(
chat_id: int, chat_id: int,
question: str, question: str,