fix(project): 统一 CLI、MCP 与 Skill 的服务端 Token 流程

mcp_server/README.md

@@ -21,10 +21,18 @@
 成功返回 JSON（Pydantic）：`markdown_path`、`pages_tree_path`、`content_preview`（默认 2000 字）、`message` 等。
 调用方应以返回的 `output_dir` 为准；服务不会复用或删除已有目录。
 
+本地 PDF 固定通过 `trigger_file` 直接 multipart 上传，不使用 TOS 或自动回退。MCP 触发时固定发送
+`token=None` 和 `server_generated_token=True`；同步请求使用 `(60, 1860)`，异步请求使用 `(60, 60)`，
+只有服务端确认的 token 才能进入结果轮询。
+
 > 出于安全原因，根目录、HOME、当前工作目录及 Git 元数据目录不能作为首选输出目录。
 
 健康检查、版本查询、按 token 手动恢复请使用 CLI：`uniparser health`、`uniparser version`、`uniparser fetch`。
 
+失败返回的 `error.code` 可能为 `CONFIG_ERROR`、`INPUT_ERROR`、`UPLOAD_ERROR`、`PARSE_ERROR` 或
+`TOKEN_NOT_FOUND`。触发失败时可能保留标准 `token` 供诊断，但 MCP 不会使用它或进入恢复流程；只有
+成功触发返回的 token 才会被轮询。`undefined` 最多检查三次，不会持续等待 1800 秒。
+
 ## 环境变量
 
 | 变量 | 说明 | 默认 |
@@ -81,5 +89,5 @@ uv run pytest tests/ -v
 
 `UNIPARSER_BASE_URL` 可省略（默认云服务）；本地自托管时设置为 `http://127.0.0.1:40001` 等。
 
-同步解析使用 `UniParserClient.sync_request_timeout`，轮询和结果获取使用
-`request_timeout`；二者均为客户端 HTTP 超时，与服务端解析预算相互独立。
+本地 PDF 同步直传显式使用 `(60, 1860)`，异步直传显式使用 `(60, 60)`；轮询和结果获取使用
+`UniParserClient.request_timeout`。这些都是客户端 HTTP 超时，与服务端解析预算相互独立。

mcp_server/tests/test_pipeline_parse.py

@@ -1,39 +1,35 @@
 import asyncio
+import json
 from pathlib import Path
 from unittest.mock import MagicMock
 
 from uniparser_mcp.pipeline.parse import run_parse
 from uniparser_mcp.schemas import ParseRequest
 
 
-def test_duplicate_triggers_internal_complete(tmp_path: Path):
+def test_failed_trigger_does_not_attempt_token_recovery(tmp_path: Path):
     pdf = tmp_path / "paper.pdf"
     pdf.write_bytes(b"%PDF-1.4")
 
     client = MagicMock()
-    client.to_token.return_value = "tok123"
     client.trigger_file.return_value = {
         "status": "error",
-        "message": "Token is duplicated",
-        "token": "tok123",
-    }
-    client.get_result.side_effect = [
-        {"status": "success"},
-        {"status": "success", "pages_tree": True},
-    ]
-    client.get_formatted.return_value = {
-        "status": "success",
-        "content": "# Hello",
+        "token": "diagnostic-token",
+        "message": "trigger failed",
     }
 
-    req = ParseRequest(file_path=str(pdf), output_dir=str(tmp_path / "out"))
+    out = tmp_path / "out"
+    req = ParseRequest(file_path=str(pdf), output_dir=str(out))
     result = asyncio.run(run_parse(client, req))
 
-    assert result.ok is True
-    assert result.token == "tok123"
-    assert result.trigger_meta_path is None
-    assert Path(result.markdown_path).is_file()
+    assert result.ok is False
+    assert result.error.code == "PARSE_ERROR"
+    assert result.error.token == "diagnostic-token"
+    saved_error = json.loads((out / "trigger_error.json").read_text(encoding="utf-8"))
+    assert saved_error["token"] == "diagnostic-token"
     client.trigger_file.assert_called_once()
