From 035f6ba217c0ad6d02d7b4894dbc95db54b1bb4c Mon Sep 17 00:00:00 2001
From: Soulter <905617992@qq.com>
Date: Tue, 24 Mar 2026 11:12:30 +0800
Subject: [PATCH] feat: enhance Open API with file upload and retrieval
 endpoints, and update LiveChat message handling

---
 astrbot/dashboard/routes/live_chat.py | 172 ++++++++++++++++++++++--
 astrbot/dashboard/routes/open_api.py  |  10 +-
 docs/live-api/README.md               |  27 ++++
 docs/public/openapi.json              |  45 +++++++
 scripts/run_live_upload_flow.py       | 181 ++++++++++++++++++++++++++
 tests/test_api_key_open_api.py        |  58 +++++++++
 6 files changed, 479 insertions(+), 14 deletions(-)
 create mode 100644 scripts/run_live_upload_flow.py

diff --git a/astrbot/dashboard/routes/live_chat.py b/astrbot/dashboard/routes/live_chat.py
index f089b659f..3b00d3ca5 100644
--- a/astrbot/dashboard/routes/live_chat.py
+++ b/astrbot/dashboard/routes/live_chat.py
@@ -694,6 +694,34 @@ class LiveChatRoute(Route):
             strict=False,
         )
 
+    @staticmethod
+    def _format_live_input_preview(message_parts: list[dict]) -> str:
+        plain_texts = [
+            str(part.get("text", "")).strip()
+            for part in message_parts
+            if part.get("type") == "plain"
+            and isinstance(part.get("text"), str)
+            and str(part.get("text", "")).strip()
+        ]
+        if plain_texts:
+            return "\n".join(plain_texts)
+
+        media_counts: dict[str, int] = {}
+        for part in message_parts:
+            part_type = part.get("type")
+            if part_type in {"image", "record", "file", "video"}:
+                media_type = str(part_type)
+                media_counts[media_type] = media_counts.get(media_type, 0) + 1
+
+        if not media_counts:
+            return ""
+
+        snippets = [
+            f"{count} {media_type}"
+            for media_type, count in sorted(media_counts.items())
+        ]
+        return "[attachment] " + ", ".join(snippets)
+
     async def _handle_message(self, session: LiveChatSession, message: dict) -> None:
         """处理 WebSocket 消息"""
         msg_type = message.get("t")  # 使用 t 代替 type
@@ -758,10 +786,32 @@ class LiveChatRoute(Route):
                 return
 
             user_text = message.get("text")
-            if not isinstance(user_text, str):
-                user_text = message.get("message")
+            user_message = message.get("message")
+            message_parts = None
+            if isinstance(user_message, list):
+                message_parts = await self._build_chat_message_parts(user_message)
+                if not message_parts or not webchat_message_parts_have_content(
+                    message_parts
+                ):
+                    await websocket.send_json(
+                        {
+                            "t": "error",
+                            "data": "message must include plain text or attachment_id media",
+                            "code": "INVALID_MESSAGE_FORMAT",
+                        }
+                    )
+                    return
+                if isinstance(user_text, str):
+                    user_text = user_text.strip()
+                else:
+                    user_text = ""
+            elif isinstance(user_text, str):
+                user_text = user_text.strip()
 
-            if not isinstance(user_text, str) or not user_text.strip():
+            if not isinstance(user_text, str):
+                user_text = message.get("text")
+
+            if (not user_text or not str(user_text).strip()) and message_parts is None:
                 await websocket.send_json(
                     {
                         "t": "error",
@@ -771,12 +821,37 @@ class LiveChatRoute(Route):
                 )
                 return
 
-            await self._process_live_user_text(
-                session,
-                user_text=user_text.strip(),
-                initial_metrics={"input_type": "text"},
-                processing_start_time=time.time(),
-            )
+            if message_parts is None and (
+                not isinstance(user_text, str) or not user_text.strip()
+            ):
+                await websocket.send_json(
+                    {
+                        "t": "error",
+                        "data": "message must be non-empty text",
+                        "code": "INVALID_MESSAGE_FORMAT",
+                    }
+                )
+                return
+            if message_parts is not None and not user_text:
+                user_text = self._format_live_input_preview(message_parts)
+                if not user_text:
+                    await websocket.send_json(
+                        {
+                            "t": "error",
+                            "data": "message must include plain text or attachment_id media",
+                            "code": "INVALID_MESSAGE_FORMAT",
+                        }
+                    )
+                    return
+
+            if user_text:
+                await self._process_live_user_text(
+                    session,
+                    user_text=user_text.strip(),
+                    user_message_parts=message_parts,
+                    initial_metrics={"input_type": "text"},
+                    processing_start_time=time.time(),
+                )
 
         elif msg_type == "interrupt":
             # 用户打断
@@ -787,10 +862,27 @@ class LiveChatRoute(Route):
         self,
         session: LiveChatSession,
         user_text: str,
+        user_message_parts: list[dict] | None = None,
         initial_metrics: dict[str, Any] | None = None,
         processing_start_time: float | None = None,
     ) -> None:
         """处理 Live 用户文本：走 run_live_agent pipeline 并回传流式 TTS."""
