feat: return media as MCP content

Author: alexesipenko

telegram_mcp/tools/media.py

@@ -1,8 +1,47 @@
 """Media MCP tools."""
 
+from mcp.server.fastmcp import Audio, Image
+
 from telegram_mcp.runtime import *
 
 
+MAX_INLINE_MEDIA_BYTES = 20 * 1024 * 1024
+
+
+@mcp.tool(annotations=ToolAnnotations(title="Get Media", openWorldHint=True, readOnlyHint=True))
+@with_account(readonly=True)
+@validate_id("chat_id")
+async def get_media(
+    chat_id: Union[int, str], message_id: int, account: str = None
+) -> Any:
+    """Return a Telegram image or audio attachment as native MCP media content."""
+    try:
+        cl = get_client(account)
+        entity = await resolve_entity(chat_id, cl)
+        msg = await cl.get_messages(entity, ids=message_id)
+        if not msg or not msg.media:
+            return "No media found in the specified message."
+
+        mime = getattr(getattr(msg, "file", None), "mime_type", None)
+        if not mime and getattr(msg, "photo", None) is not None:
+            mime = "image/jpeg"
+        if not mime or not (mime.startswith("image/") or mime.startswith("audio/")):
+            return "Native MCP media is available for images and audio only."
+
+        data = await cl.download_media(msg, bytes)
+        if not data:
+            return f"Download failed for message {message_id}."
+        if len(data) > MAX_INLINE_MEDIA_BYTES:
+            return f"Media is too large for inline MCP delivery ({len(data)} bytes)."
+
+        media_type, media_format = mime.split("/", 1)
+        if media_type == "image":
+            return Image(data=data, format=media_format)
+        return Audio(data=data, format=media_format)
+    except Exception as e:
+        return log_and_format_error("get_media", e, chat_id=chat_id, message_id=message_id)
+
+
 @mcp.tool(annotations=ToolAnnotations(title="Send File", openWorldHint=True, destructiveHint=True))
 @with_account(readonly=False)
 @validate_id("chat_id")

tests/test_media_album.py

@@ -16,6 +16,36 @@ async def send_file(self, entity, file_paths, caption=None):
         }
 
 
+@pytest.mark.asyncio
+async def test_get_media_returns_native_image_content(monkeypatch):
+    class Client:
+        async def get_messages(self, entity, ids):
+            assert (entity, ids) == ("chat", 42)
+            return type(
+                "Message",
+                (),
+                {"media": object(), "photo": object(), "file": type("File", (), {"mime_type": None})()},
+            )()
+
+        async def download_media(self, message, output):
+            assert output is bytes
+            return b"image-bytes"
+
+    client = Client()
+    monkeypatch.setattr(media, "get_client", lambda account=None: client)
+
+    async def resolve(chat_id, cl):
+        assert (chat_id, cl) == (123, client)
+        return "chat"
+
+    monkeypatch.setattr(media, "resolve_entity", resolve)
+
+    result = await media.get_media(123, 42)
+
+    assert result.data == b"image-bytes"
+    assert result._format == "jpeg"
+
+
 @pytest.mark.asyncio
 @pytest.mark.parametrize("tool_name", ["send_album", "send_file"])
 async def test_album_mode_sends_multiple_files_as_one_media_group(