Merge pull request chigwell#127 from JosXa/send-album

Add Telegram album sending tool

Author: chigwell

telegram_mcp/tools/media.py

@@ -8,7 +8,7 @@
 @validate_id("chat_id")
 async def send_file(
     chat_id: Union[int, str],
-    file_path: str,
+    file_path: Union[str, List[str]],
     caption: str = None,
     ctx: Optional[Context] = None,
     account: str = None,
@@ -18,9 +18,19 @@ async def send_file(
     Args:
         chat_id: The chat ID or username.
         file_path: Absolute or relative path to the file under allowed roots.
-        caption: Optional caption for the file.
+            Pass a list of 2-10 paths to send them as one Telegram media group.
+        caption: Optional caption for the file or media group.
     """
     try:
+        if isinstance(file_path, list):
+            return await _send_album(
+                chat_id=chat_id,
+                file_paths=file_path,
+                caption=caption,
+                ctx=ctx,
+                account=account,
+            )
+
         cl = get_client(account)
         safe_path, path_error = await _resolve_readable_file_path(
             raw_path=file_path,
@@ -38,6 +48,69 @@ async def send_file(
         )
 
 
+async def _send_album(
+    chat_id: Union[int, str],
+    file_paths: List[str],
+    caption: str = None,
+    ctx: Optional[Context] = None,
+    account: str = None,
+) -> str:
+    if not 2 <= len(file_paths) <= 10:
+        return "Albums must contain between 2 and 10 files."
+
+    cl = get_client(account)
+    safe_paths = []
+    for file_path in file_paths:
+        safe_path, path_error = await _resolve_readable_file_path(
+            raw_path=file_path,
+            ctx=ctx,
+            tool_name="send_file",
+        )
+        if path_error:
+            return path_error
+        safe_paths.append(str(safe_path))
+
+    entity = await resolve_entity(chat_id, cl)
+    await cl.send_file(entity, safe_paths, caption=caption)
+    return f"Album sent to chat {chat_id} with {len(safe_paths)} files."
+
+
+@mcp.tool(
+    annotations=ToolAnnotations(title="Send Album", openWorldHint=True, destructiveHint=True)
+)
+@with_account(readonly=False)
+@validate_id("chat_id")
+async def send_album(
+    chat_id: Union[int, str],
+    file_paths: List[str],
+    caption: str = None,
+    ctx: Optional[Context] = None,
+    account: str = None,
+) -> str:
+    """
+    Send multiple photos/videos as one Telegram media group (album).
+
+    Args:
+        chat_id: The chat ID or username.
+        file_paths: 2-10 absolute or relative file paths under allowed roots.
+        caption: Optional caption for the album. Telegram displays it on the first item.
+    """
+    try:
+        if not isinstance(file_paths, list):
+            return "file_paths must be a list of file paths."
+        return await _send_album(
+            chat_id=chat_id,
+            file_paths=file_paths,
+            caption=caption,
+            ctx=ctx,
+            account=account,
+        )
+    except Exception as e:
+        return log_and_format_error(
+            "send_album", e, chat_id=chat_id, file_paths=file_paths, caption=caption
+        )
+
+
 @mcp.tool(
     annotations=ToolAnnotations(title="Download Media", openWorldHint=True, destructiveHint=True)
 )
@@ -347,6 +420,7 @@ async def send_gif(chat_id: Union[int, str], gif_id: int, account: str = None) -
 
 __all__ = [
     "send_file",
+    "send_album",
     "download_media",
     "send_voice",
     "upload_file",

tests/test_media_album.py

@@ -0,0 +1,90 @@
+import pytest
+
+from telegram_mcp import runtime
+from telegram_mcp.tools import media
+
+
+class _DummyClient:
+    def __init__(self):
+        self.sent = None
+
+    async def send_file(self, entity, file_paths, caption=None):
+        self.sent = {
+            "entity": entity,
+            "file_paths": file_paths,
+            "caption": caption,
+        }
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("tool_name", ["send_album", "send_file"])
+async def test_album_mode_sends_multiple_files_as_one_media_group(
+    tmp_path, monkeypatch, tool_name
+):
+    root = (tmp_path / "root").resolve()
+    root.mkdir()
+    first = root / "one.png"
+    second = root / "two.png"
+    first.write_bytes(b"png-one")
+    second.write_bytes(b"png-two")
+
+    client = _DummyClient()
+    monkeypatch.setattr(runtime, "SERVER_ALLOWED_ROOTS", [root])
+    monkeypatch.setattr(media, "clients", {"default": client})
+    monkeypatch.setattr(media, "get_client", lambda account=None: client)
+
+    async def _resolve_entity(chat_id, cl):
+        assert chat_id == "AgenticAIChat"
+        assert cl is client
+        return "entity:AgenticAIChat"
+
+    monkeypatch.setattr(media, "resolve_entity", _resolve_entity)
+
+    tool = getattr(media, tool_name)
+    result = await tool(
+        "AgenticAIChat",
+        ["one.png", str(second)],
+        caption="pick one",
+    )
+
+    assert result == "Album sent to chat AgenticAIChat with 2 files."
+    assert client.sent == {
+        "entity": "entity:AgenticAIChat",
+        "file_paths": [str(first), str(second)],
+        "caption": "pick one",
+    }
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("file_paths", "expected"),
+    [
+        ("not-a-list", "file_paths must be a list of file paths."),
+        (["one.png"], "Albums must contain between 2 and 10 files."),
+        ([f"{index}.png" for index in range(11)], "Albums must contain between 2 and 10 files."),
+    ],
+)
+async def test_send_album_validates_album_file_count(file_paths, expected, monkeypatch):
+    monkeypatch.setattr(media, "clients", {"default": _DummyClient()})
+
+    result = await media.send_album("AgenticAIChat", file_paths)
+
+    assert result == expected
+
+
+@pytest.mark.asyncio
+async def test_send_album_reuses_readable_path_security(tmp_path, monkeypatch):
+    root = (tmp_path / "root").resolve()
+    outside = (tmp_path / "outside").resolve()
+    root.mkdir()
+    outside.mkdir()
+    (root / "one.png").write_bytes(b"png-one")
+    outside_file = outside / "two.png"
+    outside_file.write_bytes(b"png-two")
+
+    monkeypatch.setattr(runtime, "SERVER_ALLOWED_ROOTS", [root])
+    monkeypatch.setattr(media, "clients", {"default": _DummyClient()})
+
+    result = await media.send_album("AgenticAIChat", ["one.png", str(outside_file)])
+
+    assert result == "Path is outside allowed roots."