+    client.get_result.assert_not_called()
+    client.get_formatted.assert_not_called()
 
 
 def test_success_parse_writes_trigger_meta(tmp_path: Path):
@@ -58,6 +54,60 @@ def test_success_parse_writes_trigger_meta(tmp_path: Path):
     assert result.ok is True
     assert result.trigger_meta_path is not None
     assert Path(result.trigger_meta_path).is_file()
+    trigger_args, trigger_kwargs = client.trigger_file.call_args
+    assert trigger_args == (str(pdf.resolve()),)
+    assert trigger_kwargs["token"] is None
+    assert trigger_kwargs["server_generated_token"] is True
+    assert trigger_kwargs["http_timeout"] == (60.0, 1860.0)
+
+
+def test_async_local_pdf_uses_direct_upload_timeout(tmp_path: Path):
+    pdf = tmp_path / "paper.pdf"
+    pdf.write_bytes(b"%PDF-1.4")
+    client = MagicMock()
+    client.trigger_file.return_value = {"status": "error", "description": "stop"}
+
+    req = ParseRequest(file_path=str(pdf), output_dir=str(tmp_path / "out"), async_mode=True)
+    result = asyncio.run(run_parse(client, req))
+
+    assert result.ok is False
+    _, trigger_kwargs = client.trigger_file.call_args
+    assert trigger_kwargs["token"] is None
+    assert trigger_kwargs["server_generated_token"] is True
+    assert trigger_kwargs["http_timeout"] == (60.0, 60.0)
+    assert trigger_kwargs["sync"] is False
+
+
+def test_direct_upload_transport_failure_uses_upload_error(tmp_path: Path):
+    pdf = tmp_path / "paper.pdf"
+    pdf.write_bytes(b"%PDF-1.4")
+    client = MagicMock()
+    client.trigger_file.return_value = {
+        "status": "error",
+        "description": "The write operation timed out",
+        "error_type": "WriteTimeout",
+    }
+
+    req = ParseRequest(file_path=str(pdf), output_dir=str(tmp_path / "out"))
+    result = asyncio.run(run_parse(client, req))
+
+    assert result.ok is False
+    assert result.error.code == "UPLOAD_ERROR"
+    assert result.error.stage == "trigger_file"
+    assert result.error.token is None
+
+
+def test_url_trigger_uses_server_generated_token(tmp_path: Path):
+    client = MagicMock()
+    client.trigger_url.return_value = {"status": "error", "description": "stop"}
+
+    req = ParseRequest(pdf_url="https://example.com/paper.pdf", output_dir=str(tmp_path / "out"))
+    result = asyncio.run(run_parse(client, req))
+
+    assert result.ok is False
+    _, trigger_kwargs = client.trigger_url.call_args
+    assert trigger_kwargs["token"] is None
+    assert trigger_kwargs["server_generated_token"] is True
 
 
 def test_existing_output_is_preserved_and_result_uses_suffixed_sibling(tmp_path: Path):

mcp_server/tests/test_pipeline_poll.py

@@ -0,0 +1,39 @@
+import asyncio
+from unittest.mock import MagicMock
+
+import pytest
+from uniparser_mcp.pipeline.poll import poll_until_success
+
+
+def test_undefined_token_stops_after_bounded_checks(monkeypatch: pytest.MonkeyPatch):
+    async def no_sleep(_seconds: float) -> None:
+        return None
+
+    monkeypatch.setattr("uniparser_mcp.pipeline.poll.asyncio.sleep", no_sleep)
+    client = MagicMock()
+    client.get_result.return_value = {"status": "undefined"}
+
+    result = asyncio.run(poll_until_success(client, "missing-token"))
+
+    assert result.ok is False
+    assert result.error.code == "TOKEN_NOT_FOUND"
+    assert result.error.token == "missing-token"
+    assert client.get_result.call_count == 3
+
+
+def test_pending_statuses_still_reach_success(monkeypatch: pytest.MonkeyPatch):
+    async def no_sleep(_seconds: float) -> None:
+        return None
+
+    monkeypatch.setattr("uniparser_mcp.pipeline.poll.asyncio.sleep", no_sleep)
+    client = MagicMock()
+    client.get_result.side_effect = [
+        {"status": "waiting"},
+        {"status": "processing"},
+        {"status": "success"},
+    ]
+
+    result = asyncio.run(poll_until_success(client, "valid-token"))
+
+    assert result == {"status": "success"}
+    assert client.get_result.call_count == 3

