refactor(tool): wrap media content with descriptive tags in ReadMediaFile (#744)

Signed-off-by: Richard Chien <[email protected]>

Author: stdrc

CHANGELOG.md

@@ -12,6 +12,7 @@ Only write entries that are worth mentioning to users.
 ## Unreleased
 
 - Auth: Fix authentication issue during agent turns
+- Tool: Wrap media content with descriptive tags in `ReadMediaFile` for better path traceability
 
 ## 1.2 (2026-01-27)
 

docs/en/release-notes/changelog.md

@@ -5,6 +5,7 @@ This page documents the changes in each Kimi Code CLI release.
 ## Unreleased
 
 - Auth: Fix authentication issue during agent turns
+- Tool: Wrap media content with descriptive tags in `ReadMediaFile` for better path traceability
 
 ## 1.2 (2026-01-27)
 

docs/zh/release-notes/changelog.md

@@ -4,7 +4,8 @@
 
 ## 未发布
 
-- Auth: 修复 Agent 轮次期间的认证问题
+- Auth：修复 Agent 轮次期间的认证问题
+- Tool：为 `ReadMediaFile` 中的媒体内容添加描述性标签，提高路径可追溯性
 
 ## 1.2 (2026-01-27)
 

src/kimi_cli/tools/file/read_media.py

@@ -12,6 +12,7 @@
 from kimi_cli.tools import SkipThisTool
 from kimi_cli.tools.file.utils import MEDIA_SNIFF_BYTES, FileType, detect_file_type
 from kimi_cli.tools.utils import load_desc_jinja
+from kimi_cli.utils.media_tags import wrap_media_part
 from kimi_cli.utils.path import is_within_directory
 from kimi_cli.wire.types import ImageURLPart, VideoURLPart
 
@@ -86,7 +87,7 @@ async def _validate_path(self, path: KaosPath) -> ToolError | None:
     async def _read_media(self, path: KaosPath, file_type: FileType) -> ToolReturnValue:
         assert file_type.kind in ("image", "video")
 
-        media_id = str(path)
+        media_path = str(path)
         stat = await path.stat()
         size = stat.st_size
         if size == 0:
@@ -107,7 +108,8 @@ async def _read_media(self, path: KaosPath, file_type: FileType) -> ToolReturnVa
             case "image":
                 data = await path.read_bytes()
                 data_url = _to_data_url(file_type.mime_type, data)
-                part = ImageURLPart(image_url=ImageURLPart.ImageURL(url=data_url, id=media_id))
+                part = ImageURLPart(image_url=ImageURLPart.ImageURL(url=data_url))
+                wrapped = wrap_media_part(part, tag="image", attrs={"path": media_path})
                 image_size = _extract_image_size(data)
             case "video":
                 data = await path.read_bytes()
@@ -116,10 +118,11 @@ async def _read_media(self, path: KaosPath, file_type: FileType) -> ToolReturnVa
                         data=data,
                         mime_type=file_type.mime_type,
                     )
-                    part.video_url.id = media_id
+                    wrapped = wrap_media_part(part, tag="video", attrs={"path": media_path})
                 else:
                     data_url = _to_data_url(file_type.mime_type, data)
-                    part = VideoURLPart(video_url=VideoURLPart.VideoURL(url=data_url, id=media_id))
+                    part = VideoURLPart(video_url=VideoURLPart.VideoURL(url=data_url))
+                    wrapped = wrap_media_part(part, tag="video", attrs={"path": media_path})
                 image_size = None
 
         size_hint = ""
