feat: add forum topic management tools (chigwell#5)

Author: artgas1

telegram_mcp/tools/chats.py

@@ -1,5 +1,6 @@
 """Chats MCP tools."""
 
+import secrets
 import struct
 
 from telethon.tl.tlobject import TLObject, TLRequest
@@ -71,6 +72,87 @@ def from_reader(cls, reader):
         )
 
 
+class CreateForumTopicRequest(TLRequest):
+    """Raw request for messages.createForumTopic missing in Telethon 1.42."""
+
+    CONSTRUCTOR_ID = 0x2F98C3D5
+    SUBCLASS_OF_ID = 0x0
+
+    def __init__(
+        self,
+        peer,
+        title,
+        random_id,
+        icon_color=None,
+        icon_emoji_id=None,
+        send_as=None,
+    ):
+        self.peer = peer
+        self.title = title
+        self.icon_color = icon_color
+        self.icon_emoji_id = icon_emoji_id
+        self.random_id = random_id
+        self.send_as = send_as
+
+    async def resolve(self, client, utils):
+        self.peer = utils.get_input_peer(await client.get_input_entity(self.peer))
+        if self.send_as is not None:
+            self.send_as = utils.get_input_peer(await client.get_input_entity(self.send_as))
+
+    def to_dict(self):
+        return {
+            "_": "CreateForumTopicRequest",
+            "peer": self.peer.to_dict() if isinstance(self.peer, TLObject) else self.peer,
+            "title": self.title,
+            "icon_color": self.icon_color,
+            "icon_emoji_id": self.icon_emoji_id,
+            "random_id": self.random_id,
+            "send_as": (
+                self.send_as.to_dict() if isinstance(self.send_as, TLObject) else self.send_as
+            ),
+        }
+
+    def _bytes(self):
+        flags = 0
+        if self.icon_color is not None:
+            flags |= 1 << 0
+        if self.send_as is not None:
+            flags |= 1 << 2
+        if self.icon_emoji_id is not None:
+            flags |= 1 << 3
+
+        return b"".join(
+            (
+                struct.pack("<I", self.CONSTRUCTOR_ID),
+                struct.pack("<I", flags),
+                self.peer._bytes(),
+                self.serialize_bytes(self.title),
+                b"" if self.icon_color is None else struct.pack("<i", self.icon_color),
+                b"" if self.icon_emoji_id is None else struct.pack("<q", self.icon_emoji_id),
+                struct.pack("<q", self.random_id),
+                b"" if self.send_as is None else self.send_as._bytes(),
+            )
+        )
+
+    @classmethod
+    def from_reader(cls, reader):
+        flags = reader.read_int()
+        peer = reader.tgread_object()
+        title = reader.tgread_string()
+        icon_color = reader.read_int() if flags & (1 << 0) else None
+        icon_emoji_id = reader.read_long() if flags & (1 << 3) else None
+        random_id = reader.read_long()
+        send_as = reader.tgread_object() if flags & (1 << 2) else None
+        return cls(
+            peer=peer,
+            title=title,
+            random_id=random_id,
+            icon_color=icon_color,
+            icon_emoji_id=icon_emoji_id,
+            send_as=send_as,
+        )
+
+
 @mcp.tool(annotations=ToolAnnotations(title="Get Chats", openWorldHint=True, readOnlyHint=True))
 @with_account(readonly=True)
 async def get_chats(account: str = None, page: int = 1, page_size: int = 20) -> str:
@@ -231,6 +313,139 @@ async def list_topics(
         )
 
 
