fix(mcp): 优先读取本地高清图片，补齐远程补图与转发定位参数

docs/image-mcp-dev-verification-2026-09-24.md

@@ -0,0 +1,41 @@
+# MCP 历史图片 dev 验证（2026-09-24）
+
+## 运行环境
+
+- 使用 Node 24.21.0 执行 `desktop/scripts/dev.cjs`，启动 Nuxt、Electron 和 Python 源码后端。
+- 前端 `http://127.0.0.1:3000`，后端 `http://127.0.0.1:10392`。
+- 系统默认 Node 18 启动失败；改用本机已有 Node 24，并补齐前端缺失依赖后启动成功。没有修改依赖清单或锁文件。
+- `/api/health` 返回 healthy，聊天页 HTTP 200；原生桌面确认聊天记录、图片筛选、图片查看器均正常显示。
+
+## 自动化验证
+
+运行以下测试：
+
+```powershell
+.venv/Scripts/python.exe -m pytest tests/test_mcp_router.py tests/test_chat_media_image_cache_upgrade.py tests/test_chat_large_image_frontend.py tests/test_chat_media_file_id_scope.py tests/test_chat_image_group_info.py -q
+```
+
+结果：62 passed，145 subtests passed。
+
+覆盖 MCP 参数声明、默认高清优先、显式关闭参数、转发图片定位参数，以及 MCP 调用到图片接口的补图链路。远程成功下载、额度不足、限流、账号冻结与远端失败等自动化场景使用模拟服务响应。
+
+## 真实 dev 服务验证
+
+通过运行中服务的 MCP 读取真实历史图片消息，再调用图片链接工具并下载、解码返回文件。没有调用 AI 模型。
+
+| 样本 | 返回像素尺寸 | HTTP |
+| --- | --- | --- |
+| 实时数据库中的历史图片 1 | 1722 × 1169 | 200 |
+| 实时数据库中的历史图片 2 | 872 × 1577 | 200 |
+| 实时数据库中的历史图片 3 | 938 × 1502 | 200 |
+| 解密快照中的历史图片 1 | 800 × 548 | 200 |
+| 解密快照中的历史图片 2 | 238 × 274 | 200 |
+| 解密快照中的历史图片 3 | 319 × 268 | 200 |
+
+前三张确认外部 MCP 读取未被限制为缩略图，也没有走内置 AI 的 1600 像素压缩函数。小尺寸样本只能证明当前可读取该尺寸文件，不能据此宣称它们是高清原图。
+
+对另一张本地只有 210 × 118 图片的群聊历史消息，实际设置 `fetch_remote=true`。后端成功获取 CDN token 并触发远程请求，最终返回 HTTP 404：`Large image not found locally or via CDN.`。该样本的真实远程原图下载未成功，不能把模拟下载测试等同于真实原图恢复成功。
+
+## 结论
+
+MCP 参数修复、本地大图读取、dev 桌面显示及自动化回归通过。真实旧图片的远程补图链路已触发，但所测缺图样本未恢复原图；不承诺所有历史图片均可恢复高清。

skills/wechat-mcp-copilot/references/media.md

@@ -26,4 +26,8 @@ Use this for image, video, emoji, file, link, and voice resources.
 - For phone clients, prefer `wechat.mobile.get_media_links` first.
 - MCP does not open local folders or download media into cache; use returned URLs in the client.
 - Locate the message first, then fetch media URL by message fields such as `server_id`, `username`, `md5`, or returned media references.
+- 聊天图片默认使用 `prefer_live=true`，比较本地候选图片的实际尺寸，避免重复读取旧缩略图。返回 URL 不等于已读取图片，客户端需访问 URL 并查看实际图片。
+- 需要高清图时，调用 `wechat.media.get_chat_image_url`，传入 `fetch_remote=true` 及原消息的 `server_id`、`username`、`account` 和已有的 `md5`/`file_id`。本地有大图时优先使用本地文件，否则尝试 CDN 补下载；下载可能消耗服务端额度。
+- 保留消息返回的 `src_create_time`、`file_size`、`record_index`、`record_index_path`、`record_attach` 等定位字段，尤其是合并转发中的图片。
+- 补图失败时按实际错误说明原因，不要承诺恢复高清图，也不要把缩略图当成原图。
 - For Moments, prefer local media URL fields from timeline records. Use remote video/article helpers only when the timeline record has a remote URL or article URL.

src/wechat_decrypt_tool/ai/media.py

