feat(agent): expose self-describing service operations

app/agent/policy/api_mcp_contract.py

@@ -0,0 +1,295 @@
+"""从业务 OpenAPI 构建 moviepilot_api 的外部 MCP 输入合同。"""
+
+from __future__ import annotations
+
+from copy import deepcopy
+from typing import Any, Mapping, Sequence
+
+
+def _rewrite_schema_refs(
+    schema: Mapping[str, Any],
+    *,
+    components: Mapping[str, Any],
+    definitions: dict[str, Any],
+) -> dict[str, Any]:
+    """把 OpenAPI components 引用改写为独立 MCP schema 的 $defs 引用。"""
+    reference = schema.get("$ref")
+    if isinstance(reference, str) and reference.startswith("#/components/schemas/"):
+        name = reference.rsplit("/", 1)[-1]
+        if name not in definitions:
+            definitions[name] = {}
+            source = components.get(name)
+            if not isinstance(source, Mapping):
+                raise ValueError(f"OpenAPI 缺少请求模型: {name}")
+            definitions[name] = _rewrite_schema_refs(
+                source,
+                components=components,
+                definitions=definitions,
+            )
+        return {"$ref": f"#/$defs/{name}"}
+
+    rewritten: dict[str, Any] = {}
+    for key, value in schema.items():
+        if isinstance(value, Mapping):
+            rewritten[key] = _rewrite_schema_refs(
+                value,
+                components=components,
+                definitions=definitions,
+            )
+        elif isinstance(value, list):
+            rewritten[key] = [
+                _rewrite_schema_refs(
+                    item,
+                    components=components,
+                    definitions=definitions,
+                )
+                if isinstance(item, Mapping)
+                else deepcopy(item)
+                for item in value
+            ]
+        else:
+            rewritten[key] = deepcopy(value)
+    return rewritten
+
+
+def _parameter_object_schema(
+    parameters: Sequence[Mapping[str, Any]],
+    *,
+    location: str,
+    components: Mapping[str, Any],
+    definitions: dict[str, Any],
+) -> dict[str, Any] | None:
+    """把 OpenAPI path/query 参数投影为网关结构化对象。"""
+    selected = [parameter for parameter in parameters if parameter.get("in") == location]
+    if not selected:
+        return None
+    properties: dict[str, Any] = {}
+    required: list[str] = []
+    for parameter in selected:
+        name = str(parameter["name"])
+        raw_schema = parameter.get("schema")
+        if not isinstance(raw_schema, Mapping):
+            raw_schema = {}
+        field_schema = _rewrite_schema_refs(
+            raw_schema,
+            components=components,
+            definitions=definitions,
+        )
+        if parameter.get("description") and "description" not in field_schema:
+            field_schema["description"] = parameter["description"]
+        properties[name] = field_schema
+        if parameter.get("required"):
+            required.append(name)
+    result: dict[str, Any] = {
+        "type": "object",
+        "properties": properties,
+        "additionalProperties": False,
+    }
+    if required:
+        result["required"] = required
+    return result
+
+
+def _request_body_schema(
+    operation: Mapping[str, Any],
+    *,
+    components: Mapping[str, Any],
+    definitions: dict[str, Any],
+) -> tuple[dict[str, Any] | None, bool]:
+    """读取一个 OpenAPI operation 的 JSON 请求体及必填性。"""
+    request_body = operation.get("requestBody")
+    if not isinstance(request_body, Mapping):
+        return None, False
+    content = request_body.get("content")
+    if not isinstance(content, Mapping):
+        return None, bool(request_body.get("required"))
+    media = content.get("application/json")
+    if not isinstance(media, Mapping):
+        media = next((item for item in content.values() if isinstance(item, Mapping)), None)
+    raw_schema = media.get("schema") if isinstance(media, Mapping) else None
+    if not isinstance(raw_schema, Mapping):
+        return None, bool(request_body.get("required"))
+    return (
+        _rewrite_schema_refs(
+            raw_schema,
+            components=components,
+            definitions=definitions,
+        ),
+        bool(request_body.get("required")),
+    )
+
+
+def _person_credits_operation(openapi: Mapping[str, Any]) -> dict[str, Any]:
+    """合并四个来源端点为稳定 media.person.credits 网关合同。"""
+    paths = openapi.get("paths", {})
+    source_paths = {
+        "douban": "/api/v1/douban/person/credits/{person_id}",
+        "tmdb": "/api/v1/tmdb/person/credits/{person_id}",
+        "bangumi": "/api/v1/bangumi/person/credits/{person_id}",
+        "anilist": "/api/v1/anilist/person/credits/{person_id}",
+    }
+    for source_path in source_paths.values():
+        if source_path not in paths:
+            raise ValueError(f"OpenAPI 缺少人物作品端点: {source_path}")
+    return {
+        "summary": "读取人物作品",
+        "parameters": [
+            {
+                "name": "source",
+                "in": "path",
+                "required": True,
+                "schema": {"type": "string", "enum": list(source_paths)},
+                "description": "人物数据来源。",
+            },
+            {
+                "name": "person_id",
+                "in": "path",
+                "required": True,
+                "schema": {"type": "integer"},
+                "description": "来源原生人物 ID。",
+            },
+            {
+                "name": "page",
+                "in": "query",
+                "required": False,
+                "schema": {"type": "integer", "minimum": 1, "default": 1},
+            },
+            {
+                "name": "count",
+                "in": "query",
+                "required": False,
+                "schema": {"type": "integer", "minimum": 1, "maximum": 50, "default": 20},
+                "description": "Bangumi 与 AniList 支持的每页条数；其他来源忽略。",
+            },
+        ],
+    }
+
+
+def _resolve_openapi_operation(
+    operation_id: str,
+    route: Any,
+    openapi: Mapping[str, Any],
+) -> Mapping[str, Any]:
+    """按固定路由读取 OpenAPI operation，并处理多来源合成路由。"""
+    if operation_id == "media.person.credits":
+        return _person_credits_operation(openapi)
+    path_item = openapi.get("paths", {}).get(route.path)
+    if not isinstance(path_item, Mapping):
+        raise ValueError(f"OpenAPI 缺少 operation 路径: {operation_id} -> {route.path}")
+    operation = path_item.get(str(route.method).lower())
+    if not isinstance(operation, Mapping):
+        raise ValueError(f"OpenAPI 缺少 operation 方法: {operation_id} -> {route.method} {route.path}")
+    return operation
+
+
+def _apply_operation_overrides(
+    operation_id: str,
+    query_schema: dict[str, Any] | None,
+) -> None:
+    """补充同一路由多 operation 时无法由 FastAPI 自动表达的语义约束。"""
+    if operation_id != "media.person.search" or query_schema is None:
+        return
+    query_schema["properties"]["type"] = {
+        "type": "string",
+        "const": "person",
+        "description": "人物搜索固定传 person。",
+    }
+    required = query_schema.setdefault("required", [])
+    if "type" not in required:
+        required.append("type")
+
+
+def build_api_mcp_input_schema(
+    *,
+    openapi: Mapping[str, Any],
+    routes: Mapping[str, Any],
+    specs: Sequence[Any],
+) -> dict[str, Any]:
+    """构建 59 个白名单 operation 的完整 MCP oneOf 输入合同。"""
+    components = openapi.get("components", {}).get("schemas", {})
+    if not isinstance(components, Mapping):
+        components = {}
+    definitions: dict[str, Any] = {}
+    spec_by_id = {spec.operation_id: spec for spec in specs}
+    if set(spec_by_id) != set(routes):
+        raise ValueError("API operation 策略与路由注册表不一致")
+
+    branches: list[dict[str, Any]] = []
+    for operation_id in sorted(routes):
+        route = routes[operation_id]
+        operation = _resolve_openapi_operation(operation_id, route, openapi)
+        parameters = operation.get("parameters")
+        if not isinstance(parameters, list):
+            parameters = []
+        path_schema = _parameter_object_schema(
+            parameters,
+            location="path",
+            components=components,
+            definitions=definitions,
+        )
+        query_schema = _parameter_object_schema(
+            parameters,
+            location="query",
+            components=components,
+            definitions=definitions,
+        )
+        _apply_operation_overrides(operation_id, query_schema)
+        body_schema, body_required = _request_body_schema(
+            operation,
+            components=components,
+            definitions=definitions,
+        )
+
+        properties: dict[str, Any] = {
+            "operation_id": {"type": "string", "const": operation_id},
+        }
+        required = ["operation_id"]
+        if path_schema is not None:
+            properties["path_params"] = path_schema
+            if path_schema.get("required"):
+                required.append("path_params")
+        if query_schema is not None:
+            properties["query"] = query_schema
+            if query_schema.get("required"):
+                required.append("query")
+        if body_schema is not None:
+            properties["body"] = body_schema
+            if body_required:
+                required.append("body")
+
+        summary = str(operation.get("summary") or operation.get("description") or operation_id)
+        spec = spec_by_id[operation_id]
+        branches.append(
+            {
+                "type": "object",
+                "title": operation_id,
+                "description": (
+                    f"{summary} Method: {route.method}. Path: {route.path}. "
+                    f"Effect: {spec.effect.value}."
+                ),
+                "properties": properties,
+                "required": required,
+                "additionalProperties": False,
+            }
+        )
+
+    schema: dict[str, Any] = {
+        "$schema": "https://json-schema.org/draft/2020-12/schema",
+        "title": "moviepilot_api",
+        "type": "object",
+        "description": "Select the oneOf branch matching operation_id and send exactly its documented fields.",
+        "properties": {
+            "operation_id": {"type": "string", "enum": sorted(routes)},
+            "path_params": {"type": "object"},
+            "query": {"type": "object"},
+            "body": {"type": "object"},
+        },
+        "required": ["operation_id"],
+        "oneOf": branches,
+    }
+    if definitions:
+        schema["$defs"] = definitions
+    return schema
+
+
+__all__ = ["build_api_mcp_input_schema"]

app/agent/policy/registry.py

@@ -47,6 +47,8 @@ BUILTIN_LEGACY_SHADOW_INVENTORY = frozenset(
         "persona",
         "write_file",
         "moviepilot_api",
+        "downloader_operation",
+        "mediaserver_operation",
     }
 )
 

app/agent/tools/factory.py

@@ -17,6 +17,7 @@ from app.agent.tools.impl.search_web import SearchWebTool
 from app.agent.tools.impl.send_local_file import SendLocalFileTool
 from app.agent.tools.impl.send_message import SendMessageTool
 from app.agent.tools.impl.send_voice_message import SendVoiceMessageTool
+from app.agent.tools.impl.service_operation import DownloaderOperationTool, MediaServerOperationTool
 from app.agent.tools.impl.write_file import WriteFileTool
 from app.application.agent import AgentDataContext
 from app.application.plugin.runtime import get_plugin_manager
@@ -59,6 +60,12 @@ class MoviePilotToolFactory:
         BrowseWebpageTool,
         QueryDoctorReportTool,
     )
+    # 下载器与媒体服务器原生操作通过 Skill 按需提供给内置 Agent；只有外部
+    # HTTP/MCP 工具管理器需要常驻、自描述的结构化入口。
+    EXTERNAL_SERVICE_TOOL_CLASSES: tuple[Type[MoviePilotTool], ...] = (
+        DownloaderOperationTool,
+        MediaServerOperationTool,
+    )
 
     # 这些通用工具需要始终保留，避免大工具集裁剪后让 Agent 丢失基础的
     # 文件系统、命令执行、历史检索或交互确认能力。AskUserChoiceTool 仅在支持按钮
@@ -79,7 +86,10 @@ class MoviePilotToolFactory:
     @classmethod
     def catalog_factory_revision(cls) -> str:
         """返回当前内置工具工厂定义的稳定摘要。"""
-        identities = (f"{tool_class.__module__}.{tool_class.__qualname__}" for tool_class in cls.BUILTIN_TOOL_CLASSES)
+        identities = (
+            f"{tool_class.__module__}.{tool_class.__qualname__}"
+            for tool_class in (*cls.BUILTIN_TOOL_CLASSES, *cls.EXTERNAL_SERVICE_TOOL_CLASSES)
+        )
         return hashlib.sha256("\n".join(identities).encode("utf-8")).hexdigest()
 
     @staticmethod
@@ -141,6 +151,7 @@ class MoviePilotToolFactory:
         stream_handler: Callable = None,
         agent_context: dict = None,
         allow_message_tools: bool = True,
+        include_external_service_tools: bool = False,
         data: Optional[AgentDataContext] = None,
     ) -> List[MoviePilotTool]:
         """
@@ -152,6 +163,8 @@ class MoviePilotToolFactory:
         """
         tools = []
         tool_definitions = cls._get_builtin_tool_classes(channel)
+        if include_external_service_tools:
+            tool_definitions.extend(cls.EXTERNAL_SERVICE_TOOL_CLASSES)
         # 创建内置工具
         for ToolClass in tool_definitions:
             tool = ToolClass(session_id=session_id, user_id=user_id, data=data)

app/agent/tools/impl/api.py

@@ -1,6 +1,9 @@
 """MoviePilot 结构化 API 网关工具。"""
 
 import json
+from copy import deepcopy
+from functools import lru_cache
+from pathlib import Path
 from typing import Any, Dict, Optional, Type
 
 from pydantic import BaseModel, Field, PrivateAttr