mcp_server/uniparser_mcp/defaults.py

@@ -2,8 +2,11 @@
 
 POLL_INTERVAL_SEC = 3
 POLL_TIMEOUT_SEC = 1800
+UNDEFINED_MAX_POLLS = 3
+DIRECT_UPLOAD_REQUEST_TIMEOUT = (60.0, 60.0)
+DIRECT_SYNC_UPLOAD_REQUEST_TIMEOUT = (60.0, 1860.0)
 
-PENDING_STATUSES = frozenset({"undefined", "waiting", "processing"})
+PENDING_STATUSES = frozenset({"waiting", "processing"})
 
 IMAGE_SUFFIXES = frozenset({".png", ".jpg", ".jpeg", ".webp", ".gif", ".bmp", ".tif", ".tiff"})
 

mcp_server/uniparser_mcp/errors.py

@@ -13,12 +13,31 @@ def input_error(message: str) -> ErrorResult:
     return ErrorResult(error=ErrorDetail(code="INPUT_ERROR", message=message))
 
 
-def parse_error(stage: str, result: dict) -> ErrorResult:
+def _operation_error(code: str, stage: str, result: dict) -> ErrorResult:
     return ErrorResult(
         error=ErrorDetail(
-            code="PARSE_ERROR",
+            code=code,
             message=result.get("description") or result.get("message") or str(result),
             stage=stage,
             token=result.get("token"),
         )
     )
+
+
+def parse_error(stage: str, result: dict) -> ErrorResult:
+    return _operation_error("PARSE_ERROR", stage, result)
+
+
+def upload_error(stage: str, result: dict) -> ErrorResult:
+    return _operation_error("UPLOAD_ERROR", stage, result)
+
+
+def token_not_found_error(token: str, *, attempts: int) -> ErrorResult:
+    return ErrorResult(
+        error=ErrorDetail(
+            code="TOKEN_NOT_FOUND",
+            message=f"The service did not recognize this token after {attempts} checks.",
+            stage="get_result_poll",
+            token=token,
+        ),
+    )

mcp_server/uniparser_mcp/instructions.py

@@ -21,5 +21,8 @@
 
 ## After calling uniparser_parse
 