+        payload_message = (
+            user_message_parts
+            if user_message_parts is not None
+            else [{"type": "plain", "text": user_text}]
+        )
+        if not payload_message:
+            await websocket.send_json(
+                {"t": "error", "data": "Message content is empty"}
+            )
+            return
+        display_user_text = (
+            user_text
+            if user_message_parts is None or user_text
+            else self._format_live_input_preview(user_message_parts)
+        )
+
         try:
             if initial_metrics:
                 await websocket.send_json({"t": "metrics", "data": initial_metrics})
@@ -802,7 +894,10 @@ class LiveChatRoute(Route):
             await websocket.send_json(
                 {
                     "t": "user_msg",
-                    "data": {"text": user_text, "ts": int(time.time() * 1000)},
+                    "data": {
+                        "text": display_user_text,
+                        "ts": int(time.time() * 1000),
+                    },
                 }
             )
 
@@ -814,7 +909,7 @@ class LiveChatRoute(Route):
             message_id = str(uuid.uuid4())
             payload = {
                 "message_id": message_id,
-                "message": [{"type": "plain", "text": user_text}],  # 直接发送文本
+                "message": payload_message,
                 "action_type": "live",  # 标记为 live mode
             }
 
@@ -833,7 +928,7 @@ class LiveChatRoute(Route):
                         logger.info("[Live Chat] 检测到用户打断")
                         await websocket.send_json({"t": "stop_play"})
                         await self._save_interrupted_message(
-                            session, user_text, bot_text
+                            session, display_user_text, bot_text
                         )
                         while not back_queue.empty():
                             try:
@@ -897,6 +992,18 @@ class LiveChatRoute(Route):
                             )
                         continue
 
+                    if result_chain_type in ["tool_call", "tool_call_result"]:
+                        await websocket.send_json(
+                            {
+                                "t": result_chain_type,
+                                "data": data,
+                                "chain_type": result_chain_type,
+                                "streaming": result_streaming,
+                                "message_id": message_id,
+                            }
+                        )
+                        continue
+
                     if result_type == "plain":
                         if (
                             result_streaming
@@ -911,6 +1018,47 @@ class LiveChatRoute(Route):
                             )
                         bot_text += data
 
+                    elif result_type in ["image", "record", "file", "video"]:
+                        filename = str(data)
+                        part = None
+                        if result_type == "image":
+                            filename = filename.replace("[IMAGE]", "")
+                        elif result_type == "record":
+                            filename = filename.replace("[RECORD]", "")
+                        elif result_type == "file":
+                            filename = filename.replace("[FILE]", "")
+                        else:
+                            filename = filename.replace("[VIDEO]", "")
+
+                        if filename:
+                            part = await self._create_attachment_from_file(
+                                filename=filename,
+                                attach_type=result_type,
+                            )
+
+                        if part is not None:
+                            await websocket.send_json(
+                                {
+                                    "t": result_type,
+                                    "data": {
+                                        "attachment_id": part.get("attachment_id"),
+                                        "filename": part.get("filename"),
+                                        "type": part.get("type", result_type),
+                                    },
+                                    "message_id": message_id,
+                                    "chain_type": result_type,
+                                    "streaming": result_streaming,
+                                }
+                            )
+                        elif str(data):
+                            await websocket.send_json(
+                                {
+                                    "t": result_type,
+                                    "data": {"raw": str(data)},
+                                    "message_id": message_id,
+                                }
+                            )
+
                     elif result_type == "audio_chunk":
                         if not audio_playing:
                             audio_playing = True