@@ -12,6 +15,16 @@ from app.agent.tools.tags import ToolTag
 from app.schemas.types import NotificationChannel
 
 
+@lru_cache(maxsize=1)
+def _load_api_mcp_input_schema() -> dict[str, Any]:
+    """读取由业务 OpenAPI 生成并经漂移测试锁定的外部 MCP schema。"""
+    schema_path = Path(__file__).resolve().parents[2] / "policy" / "api_mcp_schema.json"
+    payload = json.loads(schema_path.read_text(encoding="utf-8"))
+    if not isinstance(payload, dict) or not isinstance(payload.get("oneOf"), list):
+        raise RuntimeError("moviepilot_api MCP schema 无效")
+    return payload
+
+
 class MoviePilotApiInput(BaseModel):  # type: ignore[misc]
     """MoviePilot API 网关的结构化输入参数。"""
 
@@ -54,7 +67,8 @@ class MoviePilotApiTool(MoviePilotTool):
     ]
     description: str = (
         "调用经过白名单审核的 MoviePilot 业务 API。使用领域 Skill 获取 operation_id、"
-        "参数和失败处理；不能调用任意 URL、命令或认证接口。"
+        "参数和失败处理；外部 MCP tools/list 会为每个 operation 提供完整 oneOf 参数合同；"
+        "不能调用任意 URL、命令或认证接口。"
     )
     require_admin: bool = False
     args_schema: Type[BaseModel] = MoviePilotApiInput
@@ -78,6 +92,10 @@ class MoviePilotApiTool(MoviePilotTool):
         operation_id = kwargs.get("operation_id") or "未知操作"
         return f"调用 MoviePilot API：{operation_id}"
 
+    def get_mcp_input_schema(self) -> dict[str, Any]:
+        """返回包含全部白名单 operation 精确参数的 MCP JSON Schema。"""
+        return deepcopy(_load_api_mcp_input_schema())
+
     async def _resolve_api_identity(self) -> tuple[str, Optional[str], bool]:
         """把 Web 或渠道身份解析为真实 MoviePilot 用户身份。"""
         raw_user_id = str(self._user_id or "")

app/agent/tools/impl/service_operation.py

@@ -0,0 +1,391 @@
+"""向 HTTP/MCP 管理入口暴露自描述的服务操作工具。"""
+
+from __future__ import annotations
+
+import json
+import runpy
+import subprocess
+import sys
+from functools import lru_cache
+from pathlib import Path
+from typing import Any, ClassVar, Dict, Optional, Type
+
+from pydantic import BaseModel, ConfigDict, Field
+
+from app.agent.tools.base import MoviePilotTool
+from app.agent.tools.tags import ToolTag
+from app.runtime.settings import get_runtime_setting
+
+
+class DownloaderOperationInput(BaseModel):
+    """下载器操作工具的运行时输入模型。"""
+
+    client: Optional[str] = Field(default=None, description="下载器实例名；省略时使用默认或唯一实例。")
+    action: str = Field(description="固定下载器 action；MCP inputSchema 会按 action 展示精确枚举和参数。")
+    arguments: Dict[str, Any] = Field(default_factory=dict, description="当前 action 的结构化参数对象。")
+
+    model_config = ConfigDict(extra="forbid")
+
+
+class MediaServerOperationInput(BaseModel):
+    """媒体服务器操作工具的运行时输入模型。"""
+
+    server: Optional[str] = Field(default=None, description="媒体服务器实例名；省略时使用唯一启用实例。")
+    action: str = Field(description="固定媒体服务器 action；MCP inputSchema 会按 action 展示精确枚举和参数。")
+    arguments: Dict[str, Any] = Field(default_factory=dict, description="当前 action 的结构化参数对象。")
+
+    model_config = ConfigDict(extra="forbid")
+
+
+@lru_cache(maxsize=2)
+def _load_action_contracts(script_path: str) -> dict[str, Any]:
+    """从固定 Skill 脚本加载无配置副作用的 action 注册表。"""
+    namespace = runpy.run_path(script_path)
+    actions = namespace.get("ACTIONS")
+    if not isinstance(actions, dict):
+        raise RuntimeError("服务操作脚本缺少 ACTIONS 合同")
+    return actions
+
+
+def _compact_type_schema(declared_type: str) -> dict[str, Any]:
+    """把 Skill 的紧凑参数类型转换为标准 JSON Schema。"""
+    variants: list[dict[str, Any]] = []
+    for candidate in declared_type.split("|"):
+        if candidate.endswith("[]"):
+            variants.append(
+                {
+                    "type": "array",
+                    "items": _compact_type_schema(candidate[:-2]),
+                }
+            )
+        else:
+            variants.append({"type": candidate})
+    return variants[0] if len(variants) == 1 else {"anyOf": variants}
+
+
+def _argument_schema(argument: dict[str, Any]) -> dict[str, Any]:
+    """构建一个包含类型、说明、默认值和枚举的参数 schema。"""
+    schema = _compact_type_schema(str(argument["type"]))
+    name = str(argument["name"])
+    schema["description"] = str(argument.get("description") or "")
+    if "default" in argument:
+        schema["default"] = argument["default"]
+    if argument.get("enum"):
+        schema["enum"] = list(argument["enum"])
+    if name == "offset":
+        schema["minimum"] = 0
+    if name == "limit":
+        schema.update({"minimum": 1, "maximum": 200})
+    if name in {"upload_limit", "download_limit", "ratio_limit", "seeding_time_limit"}:
+        schema["minimum"] = 0
+    if name in {
+        "task_id",
+        "item_id",
+        "content",
+        "location",
+        "category",
+    }:
+        schema["minLength"] = 1
+    if name in {
+        "task_ids",
+        "wanted_file_ids",
+        "unwanted_file_ids",
+        "tags",
+        "trackers",
+        "items",
+    }:
+        schema["minItems"] = 1
+    return schema
+
+
+def _metadata_refresh_items_schema() -> dict[str, Any]:
+    """返回 metadata.refresh.items 的完整嵌套条目合同。"""
+    return {
+        "type": "array",
+        "items": {
+            "type": "object",
+            "properties": {
+                "title": {"type": "string", "description": "媒体标题。"},
+                "year": {
+                    "anyOf": [{"type": "string"}, {"type": "integer"}],
+                    "description": "媒体年份。",
+                },
+                "type": {
+                    "type": "string",
+                    "enum": ["电影", "电视剧", "音乐"],
+                    "description": "MoviePilot 媒体类型。",
+                },
+                "category": {"type": "string", "description": "媒体分类。"},
+                "target_path": {"type": "string", "description": "媒体文件或目录路径。"},
+            },
+            "additionalProperties": False,
+        },
+    }
+
+
+def _add_action_argument_rules(action: str, schema: dict[str, Any]) -> None:
+    """把跨字段约束补充为机器可校验的 JSON Schema。"""
+    task_selector_actions = {
+        "tasks.start",
+        "tasks.stop",
+        "tasks.delete",
+        "tasks.recheck",
+        "tasks.reannounce",
+        "tasks.queue.move",
+        "tasks.force_start.set",
+        "tasks.tags.set",
+    }
+    if action in task_selector_actions:
+        schema["oneOf"] = [
+            {"required": ["task_id"], "not": {"required": ["task_ids"]}},
+            {"required": ["task_ids"], "not": {"required": ["task_id"]}},
+        ]
+    if action == "tasks.files.selection.set":
+        schema["anyOf"] = [
+            {"required": ["wanted_file_ids"]},
+            {"required": ["unwanted_file_ids"]},
+        ]
+    if action == "tasks.properties.set":
+        schema["anyOf"] = [
+            {"required": [name]}
+            for name in ("upload_limit", "download_limit", "ratio_limit", "seeding_time_limit")
+        ]
+    if action == "session.speed_limits.set":
+        schema["anyOf"] = [
+            {"required": ["download_limit"]},
+            {"required": ["upload_limit"]},
+        ]
+    if action == "items.music.search":
+        schema["anyOf"] = [
+            {"required": ["title"]},
+            {"required": ["artist"]},
+            {"required": ["album"]},
+        ]
+    if action == "items.season_episodes":
+        schema["anyOf"] = [{"required": ["item_id"]}, {"required": ["title"]}]
+
+
+def _build_arguments_schema(action: str, spec: Any) -> dict[str, Any]:
+    """把一个 action 的脚本合同转换为 MCP arguments schema。"""
+    contract = spec.to_dict(action)
+    properties = {
+        argument["name"]: _argument_schema(argument)
+        for argument in contract["arguments"]
+    }
+    if action == "metadata.refresh":
+        properties["items"] = _metadata_refresh_items_schema()
+        properties["items"]["description"] = contract["arguments"][0]["description"]
+    schema: dict[str, Any] = {
+        "type": "object",
+        "description": "；".join(contract.get("argument_rules") or []),
+        "properties": properties,
+        "required": list(contract.get("required_arguments") or []),
+        "additionalProperties": False,
+    }
+    _add_action_argument_rules(action, schema)
+    return schema
+
+
+def _build_mcp_input_schema(
+    *,
+    actions: dict[str, Any],
+    selector_name: str,
+    selector_description: str,
+    title: str,
+) -> dict[str, Any]:
+    """构建按 action 分支且可由外部 MCP Client 直接发现的输入合同。"""
+    action_names = sorted(actions)
+    branches = []
+    for action in action_names:
+        contract = actions[action].to_dict(action)
+        branches.append(
+            {
+                "type": "object",
+                "title": action,
+                "description": (
+                    f"{contract['description']} Effect: {contract['effect']}. "
+                    f"Providers: {', '.join(contract['providers'])}."
+                ),
+                "properties": {
+                    selector_name: {"type": "string", "description": selector_description},
+                    "action": {"type": "string", "const": action},
+                    "arguments": _build_arguments_schema(action, actions[action]),
+                },
+                "required": ["action", "arguments"],
+                "additionalProperties": False,
+            }
+        )
+    return {
+        "$schema": "https://json-schema.org/draft/2020-12/schema",
+        "title": title,
+        "type": "object",
+        "properties": {
+            selector_name: {"type": "string", "description": selector_description},
+            "action": {"type": "string", "enum": action_names},
+            "arguments": {
+                "type": "object",
+                "description": "参数必须匹配所选 action 的 oneOf 分支。",
+            },
+        },
+        "required": ["action", "arguments"],
+        "oneOf": branches,
+    }
+
+
+def _parse_script_payload(stdout: str) -> dict[str, Any]:
+    """从可能带启动日志的 stdout 中提取最后一个完整 JSON 对象。"""
+    decoder = json.JSONDecoder()
+    for index in range(len(stdout) - 1, -1, -1):
+        if stdout[index] != "{":
+            continue
+        try:
+            payload, end = decoder.raw_decode(stdout[index:])
+        except json.JSONDecodeError:
+            continue
+        if not stdout[index + end :].strip() and isinstance(payload, dict):
+            return payload
+    raise RuntimeError("服务操作脚本未返回有效 JSON")
+
+
+def _run_service_script(
+    *,
+    relative_script: str,
+    selector_flag: str,
+    selector_value: Optional[str],
+    action: str,
+    arguments: Dict[str, Any],
+) -> dict[str, Any]:
+    """不经 shell 调用固定 Skill 脚本，并只返回其结构化 JSON envelope。"""
+    root_path = Path(get_runtime_setting("ROOT_PATH"))
+    script_path = root_path / relative_script
+    command = [
+        sys.executable,
+        str(script_path),
+        "call",
+        "--action",
+        action,
+        "--arguments",
+        json.dumps(arguments, ensure_ascii=False, separators=(",", ":")),
+    ]
+    if selector_value:
+        command.extend([selector_flag, selector_value])
+    completed = subprocess.run(
+        command,
+        cwd=root_path,
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    return _parse_script_payload(completed.stdout)
+
+
+class _ServiceOperationTool(MoviePilotTool):
+    """固定 Skill 脚本的 MCP-only 安全包装基类。"""
+
+    require_admin: bool = True
+    _relative_script: ClassVar[str]
+    _selector_name: ClassVar[str]
+    _selector_flag: ClassVar[str]
+    _blocking_bucket: ClassVar[str]
+
+    def get_mcp_input_schema(self) -> dict[str, Any]:
+        """返回保留 action 条件分支的完整 MCP JSON Schema。"""
+        root_path = Path(get_runtime_setting("ROOT_PATH"))
+        actions = _load_action_contracts(str(root_path / self._relative_script))
+        selector_description = self.args_schema.model_json_schema()["properties"][self._selector_name][
+            "description"
+        ]
+        return _build_mcp_input_schema(
+            actions=actions,
+            selector_name=self._selector_name,
+            selector_description=selector_description,
+            title=self.name,
+        )
+
+    async def _run_operation(
+        self,
+        *,
+        selector_value: Optional[str],
+        action: str,
+        arguments: Dict[str, Any],
+    ) -> str:
+        """在线程池中运行固定脚本并序列化安全结果。"""
+        payload = await self.run_blocking(
+            self._blocking_bucket,
+            _run_service_script,
+            relative_script=self._relative_script,
+            selector_flag=self._selector_flag,
+            selector_value=selector_value,
+            action=action,
+            arguments=arguments,
+        )
+        return json.dumps(payload, ensure_ascii=False, indent=2)
+
+
+class DownloaderOperationTool(_ServiceOperationTool):
+    """外部 HTTP/MCP 调用下载器原生能力的结构化工具。"""
+
+    name: str = "downloader_operation"
+    description: str = (
+        "Operate a configured qBittorrent, Transmission, or rTorrent instance. "
+        "The input schema contains one exact branch per action, including providers, "
+        "effects, required fields, types, defaults, enums, and cross-field constraints."
+    )
+    tags: list[str] = [ToolTag.Download]
+    args_schema: Type[BaseModel] = DownloaderOperationInput
+    _relative_script = "skills/downloader-operation/scripts/mp-downloader.py"
+    _selector_name = "client"
+    _selector_flag = "--client"
+    _blocking_bucket = "downloader"
+
+    async def run(
+        self,
+        action: str,
+        arguments: Optional[Dict[str, Any]] = None,
+        client: Optional[str] = None,
+    ) -> str:
+        """执行一次自描述的下载器操作。"""
+        return await self._run_operation(
+            selector_value=client,
+            action=action,
+            arguments=arguments or {},
+        )
+
+
+class MediaServerOperationTool(_ServiceOperationTool):
+    """外部 HTTP/MCP 调用媒体服务器原生能力的结构化工具。"""
+
+    name: str = "mediaserver_operation"
+    description: str = (
+        "Operate a configured Emby, Jellyfin, Plex, ZSpace, UGREEN, TrimeMedia, "
+        "or Navidrome server. The input schema contains one exact branch per action, "
+        "including providers, effects, required fields, types, defaults, enums, nested "
+        "item fields, and cross-field constraints."
+    )
+    tags: list[str] = [ToolTag.Media]
+    args_schema: Type[BaseModel] = MediaServerOperationInput
+    _relative_script = "skills/mediaserver-operation/scripts/mp-mediaserver.py"
+    _selector_name = "server"
+    _selector_flag = "--server"
+    _blocking_bucket = "mediaserver"
+
+    async def run(
+        self,
+        action: str,
+        arguments: Optional[Dict[str, Any]] = None,
+        server: Optional[str] = None,
+    ) -> str:
+        """执行一次自描述的媒体服务器操作。"""
+        return await self._run_operation(
+            selector_value=server,
+            action=action,
+            arguments=arguments or {},
+        )
+
+
+__all__ = [
+    "DownloaderOperationInput",
+    "DownloaderOperationTool",
+    "MediaServerOperationInput",
+    "MediaServerOperationTool",
+]

app/agent/tools/manager.py

@@ -108,6 +108,7 @@ class MoviePilotToolsManager:
                 username="API Client",
                 stream_handler=None,
                 agent_context={"is_admin": self.is_admin},
+                include_external_service_tools=True,
                 data=self._data,
             )
             self.catalog = catalog