-Always include the `message` field from the tool response in your reply to the user.
+On success, always include the `message` field from the tool response in your reply to the user.
+On failure, report `error.message`. Never poll a token from a failed trigger; it is diagnostic only. Only a
+token returned with `status=success` may be polled. Do not attempt recovery from a failed trigger.
+Stop retrying when the error code is `TOKEN_NOT_FOUND`.
 """

mcp_server/uniparser_mcp/pipeline/parse.py

@@ -7,7 +7,11 @@
 
 from mcp.server.fastmcp import Context
 
-from uniparser_mcp.errors import input_error, parse_error
+from uniparser_mcp.defaults import (
+    DIRECT_SYNC_UPLOAD_REQUEST_TIMEOUT,
+    DIRECT_UPLOAD_REQUEST_TIMEOUT,
+)
+from uniparser_mcp.errors import input_error, parse_error, upload_error
 from uniparser_mcp.input import InputKind, ResolvedInput, display_label, resolve_request
 from uniparser_mcp.parse_options import resolve_trigger_kwargs, serialize_trigger_kwargs
 from uniparser_mcp.pipeline.output import (
@@ -20,14 +24,7 @@
 from uniparser_mcp.schemas import ErrorResult, ParseRequest, ParseResult
 from uniparser_mcp.tools.response import build_parse_success
 from uniparser_tools.api.clients import UniParserClient
-from uniparser_tools.common.constant import ErrorFlag, FormatFlag
-
-
-def _is_token_duplicated(trigger: dict[str, Any]) -> bool:
-    if trigger.get("status") == "success":
-        return False
-    message = str(trigger.get("message") or trigger.get("description") or "")
-    return message == ErrorFlag.Token_Duplicated or "Token is duplicated" in message
+from uniparser_tools.common.constant import FormatFlag
 
 
 async def _ctx_info(ctx: Context | None, message: str) -> None:
@@ -48,21 +45,34 @@ async def _trigger_input(
             client.trigger_file,
             str(resolved.path),
             token=None,
+            server_generated_token=True,
+            http_timeout=(
+                DIRECT_SYNC_UPLOAD_REQUEST_TIMEOUT
+                if trigger_kwargs.get("sync", True)
+                else DIRECT_UPLOAD_REQUEST_TIMEOUT
+            ),
             **trigger_kwargs,
         )
+        if not isinstance(trigger, dict):
+            trigger = {
+                "status": "error",
+                "message": "Direct upload returned an invalid response",
+            }
         return trigger, "trigger_file"
     if resolved.kind is InputKind.IMAGE:
         trigger = await asyncio.to_thread(
             client.trigger_snip,
             str(resolved.path),
             token=None,
+            server_generated_token=True,
             **trigger_kwargs,
         )
         return trigger, "trigger_snip"
     trigger = await asyncio.to_thread(
         client.trigger_url,
         resolved.raw,
         token=None,
+        server_generated_token=True,
         **trigger_kwargs,
     )
     return trigger, "trigger_url"
@@ -156,19 +166,10 @@ async def run_parse(client: UniParserClient, req: ParseRequest, ctx: Context | N
     await _ctx_info(ctx, f"Parsing {display_label(resolved)}")
     trigger, stage = await _trigger_input(client, resolved, trigger_kwargs=trigger_kwargs)
 
-    if _is_token_duplicated(trigger):
-        token = trigger.get("token") or client.to_token(resolved.token_seed)
-        return await _complete_by_token(
-            client,
-            token,
-            resolved=resolved,
-            out_dir=out_dir,
-            ctx=ctx,
-            write_trigger_meta_file=False,
-        )
-
     if trigger.get("status") != "success":
         save_stage_error(out_dir, "trigger_error.json", trigger)
+        if stage == "trigger_file" and trigger.get("error_type"):
+            return upload_error(stage, trigger)
         return parse_error(stage, trigger)
 
     token = trigger.get("token")

mcp_server/uniparser_mcp/pipeline/poll.py

@@ -6,17 +6,18 @@
 import time
 from typing import Any
 
-from uniparser_mcp.defaults import PENDING_STATUSES, POLL_INTERVAL_SEC, POLL_TIMEOUT_SEC
-from uniparser_mcp.errors import parse_error
+from uniparser_mcp.defaults import PENDING_STATUSES, POLL_INTERVAL_SEC, POLL_TIMEOUT_SEC, UNDEFINED_MAX_POLLS
+from uniparser_mcp.errors import parse_error, token_not_found_error
 from uniparser_mcp.schemas import ErrorResult
 from uniparser_tools.api.clients import UniParserClient
 
 
 async def poll_until_success(client: UniParserClient, token: str) -> dict[str, Any] | ErrorResult:
-    deadline = time.time() + POLL_TIMEOUT_SEC
+    deadline = time.monotonic() + POLL_TIMEOUT_SEC
     last: dict[str, Any] = {}
+    undefined_polls = 0
 
-    while time.time() < deadline:
+    while time.monotonic() < deadline:
         last = await asyncio.to_thread(
             client.get_result,
             token,
@@ -30,7 +31,14 @@ async def poll_until_success(client: UniParserClient, token: str) -> dict[str, A
             return last
         if status == "error":
             return parse_error("get_result_poll", last)
-        if status in PENDING_STATUSES or status is None:
+        if status == "undefined":
+            undefined_polls += 1
+            if undefined_polls >= UNDEFINED_MAX_POLLS:
+                return token_not_found_error(token, attempts=undefined_polls)
+            await asyncio.sleep(POLL_INTERVAL_SEC)
+            continue
+        if status in PENDING_STATUSES:
+            undefined_polls = 0
             await asyncio.sleep(POLL_INTERVAL_SEC)
             continue
         return parse_error("get_result_poll", last)