@@ -154,7 +154,7 @@ def resolve_media(account, message, max_mb):
     if path is None and kind == "image" and raw.get("imageFileId"):
         path = _fallback_search_media_by_file_id(str(_resolve_account_wxid_dir(account_dir) or ''), raw["imageFileId"], kind="image", username=message["username"], allow_global_scan=False)
     if path is None:
-        raise ValueError("本机附件或图片缺失，请先在微信中下载并刷新")
+        raise ValueError("本机附件或图片缺失")
     path = Path(path)
     if path.stat().st_size > max_mb * 1024 * 1024:
         raise ValueError(f"附件超过 {max_mb} MB，请调整上限后重试")

src/wechat_decrypt_tool/mcp/tools.py

@@ -1357,7 +1357,13 @@ def _avatar_url(args: dict[str, Any], ctx: McpToolContext) -> dict[str, Any]:
 
 
 def _chat_image_url(args: dict[str, Any], ctx: McpToolContext) -> dict[str, Any]:
-    return _media_url("/api/chat/media/image", args, ctx, ["md5", "file_id", "server_id", "account", "username", "deep_scan", "prefer_live"])
+    # 默认比较本地候选图片的实际尺寸，避免旧缩略图缓存遮住已下载的大图。
+    args = {"prefer_live": True, **args}
+    return _media_url("/api/chat/media/image", args, ctx, [
+        "md5", "file_id", "server_id", "account", "username",
+        "src_create_time", "file_size", "record_index", "record_index_path", "record_attach",
+        "deep_scan", "prefer_live", "fetch_remote",
+    ])
 
 
 def _chat_emoji_url(args: dict[str, Any], ctx: McpToolContext) -> dict[str, Any]:
@@ -1444,7 +1450,27 @@ def _install_tools() -> None:
     _register("wechat.analytics.get_wrapped_annual", "Return full annual wrapped data. Prefer meta/card for mobile clients.", object_schema({**COMMON_ACCOUNT, "year": int_schema("Optional year.")}), _wrapped_annual, package="wechat.analytics")
 
     _register("wechat.media.get_avatar_url", "Build a URL for a contact avatar.", object_schema({**COMMON_ACCOUNT, "username": string_schema("Contact username.")}, required=["username"]), _avatar_url, package="wechat.media")
-    _register("wechat.media.get_chat_image_url", "Build a URL for a chat image message resource.", object_schema(additional_properties=True), _chat_image_url, package="wechat.media")
+    _register(
+        "wechat.media.get_chat_image_url",
+        "获取聊天图片链接，默认优先本地较高清版本。需要大图时设置 fetch_remote=true，并提供 server_id 和 username；本地缺失时尝试远程补图，可能消耗下载额度。返回链接后需实际读取图片，普通请求可能仍返回缩略图。",
+        object_schema({
+            **COMMON_ACCOUNT,
+            "md5": string_schema("图片 MD5。"),
+            "file_id": string_schema("图片文件标识。"),
+            "server_id": int_schema("原图片消息的服务端 ID，远程补图需要。"),
+            "msg_svr_id": int_schema("server_id 的兼容别名。"),
+            "username": string_schema("图片所属会话。"),
+            "src_create_time": int_schema("原消息时间戳。"),
+            "file_size": int_schema("原图片文件大小。", minimum=0),
+            "record_index": int_schema("合并转发中的图片索引。", minimum=0),
+            "record_index_path": string_schema("嵌套合并转发中的索引路径。"),
+            "record_attach": string_schema("消息返回的附件定位信息。"),
+            "deep_scan": bool_schema("允许扩大本地文件搜索范围。", default=False),
+            "prefer_live": bool_schema("比较本地候选图片尺寸，优先较高清版本。", default=True),
+            "fetch_remote": bool_schema("明确请求大图；本地缺失时尝试 CDN 下载，失败会报错。", default=False),
+        }, additional_properties=True),
+        _chat_image_url, package="wechat.media",
+    )
     _register("wechat.media.get_chat_emoji_url", "Build a URL for a chat emoji message resource.", object_schema(additional_properties=True), _chat_emoji_url, package="wechat.media")
     _register("wechat.media.get_chat_video_thumb_url", "Build a URL for a chat video thumbnail.", object_schema(additional_properties=True), _chat_video_thumb_url, package="wechat.media")
     _register("wechat.media.get_chat_video_url", "Build a URL for a chat video resource.", object_schema(additional_properties=True), _chat_video_url, package="wechat.media")

tests/test_chat_media_image_cache_upgrade.py

@@ -368,18 +368,26 @@ def test_explicit_large_image_request_uses_cdn_when_only_cached_thumb_exists(sel
                     ),
                     patch.object(chat_media.cdn_image_service, "download_original_image", downloader),
                 ):