@@ -132,7 +135,7 @@ async def _read_media(self, path: KaosPath, file_type: FileType) -> ToolReturnVa
             "before continuing."
         )
         return ToolOk(
-            output=part,
+            output=wrapped,
             message=(
                 f"Loaded {file_type.kind} file `{path}` "
                 f"({file_type.mime_type}, {size} bytes{size_hint}).{note}"

src/kimi_cli/ui/shell/debug.py

@@ -51,8 +51,7 @@ def _format_content_part(part: ContentPart) -> Text | Panel | Group:
 
         case ImageURLPart(image_url=img):
             url_display = img.url[:80] + "..." if len(img.url) > 80 else img.url
-            id_text = f" (id: {img.id})" if img.id else ""
-            return Text(f"[Image{id_text}] {url_display}", style="blue")
+            return Text(f"[Image] {url_display}", style="blue")
 
         case AudioURLPart(audio_url=audio):
             url_display = audio.url[:80] + "..." if len(audio.url) > 80 else audio.url
@@ -61,8 +60,7 @@ def _format_content_part(part: ContentPart) -> Text | Panel | Group:
 
         case VideoURLPart(video_url=video):
             url_display = video.url[:80] + "..." if len(video.url) > 80 else video.url
-            id_text = f" (id: {video.id})" if video.id else ""
-            return Text(f"[Video{id_text}] {url_display}", style="blue")
+            return Text(f"[Video] {url_display}", style="blue")
 
         case _:
             return Text(f"[Unknown content type: {type(part).__name__}]", style="red")

src/kimi_cli/ui/shell/prompt.py

@@ -47,6 +47,7 @@
 from kimi_cli.ui.shell.console import console
 from kimi_cli.utils.clipboard import grab_image_from_clipboard, is_clipboard_available
 from kimi_cli.utils.logging import logger
+from kimi_cli.utils.media_tags import wrap_media_part
 from kimi_cli.utils.slashcmd import SlashCommand
 from kimi_cli.utils.string import random_string
 from kimi_cli.wire.types import ContentPart, ImageURLPart, TextPart
@@ -508,12 +509,11 @@ def _guess_image_mime(path: Path) -> str:
     return "image/png"
 
 
-def _build_image_part(image_bytes: bytes, image_id: str, mime_type: str) -> ImageURLPart:
+def _build_image_part(image_bytes: bytes, mime_type: str) -> ImageURLPart:
     image_base64 = base64.b64encode(image_bytes).decode("ascii")
     return ImageURLPart(
         image_url=ImageURLPart.ImageURL(
             url=f"data:{mime_type};base64,{image_base64}",
-            id=image_id,
         )
     )
 
@@ -607,16 +607,17 @@ def load_bytes(
             )
             return None
 
-    def load_content_part(
+    def load_content_parts(
         self, kind: CachedAttachmentKind, attachment_id: str
-    ) -> ContentPart | None:
+    ) -> list[ContentPart] | None:
         if kind == "image":
             payload = self.load_bytes(kind, attachment_id)
             if payload is None:
                 return None
             path, image_bytes = payload
             mime_type = _guess_image_mime(path)
-            return _build_image_part(image_bytes, str(path), mime_type)
+            part = _build_image_part(image_bytes, mime_type)
+            return wrap_media_part(part, tag="image", attrs={"path": str(path)})
         return None
 
 
@@ -841,9 +842,9 @@ async def prompt(self) -> UserInput:
             attachment_kind = _parse_attachment_kind(match.group("type"))
             part = None
             if attachment_kind is not None:
-                part = self._attachment_cache.load_content_part(attachment_kind, attachment_id)
+                part = self._attachment_cache.load_content_parts(attachment_kind, attachment_id)
             if part is not None:
-                content.append(part)
+                content.extend(part)
             else:
                 logger.warning(
                     "Attachment placeholder found but no matching attachment part: {placeholder}",

src/kimi_cli/utils/media_tags.py

@@ -0,0 +1,29 @@
+from __future__ import annotations
+
+from collections.abc import Mapping
+from html import escape
+
+from kimi_cli.wire.types import ContentPart, TextPart
+
+
+def _format_tag(tag: str, attrs: Mapping[str, str | None] | None = None) -> str:
+    if not attrs:
+        return f"<{tag}>"
+    rendered: list[str] = []
+    for key, value in sorted(attrs.items()):
+        if not value:
+            continue
+        rendered.append(f'{key}="{escape(str(value), quote=True)}"')
+    if not rendered:
+        return f"<{tag}>"
+    return f"<{tag} " + " ".join(rendered) + ">"
+
+
+def wrap_media_part(
+    part: ContentPart, *, tag: str, attrs: Mapping[str, str | None] | None = None
+) -> list[ContentPart]:
+    return [
+        TextPart(text=_format_tag(tag, attrs)),
+        part,
+        TextPart(text=f"</{tag}>"),
+    ]

src/kimi_cli/utils/message.py

@@ -13,14 +13,12 @@ def message_stringify(message: Message) -> str:
         if isinstance(part, TextPart):
             parts.append(part.text)
         elif isinstance(part, ImageURLPart):
-            suffix = f":{part.image_url.id}" if part.image_url.id else ""
-            parts.append(f"[image{suffix}]")
+            parts.append("[image]")
         elif isinstance(part, AudioURLPart):
             suffix = f":{part.audio_url.id}" if part.audio_url.id else ""
             parts.append(f"[audio{suffix}]")
         elif isinstance(part, VideoURLPart):
-            suffix = f":{part.video_url.id}" if part.video_url.id else ""
-            parts.append(f"[video{suffix}]")
+            parts.append("[video]")
         else:
             parts.append(f"[{part.type}]")
     return "".join(parts)

tests/test_attachment_cache.py

@@ -5,7 +5,7 @@
 from PIL import Image
 
 from kimi_cli.ui.shell.prompt import AttachmentCache, _parse_attachment_kind
-from kimi_cli.wire.types import ImageURLPart
+from kimi_cli.wire.types import ImageURLPart, TextPart
 
 
 def _make_image() -> Image.Image:
@@ -21,12 +21,15 @@ def test_attachment_cache_roundtrip(tmp_path) -> None:
     assert cached.path.exists()
     assert cached.path.parent == tmp_path / "images"
 
-    part = cache.load_content_part("image", cached.attachment_id)
-    assert isinstance(part, ImageURLPart)
-    assert part.image_url.id == str(cached.path)
-    assert part.image_url.url.startswith("data:image/png;base64,")
+    parts = cache.load_content_parts("image", cached.attachment_id)
+    assert parts is not None
+    assert len(parts) == 3
+    assert parts[0] == TextPart(text=f'<image path="{cached.path}">')
+    assert isinstance(parts[1], ImageURLPart)
+    assert parts[2] == TextPart(text="</image>")
+    assert parts[1].image_url.url.startswith("data:image/png;base64,")
 
-    encoded = part.image_url.url.split(",", 1)[1]
+    encoded = parts[1].image_url.url.split(",", 1)[1]
     assert base64.b64decode(encoded).startswith(b"\x89PNG")
 
 

tests/tools/test_read_media_file.py

@@ -12,7 +12,7 @@
 from kimi_cli.llm import ModelCapability
 from kimi_cli.soul.agent import Runtime
 from kimi_cli.tools.file.read_media import Params, ReadMediaFile
-from kimi_cli.wire.types import ImageURLPart, VideoURLPart
+from kimi_cli.wire.types import ImageURLPart, TextPart, VideoURLPart
 
 
 async def test_read_image_file(read_media_file_tool: ReadMediaFile, temp_work_dir: KaosPath):
@@ -25,8 +25,10 @@ async def test_read_image_file(read_media_file_tool: ReadMediaFile, temp_work_di
 
     assert not result.is_error
     assert isinstance(result.output, list)
-    assert len(result.output) == 1
-    part = result.output[0]
+    assert len(result.output) == 3
+    assert result.output[0] == TextPart(text=f'<image path="{image_file}">')
+    assert result.output[2] == TextPart(text="</image>")
+    part = result.output[1]
     assert isinstance(part, ImageURLPart)
     assert part.image_url.url.startswith("data:image/png;base64,")
     assert result.message == snapshot(
@@ -50,8 +52,10 @@ async def test_read_extensionless_image_file(
 
     assert not result.is_error
     assert isinstance(result.output, list)
-    assert len(result.output) == 1
-    part = result.output[0]
+    assert len(result.output) == 3
+    assert result.output[0] == TextPart(text=f'<image path="{image_file}">')
+    assert result.output[2] == TextPart(text="</image>")
+    part = result.output[1]
     assert isinstance(part, ImageURLPart)
     assert part.image_url.url.startswith("data:image/png;base64,")
     assert result.message == snapshot(
@@ -97,8 +101,10 @@ async def test_read_video_file(read_media_file_tool: ReadMediaFile, temp_work_di
 
     assert not result.is_error
     assert isinstance(result.output, list)
-    assert len(result.output) == 1
-    part = result.output[0]
+    assert len(result.output) == 3
+    assert result.output[0] == TextPart(text=f'<video path="{video_file}">')
+    assert result.output[2] == TextPart(text="</video>")
+    part = result.output[1]
     assert isinstance(part, VideoURLPart)
     assert part.video_url.url.startswith("data:video/mp4;base64,")
     assert result.message == snapshot(