+@mcp.tool(
+    annotations=ToolAnnotations(
+        title="Enable Forum Topics", openWorldHint=True, destructiveHint=True, idempotentHint=True
+    )
+)
+@with_account(readonly=False)
+@validate_id("chat_id")
+async def enable_forum_topics(
+    chat_id: Union[int, str], tabs: bool = True, account: str = None
+) -> str:
+    """
+    Enable Telegram forum topics for a supergroup.
+
+    Args:
+        chat_id: The supergroup ID or username.
+        tabs: Whether Telegram should display topics as tabs (default True).
+
+    The caller must be an admin with permission to change chat info.
+    """
+    try:
+        cl = get_client(account)
+        entity = await resolve_entity(chat_id, cl)
+
+        if not isinstance(entity, Channel) or not getattr(entity, "megagroup", False):
+            return "The specified chat is not a supergroup."
+
+        if getattr(entity, "forum", False):
+            title = sanitize_name(getattr(entity, "title", str(chat_id)))
+            return f"Forum topics already enabled for {title}."
+
+        await cl(functions.channels.ToggleForumRequest(channel=entity, enabled=True, tabs=tabs))
+        # Keep the resolved entity in sync for callers/tests that reuse it.
+        try:
+            entity.forum = True
+        except Exception:
+            pass
+
+        title = sanitize_name(getattr(entity, "title", str(chat_id)))
+        return f"Forum topics enabled for {title}."
+    except Exception as e:
+        return log_and_format_error("enable_forum_topics", e, chat_id=chat_id, tabs=tabs)
+
+
+@mcp.tool(
+    annotations=ToolAnnotations(
+        title="Create Forum Topic", openWorldHint=True, destructiveHint=True
+    )
+)
+@with_account(readonly=False)
+@validate_id("chat_id")
+async def create_forum_topic(
+    chat_id: Union[int, str],
+    title: str,
+    icon_color: int = None,
+    icon_emoji_id: int = None,
+    account: str = None,
+) -> str:
+    """
+    Create a Telegram forum topic in a forum-enabled supergroup.
+
+    Args:
+        chat_id: The forum-enabled supergroup ID or username.
+        title: Topic title.
+        icon_color: Optional Telegram topic icon color integer.
+        icon_emoji_id: Optional custom emoji document ID for the topic icon.
+
+    Returns a JSON result with chat_id, topic_id (when Telegram returns it), and title.
+    """
+    try:
+        cl = get_client(account)
+        entity = await resolve_entity(chat_id, cl)
+
+        if not isinstance(entity, Channel) or not getattr(entity, "megagroup", False):
+            return "The specified chat is not a supergroup."
+
+        if not getattr(entity, "forum", False):
+            return (
+                "The specified supergroup does not have forum topics enabled. "
+                "Use enable_forum_topics first."
+            )
+
+        clean_title = sanitize_user_content(title, max_length=128)
+        result = await cl(
+            CreateForumTopicRequest(
+                peer=entity,
+                title=clean_title,
+                random_id=secrets.randbits(63),
+                icon_color=icon_color,
+                icon_emoji_id=icon_emoji_id,
+            )
+        )
+
+        topic_id = _extract_created_topic_id(result)
+        record = {
+            "chat_id": get_marked_id(entity),
+            "title": clean_title,
+        }
+        if topic_id is not None:
+            record["topic_id"] = topic_id
+
+        return format_tool_result([record])
+    except Exception as e:
+        return log_and_format_error(
+            "create_forum_topic",
+            e,
+            chat_id=chat_id,
+            title=title,
+            icon_color=icon_color,
+            icon_emoji_id=icon_emoji_id,
+        )
+
+
+def _extract_created_topic_id(result) -> Optional[int]:
+    """Best-effort extraction of the top message/topic ID from Updates."""
+    updates = getattr(result, "updates", None) or []
+    for update in updates:
+        message = getattr(update, "message", None)
+        message_id = getattr(message, "id", None)
+        if isinstance(message_id, int):
+            return message_id
+
+        update_id = getattr(update, "id", None)
+        if isinstance(update_id, int):
+            return update_id
+
+    message = getattr(result, "message", None)
+    message_id = getattr(message, "id", None)
+    if isinstance(message_id, int):
+        return message_id
+
+    return None
+
+
 @mcp.tool(annotations=ToolAnnotations(title="List Chats", openWorldHint=True, readOnlyHint=True))
 @with_account(readonly=True)
 async def list_chats(
@@ -877,6 +1092,8 @@ async def get_message_link(
 __all__ = [
     "get_chats",
     "list_topics",
+    "enable_forum_topics",
+    "create_forum_topic",
     "list_chats",
     "get_chat",
     "subscribe_public_channel",

tests/test_forum_topics.py

@@ -0,0 +1,118 @@
+import json
+from types import SimpleNamespace
+
+import pytest
+from telethon.tl import functions
+from telethon.tl.types import Channel
+
+from telegram_mcp.tools import chats
+
+
+def _supergroup(*, forum=False):
+    return Channel(
+        id=12345,
+        title="Hermes Topics",
+        photo=None,
+        date=None,
+        creator=True,
+        left=False,
+        broadcast=False,
+        verified=False,
+        megagroup=True,
+        restricted=False,
+        signatures=False,
+        min=False,
+        scam=False,
+        has_link=False,
+        has_geo=False,
+        slowmode_enabled=False,
+        call_active=False,
+        call_not_empty=False,
+        fake=False,
+        gigagroup=False,
+        noforwards=False,
+        join_to_send=False,
+        join_request=False,
+        forum=forum,
+        stories_hidden=False,
+        stories_hidden_min=False,
+        stories_unavailable=False,
+        access_hash=67890,
+    )
+
+
+class RecordingClient:
+    def __init__(self, result=None):
+        self.requests = []
+        self.result = result or SimpleNamespace(updates=[])
+
+    async def __call__(self, request):
+        self.requests.append(request)
+        return self.result
+
+
+@pytest.mark.asyncio
+async def test_enable_forum_topics_sends_toggle_forum_request(monkeypatch):
+    entity = _supergroup(forum=False)
+    client = RecordingClient()
+
+    async def fake_resolve(chat_id, cl):
+        return entity
+
+    monkeypatch.setattr(chats, "get_client", lambda account=None: client)
+    monkeypatch.setattr(chats, "resolve_entity", fake_resolve)
+
+    result = await chats.enable_forum_topics(chat_id=12345)
+
+    assert result == "Forum topics enabled for Hermes Topics."
+    assert len(client.requests) == 1
+    request = client.requests[0]
+    assert isinstance(request, functions.channels.ToggleForumRequest)
+    assert request.channel is entity
+    assert request.enabled is True
+    assert request.tabs is True
+    assert entity.forum is True
+
+
+@pytest.mark.asyncio
+async def test_create_forum_topic_sends_raw_create_forum_topic_request(monkeypatch):
+    entity = _supergroup(forum=True)
+    client = RecordingClient(SimpleNamespace(updates=[SimpleNamespace(id=777)]))
+
+    async def fake_resolve(chat_id, cl):
+        return entity
+
+    monkeypatch.setattr(chats, "get_client", lambda account=None: client)
+    monkeypatch.setattr(chats, "resolve_entity", fake_resolve)
+
+    result = await chats.create_forum_topic(chat_id=12345, title="Dev", icon_color=0x6FB9F0)
+
+    payload = json.loads(result)
+    assert payload["results"] == [{"chat_id": -1000000012345, "topic_id": 777, "title": "Dev"}]
+    assert len(client.requests) == 1
+    request = client.requests[0]
+    assert isinstance(request, chats.CreateForumTopicRequest)
+    assert request.peer is entity
+    assert request.title == "Dev"
+    assert request.icon_color == 0x6FB9F0
+    assert isinstance(request.random_id, int)
+
+
+@pytest.mark.asyncio
+async def test_create_forum_topic_requires_forum_enabled(monkeypatch):
+    entity = _supergroup(forum=False)
+    client = RecordingClient()
+
+    async def fake_resolve(chat_id, cl):
+        return entity
+
+    monkeypatch.setattr(chats, "get_client", lambda account=None: client)
+    monkeypatch.setattr(chats, "resolve_entity", fake_resolve)
+
+    result = await chats.create_forum_topic(chat_id=12345, title="Dev")
+
+    assert (
+        result
+        == "The specified supergroup does not have forum topics enabled. Use enable_forum_topics first."
+    )
+    assert client.requests == []