diff --git a/astrbot/dashboard/routes/open_api.py b/astrbot/dashboard/routes/open_api.py
index 4f790ac18..c5008b058 100644
--- a/astrbot/dashboard/routes/open_api.py
+++ b/astrbot/dashboard/routes/open_api.py
@@ -43,7 +43,10 @@ class OpenApiRoute(Route):
             "/v1/chat": ("POST", self.chat_send),
             "/v1/chat/sessions": ("GET", self.get_chat_sessions),
             "/v1/configs": ("GET", self.get_chat_configs),
-            "/v1/file": ("POST", self.upload_file),
+            "/v1/file": [
+                ("POST", self.openapi_upload_file),
+                ("GET", self.openapi_get_file),
+            ],
             "/v1/im/message": ("POST", self.send_message),
             "/v1/im/bots": ("GET", self.get_bots),
         }
@@ -571,9 +574,12 @@ class OpenApiRoute(Route):
             force_ct=force_ct,
         )
 
-    async def upload_file(self):
+    async def openapi_upload_file(self):
         return await self.chat_route.post_file()
 
+    async def openapi_get_file(self):
+        return await self.chat_route.get_attachment()
+
     async def get_chat_sessions(self):
         username, username_err = self._resolve_open_username(
             request.args.get("username")
diff --git a/docs/live-api/README.md b/docs/live-api/README.md
index 7dc897284..48c670798 100644
--- a/docs/live-api/README.md
+++ b/docs/live-api/README.md
@@ -139,6 +139,20 @@ Send a plain text input directly while using `ct=live`. The server will still ro
 }
 ```
 
+You can also send message parts and use attachment IDs (same segment format as other APIs), e.g. image/file references:
+
+```json
+{
+  "t": "text_input",
+  "message": [
+    { "type": "plain", "text": "参考这张图" },
+    { "type": "image", "attachment_id": "att_1234567890" }
+  ]
+}
+```
+
+Attachment-based inputs are accepted only when `ct=live`; this is converted to the same internal message format as chat mode and then processed by the live pipeline.
+
 #### `interrupt`
 
 Interrupt the current model or TTS response.
@@ -231,6 +245,19 @@ One TTS audio chunk, Base64 encoded.
 }
 ```
 
+Attachment results can also be returned as attachment events when produced by the model:
+
+```json
+{
+  "t": "image",
+  "data": {
+    "attachment_id": "att_1234567890",
+    "filename": "abc.jpg",
+    "type": "image"
+  }
+}
+```
+
 #### `bot_msg`
 
 Final bot text when the response completed without audio streaming.