-                    resp = client.get(
-                        "/api/chat/media/image",
-                        params={
-                            "account": account,
-                            "md5": md5,
-                            "server_id": server_id,
-                            "username": username,
-                            "prefer_live": "true",
-                            "deep_scan": "true",
-                            "fetch_remote": "true",
-                        },
-                    )
+                    # 走完整 MCP 调用和返回链接，防止链接构造时再次丢失补图参数。
+                    from wechat_decrypt_tool.routers.mcp import router as mcp_router
+
+                    client.app.include_router(mcp_router)
+                    token = "test-image-mcp-token-1234567890"
+                    with patch.dict(os.environ, {"WECHAT_TOOL_MCP_TOKEN": token}):
+                        result = client.post("/mcp", headers={"Authorization": f"Bearer {token}"}, json={
+                            "jsonrpc": "2.0", "id": 1, "method": "tools/call",
+                            "params": {
+                                "name": "wechat.media.get_chat_image_url",
+                                "arguments": {
+                                    "account": account, "md5": md5,
+                                    "server_id": server_id, "username": username,
+                                    "fetch_remote": True,
+                                },
+                            },
+                        })
+                        self.assertEqual(result.status_code, 200)
+                        image_url = result.json()["result"]["structuredContent"]["url"]
+                        resp = client.get(image_url)
 
                 self.assertEqual(resp.status_code, 200)
                 self.assertEqual(resp.content, remote_original)

tests/test_mcp_router.py

@@ -4,6 +4,7 @@
 import unittest
 from pathlib import Path
 from unittest.mock import patch
+from urllib.parse import parse_qs, urlsplit
 
 from fastapi import FastAPI
 from fastapi.testclient import TestClient
@@ -706,6 +707,12 @@ def test_media_url_helpers_pass_supported_parameters(self):
                     "account": "wxid_acc",
                     "deep_scan": True,
                     "prefer_live": True,
+                    "fetch_remote": True,
+                    "src_create_time": 1735689600,
+                    "file_size": 2048,
+                    "record_index": 0,
+                    "record_index_path": "0/1",
+                    "record_attach": "folder/image & original.dat",
                 },
             ),
         )
@@ -715,6 +722,13 @@ def test_media_url_helpers_pass_supported_parameters(self):
         self.assertEqual(image["params"]["file_id"], "fid")
         self.assertTrue(image["params"]["deep_scan"])
         self.assertTrue(image["params"]["prefer_live"])
+        query = parse_qs(urlsplit(image["url"]).query)
+        self.assertEqual(query["fetch_remote"], ["True"])
+        self.assertEqual(query["src_create_time"], ["1735689600"])
+        self.assertEqual(query["file_size"], ["2048"])
+        self.assertEqual(query["record_index"], ["0"])
+        self.assertEqual(query["record_index_path"], ["0/1"])
+        self.assertEqual(query["record_attach"], ["folder/image & original.dat"])
 
         moments_resp = client.post(
             "/mcp",
@@ -730,6 +744,28 @@ def test_media_url_helpers_pass_supported_parameters(self):
         self.assertNotIn("use_cache", moments["params"])
         self.assertNotIn("use_cache", moments["url"])
 
+    def test_image_tool_exposes_large_image_options_and_defaults(self):
+        with self._client() as client:
+            listed = client.post("/mcp", json=self._rpc("tools/list")).json()["result"]["tools"]
+            tool = next(item for item in listed if item["name"] == "wechat.media.get_chat_image_url")
+            properties = tool["inputSchema"]["properties"]
+            self.assertTrue(properties["prefer_live"]["default"])
+            self.assertFalse(properties["fetch_remote"]["default"])
+            self.assertIn("server_id", properties)
+            self.assertIn("record_index_path", properties)
+            for options, expected in [({}, "True"), ({"prefer_live": False, "fetch_remote": False}, "False")]:
+                with self.subTest(options=options):
+                    result = client.post("/mcp", json=self._rpc("tools/call", {
+                        "name": "wechat.media.get_chat_image_url",
+                        "arguments": {"md5": "abc", **options},
+                    })).json()["result"]["structuredContent"]
+                    query = parse_qs(urlsplit(result["url"]).query)
+                    self.assertEqual(query["prefer_live"], [expected])
+                    if options:
+                        self.assertEqual(query["fetch_remote"], ["False"])
+                    else:
+                        self.assertNotIn("fetch_remote", query)
+
     def test_completed_mcp_packages_and_mobile_facade_are_listed(self):
         client = self._client()