fix(agent): make API input errors actionable

app/agent/api/arguments.py

@@ -1,5 +1,6 @@
 """按已生成的 API 输入合同规范实际请求，使相同写入使用稳定参数身份。"""
 
+import re
 from copy import deepcopy
 from typing import Any
 
@@ -17,6 +18,49 @@ _MISSING = object()
 _CONTRACT_DESCRIPTION_MAX_CHARS = 180
 
 
+class ApiArgumentError(ValueError):
+    """只携带字段位置和合同说明，不回显模型提交的参数值。"""
+
+    def __init__(self, message: str, *, path: str = "") -> None:
+        """保留独立字段路径，便于嵌套对象逐层补齐错误位置。"""
+        self.detail = message
+        self.path = path
+        super().__init__(f"{path}: {message}" if path else message)
+
+
+def _field_error(field: str, error: ValueError) -> ApiArgumentError:
+    """拼接由合同声明的字段路径，同时移除 Pydantic 错误中的原始输入。"""
+    if isinstance(error, ApiArgumentError):
+        path = f"{field}.{error.path}" if error.path else field
+        return ApiArgumentError(error.detail, path=path)
+    detail = "API 参数类型不符合输入合同" if isinstance(error, ValidationError) else str(error)
+    return ApiArgumentError(detail, path=field)
+
+
+def _unknown_fields_message(fields: set[str]) -> str:
+    """仅展示有界参数标识符，避免把误作字段名的 URL 或凭据带回模型。"""
+    names = sorted(name for name in fields if re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]{0,63}", name))
+    suffix = "；其余字段名已省略" if len(names) != len(fields) or len(names) > 8 else ""
+    return f"API 参数包含合同未声明的字段：{', '.join(names[:8])}{suffix}"
+
+
+def _validate_argument_locations(arguments: dict[str, Any], schema: dict[str, Any]) -> None:
+    """在通用模型丢弃额外字段前拒绝平铺参数，并按精确合同提示合法位置。"""
+    unknown = set(arguments) - {"operation_id", "path_params", "query", "body"}
+    if not unknown:
+        return
+    contract = api_input_contract(str(arguments.get("operation_id") or ""), schema)
+    hints = []
+    for container in ("path_params", "query", "body"):
+        fields = contract.get(container, {}).get("fields", {})
+        for name in sorted(unknown.intersection(fields))[:8]:
+            hints.append(f"{name} 应移入 {container}.{name}")
+    message = _unknown_fields_message(unknown)
+    if hints:
+        message += "；" + "；".join(hints)
+    raise ApiArgumentError(message)
+
+
 def _resolve_schema(schema: dict[str, Any], definitions: dict[str, Any]) -> dict[str, Any]:
     """只展开缓存合同的本地引用，拒绝循环引用和任何远程 schema。"""
     seen = set()