@@ -211,14 +212,17 @@ class MoviePilotToolsManager:
         for tool in tools:
             if getattr(tool, "_require_admin", False) and not self.is_admin:
                 continue
-            # 获取工具的输入参数模型
-            args_schema = getattr(tool, "args_schema", None)
-            if args_schema:
-                # 将Pydantic模型转换为JSON Schema
-                input_schema = self._convert_to_json_schema(args_schema)
+            # MCP-only 复杂工具可以保留 oneOf、嵌套对象和跨字段约束；普通工具
+            # 继续沿用兼容投影，避免一次性改变已有外部合同。
+            schema_factory = getattr(tool, "get_mcp_input_schema", None)
+            if callable(schema_factory):
+                input_schema = schema_factory()
             else:
-                # 如果没有args_schema，使用基本信息
-                input_schema = {"type": "object", "properties": {}, "required": []}
+                args_schema = getattr(tool, "args_schema", None)
+                if args_schema:
+                    input_schema = self._convert_to_json_schema(args_schema)
+                else:
+                    input_schema = {"type": "object", "properties": {}, "required": []}
 
             tools_list.append(
                 ToolDefinition(

scripts/generate_agent_api_mcp_schema.py

@@ -0,0 +1,46 @@
+#!/usr/bin/env python3
+"""从 FastAPI OpenAPI 生成 moviepilot_api 的外部 MCP 输入合同。"""
+
+from __future__ import annotations
+
+import json
+import sys
+from pathlib import Path
+
+from fastapi import FastAPI
+
+PROJECT_ROOT = Path(__file__).resolve().parents[1]
+if str(PROJECT_ROOT) not in sys.path:
+    sys.path.insert(0, str(PROJECT_ROOT))
+
+from app.agent.policy.api import API_OPERATION_ROUTES, API_OPERATION_SPECS  # noqa: E402
+from app.agent.policy.api_mcp_contract import build_api_mcp_input_schema  # noqa: E402
+from app.api.apiv1 import api_router  # noqa: E402
+
+OUTPUT_PATH = PROJECT_ROOT / "app/agent/policy/api_mcp_schema.json"
+
+
+def generate_schema() -> dict:
+    """聚合 v1 路由 OpenAPI 并生成稳定的网关 MCP schema。"""
+    app = FastAPI()
+    app.include_router(api_router, prefix="/api/v1")
+    return build_api_mcp_input_schema(
+        openapi=app.openapi(),
+        routes=API_OPERATION_ROUTES,
+        specs=API_OPERATION_SPECS,
+    )
+
+
+def main() -> int:
+    """写入格式稳定的生成文件。"""
+    schema = generate_schema()
+    OUTPUT_PATH.write_text(
+        json.dumps(schema, ensure_ascii=False, indent=2, sort_keys=True) + "\n",
+        encoding="utf-8",
+    )
+    print(f"generated {OUTPUT_PATH.relative_to(PROJECT_ROOT)}")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

skills/downloader-operation/SKILL.md

@@ -1,6 +1,6 @@
 ---
 name: downloader-operation
-version: 1
+version: 2
 description: >-
   Use this skill when the user asks to inspect, diagnose, or directly control a
   configured qBittorrent, Transmission, or rTorrent instance. It exposes
@@ -29,7 +29,30 @@ username, password, API key, Cookie, or arbitrary URL.
 - Paths passed to `tasks.location.set` and `tasks.add.direct` are downloader-side
   paths, not MoviePilot storage paths.
 
-## Discover First
+## Instance And Provider Discovery
+
+### Fast path: call directly
+
+Do not routinely call `instances` or `capabilities` before an operation. This
+Skill already contains the full action contract, and the helper performs
+instance resolution, provider support checks, complete argument validation, and
+the action in one `call` invocation.
+
+- If the user or prior context provides the exact client name, pass it with
+  `--client` and call the action immediately.
+- If no client name is known, omit `--client`. The helper automatically uses the
+  single default downloader, or the only enabled downloader.
+- If multiple clients remain ambiguous, the failed call lists every valid
+  client name. Reuse that list for the next direct call; do not add a separate
+  `instances` call unless the user explicitly asks to inspect instances.
+- Do not probe an action with empty or guessed arguments. Compose the complete
+  JSON object from the contract below before calling.
+
+The helper rejects unknown fields and reports all detectable argument errors in
+one response before connecting to the provider, so correct every reported field
+together instead of retrying one field at a time.
+
+### Optional discovery
 
 List configured instances without secrets:
 
@@ -44,8 +67,19 @@ python skills/downloader-operation/scripts/mp-downloader.py capabilities
 python skills/downloader-operation/scripts/mp-downloader.py capabilities --client "main-qb"
 ```
 
-Do not guess a provider-specific action. Call `capabilities` when the current
-instance, provider, argument contract, or side-effect level is uncertain.
+The complete action and argument contract is documented below. Use
+`capabilities` only to confirm which documented actions a configured provider
+supports. For a compact machine-readable copy of one action's same contract:
+
+```bash
+python skills/downloader-operation/scripts/mp-downloader.py capabilities \
+  --client "main-qb" \
+  --action tasks.properties.set
+```
+
+Do not inspect the helper source to discover arguments and do not guess a
+provider-specific action. `capabilities` is optional and should be used only
+when the configured provider itself is unknown or support must be diagnosed.
 
 ## Call Shape
 
@@ -59,22 +93,106 @@ python skills/downloader-operation/scripts/mp-downloader.py call \
 The `--arguments` value must be one JSON object. Large reads are paged with
 `offset` and `limit`; the default limit is 50 and the maximum is 200.
 
-## Core Actions
+## External MCP Contract
 
-- Read: `tasks.list`, `tasks.files`, `tasks.trackers`, `tasks.tags.get`,
-  `tasks.peers`, `session.stats`, `session.speed_limits.get`,
-  `session.details`, `session.content_layout`.
-- Reversible writes: `tasks.start`, `tasks.stop`, `tasks.recheck`,
-  `tasks.reannounce`, `tasks.queue.move`, `tasks.properties.set`,
-  `tasks.files.selection.set`, `tasks.force_start.set`, `tasks.location.set`,
-  `tasks.category.set`, `tasks.tags.set`, `tasks.trackers.update`,
-  `session.speed_limits.set`.
-- External/destructive: `tasks.add.direct`, `tasks.delete`.
+External MCP clients do not receive this `SKILL.md` and cannot use the hidden
+`execute_command` tool. MoviePilot therefore exposes a separate admin-only MCP
+tool named `downloader_operation`. Its `tools/list` `inputSchema` contains one
+`oneOf` branch for every action below, including the function description,
+supported providers, effect, field types, required/default values, enums, and
+cross-field rules. The external client should select the matching branch and
+make one `tools/call`; it does not need to call a discovery tool first.
 
-For task actions, use `task_id` for one hash/ID or `task_ids` for a batch. Before
-deleting data, confirm the exact client, tasks, and `delete_files=true`. Before a
-direct add, confirm the exact magnet/URL or local torrent file, client, paused
-state, provider path, tags, and category.
+MCP call arguments use the same contract without shell quoting:
+
+```json
+{
+  "client": "main-qb",
+  "action": "tasks.properties.set",
+  "arguments": {
+    "task_id": "exact-provider-hash",
+    "download_limit": 2048,
+    "upload_limit": 512
+  }
+}
+```
+
+`client` may be omitted for the default or only enabled downloader. If multiple
+instances remain ambiguous, the result lists the valid client names.
+
+## Complete Action Contract
+
+In the tables below, `*` means required. Every listed field belongs inside the
+single `--arguments` JSON object. Do not send fields that are not listed.
+
+Shared rules:
+
+- Task batch actions require exactly one of `task_id:string` or
+  `task_ids:string[]`.
+- Paged reads accept `offset:integer=0` and `limit:integer=50`; `offset` must be
+  non-negative and `limit` is clamped to `1..200`.
+- Speed values are numbers in `KB/s`. A value of `0` means unlimited.
+- Task IDs, file indexes, tags, tracker URLs, and provider paths must come from
+  the selected downloader or the user's explicit input; never invent them.
+
+### Task reads
+
+| Action | Function and providers | `--arguments` fields |
+|---|---|---|
+| `tasks.list` | List/filter tasks; all | `task_id:string` or `task_ids:string[]`; `status:string`; `tags:string\|string[]`; `offset:integer=0`; `limit:integer=50` |
+| `tasks.files` | List files and priorities for one task; all | `task_id*:string`; `offset:integer=0`; `limit:integer=50` |
+| `tasks.trackers` | List tracker URLs; qBittorrent, Transmission | `task_id*:string` |
+| `tasks.tags.get` | Read tags/labels for one task; all | `task_id*:string` |
+| `tasks.peers` | Read peer synchronization data; qBittorrent | `task_id*:string` |
+
+### Task control
+
+| Action | Function and effect | `--arguments` fields |
+|---|---|---|
+| `tasks.start` | Start/resume tasks; reversible write | exactly one of `task_id:string`, `task_ids:string[]` |
+| `tasks.stop` | Pause tasks; reversible write | exactly one of `task_id:string`, `task_ids:string[]` |
+| `tasks.recheck` | Force data verification; external side effect | exactly one of `task_id:string`, `task_ids:string[]` |
+| `tasks.reannounce` | Force tracker reannounce; qBittorrent/Transmission, external side effect | exactly one of `task_id:string`, `task_ids:string[]` |
+| `tasks.queue.move` | Move queue position; qBittorrent/Transmission, reversible write | exactly one of `task_id:string`, `task_ids:string[]`; `position*:string` = `top\|up\|down\|bottom` |
+| `tasks.force_start.set` | Toggle force-start; qBittorrent, reversible write | exactly one of `task_id:string`, `task_ids:string[]`; `enabled*:boolean` |
+| `tasks.files.selection.set` | Select files within one task; reversible write | `task_id*:string`; `wanted_file_ids:integer[]`; `unwanted_file_ids:integer[]`; at least one list, with no overlapping index |
+| `tasks.properties.set` | Set per-task limits; reversible write | `task_id*:string`; at least one of `upload_limit:number`, `download_limit:number`, `ratio_limit:number`, `seeding_time_limit:integer` minutes. rTorrent supports only speed fields |
+| `tasks.location.set` | Move/retarget data to a downloader-side path; external side effect | `task_id*:string`; `location*:string` |
+| `tasks.category.set` | Set a non-empty category; qBittorrent, reversible write | `task_id*:string`; `category*:string` |
+| `tasks.tags.set` | Set/add tags or labels; reversible write | exactly one of `task_id:string`, `task_ids:string[]`; `tags*:string[]` |
+| `tasks.trackers.update` | Add/replace trackers; qBittorrent/Transmission, reversible write | `task_id*:string`; `trackers*:string[]` of URLs |
+| `tasks.delete` | Delete tasks and optionally data; destructive write | exactly one of `task_id:string`, `task_ids:string[]`; `delete_files:boolean=false` |
+| `tasks.add.direct` | Submit directly to provider, bypassing MoviePilot orchestration; external side effect | `content*:string` magnet/URL/path; `torrent_file:boolean=false`; `paused:boolean=false`; `download_dir:string`; `tags:string[]`; `category:string` (qBittorrent only) |
+
+### Session operations
+
+| Action | Function and providers | `--arguments` fields |
+|---|---|---|
+| `session.stats` | Read transfer/session statistics; all | none (`{}`) |
+| `session.speed_limits.get` | Read global download/upload limits; qBittorrent, Transmission | none (`{}`) |
+| `session.speed_limits.set` | Set global limits; qBittorrent, Transmission | at least one of `download_limit:number`, `upload_limit:number`; use explicit `0` to clear a limit |
+| `session.details` | Read Transmission session configuration/capacity; Transmission | none (`{}`) |
+| `session.content_layout` | Read default torrent content layout; qBittorrent | none (`{}`) |
+
+Examples:
+
+```bash
+# Read one task's files.
+python skills/downloader-operation/scripts/mp-downloader.py call \
+  --client "main-qb" \
+  --action tasks.files \
+  --arguments '{"task_id":"exact-provider-hash","offset":0,"limit":50}'
+
+# Limit one task to 2048 KB/s download and 512 KB/s upload.
+python skills/downloader-operation/scripts/mp-downloader.py call \
+  --client "main-qb" \
+  --action tasks.properties.set \
+  --arguments '{"task_id":"exact-provider-hash","download_limit":2048,"upload_limit":512}'
+```
+
+Before deleting data, confirm the exact client, tasks, and `delete_files=true`.
+Before a direct add, confirm the exact magnet/URL or local torrent file, client,
+paused state, provider path, tags, and category.
 
 For `tasks.files.selection.set`, pass provider file indexes from `tasks.files`
 through `wanted_file_ids` and/or `unwanted_file_ids`; never infer indexes from

skills/downloader-operation/scripts/mp-downloader.py

@@ -22,6 +22,37 @@ PROVIDER_CLASSES = {
     "transmission": "app.modules.transmission.transmission:Transmission",
     "rtorrent": "app.modules.rtorrent.rtorrent:Rtorrent",
 }
+_UNSET = object()
+
+
+class OperationError(RuntimeError):
+    """可安全返回给 Agent 的下载器操作错误。"""
+
+
+@dataclass(frozen=True, slots=True)
+class ArgumentSpec:
+    """描述一个 action 参数的公开调用合同。"""
+
+    name: str
+    type: str
+    description: str
+    required: bool = False
+    default: Any = _UNSET
+    enum: tuple[Any, ...] = ()
+
+    def to_dict(self) -> dict[str, Any]:
+        """返回可直接交给 Agent 的参数 schema。"""
+        result: dict[str, Any] = {
+            "name": self.name,
+            "type": self.type,
+            "required": self.required,
+            "description": self.description,
+        }
+        if self.default is not _UNSET:
+            result["default"] = self.default
+        if self.enum:
+            result["enum"] = list(self.enum)
+        return result
 
 
 @dataclass(frozen=True, slots=True)
@@ -31,7 +62,13 @@ class ActionSpec:
     description: str
     effect: str
     providers: tuple[str, ...] = ALL_PROVIDERS
-    required: tuple[str, ...] = ()
+    arguments: tuple[ArgumentSpec, ...] = ()
+    argument_rules: tuple[str, ...] = ()
+
+    @property
+    def required(self) -> tuple[str, ...]:
+        """返回保持旧能力合同兼容的必填参数名。"""
+        return tuple(argument.name for argument in self.arguments if argument.required)
 
     def to_dict(self, name: str) -> dict[str, Any]:
         """返回不包含实现对象的公开能力描述。"""
@@ -41,65 +78,194 @@ class ActionSpec:
             "effect": self.effect,
             "providers": list(self.providers),
             "required_arguments": list(self.required),
+            "arguments": [argument.to_dict() for argument in self.arguments],
+            "argument_rules": list(self.argument_rules),
         }
 
 
+TASK_ID = ArgumentSpec("task_id", "string", "单个任务的 provider 原生 hash 或 ID。")
+TASK_IDS = ArgumentSpec("task_ids", "string[]", "多个任务的 provider 原生 hash 或 ID；与 task_id 二选一。")
+OFFSET = ArgumentSpec("offset", "integer", "列表起始偏移，必须大于等于 0。", default=0)
+LIMIT = ArgumentSpec("limit", "integer", "返回条数，范围 1..200。", default=DEFAULT_LIMIT)
+
+
 ACTIONS: dict[str, ActionSpec] = {
-    "tasks.list": ActionSpec("List and filter downloader tasks.", "safe_read"),
-    "tasks.files": ActionSpec("List files and priorities for one task.", "safe_read", required=("task_id",)),
+    "tasks.list": ActionSpec(
+        "List and filter downloader tasks.",
+        "safe_read",
+        arguments=(
+            TASK_ID,
+            TASK_IDS,
+            ArgumentSpec("status", "string", "按 provider 原生任务状态过滤。"),
+            ArgumentSpec("tags", "string|string[]", "只返回同时包含这些标签的任务。"),
+            OFFSET,
+            LIMIT,
+        ),
+    ),
+    "tasks.files": ActionSpec(
+        "List files and priorities for one task.",
+        "safe_read",
+        arguments=(ArgumentSpec("task_id", "string", TASK_ID.description, required=True), OFFSET, LIMIT),
+    ),
     "tasks.files.selection.set": ActionSpec(
         "Select wanted and unwanted files within one task.",
         "reversible_write",
-        required=("task_id",),
+        arguments=(
+            ArgumentSpec("task_id", "string", TASK_ID.description, required=True),
+            ArgumentSpec("wanted_file_ids", "integer[]", "要下载的 provider 文件索引；与 unwanted_file_ids 至少提供一项。"),
+            ArgumentSpec("unwanted_file_ids", "integer[]", "跳过的 provider 文件索引；与 wanted_file_ids 至少提供一项。"),
+        ),
+        argument_rules=("wanted_file_ids 与 unwanted_file_ids 至少提供一项，且同一索引不能同时出现。",),
     ),
     "tasks.trackers": ActionSpec(
-        "List trackers for one task.", "safe_read", ("qbittorrent", "transmission"), ("task_id",)
+        "List trackers for one task.",
+        "safe_read",
+        ("qbittorrent", "transmission"),
+        (ArgumentSpec("task_id", "string", TASK_ID.description, required=True),),
+    ),
+    "tasks.tags.get": ActionSpec(
+        "Read task tags or labels.",
+        "safe_read",
+        arguments=(ArgumentSpec("task_id", "string", TASK_ID.description, required=True),),
     ),
-    "tasks.tags.get": ActionSpec("Read task tags or labels.", "safe_read", required=("task_id",)),
     "tasks.peers": ActionSpec(
-        "Read qBittorrent peer synchronization data.", "safe_read", ("qbittorrent",), ("task_id",)
+        "Read qBittorrent peer synchronization data.",
+        "safe_read",
+        ("qbittorrent",),
+        (ArgumentSpec("task_id", "string", TASK_ID.description, required=True),),
+    ),
+    "tasks.start": ActionSpec(
+        "Start or resume one or more tasks.",
+        "reversible_write",
+        arguments=(TASK_ID, TASK_IDS),
+        argument_rules=("task_id 与 task_ids 必须提供且只能选择一种。",),
+    ),
+    "tasks.stop": ActionSpec(
+        "Pause one or more tasks.",
+        "reversible_write",
+        arguments=(TASK_ID, TASK_IDS),
+        argument_rules=("task_id 与 task_ids 必须提供且只能选择一种。",),
+    ),
+    "tasks.delete": ActionSpec(
+        "Delete tasks and optionally their data.",
+        "destructive_write",
+        arguments=(
+            TASK_ID,
+            TASK_IDS,
+            ArgumentSpec("delete_files", "boolean", "同时永久删除任务数据文件。", default=False),
+        ),
+        argument_rules=("task_id 与 task_ids 必须提供且只能选择一种。",),
+    ),
+    "tasks.recheck": ActionSpec(
+        "Force data verification for tasks.",
+        "external_side_effect",
+        arguments=(TASK_ID, TASK_IDS),
+        argument_rules=("task_id 与 task_ids 必须提供且只能选择一种。",),
     ),
-    "tasks.start": ActionSpec("Start or resume one or more tasks.", "reversible_write", required=("task_ids",)),
-    "tasks.stop": ActionSpec("Pause one or more tasks.", "reversible_write", required=("task_ids",)),
-    "tasks.delete": ActionSpec("Delete tasks and optionally their data.", "destructive_write", required=("task_ids",)),
-    "tasks.recheck": ActionSpec("Force data verification for tasks.", "external_side_effect", required=("task_ids",)),
     "tasks.reannounce": ActionSpec(
-        "Force tracker reannounce.", "external_side_effect", ("qbittorrent", "transmission"), ("task_ids",)
+        "Force tracker reannounce.",
+        "external_side_effect",
+        ("qbittorrent", "transmission"),
+        (TASK_ID, TASK_IDS),
+        ("task_id 与 task_ids 必须提供且只能选择一种。",),
     ),
     "tasks.queue.move": ActionSpec(
         "Move tasks to top, up, down, or bottom of the queue.",
         "reversible_write",
         ("qbittorrent", "transmission"),
-        ("task_ids", "position"),
+        (
+            TASK_ID,
+            TASK_IDS,
+            ArgumentSpec(
+                "position",
+                "string",
+                "目标队列位置。",
+                required=True,
+                enum=("top", "up", "down", "bottom"),
+            ),
+        ),
+        ("task_id 与 task_ids 必须提供且只能选择一种。",),
     ),
     "tasks.force_start.set": ActionSpec(
         "Enable or disable qBittorrent force-start for tasks.",
         "reversible_write",
         ("qbittorrent",),
-        ("task_ids", "enabled"),
+        (
+            TASK_ID,
+            TASK_IDS,
+            ArgumentSpec("enabled", "boolean", "是否启用强制开始。", required=True),
+        ),
+        ("task_id 与 task_ids 必须提供且只能选择一种。",),
     ),
     "tasks.properties.set": ActionSpec(
-        "Set task speed, ratio, or seeding-time limits.", "reversible_write", required=("task_id",)
+        "Set task speed, ratio, or seeding-time limits.",
+        "reversible_write",
+        arguments=(
+            ArgumentSpec("task_id", "string", TASK_ID.description, required=True),
+            ArgumentSpec("upload_limit", "number", "上传限速，单位 KB/s；0 表示不限速。"),
+            ArgumentSpec("download_limit", "number", "下载限速，单位 KB/s；0 表示不限速。"),
+            ArgumentSpec("ratio_limit", "number", "分享率上限；rTorrent 不支持。"),
+            ArgumentSpec("seeding_time_limit", "integer", "做种时间上限，单位分钟；rTorrent 不支持。"),
+        ),
     ),
     "tasks.location.set": ActionSpec(
-        "Move or retarget one task to a provider-side path.", "external_side_effect", required=("task_id", "location")
+        "Move or retarget one task to a provider-side path.",
+        "external_side_effect",
+        arguments=(
+            ArgumentSpec("task_id", "string", TASK_ID.description, required=True),
+            ArgumentSpec("location", "string", "下载器侧的新保存路径。", required=True),
+        ),
     ),
     "tasks.category.set": ActionSpec(
-        "Set qBittorrent category.", "reversible_write", ("qbittorrent",), ("task_id", "category")
+        "Set qBittorrent category.",
+        "reversible_write",
+        ("qbittorrent",),
+        (
+            ArgumentSpec("task_id", "string", TASK_ID.description, required=True),
+            ArgumentSpec("category", "string", "非空分类名称。", required=True),
+        ),
+    ),
+    "tasks.tags.set": ActionSpec(
+        "Set or add task tags/labels.",
+        "reversible_write",
+        arguments=(
+            TASK_ID,
+            TASK_IDS,
+            ArgumentSpec("tags", "string[]", "要设置或添加的标签列表。", required=True),
+        ),
+        argument_rules=("task_id 与 task_ids 必须提供且只能选择一种。",),
     ),
-    "tasks.tags.set": ActionSpec("Set or add task tags/labels.", "reversible_write", required=("task_ids", "tags")),
     "tasks.trackers.update": ActionSpec(
-        "Add or replace task trackers.", "reversible_write", ("qbittorrent", "transmission"), ("task_id", "trackers")
+        "Add or replace task trackers.",
+        "reversible_write",
+        ("qbittorrent", "transmission"),
+        (
+            ArgumentSpec("task_id", "string", TASK_ID.description, required=True),
+            ArgumentSpec("trackers", "string[]", "Tracker URL 列表。", required=True),
+        ),
     ),
     "tasks.add.direct": ActionSpec(
         "Submit a magnet, URL, or local torrent file directly to the provider.",
         "external_side_effect",
-        required=("content",),
+        arguments=(
+            ArgumentSpec("content", "string", "Magnet、torrent URL，或 torrent_file=true 时的本地种子文件路径。", required=True),
+            ArgumentSpec("torrent_file", "boolean", "将 content 解释为本地种子文件路径。", default=False),
+            ArgumentSpec("paused", "boolean", "以暂停状态添加任务。", default=False),
+            ArgumentSpec("download_dir", "string", "下载器侧保存路径。"),
+            ArgumentSpec("tags", "string[]", "添加到任务的标签。"),
+            ArgumentSpec("category", "string", "qBittorrent 分类；其他 provider 忽略。"),
+        ),
     ),
     "session.stats": ActionSpec("Read provider transfer/session statistics.", "safe_read"),
     "session.speed_limits.get": ActionSpec("Read global speed limits.", "safe_read", ("qbittorrent", "transmission")),
     "session.speed_limits.set": ActionSpec(
-        "Set global speed limits in KB/s.", "reversible_write", ("qbittorrent", "transmission")
+        "Set global speed limits in KB/s.",
+        "reversible_write",
+        ("qbittorrent", "transmission"),
+        (
+            ArgumentSpec("download_limit", "number", "全局下载限速，单位 KB/s；0 或省略表示不限速。"),
+            ArgumentSpec("upload_limit", "number", "全局上传限速，单位 KB/s；0 或省略表示不限速。"),
+        ),
     ),
     "session.details": ActionSpec(
         "Read Transmission session configuration and capacity details.",
@@ -126,8 +292,7 @@ def _load_configs() -> list[Any]:
     _ensure_project_import()
     from app.db.oper.systemconfig import SystemConfigOper
     from app.db.session import SessionFactory
-    from app.runtime.extensions.service import ServiceConfigHelper
-    from app.runtime.extensions.service import configure_service_config_reader
+    from app.runtime.extensions.service import ServiceConfigHelper, configure_service_config_reader
 
     system_config = SystemConfigOper()
     # Skill 在独立 CLI 进程中运行，没有 lifespan 为无会话 Oper 装配事务执行器。
@@ -151,12 +316,15 @@ def _select_config(client_name: Optional[str]) -> Any:
             if config.name == client_name:
                 return config
         raise ValueError(f"未找到已启用下载器实例: {client_name}")
+    if not enabled:
+        raise ValueError("没有已启用的下载器实例")
     defaults = [config for config in enabled if config.default]
     if len(defaults) == 1:
         return defaults[0]
     if len(enabled) == 1:
         return enabled[0]
-    raise ValueError("存在多个下载器实例，请显式提供 --client")
+    names = "、".join(str(config.name) for config in enabled)
+    raise ValueError(f"存在多个下载器实例，请用 --client 指定以下之一：{names}")
 
 
 def _build_client(config: Any) -> Any:
@@ -169,7 +337,7 @@ def _build_client(config: Any) -> Any:
     if client.is_inactive():
         client.reconnect()
     if client.is_inactive():
-        raise RuntimeError("下载器连接不可用")
+        raise OperationError("下载器连接不可用")
     return client
 
 
@@ -253,6 +421,99 @@ def _require(arguments: Mapping[str, Any], name: str) -> Any:
     return value
 
 
+def _matches_argument_type(value: Any, declared_type: str) -> bool:
+    """判断 JSON 值是否符合公开参数合同中的紧凑类型表达式。"""
+    for candidate in declared_type.split("|"):
+        if candidate.endswith("[]"):
+            if isinstance(value, list) and all(
+                _matches_argument_type(item, candidate[:-2]) for item in value
+            ):
+                return True
+            continue
+        if candidate == "string" and isinstance(value, str):
+            return True
+        if candidate == "integer" and isinstance(value, int) and not isinstance(value, bool):
+            return True
+        if candidate == "number" and isinstance(value, (int, float)) and not isinstance(value, bool):
+            return True
+        if candidate == "boolean" and isinstance(value, bool):
+            return True
+        if candidate == "object" and isinstance(value, Mapping):
+            return True
+    return False
+
+
+def _validate_action_arguments(action: str, spec: ActionSpec, arguments: Mapping[str, Any]) -> None:
+    """一次性校验 action 的全部参数，避免 Agent 按单个错误反复试调用。"""
+    errors: list[str] = []
+    argument_specs = {argument.name: argument for argument in spec.arguments}
+    unknown = sorted(set(arguments) - set(argument_specs))
+    if unknown:
+        errors.append(f"未知参数: {', '.join(unknown)}")
+    for name, argument in argument_specs.items():
+        value = arguments.get(name)
+        if argument.required and (value is None or value == "" or value == []):
+            errors.append(f"缺少必填参数: {name}")
+            continue
+        if value is not None and not _matches_argument_type(value, argument.type):
+            errors.append(f"参数 {name} 必须是 {argument.type}")
+        if value is not None and argument.enum and value not in argument.enum:
+            errors.append(f"参数 {name} 仅支持: {', '.join(map(str, argument.enum))}")
+
+    task_selector_actions = {
+        "tasks.start",
+        "tasks.stop",
+        "tasks.delete",
+        "tasks.recheck",
+        "tasks.reannounce",
+        "tasks.queue.move",
+        "tasks.force_start.set",
+        "tasks.tags.set",
+    }
+    if action in task_selector_actions:
+        selector_count = int(bool(arguments.get("task_id"))) + int(bool(arguments.get("task_ids")))
+        if selector_count != 1:
+            errors.append("task_id 与 task_ids 必须提供且只能选择一种")
+    if action == "tasks.files.selection.set":
+        wanted = arguments.get("wanted_file_ids") or []
+        unwanted = arguments.get("unwanted_file_ids") or []
+        if not wanted and not unwanted:
+            errors.append("wanted_file_ids 与 unwanted_file_ids 至少提供一项")
+        comparable_indexes = isinstance(wanted, list) and isinstance(unwanted, list) and all(
+            isinstance(item, int) and not isinstance(item, bool) for item in [*wanted, *unwanted]
+        )
+        if comparable_indexes and set(wanted) & set(unwanted):
+            errors.append("同一文件不能同时出现在 wanted_file_ids 和 unwanted_file_ids")
+    if action == "tasks.properties.set" and not any(
+        arguments.get(name) is not None
+        for name in ("upload_limit", "download_limit", "ratio_limit", "seeding_time_limit")
+    ):
+        errors.append("至少提供一个要修改的任务属性")
+    if action == "session.speed_limits.set" and not any(
+        arguments.get(name) is not None for name in ("download_limit", "upload_limit")
+    ):
+        errors.append("至少提供 download_limit 或 upload_limit；清除限速请显式传 0")
+    if errors:
+        raise ValueError("参数校验失败：" + "；".join(errors))
+
+
+def _validate_provider_arguments(action: str, provider: str, arguments: Mapping[str, Any]) -> None:
+    """拒绝会被特定 provider 静默忽略的参数。"""
+    errors: list[str] = []
+    if action == "tasks.properties.set" and provider == "rtorrent":
+        unsupported = [
+            name
+            for name in ("ratio_limit", "seeding_time_limit")
+            if arguments.get(name) is not None
+        ]
+        if unsupported:
+            errors.append(f"rTorrent 不支持参数: {', '.join(unsupported)}")
+    if action == "tasks.add.direct" and provider != "qbittorrent" and arguments.get("category") is not None:
+        errors.append(f"{provider} 不支持参数: category")
+    if errors:
+        raise ValueError("参数校验失败：" + "；".join(errors))
+
+
 def _tasks_list(client: Any, arguments: Mapping[str, Any]) -> dict[str, Any]:
     """查询并分页返回下载任务。"""
     tasks, error = client.get_torrents(
@@ -261,7 +522,7 @@ def _tasks_list(client: Any, arguments: Mapping[str, Any]) -> dict[str, Any]:
         tags=arguments.get("tags"),
     )
     if error:
-        raise RuntimeError("下载器任务查询失败")
+        raise OperationError("下载器任务查询失败")
     return _page(tasks or [], arguments)
 
 
@@ -273,7 +534,7 @@ def _tags_get(client: Any, provider: str, arguments: Mapping[str, Any]) -> Any:
         return getter(task_id)
     tasks, error = client.get_torrents(ids=task_id)
     if error or not tasks:
-        raise RuntimeError("任务标签查询失败")
+        raise OperationError("任务标签查询失败")
     task = _jsonable(tasks[0])
     if provider == "qbittorrent":
         tags = task.get("tags") if isinstance(task, dict) else None
@@ -469,12 +730,20 @@ def list_instances() -> dict[str, Any]:
     return {"success": True, "instances": instances}
 
 
-def list_capabilities(client_name: Optional[str]) -> dict[str, Any]:
-    """返回全部或指定实例支持的 action 清单。"""
+def list_capabilities(client_name: Optional[str], action: Optional[str] = None) -> dict[str, Any]:
+    """返回全部或指定实例支持的 action 及完整参数合同。"""
     provider = None
     if client_name:
         provider = str(_select_config(client_name).type or "").lower()
-    actions = [spec.to_dict(name) for name, spec in ACTIONS.items() if provider is None or provider in spec.providers]
+    if action and action not in ACTIONS:
+        raise ValueError(f"未知 downloader action: {action}")
+    actions = [
+        spec.to_dict(name)
+        for name, spec in ACTIONS.items()
+        if (not action or name == action) and (provider is None or provider in spec.providers)
+    ]
+    if action and not actions:
+        raise ValueError(f"{provider} 不支持 action: {action}")
     return {
         "success": True,
         "client": client_name,
@@ -488,18 +757,15 @@ def call_action(client_name: Optional[str], action: str, arguments: Mapping[str,
     spec = ACTIONS.get(action)
     if spec is None:
         raise ValueError(f"未知 downloader action: {action}")
+    _validate_action_arguments(action, spec, arguments)
     config = _select_config(client_name)
     provider = str(config.type or "").lower()
     if provider not in spec.providers:
         raise ValueError(f"{provider} 不支持 action: {action}")
-    for name in spec.required:
-        if name == "task_ids":
-            _task_ids(arguments)
-        else:
-            _require(arguments, name)
+    _validate_provider_arguments(action, provider, arguments)
     result = _dispatch(_build_client(config), provider, action, arguments)
     if spec.effect != "safe_read" and result is False:
-        raise RuntimeError("下载器 action 返回失败")
+        raise OperationError("下载器 action 返回失败")
     return {
         "success": True,
         "client": config.name,
@@ -525,9 +791,10 @@ def _build_parser() -> argparse.ArgumentParser:
     subparsers.add_parser("instances", help="list configured instances")
     capabilities = subparsers.add_parser("capabilities", help="list allowed actions")
     capabilities.add_argument("--client")
+    capabilities.add_argument("--action")
     call = subparsers.add_parser("call", help="call one allowed action")
     call.add_argument("--client")
-    call.add_argument("--action", required=True, choices=sorted(ACTIONS))
+    call.add_argument("--action", required=True)
     call.add_argument("--arguments", default="{}")
     return parser
 
@@ -540,7 +807,7 @@ def main() -> int:
         if args.command == "instances":
             payload = list_instances()
         elif args.command == "capabilities":
-            payload = list_capabilities(args.client)
+            payload = list_capabilities(args.client, args.action)
         elif args.command == "call":
             payload = call_action(args.client, args.action, _parse_arguments(args.arguments))
         else:
@@ -550,7 +817,7 @@ def main() -> int:
         payload = {
             "success": False,
             "error_type": type(error).__name__,
-            "message": str(error) if isinstance(error, ValueError) else "下载器调用失败",
+            "message": str(error) if isinstance(error, (ValueError, OperationError)) else "下载器调用失败",
         }
         print(json.dumps(payload, ensure_ascii=False, indent=2))
         return 1

skills/mediaserver-operation/SKILL.md

@@ -1,6 +1,6 @@
 ---
 name: mediaserver-operation
-version: 1
+version: 2
 description: >-
   Use this skill when the user asks to inspect, diagnose, or directly operate a
   configured Emby, Jellyfin, Plex, ZSpace, UGREEN, TrimeMedia, or Navidrome
@@ -27,7 +27,30 @@ username, password, API key, token, Cookie, or arbitrary URL.
 - A provider result is not automatically a MoviePilot transfer, subscription,
   or history fact. Use the appropriate MoviePilot API for those workflows.
 
-## Discover First
+## Instance And Provider Discovery
+
+### Fast path: call directly
+
+Do not routinely call `instances` or `capabilities` before an operation. This
+Skill already contains the full action contract, and the helper performs
+instance resolution, provider support checks, complete argument validation, and
+the action in one `call` invocation.
+
+- If the user or prior context provides the exact server name, pass it with
+  `--server` and call the action immediately.
+- If no server name is known, omit `--server`. The helper automatically uses the
+  only enabled media server.
+- If multiple servers remain ambiguous, the failed call lists every valid
+  server name. Reuse that list for the next direct call; do not add a separate
+  `instances` call unless the user explicitly asks to inspect instances.
+- Do not probe an action with empty or guessed arguments. Compose the complete
+  JSON object from the contract below before calling.
+
+The helper rejects unknown fields and reports all detectable argument errors in
+one response before connecting to the provider, so correct every reported field
+together instead of retrying one field at a time.
+
+### Optional discovery
 
 ```bash
 python skills/mediaserver-operation/scripts/mp-mediaserver.py instances
@@ -35,8 +58,19 @@ python skills/mediaserver-operation/scripts/mp-mediaserver.py capabilities
 python skills/mediaserver-operation/scripts/mp-mediaserver.py capabilities --server "living-room"
 ```
 
-`capabilities` returns namespaced actions, argument requirements, providers, and
-side-effect levels. Call it before using an unfamiliar server or advanced action.
+The complete action and argument contract is documented below. Use
+`capabilities` only to confirm which documented actions a configured provider
+supports. For a compact machine-readable copy of one action's same contract:
+
+```bash
+python skills/mediaserver-operation/scripts/mp-mediaserver.py capabilities \
+  --server "living-room" \
+  --action items.season_episodes
+```
+
+Do not inspect the helper source to discover arguments and do not guess a
+provider-specific action. `capabilities` is optional and should be used only
+when the configured provider itself is unknown or support must be diagnosed.
 
 ## Call Shape
 
@@ -50,14 +84,100 @@ python skills/mediaserver-operation/scripts/mp-mediaserver.py call \
 The `--arguments` value must be one JSON object. List reads default to 50 items
 and cap at 200.
 
-## Actions
+## External MCP Contract
 
-- Read: `server.statistics`, `server.users.count`,
-  `server.user.library_folders`, `libraries.list`, `items.list`, `items.count`,
-  `items.detail`, `items.movies.search`, `items.music.search`,
-  `items.season_episodes`, `activity.latest`, `activity.resume`,
-  `activity.backdrops`, `playback.sessions`, and `playback.url`.
-- External side effects: `library.scan` and `metadata.refresh`.
+External MCP clients do not receive this `SKILL.md` and cannot use the hidden
+`execute_command` tool. MoviePilot therefore exposes a separate admin-only MCP
+tool named `mediaserver_operation`. Its `tools/list` `inputSchema` contains one
+`oneOf` branch for every action below, including the function description,
+supported providers, effect, field types, required/default values, enums,
+nested `metadata.refresh` item fields, and cross-field rules. The external
+client should select the matching branch and make one `tools/call`; it does not
+need to call a discovery tool first.
+
+MCP call arguments use the same contract without shell quoting:
+
+```json
+{
+  "server": "living-room",
+  "action": "items.season_episodes",
+  "arguments": {
+    "item_id": "exact-series-id",
+    "season": 2
+  }
+}
+```
+
+`server` may be omitted when only one media server is enabled. If multiple
+instances remain ambiguous, the result lists the valid server names.
+
+## Complete Action Contract
+
+In the tables below, `*` means required. Every listed field belongs inside the
+single `--arguments` JSON object. Do not send fields that are not listed.
+
+Shared rules:
+
+- Paged reads accept `offset:integer=0` where documented and
+  `limit:integer=50`; `offset` must be non-negative and `limit` is clamped to
+  `1..200`.
+- `parent`, `item_id`, library IDs, and usernames are native to the selected
+  server. Obtain them from that server's earlier response; never reuse IDs from
+  another instance.
+- All actions support only the providers shown by `capabilities`. The provider
+  list below lets the Agent choose without inspecting source; query the selected
+  instance only when provider support must be confirmed.
+
+Provider abbreviations used below: all = Emby, Jellyfin, Plex, ZSpace, UGREEN,
+TrimeMedia, and Navidrome.
+
+### Server and library reads
+
+| Action | Function and providers | `--arguments` fields |
+|---|---|---|
+| `server.statistics` | Read media counts/provider statistics; all | none (`{}`) |
+| `server.users.count` | Read provider user count; Emby, Jellyfin, ZSpace, UGREEN, TrimeMedia, Navidrome | none (`{}`) |
+| `server.user.library_folders` | Read current user's visible folders; Emby, Jellyfin, ZSpace | none (`{}`) |
+| `libraries.list` | List visible libraries; all | `hidden:boolean=false` (true = configured sync scope only); `username:string` only for Emby/Jellyfin/ZSpace |
+| `items.list` | Page items below a library/parent; all | `parent:string\|integer` required except Navidrome; `offset:integer=0`; `limit:integer=50` |
+| `items.count` | Count items below a library/parent; all | `parent:string\|integer` required except Navidrome; omitted on Navidrome uses `music` |
+| `items.detail` | Read one provider item; all | `item_id*:string` |
+
+### Native search and activity
+
+| Action | Function and providers | `--arguments` fields |
+|---|---|---|
+| `items.movies.search` | Search movies; Emby, Jellyfin, Plex, ZSpace, UGREEN, TrimeMedia | `title*:string`; `year:string\|integer` |
+| `items.music.search` | Search music; Emby, Jellyfin, Plex, ZSpace, UGREEN, Navidrome | `title:string`; `artist:string`; `album:string`; at least one is required |
+| `items.season_episodes` | Read existing episode coverage for a series; Emby, Jellyfin, Plex, ZSpace, UGREEN, TrimeMedia | `item_id:string`; `title:string`; at least one is required; optional `year:string\|integer`; `season:integer` |
+| `activity.latest` | Read recently added items; all | `limit:integer=50`; `username:string` only for Emby/Jellyfin/ZSpace |
+| `activity.resume` | Read in-progress/resumable items; all | `limit:integer=50`; `username:string` only for Emby/Jellyfin/ZSpace |
+| `activity.backdrops` | Read recent backdrop URLs; UGREEN, TrimeMedia | `limit:integer=50`; `remote:boolean=false` |
+
+### Playback and writes
+
+| Action | Function, providers, and effect | `--arguments` fields |
+|---|---|---|
+| `playback.sessions` | Read active sessions; Emby, Jellyfin, Plex; safe read | none (`{}`) |
+| `playback.url` | Build provider play URL; all; safe read | `item_id*:string` |
+| `library.scan` | Trigger provider root-library scan; all; external side effect | `scan_mode:string\|integer` only for UGREEN; otherwise omit |
+| `metadata.refresh` | Refresh metadata for mapped items; Emby, Plex, ZSpace, UGREEN, TrimeMedia; external side effect | `items*:object[]`; each object supports `title:string`, `year:string\|integer`, `type:string` (`电影\|电视剧\|音乐`), `category:string`, `target_path:string` |
+
+Examples:
+
+```bash
+# List the first page below an exact library ID.
+python skills/mediaserver-operation/scripts/mp-mediaserver.py call \
+  --server "living-room" \
+  --action items.list \
+  --arguments '{"parent":"exact-library-id","offset":0,"limit":50}'
+
+# Read season 2 coverage using an exact provider series ID.
+python skills/mediaserver-operation/scripts/mp-mediaserver.py call \
+  --server "living-room" \
+  --action items.season_episodes \
+  --arguments '{"item_id":"exact-series-id","season":2}'
+```
 
 Use the exact `server` and item/library IDs returned by earlier calls. Do not
 invent IDs or reuse IDs across different server instances. `items.list` expects

skills/mediaserver-operation/scripts/mp-mediaserver.py

@@ -35,6 +35,37 @@ PROVIDER_CLASSES = {
     "trimemedia": "app.modules.trimemedia.trimemedia:TrimeMedia",
     "navidrome": "app.modules.navidrome.navidrome:Navidrome",
 }
+_UNSET = object()
+
+
+class OperationError(RuntimeError):
+    """可安全返回给 Agent 的媒体服务器操作错误。"""
+
+
+@dataclass(frozen=True, slots=True)
+class ArgumentSpec:
+    """描述一个 action 参数的公开调用合同。"""
+
+    name: str
+    type: str
+    description: str
+    required: bool = False
+    default: Any = _UNSET
+    enum: tuple[Any, ...] = ()
+
+    def to_dict(self) -> dict[str, Any]:
+        """返回可直接交给 Agent 的参数 schema。"""
+        result: dict[str, Any] = {
+            "name": self.name,
+            "type": self.type,
+            "required": self.required,
+            "description": self.description,
+        }
+        if self.default is not _UNSET:
+            result["default"] = self.default
+        if self.enum:
+            result["enum"] = list(self.enum)
+        return result
 
 
 @dataclass(frozen=True, slots=True)
@@ -44,7 +75,13 @@ class ActionSpec:
     description: str
     effect: str
     providers: tuple[str, ...] = ALL_PROVIDERS
-    required: tuple[str, ...] = ()
+    arguments: tuple[ArgumentSpec, ...] = ()
+    argument_rules: tuple[str, ...] = ()
+
+    @property
+    def required(self) -> tuple[str, ...]:
+        """返回保持旧能力合同兼容的必填参数名。"""
+        return tuple(argument.name for argument in self.arguments if argument.required)
 
     def to_dict(self, name: str) -> dict[str, Any]:
         """返回不包含实现对象的公开能力描述。"""
@@ -54,9 +91,17 @@ class ActionSpec:
             "effect": self.effect,
             "providers": list(self.providers),
             "required_arguments": list(self.required),
+            "arguments": [argument.to_dict() for argument in self.arguments],
+            "argument_rules": list(self.argument_rules),
         }
 
 
+ITEM_ID = ArgumentSpec("item_id", "string", "当前媒体服务器返回的 provider 原生条目 ID。")
+PARENT = ArgumentSpec("parent", "string|integer", "媒体库或父条目 ID；Navidrome 可省略并使用 music。")
+OFFSET = ArgumentSpec("offset", "integer", "列表起始偏移，必须大于等于 0。", default=0)
+LIMIT = ArgumentSpec("limit", "integer", "返回条数，范围 1..200。", default=DEFAULT_LIMIT)
+
+
 ACTIONS: dict[str, ActionSpec] = {
     "server.statistics": ActionSpec("Read media counts and provider statistics.", "safe_read"),
     "server.users.count": ActionSpec(
@@ -69,41 +114,107 @@ ACTIONS: dict[str, ActionSpec] = {
         "safe_read",
         ("emby", "jellyfin", "zspace"),
     ),
-    "libraries.list": ActionSpec("List visible provider libraries.", "safe_read"),
-    "items.list": ActionSpec("Page items below one library or parent.", "safe_read"),
-    "items.count": ActionSpec("Count items below one library or parent.", "safe_read"),
-    "items.detail": ActionSpec("Read one provider item by native ID.", "safe_read", required=("item_id",)),
+    "libraries.list": ActionSpec(
+        "List visible provider libraries.",
+        "safe_read",
+        arguments=(
+            ArgumentSpec("hidden", "boolean", "仅返回配置为同步范围的媒体库。", default=False),
+            ArgumentSpec("username", "string", "按用户名读取可见媒体库；仅 Emby、Jellyfin、ZSpace 支持。"),
+        ),
+    ),
+    "items.list": ActionSpec(
+        "Page items below one library or parent.",
+        "safe_read",
+        arguments=(PARENT, OFFSET, LIMIT),
+        argument_rules=("除 Navidrome 外必须提供 parent；Navidrome 忽略 parent。",),
+    ),
+    "items.count": ActionSpec(
+        "Count items below one library or parent.",
+        "safe_read",
+        arguments=(PARENT,),
+        argument_rules=("除 Navidrome 外必须提供 parent；Navidrome 省略时使用 music。",),
+    ),
+    "items.detail": ActionSpec(
+        "Read one provider item by native ID.",
+        "safe_read",
+        arguments=(ArgumentSpec("item_id", "string", ITEM_ID.description, required=True),),
+    ),
     "items.movies.search": ActionSpec(
         "Search provider-native movie items by title and optional year.",
         "safe_read",
         ("emby", "jellyfin", "plex", "zspace", "ugreen", "trimemedia"),
-        ("title",),
+        (
+            ArgumentSpec("title", "string", "电影标题。", required=True),
+            ArgumentSpec("year", "string|integer", "可选发行年份。"),
+        ),
     ),
     "items.music.search": ActionSpec(
         "Search provider-native music by title, artist, or album.",
         "safe_read",
         ("emby", "jellyfin", "plex", "zspace", "ugreen", "navidrome"),
+        (
+            ArgumentSpec("title", "string", "歌曲、专辑或音乐条目标题。"),
+            ArgumentSpec("artist", "string", "艺人名称。"),
+            ArgumentSpec("album", "string", "专辑名称；title、artist、album 至少提供一项。"),
+        ),
+        ("title、artist、album 至少提供一项。",),
     ),
     "items.season_episodes": ActionSpec(
         "Read native episode coverage for one series and optional season.",
         "safe_read",
         ("emby", "jellyfin", "plex", "zspace", "ugreen", "trimemedia"),
+        (
+            ITEM_ID,
+            ArgumentSpec("title", "string", "剧集标题；与 item_id 至少提供一项。"),
+            ArgumentSpec("year", "string|integer", "可选首播年份。"),
+            ArgumentSpec("season", "integer", "可选季号。"),
+        ),
+        ("item_id 与 title 至少提供一项。",),
+    ),
+    "activity.latest": ActionSpec(
+        "Read recently added provider items.",
+        "safe_read",
+        arguments=(LIMIT, ArgumentSpec("username", "string", "按用户名读取；仅 Emby、Jellyfin、ZSpace 支持。")),
+    ),
+    "activity.resume": ActionSpec(
+        "Read in-progress/resumable provider items.",
+        "safe_read",
+        arguments=(LIMIT, ArgumentSpec("username", "string", "按用户名读取；仅 Emby、Jellyfin、ZSpace 支持。")),
     ),
-    "activity.latest": ActionSpec("Read recently added provider items.", "safe_read"),
-    "activity.resume": ActionSpec("Read in-progress/resumable provider items.", "safe_read"),
     "activity.backdrops": ActionSpec(
         "Read recent provider backdrop images.",
         "safe_read",
         ("ugreen", "trimemedia"),
+        (
+            LIMIT,
+            ArgumentSpec("remote", "boolean", "返回 provider 可远程访问的图片地址。", default=False),
+        ),
     ),
     "playback.sessions": ActionSpec("Read active playback sessions.", "safe_read", ("emby", "jellyfin", "plex")),
-    "playback.url": ActionSpec("Build the provider play URL for one item.", "safe_read", required=("item_id",)),
-    "library.scan": ActionSpec("Trigger a provider library scan.", "external_side_effect"),
+    "playback.url": ActionSpec(
+        "Build the provider play URL for one item.",
+        "safe_read",
+        arguments=(ArgumentSpec("item_id", "string", ITEM_ID.description, required=True),),
+    ),
+    "library.scan": ActionSpec(
+        "Trigger a provider library scan.",
+        "external_side_effect",
+        arguments=(
+            ArgumentSpec("scan_mode", "string|integer", "UGREEN 原生扫描模式；其他 provider 必须省略。"),
+        ),
+    ),
     "metadata.refresh": ActionSpec(
         "Refresh provider metadata for mapped items.",
         "external_side_effect",
         ("emby", "plex", "zspace", "ugreen", "trimemedia"),
-        ("items",),
+        (
+            ArgumentSpec(
+                "items",
+                "object[]",
+                "刷新条目；每项支持 title:string、year:string|integer、type:电影|电视剧|音乐、category:string、target_path:string。",
+                required=True,
+            ),
+        ),
     ),
 }
 
@@ -120,8 +231,7 @@ def _load_configs() -> list[Any]:
     _ensure_project_import()
     from app.db.oper.systemconfig import SystemConfigOper
     from app.db.session import SessionFactory
-    from app.runtime.extensions.service import ServiceConfigHelper
-    from app.runtime.extensions.service import configure_service_config_reader
+    from app.runtime.extensions.service import ServiceConfigHelper, configure_service_config_reader
 
     system_config = SystemConfigOper()
     # Skill 在独立 CLI 进程中运行，没有 lifespan 为无会话 Oper 装配事务执行器。
@@ -145,9 +255,12 @@ def _select_config(server_name: Optional[str]) -> Any:
             if config.name == server_name:
                 return config
         raise ValueError(f"未找到已启用媒体服务器实例: {server_name}")
+    if not enabled:
+        raise ValueError("没有已启用的媒体服务器实例")
     if len(enabled) == 1:
         return enabled[0]
-    raise ValueError("存在多个媒体服务器实例，请显式提供 --server")
+    names = "、".join(str(config.name) for config in enabled)
+    raise ValueError(f"存在多个媒体服务器实例，请用 --server 指定以下之一：{names}")
 
 
 def _build_client(config: Any) -> Any:
@@ -163,7 +276,7 @@ def _build_client(config: Any) -> Any:
     if client.is_inactive():
         client.reconnect()
     if client.is_inactive():
-        raise RuntimeError("媒体服务器连接不可用")
+        raise OperationError("媒体服务器连接不可用")
     return client
 
 
@@ -221,6 +334,101 @@ def _require(arguments: Mapping[str, Any], name: str) -> Any:
     return value
 
 
+def _matches_argument_type(value: Any, declared_type: str) -> bool:
+    """判断 JSON 值是否符合公开参数合同中的紧凑类型表达式。"""
+    for candidate in declared_type.split("|"):
+        if candidate.endswith("[]"):
+            if isinstance(value, list) and all(
+                _matches_argument_type(item, candidate[:-2]) for item in value
+            ):
+                return True
+            continue
+        if candidate == "string" and isinstance(value, str):
+            return True
+        if candidate == "integer" and isinstance(value, int) and not isinstance(value, bool):
+            return True
+        if candidate == "boolean" and isinstance(value, bool):
+            return True
+        if candidate == "object" and isinstance(value, Mapping):
+            return True
+    return False
+
+
+def _validate_refresh_items(items: Any) -> list[str]:
+    """校验 metadata.refresh 的嵌套条目并返回全部错误。"""
+    if not isinstance(items, list):
+        return []
+    errors: list[str] = []
+    allowed_fields = {"title", "year", "type", "category", "target_path"}
+    allowed_types = {"电影", "电视剧", "音乐"}
+    for index, item in enumerate(items):
+        if not isinstance(item, Mapping):
+            continue
+        unknown = sorted(set(item) - allowed_fields)
+        if unknown:
+            errors.append(f"items[{index}] 未知字段: {', '.join(unknown)}")
+        if item.get("type") is not None and item.get("type") not in allowed_types:
+            errors.append(f"items[{index}].type 仅支持: 电影、电视剧、音乐")
+        for name in ("title", "category", "target_path"):
+            if item.get(name) is not None and not isinstance(item.get(name), str):
+                errors.append(f"items[{index}].{name} 必须是 string")
+        year = item.get("year")
+        if year is not None and not (
+            isinstance(year, str) or (isinstance(year, int) and not isinstance(year, bool))
+        ):
+            errors.append(f"items[{index}].year 必须是 string|integer")
+    return errors
+
+
+def _validate_action_arguments(action: str, spec: ActionSpec, arguments: Mapping[str, Any]) -> None:
+    """一次性校验 action 的全部参数，避免 Agent 按单个错误反复试调用。"""
+    errors: list[str] = []
+    argument_specs = {argument.name: argument for argument in spec.arguments}
+    unknown = sorted(set(arguments) - set(argument_specs))
+    if unknown:
+        errors.append(f"未知参数: {', '.join(unknown)}")
+    for name, argument in argument_specs.items():
+        value = arguments.get(name)
+        if argument.required and (value is None or value == "" or value == []):
+            errors.append(f"缺少必填参数: {name}")
+            continue
+        if value is not None and not _matches_argument_type(value, argument.type):
+            errors.append(f"参数 {name} 必须是 {argument.type}")
+        if value is not None and argument.enum and value not in argument.enum:
+            errors.append(f"参数 {name} 仅支持: {', '.join(map(str, argument.enum))}")
+
+    if action == "items.music.search" and not any(
+        arguments.get(name) for name in ("title", "artist", "album")
+    ):
+        errors.append("title、artist、album 至少提供一项")
+    if action == "items.season_episodes" and not any(
+        arguments.get(name) for name in ("item_id", "title")
+    ):
+        errors.append("item_id 与 title 至少提供一项")
+    if action == "metadata.refresh":
+        errors.extend(_validate_refresh_items(arguments.get("items")))
+    if errors:
+        raise ValueError("参数校验失败：" + "；".join(errors))
+
+
+def _validate_provider_arguments(action: str, provider: str, arguments: Mapping[str, Any]) -> None:
+    """校验需要知道 provider 后才能确定的条件参数。"""
+    errors: list[str] = []
+    if action in {"items.list", "items.count"} and provider != "navidrome" and not arguments.get("parent"):
+        errors.append(f"{provider} 的 {action} 必须提供 parent")
+    username_providers = {"emby", "jellyfin", "zspace"}
+    if (
+        action in {"libraries.list", "activity.latest", "activity.resume"}
+        and arguments.get("username") is not None
+        and provider not in username_providers
+    ):
+        errors.append(f"{provider} 的 {action} 不支持参数: username")
+    if action == "library.scan" and arguments.get("scan_mode") is not None and provider != "ugreen":
+        errors.append(f"{provider} 的 library.scan 不支持参数: scan_mode")
+    if errors:
+        raise ValueError("参数校验失败：" + "；".join(errors))
+
+
 def _limit(arguments: Mapping[str, Any]) -> int:
     """读取安全的分页条数。"""
     return min(MAX_LIMIT, max(1, int(arguments.get("limit", DEFAULT_LIMIT))))
@@ -379,12 +587,20 @@ def list_instances() -> dict[str, Any]:
     return {"success": True, "instances": instances}
 
 
-def list_capabilities(server_name: Optional[str]) -> dict[str, Any]:
-    """返回全部或指定实例支持的 action 清单。"""
+def list_capabilities(server_name: Optional[str], action: Optional[str] = None) -> dict[str, Any]:
+    """返回全部或指定实例支持的 action 及完整参数合同。"""
     provider = None
     if server_name:
         provider = str(_select_config(server_name).type or "").lower()
-    actions = [spec.to_dict(name) for name, spec in ACTIONS.items() if provider is None or provider in spec.providers]
+    if action and action not in ACTIONS:
+        raise ValueError(f"未知 media server action: {action}")
+    actions = [
+        spec.to_dict(name)
+        for name, spec in ACTIONS.items()
+        if (not action or name == action) and (provider is None or provider in spec.providers)
+    ]
+    if action and not actions:
+        raise ValueError(f"{provider} 不支持 action: {action}")
     return {
         "success": True,
         "server": server_name,
@@ -398,15 +614,15 @@ def call_action(server_name: Optional[str], action: str, arguments: Mapping[str,
     spec = ACTIONS.get(action)
     if spec is None:
         raise ValueError(f"未知 media server action: {action}")
+    _validate_action_arguments(action, spec, arguments)
     config = _select_config(server_name)
     provider = str(config.type or "").lower()
     if provider not in spec.providers:
         raise ValueError(f"{provider} 不支持 action: {action}")
-    for name in spec.required:
-        _require(arguments, name)
+    _validate_provider_arguments(action, provider, arguments)
     result = _dispatch(_build_client(config), provider, action, arguments)
     if spec.effect != "safe_read" and result is False:
-        raise RuntimeError("媒体服务器 action 返回失败")
+        raise OperationError("媒体服务器 action 返回失败")
     return {
         "success": True,
         "server": config.name,
@@ -432,9 +648,10 @@ def _build_parser() -> argparse.ArgumentParser:
     subparsers.add_parser("instances", help="list configured instances")
     capabilities = subparsers.add_parser("capabilities", help="list allowed actions")
     capabilities.add_argument("--server")
+    capabilities.add_argument("--action")
     call = subparsers.add_parser("call", help="call one allowed action")
     call.add_argument("--server")
-    call.add_argument("--action", required=True, choices=sorted(ACTIONS))
+    call.add_argument("--action", required=True)
     call.add_argument("--arguments", default="{}")
     return parser
 
@@ -447,7 +664,7 @@ def main() -> int:
         if args.command == "instances":
             payload = list_instances()
         elif args.command == "capabilities":
-            payload = list_capabilities(args.server)
+            payload = list_capabilities(args.server, args.action)
         elif args.command == "call":
             payload = call_action(args.server, args.action, _parse_arguments(args.arguments))
         else:
@@ -457,7 +674,7 @@ def main() -> int:
         payload = {
             "success": False,
             "error_type": type(error).__name__,
-            "message": str(error) if isinstance(error, ValueError) else "媒体服务器调用失败",
+            "message": str(error) if isinstance(error, (ValueError, OperationError)) else "媒体服务器调用失败",
         }
         print(json.dumps(payload, ensure_ascii=False, indent=2))
         return 1

tests/test_agent_tool_policy.py

@@ -140,6 +140,10 @@ def test_builtin_policy_inventory_covers_every_fixed_tool() -> None:
             _tool_class_name(MoviePilotApiTool),
         }
     )
+    fixed_tool_names.update(
+        _tool_class_name(tool_class)
+        for tool_class in MoviePilotToolFactory.EXTERNAL_SERVICE_TOOL_CLASSES
+    )
 
     assert DEFAULT_TOOL_POLICY_REGISTRY.builtin_tool_inventory == fixed_tool_names
 

tests/test_builtin_skill_boundaries.py

@@ -33,8 +33,8 @@ def test_modified_builtin_skills_have_incremented_versions() -> None:
         "create-moviepilot-plugin": "5",
         "create-moviepilot-skill": "3",
         "publish-moviepilot-plugin": "3",
-        "downloader-operation": "1",
-        "mediaserver-operation": "1",
+        "downloader-operation": "2",
+        "mediaserver-operation": "2",
     }
 
     for skill_name, expected_version in expected_versions.items():

tests/test_mcp_plugin_tools.py

@@ -56,7 +56,7 @@ def test_mcp_refreshes_tools_after_plugin_lifecycle_change(
         MoviePilotToolFactory,
         "_get_builtin_tool_classes",
         return_value=[],
-    ):
+    ), patch.object(MoviePilotToolFactory, "EXTERNAL_SERVICE_TOOL_CLASSES", ()):
         tool_manager = MoviePilotToolsManager(
             session_id="mcp-plugin-test",
             user_id="api_user",

tests/test_service_operation_mcp_tools.py

@@ -0,0 +1,130 @@
+"""下载器与媒体服务器外部 MCP 工具合同测试。"""
+
+import asyncio
+import json
+from unittest.mock import patch
+
+from app.agent.tools.factory import MoviePilotToolFactory
+from app.agent.tools.impl.service_operation import (
+    DownloaderOperationTool,
+    MediaServerOperationTool,
+)
+from app.agent.tools.manager import MoviePilotToolsManager
+
+
+def _branch(schema: dict, action: str) -> dict:
+    """从 MCP oneOf schema 中读取指定 action 分支。"""
+    return next(
+        branch
+        for branch in schema["oneOf"]
+        if branch["properties"]["action"].get("const") == action
+    )
+
+
+def test_downloader_mcp_schema_exposes_every_action_argument_and_rule() -> None:
+    """外部 MCP Client 应在 tools/list 中直接看到下载器条件参数合同。"""
+    tool = DownloaderOperationTool(session_id="session", user_id="api_user")
+    schema = tool.get_mcp_input_schema()
+    branch = _branch(schema, "tasks.queue.move")
+    arguments = branch["properties"]["arguments"]
+
+    assert len(schema["oneOf"]) == len(schema["properties"]["action"]["enum"])
+    assert arguments["properties"]["task_ids"] == {
+        "type": "array",
+        "items": {"type": "string"},
+        "description": "多个任务的 provider 原生 hash 或 ID；与 task_id 二选一。",
+        "minItems": 1,
+    }
+    assert arguments["properties"]["position"]["enum"] == ["top", "up", "down", "bottom"]
+    assert arguments["required"] == ["position"]
+    assert arguments["oneOf"] == [
+        {"required": ["task_id"], "not": {"required": ["task_ids"]}},
+        {"required": ["task_ids"], "not": {"required": ["task_id"]}},
+    ]
+    assert arguments["additionalProperties"] is False
+
+
+def test_mediaserver_mcp_schema_exposes_nested_refresh_item_fields() -> None:
+    """外部 MCP Client 应直接看到 metadata.refresh 的嵌套字段与枚举。"""
+    tool = MediaServerOperationTool(session_id="session", user_id="api_user")
+    schema = tool.get_mcp_input_schema()
+    branch = _branch(schema, "metadata.refresh")
+    arguments = branch["properties"]["arguments"]
+    item_schema = arguments["properties"]["items"]["items"]
+
+    assert arguments["required"] == ["items"]
+    assert item_schema["properties"]["type"]["enum"] == ["电影", "电视剧", "音乐"]
+    assert item_schema["properties"]["year"]["anyOf"] == [
+        {"type": "string"},
+        {"type": "integer"},
+    ]
+    assert set(item_schema["properties"]) == {
+        "title",
+        "year",
+        "type",
+        "category",
+        "target_path",
+    }
+    assert item_schema["additionalProperties"] is False
+
+
+def test_direct_manager_preserves_service_operation_mcp_schema() -> None:
+    """工具管理器不得把服务操作的 oneOf 和嵌套 schema 压平成普通对象。"""
+    manager = MoviePilotToolsManager(session_id="session", user_id="api_user")
+    manager.tools = [DownloaderOperationTool(session_id="session", user_id="api_user")]
+
+    definition = manager.list_tools()[0]
+
+    assert definition.name == "downloader_operation"
+    assert definition.input_schema["oneOf"]
+    assert _branch(definition.input_schema, "tasks.files")["properties"]["arguments"][
+        "required"
+    ] == ["task_id"]
+
+
+def test_factory_only_adds_service_operation_tools_for_external_manager() -> None:
+    """内置 Agent 保持 Skill 按需加载，外部管理入口才增加两个常驻工具。"""
+    with (
+        patch.object(MoviePilotToolFactory, "BUILTIN_TOOL_CLASSES", ()),
+        patch("app.agent.tools.factory._get_plugin_agent_tools", return_value=[]),
+    ):
+        internal = MoviePilotToolFactory.create_tools(
+            session_id="session",
+            user_id="user",
+        )
+        external = MoviePilotToolFactory.create_tools(
+            session_id="session",
+            user_id="api_user",
+            include_external_service_tools=True,
+        )
+
+    internal_names = {tool.name for tool in internal}
+    external_names = {tool.name for tool in external}
+    assert "downloader_operation" not in internal_names
+    assert "mediaserver_operation" not in internal_names
+    assert {"downloader_operation", "mediaserver_operation"}.issubset(external_names)
+
+
+def test_service_operation_tool_calls_fixed_script_once() -> None:
+    """一次 MCP tools/call 应只执行一次固定脚本并返回其 JSON envelope。"""
+    tool = DownloaderOperationTool(session_id="session", user_id="api_user")
+    with patch(
+        "app.agent.tools.impl.service_operation._run_service_script",
+        return_value={"success": True, "action": "tasks.list", "data": {"items": []}},
+    ) as runner:
+        result = asyncio.run(
+            tool.run(
+                client="main",
+                action="tasks.list",
+                arguments={"limit": 20},
+            )
+        )
+
+    assert json.loads(result)["success"] is True
+    runner.assert_called_once_with(
+        relative_script="skills/downloader-operation/scripts/mp-downloader.py",
+        selector_flag="--client",
+        selector_value="main",
+        action="tasks.list",
+        arguments={"limit": 20},
+    )

tests/test_service_operation_skills.py

@@ -186,6 +186,88 @@ def test_downloader_instances_and_capabilities_do_not_expose_credentials(
     assert any(item["action"] == "tasks.peers" for item in capabilities["actions"])
 
 
+def test_downloader_capability_exposes_complete_action_arguments(
+    downloader_module: ModuleType,
+) -> None:
+    """单 action 能力查询应直接返回类型、必填性、默认值和枚举。"""
+    result = downloader_module.list_capabilities(None, "tasks.queue.move")
+
+    assert len(result["actions"]) == 1
+    action = result["actions"][0]
+    arguments = {item["name"]: item for item in action["arguments"]}
+    assert arguments["task_ids"]["type"] == "string[]"
+    assert arguments["position"] == {
+        "name": "position",
+        "type": "string",
+        "required": True,
+        "description": "目标队列位置。",
+        "enum": ["top", "up", "down", "bottom"],
+    }
+    assert action["argument_rules"] == ["task_id 与 task_ids 必须提供且只能选择一种。"]
+
+
+def test_downloader_argument_validation_reports_all_errors_before_config_load(
+    downloader_module: ModuleType,
+    monkeypatch,
+) -> None:
+    """下载器调用应一次返回全部可检测参数错误，且不连接配置或 provider。"""
+    load_configs = MagicMock()
+    monkeypatch.setattr(downloader_module, "_load_configs", load_configs)
+
+    with pytest.raises(ValueError) as error:
+        downloader_module.call_action(
+            None,
+            "tasks.queue.move",
+            {
+                "position": "sideways",
+                "task_id": "",
+                "task_ids": [],
+                "unexpected": True,
+            },
+        )
+
+    message = str(error.value)
+    assert "未知参数: unexpected" in message
+    assert "参数 position 仅支持: top, up, down, bottom" in message
+    assert "task_id 与 task_ids 必须提供且只能选择一种" in message
+    load_configs.assert_not_called()
+
+
+def test_downloader_ambiguous_instance_error_lists_reusable_names(
+    downloader_module: ModuleType,
+    monkeypatch,
+) -> None:
+    """实例歧义应直接返回可重试名称，避免额外 instances 探测。"""
+    first = _downloader_config()
+    first.name = "main"
+    first.default = False
+    second = _downloader_config("transmission")
+    second.name = "backup"
+    second.default = False
+    monkeypatch.setattr(downloader_module, "_load_configs", lambda: [first, second])
+
+    with pytest.raises(ValueError, match="main、backup"):
+        downloader_module._select_config(None)
+
+
+def test_downloader_call_uses_default_instance_without_discovery(
+    downloader_module: ModuleType,
+    monkeypatch,
+) -> None:
+    """省略 client 时应在同一次调用中选择默认实例并执行。"""
+    config = _downloader_config()
+    client = MagicMock()
+    client.get_torrents.return_value = ([], False)
+    monkeypatch.setattr(downloader_module, "_load_configs", lambda: [config])
+    monkeypatch.setattr(downloader_module, "_build_client", lambda _config: client)
+
+    result = downloader_module.call_action(None, "tasks.list", {"limit": 10})
+
+    assert result["success"] is True
+    assert result["client"] == "main"
+    client.get_torrents.assert_called_once_with(ids=None, status=None, tags=None)
+
+
 def test_downloader_task_list_is_paged_and_normalized(
     downloader_module: ModuleType,
     monkeypatch,
@@ -315,6 +397,86 @@ def test_mediaserver_capabilities_are_provider_specific(
     assert "secret" not in str(result)
 
 
+def test_mediaserver_capability_exposes_nested_refresh_contract(
+    mediaserver_module: ModuleType,
+) -> None:
+    """媒体服务器能力查询应直接描述嵌套刷新条目字段。"""
+    result = mediaserver_module.list_capabilities(None, "metadata.refresh")
+
+    action = result["actions"][0]
+    assert action["required_arguments"] == ["items"]
+    assert action["arguments"] == [
+        {
+            "name": "items",
+            "type": "object[]",
+            "required": True,
+            "description": (
+                "刷新条目；每项支持 title:string、year:string|integer、type:电影|电视剧|音乐、"
+                "category:string、target_path:string。"
+            ),
+        }
+    ]
+
+
+def test_mediaserver_argument_validation_reports_nested_errors_before_config_load(
+    mediaserver_module: ModuleType,
+    monkeypatch,
+) -> None:
+    """媒体服务器调用应一次返回顶层及嵌套参数错误，且不读取实例配置。"""
+    load_configs = MagicMock()
+    monkeypatch.setattr(mediaserver_module, "_load_configs", load_configs)
+
+    with pytest.raises(ValueError) as error:
+        mediaserver_module.call_action(
+            None,
+            "metadata.refresh",
+            {
+                "items": [{"type": "movie", "target_path": 42, "extra": True}],
+                "unexpected": True,
+            },
+        )
+
+    message = str(error.value)
+    assert "未知参数: unexpected" in message
+    assert "items[0] 未知字段: extra" in message
+    assert "items[0].type 仅支持: 电影、电视剧、音乐" in message
+    assert "items[0].target_path 必须是 string" in message
+    load_configs.assert_not_called()
+
+
+def test_mediaserver_ambiguous_instance_error_lists_reusable_names(
+    mediaserver_module: ModuleType,
+    monkeypatch,
+) -> None:
+    """媒体服务器歧义应直接返回可重试名称，避免额外 instances 探测。"""
+    first = _mediaserver_config()
+    first.name = "living-room"
+    second = _mediaserver_config("plex")
+    second.name = "study"
+    monkeypatch.setattr(mediaserver_module, "_load_configs", lambda: [first, second])
+
+    with pytest.raises(ValueError, match="living-room、study"):
+        mediaserver_module._select_config(None)
+
+
+def test_mediaserver_call_uses_only_instance_without_discovery(
+    mediaserver_module: ModuleType,
+    monkeypatch,
+) -> None:
+    """省略 server 时应在同一次调用中选择唯一实例并执行。"""
+    config = _mediaserver_config()
+    client = MagicMock()
+    client.get_medias_count.return_value = {"movie": 12}
+    monkeypatch.setattr(mediaserver_module, "_load_configs", lambda: [config])
+    monkeypatch.setattr(mediaserver_module, "_build_client", lambda _config: client)
+
+    result = mediaserver_module.call_action(None, "server.statistics", {})
+
+    assert result["success"] is True
+    assert result["server"] == "living-room"
+    client.get_medias_count.assert_called_once_with()
+
+
 def test_mediaserver_items_and_scan_use_fixed_public_methods(
     mediaserver_module: ModuleType,
     monkeypatch,
@@ -406,3 +568,36 @@ def test_skill_docs_forbid_arbitrary_network_and_preserve_high_level_workflows()
     assert "download.add" in downloader
     assert "arbitrary SDK methods" in mediaserver
     assert "library.exists" in mediaserver
+    assert "Do not routinely call `instances` or `capabilities`" in downloader
+    assert "Do not routinely call `instances` or `capabilities`" in mediaserver
+    assert "all detectable argument errors" in downloader
+    assert "all detectable argument errors" in mediaserver
+
+
+def test_service_operation_skill_docs_cover_every_action_and_argument(
+    downloader_module: ModuleType,
+    mediaserver_module: ModuleType,
+) -> None:
+    """Agent 加载 Skill 后应能直接看到每个 action 的功能和全部参数名。"""
+    pairs = (
+        (
+            downloader_module.ACTIONS,
+            PROJECT_ROOT / "skills/downloader-operation/SKILL.md",
+        ),
+        (
+            mediaserver_module.ACTIONS,
+            PROJECT_ROOT / "skills/mediaserver-operation/SKILL.md",
+        ),
+    )
+
+    for actions, path in pairs:
+        content = path.read_text(encoding="utf-8")
+        assert "## Complete Action Contract" in content
+        for action_name, spec in actions.items():
+            row = next(
+                (line for line in content.splitlines() if line.startswith(f"| `{action_name}` |")),
+                "",
+            )
+            assert row, f"{path} 缺少 {action_name} 的独立合同表格行"
+            for argument in spec.arguments:
+                assert f"`{argument.name}" in row