diff --git a/docs/public/openapi.json b/docs/public/openapi.json
index 3fe71f391..9b4b82239 100644
--- a/docs/public/openapi.json
+++ b/docs/public/openapi.json
@@ -50,6 +50,51 @@
       }
     },
     "/api/v1/file": {
+      "get": {
+        "tags": [
+          "Open API"
+        ],
+        "summary": "Get attachment file",
+        "description": "Download an attachment by attachment_id.",
+        "security": [
+          {
+            "ApiKeyHeader": []
+          }
+        ],
+        "parameters": [
+          {
+            "name": "attachment_id",
+            "in": "query",
+            "required": true,
+            "schema": {
+              "type": "string"
+            },
+            "description": "Attachment ID returned by upload API."
+          }
+        ],
+        "responses": {
+          "200": {
+            "description": "Attachment binary content.",
+            "content": {
+              "application/octet-stream": {
+                "schema": {
+                  "type": "string",
+                  "format": "binary"
+                }
+              }
+            }
+          },
+          "401": {
+            "$ref": "#/components/responses/Unauthorized"
+          },
+          "403": {
+            "$ref": "#/components/responses/Forbidden"
+          },
+          "404": {
+            "description": "Attachment not found"
+          }
+        }
+      },
       "post": {
         "tags": [
           "Open API"
diff --git a/scripts/run_live_upload_flow.py b/scripts/run_live_upload_flow.py
new file mode 100644
index 000000000..eb5fc7ecb
--- /dev/null
+++ b/scripts/run_live_upload_flow.py
@@ -0,0 +1,181 @@
+import argparse
+import asyncio
+import logging
+import json
+import sys
+from pathlib import Path
+from urllib.parse import urlencode
+
+import aiohttp
+import websockets
+
+
+def build_ws_url(base_url: str, api_key: str, username: str) -> str:
+    normalized = base_url.rstrip("/")
+    if normalized.startswith("https://"):
+        ws_base = "wss://" + normalized.removeprefix("https://").removeprefix("wss://")
+    elif normalized.startswith("http://"):
+        ws_base = "ws://" + normalized.removeprefix("http://").removeprefix("wss://")
+    else:
+        ws_base = f"ws://{normalized}"
+
+    query = urlencode(
+        {
+            "api_key": api_key,
+            "username": username,
+            "ct": "live",
+        }
+    )
+    return f"{ws_base}/api/v1/live/ws?{query}"
+
+
+def build_headers(api_key: str) -> dict[str, str]:
+    return {"X-API-Key": api_key}
+
+
+def create_logger(log_file: str | None) -> logging.Logger:
+    logger = logging.getLogger("live_upload_flow")
+    if logger.handlers:
+        return logger
+
+    logger.setLevel(logging.INFO)
+    formatter = logging.Formatter(
+        "%(asctime)s | %(levelname)s | %(name)s | %(message)s"
+    )
+
+    console_handler = logging.StreamHandler(sys.stdout)
+    console_handler.setLevel(logging.INFO)
+    console_handler.setFormatter(formatter)
+    logger.addHandler(console_handler)
+
+    if log_file:
+        file_handler = logging.FileHandler(log_file, encoding="utf-8")
+        file_handler.setLevel(logging.INFO)
+        file_handler.setFormatter(formatter)
+        logger.addHandler(file_handler)
+
+    logger.propagate = False
+    return logger
+
+
+async def upload_file(session: aiohttp.ClientSession, base_url: str, api_key: str, file_path: Path) -> str:
+    logger = logging.getLogger("live_upload_flow")
+    form = aiohttp.FormData()
+    with file_path.open("rb") as file_handle:
+        form.add_field(
+            "file",
+            file_handle,
+            filename=file_path.name,
+            content_type="application/octet-stream",
+        )
+        async with session.post(
+            f"{base_url.rstrip('/')}/api/v1/file",
+            data=form,
+            headers=build_headers(api_key),
+        ) as resp:
+            payload = await resp.json()
+
+    logger.info(
+        "[UPLOAD] status=%s, attachment_id=%s",
+        payload.get("status"),
+        payload.get("data", {}).get("attachment_id"),
+    )
+    if payload.get("status") != "ok":
+        raise RuntimeError(f"Upload failed: {payload}")
+    attachment_id = payload["data"]["attachment_id"]
+    logger.info("[UPLOAD] attachment_id=%s", attachment_id)
+    return attachment_id
+
+
+async def get_file(session: aiohttp.ClientSession, base_url: str, api_key: str, attachment_id: str) -> bytes:
+    logger = logging.getLogger("live_upload_flow")
+    url = f"{base_url.rstrip('/')}/api/v1/file?{urlencode({'attachment_id': attachment_id})}"
+    async with session.get(url, headers=build_headers(api_key)) as resp:
+        logger.info("[GET] status=%s, content_type=%s", resp.status, resp.headers.get("Content-Type"))
+        if resp.status != 200:
+            text = await resp.text()
+            raise RuntimeError(f"Failed to fetch attachment: {resp.status} {text}")
+        return await resp.read()
+
+
+async def run_live_check(base_url: str, api_key: str, username: str, attachment_id: str, text: str) -> None:
+    logger = logging.getLogger("live_upload_flow")
+    ws_url = build_ws_url(base_url, api_key, username)
+    message = {
+        "t": "text_input",
+        "message": [
+            {"type": "file", "attachment_id": attachment_id},
+            {"type": "plain", "text": text},
+        ],
+    }
+
+    async with websockets.connect(ws_url) as websocket:
+        logger.info("[WS] connected: %s", ws_url)
+        logger.info("[WS] send: %s", json.dumps(message, ensure_ascii=False))
+        await websocket.send(json.dumps(message))
+
+        try:
+            while True:
+                raw = await asyncio.wait_for(websocket.recv(), timeout=90)
+                data = json.loads(raw)
+                logger.info("[WS] recv: %s", json.dumps(data, ensure_ascii=False))
+                if data.get("t") == "end":
+                    break
+        except asyncio.TimeoutError:
+            logger.warning("[WS] timeout reached, stop collecting messages")
+
+
+async def main() -> None:
+    parser = argparse.ArgumentParser(description="Upload file and test live mode input path.")
+    parser.add_argument("--base-url", default="http://localhost:6185", help="Server base URL")
+    parser.add_argument("--api-key", required=True, help="OpenAPI key")
+    parser.add_argument("--username", default="alice", help="OpenAPI username")
+    parser.add_argument("--file", required=True, type=Path, help="Local file to upload")
+    parser.add_argument(
+        "--text",
+        default="Please analyze the uploaded file.",
+        help="Additional text for live message",
+    )
+    parser.add_argument(
+        "--log-file",
+        help="Write logs to this file in addition to terminal output",
+    )
+    parser.add_argument(
+        "--skip-download-check",
+        action="store_true",
+        help="Skip GET attachment content verification",
+    )
+
+    args = parser.parse_args()
+
+    create_logger(args.log_file)
+    logger = logging.getLogger("live_upload_flow")
+
+    if not args.file.exists():
+        raise FileNotFoundError(f"file not found: {args.file}")
+    if not args.file.is_file():
+        raise ValueError(f"not a regular file: {args.file}")
+
+    timeout = aiohttp.ClientTimeout(total=30)
+    async with aiohttp.ClientSession(timeout=timeout) as session:
+        attachment_id = await upload_file(session, args.base_url, args.api_key, args.file)
+        if not args.skip_download_check:
+            content = await get_file(session, args.base_url, args.api_key, attachment_id)
+            logger.info("[GET] attachment size=%s bytes", len(content))
+
+    await run_live_check(
+        args.base_url,
+        args.api_key,
+        args.username,
+        attachment_id,
+        args.text,
+    )
+
+
+if __name__ == "__main__":
+    create_logger(None)
+    try:
+        asyncio.run(main())
+    except Exception as e:
+        logging.getLogger("live_upload_flow").error("error: %s", e, exc_info=True)
+        raise SystemExit(1)
diff --git a/tests/test_api_key_open_api.py b/tests/test_api_key_open_api.py
index a6d594c4f..8685a0da6 100644
--- a/tests/test_api_key_open_api.py
+++ b/tests/test_api_key_open_api.py
@@ -767,3 +767,61 @@ async def test_open_file_upload_requires_file_and_can_upload(
     assert isinstance(upload_data["data"]["attachment_id"], str)
     assert upload_data["data"]["filename"] == "openapi_test.txt"
     assert upload_data["data"]["type"] == "file"
+
+
+@pytest.mark.asyncio
+async def test_open_file_get_by_attachment_id_returns_content(
+    app: Quart,
+    authenticated_header: dict,
+):
+    test_client = app.test_client()
+    raw_key, _ = await _create_api_key(
+        app,
+        authenticated_header,
+        scopes=["file"],
+        name_prefix="file-get-scope-key",
+    )
+
+    upload_res = await test_client.post(
+        "/api/v1/file",
+        files={
+            "file": FileStorage(
+                stream=BytesIO(b"openapi-get-content"),
+                filename="openapi_get.txt",
+                content_type="text/plain",
+            )
+        },
+        headers={"X-API-Key": raw_key},
+    )
+    assert upload_res.status_code == 200
+    upload_data = await upload_res.get_json()
+    attachment_id = upload_data["data"]["attachment_id"]
+
+    get_res = await test_client.get(
+        f"/api/v1/file?attachment_id={attachment_id}",
+        headers={"X-API-Key": raw_key},
+    )
+    assert get_res.status_code == 200
+    assert await get_res.get_data() == b"openapi-get-content"
+
+
+@pytest.mark.asyncio
+async def test_open_file_get_attachment_id_missing_returns_error(
+    app: Quart,
+    authenticated_header: dict,
+):
+    test_client = app.test_client()
+    raw_key, _ = await _create_api_key(
+        app,
+        authenticated_header,
+        scopes=["file"],
+        name_prefix="file-get-missing-key",
+    )
+
+    missing_res = await test_client.get(
+        "/api/v1/file",
+        headers={"X-API-Key": raw_key},
+    )
+    missing_data = await missing_res.get_json()
+    assert missing_data["status"] == "error"
+    assert missing_data["message"] == "Missing key: attachment_id"