@@ -144,6 +188,8 @@ def _normalize_union(value: Any, schema: dict[str, Any], definitions: dict[str,
     """优先选择原始类型匹配的联合，歧义时不猜测不同分支的默认值。"""
     alternatives = [_resolve_schema(item, definitions) for item in schema.get("oneOf", schema.get("anyOf", []))]
     exact = [item for item in alternatives if item.get("type") == _value_type(value)]
+    if len(exact) == 1:
+        return _normalize_value(value, exact[0], definitions, depth + 1)
     candidates = []
     for option in exact or alternatives:
         try:
@@ -166,7 +212,7 @@ def _normalize_object(value: dict[str, Any], schema: dict[str, Any], definitions
     extra = schema.get("additionalProperties")
     unknown = set(value) - set(properties)
     if unknown and extra is False:
-        raise ValueError("API 参数包含合同未声明的字段")
+        raise ApiArgumentError(_unknown_fields_message(unknown))
     normalized = {key: deepcopy(value[key]) for key in unknown} if extra is True or isinstance(extra, dict) else {}
     required = schema.get("required", [])
     for key, declaration in properties.items():
@@ -174,9 +220,12 @@ def _normalize_object(value: dict[str, Any], schema: dict[str, Any], definitions
         current = value.get(key, field_schema.get("default", _MISSING))
         if current is _MISSING:
             if key in required:
-                raise ValueError("API 参数缺少必需字段")
+                raise ApiArgumentError("API 参数缺少必需字段", path=key)
             continue
-        normalized[key] = _normalize_value(current, field_schema, definitions, depth + 1)
+        try:
+            normalized[key] = _normalize_value(current, field_schema, definitions, depth + 1)
+        except ValueError as error:
+            raise _field_error(key, error) from error
     return normalized
 
 
@@ -219,6 +268,7 @@ def canonical_api_arguments(arguments: dict[str, Any], schema: dict[str, Any]) -
     route = resolve_api_route(str(operation_id or ""))
     if branch is None or route is None:
         raise ValueError("API 操作没有规范参数合同")
+    _validate_argument_locations(arguments, schema)
     values = deepcopy(arguments)
     if route.method == "GET" and isinstance(values.get("body"), dict):
         values["query"] = {**(values.get("query") or {}), **values.pop("body")}

app/agent/tools/impl/api.py

@@ -4,9 +4,10 @@ import json
 from copy import deepcopy
 from functools import lru_cache
 from pathlib import Path
-from typing import Any, Dict, Literal, Optional, Type, Union
+from typing import Any, Dict, Literal, Optional, Type, Union, cast
 
-from pydantic import BaseModel, Field, PrivateAttr
+from langchain_core.tools import ToolException
+from pydantic import BaseModel, ConfigDict, Field, PrivateAttr
 
 from app.agent.api.arguments import api_input_contract, canonical_api_arguments
 from app.agent.api.executor import ApiExecutionContext, ApiExecutionError, MoviePilotApiExecutor
@@ -45,6 +46,8 @@ def _load_api_mcp_input_schema() -> dict[str, Any]:
 class MoviePilotApiInput(BaseModel):  # type: ignore[misc]
     """MoviePilot API 网关的结构化输入参数。"""
 
+    model_config = ConfigDict(extra="forbid")
+
     operation_id: str = Field(
         ...,
         description=(
@@ -54,11 +57,14 @@ class MoviePilotApiInput(BaseModel):  # type: ignore[misc]
     )
     path_params: Dict[str, Any] = Field(
         default_factory=dict,
-        description="Route placeholder values declared by the selected operation.",
+        description="Route placeholders only, e.g. media.detail uses {\"media_id\": \"27205\"}.",
     )
     query: Dict[str, Any] = Field(
         default_factory=dict,
-        description="Query-string fields declared by the selected operation.",
+        description=(
+            "Query fields declared by this operation, never flattened at the top level. "
+            "Pagination is operation-specific: subscription.execution.list uses limit, not page/count."
+        ),
     )
     body: MoviePilotApiBody = Field(
         default=None,
@@ -91,10 +97,18 @@ class MoviePilotApiTool(MoviePilotTool):
     description: str = (
         "Call allowlisted MoviePilot business APIs through operation-specific input contracts. "
         "Load the relevant domain Skill before calling and use operation error feedback to correct inputs. "
+        "The ONLY top-level keys are operation_id, path_params, query, body. "
+        "Examples (replace sample IDs with IDs from prior results): "
+        '{"operation_id":"media.detail","path_params":{"media_id":"27205"},'
+        '"query":{"media_source":"tmdb","type_name":"电影"}}; '
+        '{"operation_id":"subscription.execution.list","query":{"limit":10}}; '
+        '{"operation_id":"site.rss","query":{"page":1,"count":20}}. '
+        "Never invent filters or copy pagination fields from another operation. "
         "Arbitrary URLs, commands, authentication endpoints, headers, and tokens are forbidden."
     )
     require_admin: bool = False
     args_schema: Type[BaseModel] = MoviePilotApiInput
+    handle_tool_error: bool = True
 
     _executor: Optional[MoviePilotApiExecutor] = PrivateAttr(default=None)
 
@@ -141,10 +155,26 @@ class MoviePilotApiTool(MoviePilotTool):
         """返回包含全部白名单 operation 精确参数的 MCP JSON Schema。"""
         return deepcopy(_load_api_mcp_input_schema())
 
+    @property
+    def tool_call_schema(self) -> Type[BaseModel]:
+        """本工具没有注入参数，直接保留原模型，防止子集投影丢失 extra=forbid。"""
+        return MoviePilotApiInput
+
+    def _parse_input(
+        self, tool_input: Union[str, dict[str, Any]], tool_call_id: Optional[str],
+    ) -> Union[str, dict[str, Any]]:
+        """在 LangChain 解析前校验原始参数，让只读调用也能收到安全纠错回执。"""
+        if isinstance(tool_input, dict):
+            operation_id = str(tool_input.get("operation_id") or "")
+            try:
+                self.canonical_arguments(tool_input)
+            except (TypeError, ValueError) as error:
+                raise ToolException(self._invalid_input_result(operation_id, error)) from error
+        return cast(Union[str, dict[str, Any]], super()._parse_input(tool_input, tool_call_id))
+
     def canonical_arguments(self, arguments: dict[str, Any]) -> dict[str, Any]:
         """用缓存的 operation 合同生成实际执行与持久指纹共用的参数。"""
-        validated = MoviePilotApiInput.model_validate(arguments).model_dump(mode="json")
-        return canonical_api_arguments(validated, _load_api_mcp_input_schema())
+        return canonical_api_arguments(arguments, _load_api_mcp_input_schema())
 
     def get_operation_input_contract(self, operation_id: str) -> dict[str, Any]:
         """返回单个 operation 的有界参数合同，供失败回执指导模型重试。"""
@@ -285,7 +315,6 @@ class MoviePilotApiTool(MoviePilotTool):
         :param body: JSON 请求体
         :return: 结构化业务结果或安全错误消息
         """
-        del kwargs
         spec = resolve_api_operation(operation_id)
         if spec is None:
             return json.dumps(
@@ -297,7 +326,7 @@ class MoviePilotApiTool(MoviePilotTool):
                 ensure_ascii=False,
             )
         try:
-            arguments: dict[str, Any] = {"operation_id": operation_id}
+            arguments: dict[str, Any] = {"operation_id": operation_id, **kwargs}
             if path_params is not None:
                 arguments["path_params"] = path_params
             if query is not None:

docs/mcp-api.md

@@ -188,6 +188,15 @@ operation ID、权限、副作用、确认、恢复、结果敏感性及精确
 
 只允许传 `tools/list` 对应 operation 分支中声明的 `path_params`、`query` 和 `body` 字段。不得传 URL、认证头、API Token 或任意 HTTP 方法。
 
+内置 Agent 工具同样禁止额外顶层字段。业务参数不能平铺在 `operation_id` 旁边：
+例如 `media.detail` 的 `media_id` 属于 `path_params`，`media_source` 与 `type_name` 属于 `query`；
+`subscription.execution.list` 使用 `query.limit`，不能套用其它列表的 `page` / `count`；
+`site.rss` 支持 `query.page` / `query.count`，不接受 `site_id` 筛选。
+工具说明内联了这三个接口的完整 JSON 示例，示例 ID 必须替换为前序查询返回的真实 ID。
+参数错误在请求发出前返回纠错回执和该 operation 的 `input_contract`（网关错误码为 `invalid_input`），
+并指出具体缺失或错误字段（如 `query.media_source`）、未声明字段或顶层参数的正确位置，
+不回显参数值。模型应根据该合同修正参数后重试；写入参数错误不会创建执行认领。
+
 `body` 使用原生 JSON 值。对象和数组直接传入；仅在选定的 operation 允许时传 `null`。当前唯一的字符串请求体为 `system.upgrade.dev` 的固定值 `"dev"`。网关会按选定的 operation 合同校验请求体类型和字段。
 
 `search.torrents` 可能按媒体标题与别名分轮搜索站点：MoviePilot 内层请求最多等待 290 秒，Agent 工具总等待上限为 300 秒，预留 10 秒处理超时与返回结果；V3 MCP 包装层继续使用原有 `mcp_proxy_timeout` 配置（默认 600 秒）。其它 operation 沿用原有超时配置。`page` / `count` 只分页已完成的搜索结果，不会缩短站点搜索过程。

skills/moviepilot-api/SKILL.md

@@ -1,6 +1,6 @@
 ---
 name: moviepilot-api
-version: 38
+version: 39
 description: >-
   Use this skill for MoviePilot product operations such as media search, torrent
   search, downloads, subscriptions, library checks, sites, storage, workflows,
@@ -119,6 +119,11 @@ downloader mappings when switching path modes.
    identifiers and use the documented pagination fields. Pass object and array
    bodies as native JSON values, and pass null only when the selected operation
    allows it. The only string body is the literal `"dev"` for `system.upgrade.dev`.
+   Never flatten operation fields next to `operation_id`: put each in its declared
+   `path_params`, `query`, or `body` container. Do not copy pagination or filters
+   from another operation. An `invalid_input` response names the failing field
+   and includes `input_contract`; correct that field and all required fields
+   before retrying. Invalid input is rejected before the API request is sent.
 5. Obtain confirmation for confirmation-protected or side-effecting operations,
    then execute the gateway call once.
 6. Inspect `success`, `execution_outcome`, errors, empty results, and collection
@@ -126,6 +131,21 @@ downloader mappings when switching path modes.
 7. Verify writes with the category's read-back operation when the contract
    requires it; do not repeat a write whose outcome is `unknown`.
 
+### Call Shape Examples
+
+Replace example IDs with the exact identifiers returned by earlier calls. These
+are separate operation contracts, not interchangeable parameter templates.
+
+```json
+{"operation_id":"media.detail","path_params":{"media_id":"27205"},"query":{"media_source":"tmdb","type_name":"\u7535\u5f71"}}
+{"operation_id":"subscription.execution.list","query":{"limit":10}}
+{"operation_id":"site.rss","query":{"page":1,"count":20}}
+```
+
+`media.detail` requires both the source-native ID and its source/type. Recent
+subscription executions use `limit`, not `page` or `count`. `site.rss` lists
+RSS-enabled sites; it does not accept a `site_id` filter.
+
 ## API Category Index
 
 Each category file contains the complete operation contracts for its namespace.

tests/test_agent_api_gateway.py

@@ -5,6 +5,7 @@ from types import SimpleNamespace
 from unittest.mock import AsyncMock, Mock, patch
 
 import pytest
+from langchain_core.utils.function_calling import convert_to_openai_tool
 
 # pylint: disable=no-name-in-module  # 策略包根通过 __getattr__ 惰性导出，Pylint 无法静态解析。
 from app.agent.policy import (
@@ -25,7 +26,7 @@ from app.agent.policy.api import (
 )
 from app.agent.tools.factory import MoviePilotToolFactory
 from app.agent.tools.impl.agent_task import AgentTaskTool
-from app.agent.tools.impl.api import MoviePilotApiTool
+from app.agent.tools.impl.api import MoviePilotApiInput, MoviePilotApiTool
 from app.agent.tools.impl.execute_command import ExecuteCommandTool
 from app.agent.tools.manager import MoviePilotToolsManager
 from app.schemas.types import NotificationChannel
@@ -258,6 +259,56 @@ def test_operation_body_shapes_fit_local_agent_input_schema() -> None:
     assert string_bodies == {("system.upgrade.dev", "dev")}
 
 
+def test_local_api_schema_forbids_flattened_arguments_and_examples_match_contracts() -> None:
+    """模型实际接收的 schema 禁止额外顶层字段，内联示例均通过真实 operation 合同。"""
+    tool = MoviePilotApiTool(session_id="session", user_id="1")
+    definition = convert_to_openai_tool(tool)["function"]
+    schema = definition["parameters"]
+    assert definition["description"] == tool.description
+    assert schema["additionalProperties"] is False
+    assert set(schema["properties"]) == {"operation_id", "path_params", "query", "body"}
+    with pytest.raises(ValueError):
+        MoviePilotApiInput.model_validate({"operation_id": "site.rss", "page": 1})
+
+    examples = tool.description.split("Examples (replace sample IDs with IDs from prior results): ")[1]
+    decoder = json.JSONDecoder()
+    operations = []
+    for _ in range(3):
+        example, end = decoder.raw_decode(examples)
+        operations.append(tool.canonical_arguments(example)["operation_id"])
+        examples = examples[end:].lstrip("; ")
+    assert operations == ["media.detail", "subscription.execution.list", "site.rss"]
+
+
+@pytest.mark.parametrize(("arguments", "expected"), [
+    ({"operation_id": "media.detail", "media_id": "secret-value", "media_source": "tmdb"},
+     ["media_id 应移入 path_params.media_id", "media_source 应移入 query.media_source"]),
+    ({"operation_id": "subscription.execution.list", "limit": 2}, ["limit 应移入 query.limit"]),
+    ({"operation_id": "site.rss", "query": {"site_id": "secret-value"}}, ["query", "site_id"]),
+    ({"operation_id": "media.detail"}, ["path_params.media_id"]),
+    ({"operation_id": "media.detail", "path_params": {"media_id": "27205"}}, ["query.media_source"]),
+    ({"operation_id": "subscription.execution.list", "query": {"limit": "secret-value"}}, ["query.limit"]),
+    ({"operation_id": "site.list", "query": {"status": "secret-value"}}, ["query.status"]),
+    ({"operation_id": "download.add", "torrent_in": {}}, ["torrent_in 应移入 body.torrent_in"]),
+    ({"operation_id": "config.system.update", "body": {"value": "secret-value"}}, ["body.setting_key"]),
+    ({"operation_id": "site.rss", "https://secret-value.invalid/": "secret-value"}, ["字段名已省略"]),
+])
+@pytest.mark.asyncio
+async def test_gateway_raw_tool_inputs_return_actionable_errors_without_execution(arguments, expected) -> None:
+    """覆盖 LangChain 和直接调用入口，错误字段不能静默丢弃、执行或泄露参数值。"""
+    executor = AsyncMock()
+    gateway = MoviePilotApiTool(session_id="session", user_id="1", executor=executor)
+    message = await gateway.ainvoke({"name": gateway.name, "args": arguments, "id": "invalid", "type": "tool_call"})
+    assert message.status == "error"
+    for result in (message.content, await gateway.run(**arguments)):
+        payload = json.loads(result)
+        assert payload["error"] == "invalid_input"
+        assert payload["input_contract"]["operation_id"] == arguments["operation_id"]
+        assert all(item in payload["message"] for item in expected)
+        assert "secret-value" not in result
+    executor.execute.assert_not_awaited()
+
+
 def test_mcp_collection_contract_distinguishes_exact_and_unavailable_totals() -> None:
     """MCP 必须说明缺省全量、精确总数和外部无总数三种集合语义。"""
     schema = MoviePilotApiTool(session_id="session", user_id="api_user").get_mcp_input_schema()

tests/test_agent_invocation_middleware.py

@@ -332,9 +332,12 @@ async def test_unresolved_write_does_not_block_another_host_user_or_session(invo
         arguments = {**WRITE_ARGUMENTS, "user_id": "injected-user", "session_id": "injected-session"}
         graph, _model, _middleware = _graph(repository, [
             AIMessage(content="", tool_calls=[_call("same-provider-call-id", arguments)]),
+            AIMessage(content="", tool_calls=[_call("corrected-call-id", WRITE_ARGUMENTS)]),
             AIMessage(content="待核验"),
         ], context=context)
-        await _invoke(graph)
+        messages = _tool_messages(await _invoke(graph))
+        assert messages[0].status == "error"
+        assert "session_id, user_id" in messages[0].content
     assert run.await_count == 3
     assert {(row["principal_id"], row["session_id"]) for row in records()} == {
         (context.user_id, context.session_id) for context in contexts
@@ -412,6 +415,65 @@ async def test_api_effective_parameters_prevent_unknown_retry_bypass(invocation_
     assert "ignored_by_endpoint" not in (normalized.get("body") or {})
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize(("invalid", "corrected", "hint"), [
+    (
+        {"operation_id": "media.detail", "media_id": "27205", "media_source": "tmdb", "type_name": "电影"},
+        {"operation_id": "media.detail", "path_params": {"media_id": "27205"},
+         "query": {"media_source": "tmdb", "type_name": "电影"}},
+        "path_params.media_id",
+    ),
+    (
+        {"operation_id": "subscription.execution.list", "query": {"page": 1, "count": 20}},
+        {"operation_id": "subscription.execution.list", "query": {"limit": 10}},
+        "page",
+    ),
+    (
+        {"operation_id": "site.rss", "site_id": 1},
+        {"operation_id": "site.rss", "query": {"page": 1, "count": 20}},
+        "site_id",
+    ),
+])
+async def test_api_read_errors_reach_model_and_corrected_call_executes_once(invocation_runtime, invalid, corrected, hint):
+    """真实 Agent 图把纠错回执交给下一轮模型，只有修正后的只读调用到达执行器。"""
+    repository, records = invocation_runtime
+    executor = AsyncMock()
+    executor.execute.return_value = '{"success":true}'
+    tool = MoviePilotApiTool(session_id="invocation-chat", user_id="owner", executor=executor)
+    tool.set_agent_context({"is_admin": True})
+    graph, _model, _middleware = _graph(repository, [
+        AIMessage(content="", tool_calls=[_call("invalid-read", invalid)]),
+        AIMessage(content="", tool_calls=[_call("corrected-read", corrected)]),
+        AIMessage(content="查询完成"),
+    ], tools=[tool])
+
+    messages = _tool_messages(await _invoke(graph))
+    assert messages[0].status == "error"
+    failure = json.loads(messages[0].content)
+    assert failure["error"] == "invalid_input"
+    assert hint in failure["message"]
+    assert failure["input_contract"]["operation_id"] == invalid["operation_id"]
+    assert json.loads(messages[1].content)["success"] is True
+    executor.execute.assert_awaited_once()
+    assert records() == []
+
+
+@pytest.mark.asyncio
+async def test_flattened_write_arguments_fail_before_claim(invocation_runtime, monkeypatch):
+    """即使请求体本身合法，额外顶层写入字段也必须在认领前失败并提示位置。"""
+    repository, records = invocation_runtime
+    run = AsyncMock(return_value='{"success":true}')
+    monkeypatch.setattr(MoviePilotApiTool, "run", run)
+    graph, _model, _middleware = _graph(repository, [
+        AIMessage(content="", tool_calls=[_call("invalid-write", {**WRITE_ARGUMENTS, "allow_unrecognized": True})]),
+        AIMessage(content="需要修正参数"),
+    ])
+    failure = json.loads(_tool_messages(await _invoke(graph))[0].content)
+    assert "allow_unrecognized 应移入 body.allow_unrecognized" in failure["message"]
+    assert records() == []
+    run.assert_not_awaited()
+
+
 @pytest.mark.asyncio
 async def test_api_forbidden_query_keys_do_not_execute_or_claim(invocation_runtime, monkeypatch):
     """路径和查询合同禁止的字段在认领前拒绝，不能仅从指纹丢弃后仍传给服务端。"""

tests/test_builtin_skill_boundaries.py

@@ -117,7 +117,7 @@ def test_modified_builtin_skills_have_incremented_versions() -> None:
         "command-dispatch": "2",
         "database-operation": "9",
         "feedback-issue": "10",
-        "moviepilot-api": "38",
+        "moviepilot-api": "39",
         "moviepilot-update": "5",
         "organize-files": "5",
         "transfer-failed-retry": "5",