"""Versioned plugin contracts shared by the installer, API and task workers.""" from __future__ import annotations import re import sys import platform from typing import Any, Annotated, Literal from urllib.parse import urlsplit from packaging.specifiers import SpecifierSet from packaging.version import Version from pydantic import BaseModel, ConfigDict, Field, SecretStr, field_validator, model_validator Identifier = Annotated[str, Field(pattern=r"^[a-z][a-z0-9_-]{0,47}$")] class PluginModel(BaseModel): model_config = ConfigDict(extra="forbid") class ServerConfig(PluginModel): transport: Literal["stdio", "streamable_http"] = "stdio" command: str = "" args: list[str] = Field(default_factory=list) env: dict[str, str] = Field(default_factory=dict) url: str = "" headers: dict[str, str] = Field(default_factory=dict) @model_validator(mode="after") def validate_transport(self): if self.transport == "stdio" and not self.command.strip(): raise ValueError("stdio 插件必须提供启动命令") if self.transport == "streamable_http": parsed = urlsplit(self.url) if parsed.scheme not in {"http", "https"} or not parsed.hostname: raise ValueError("MCP 地址必须是 HTTP(S) 地址") if parsed.username or parsed.password or parsed.fragment: raise ValueError("MCP 地址不能包含凭据或片段；请使用密钥配置") return self class ConfigField(PluginModel): label: str = Field(min_length=1, max_length=100) description: str = Field(default="", max_length=1000) secret: bool = False required: bool = False default: str = "" ExecutionTarget = Literal["android", "computer", "mixed"] class PluginObservation(PluginModel): tool: str = Field(min_length=1) arguments: dict[str, Any] = Field(default_factory=dict) refresh_tools: list[str] = Field(default_factory=list) class PluginToolContract(PluginModel): """Host-enforced capability description; independent of a tool's prose Skill.""" effect: Literal["read", "write"] observation: bool = False idempotent: bool = False target_argument: str = "" cancellation: Literal["confirmed", "unknown"] = "unknown" evidence: list[Literal["text", "image", "file", "state"]] = Field(default_factory=list) @model_validator(mode="after") def observation_is_read_only(self): if self.observation and self.effect != "read": raise ValueError("自动观察工具必须声明为只读") return self class PluginCapabilities(PluginModel): kind: Literal["visual", "terminal", "api", "file", "other"] = "other" recording: Literal["none", "windows"] = "none" tools: dict[str, PluginToolContract] = Field(default_factory=dict) required_tools: list[str] = Field(default_factory=list) class EvidenceSource(PluginModel): kind: Literal["android", "computer", "terminal", "api", "file", "other"] = "other" target_id: str = Field(min_length=1, max_length=256) label: str = Field(default="", max_length=500) captured_at: float | None = None width: int = Field(default=0, ge=0, le=32768) height: int = Field(default=0, ge=0, le=32768) # Physical desktop coordinates, never normalized model coordinates. window: dict[str, int] | None = None screen: dict[str, int] | None = None class PluginManifest(PluginModel): schema_version: Literal[1, 2] = 1 id: Identifier name: str = Field(min_length=1, max_length=100) version: str description: str = Field(default="", max_length=4000) artemis_version: str = ">=2.0.0,<3.0.0" platforms: list[Literal["win32", "linux", "darwin"]] = Field(default_factory=lambda: ["win32"]) architectures: list[Literal["amd64", "arm64"]] = Field( default_factory=lambda: ["amd64", "arm64"] ) source: str = "" license: str = "" server: ServerConfig | None = None skill: str | None = None config: dict[Identifier, ConfigField] = Field(default_factory=dict) resources: list[str] = Field(default_factory=list) targets: list[ExecutionTarget] = Field(default_factory=lambda: ["mixed"]) observation: PluginObservation | None = None capabilities: PluginCapabilities | None = None @field_validator("version") @classmethod def valid_version(cls, value: str) -> str: Version(value) if not re.fullmatch(r"[0-9][a-zA-Z0-9.+-]{0,63}", value): raise ValueError("插件版本格式无效") return value @field_validator("artemis_version") @classmethod def valid_specifier(cls, value: str) -> str: SpecifierSet(value) return value @model_validator(mode="after") def has_capability(self): if self.server is None and self.skill is None: raise ValueError("插件至少需要 MCP 服务或 Skill") for name, field in self.config.items(): if field.secret and field.default: raise ValueError(f"密钥字段 {name} 不允许默认值") if self.schema_version == 1 and self.capabilities is not None: raise ValueError("能力契约需要 schema_version=2；旧清单保持兼容") if self.schema_version == 2: if self.capabilities is None: raise ValueError("第二版插件必须声明 capabilities") if self.observation: for tool in {self.observation.tool, *self.observation.refresh_tools}: spec = self.capabilities.tools.get(tool) if spec is None or not spec.observation or spec.effect != "read": raise ValueError(f"观察工具 {tool} 缺少只读能力声明") if self.capabilities.recording != "none" and self.capabilities.kind != "visual": raise ValueError("只有视觉插件可以声明屏幕录像") return self def check_compatible(self, artemis_version: str | None = None) -> None: from artemis._version import __version__ artemis_version = artemis_version or __version__ if Version(artemis_version) not in SpecifierSet(self.artemis_version): raise ValueError(f"插件不兼容 ARTEMIS {artemis_version}") if sys.platform not in self.platforms: raise ValueError(f"插件不支持当前平台 {sys.platform}") architecture = {"x86_64": "amd64", "aarch64": "arm64"}.get( platform.machine().lower(), platform.machine().lower() ) if architecture not in self.architectures: raise ValueError(f"插件不支持当前处理器架构 {architecture}") class PluginProfile(PluginModel): name: Identifier = "default" label: str = "默认配置" is_default: bool = False values: dict[str, str] = Field(default_factory=dict) # Only a write payload contains secrets; read responses expose configured names. secrets: dict[str, SecretStr] = Field(default_factory=dict, repr=False) allowed_tools: list[str] | None = None argument_constraints: dict[str, dict[str, list[str]]] = Field(default_factory=dict) resources: list[str] = Field(default_factory=list) authorized: bool = False class PluginSelection(PluginModel): id: Identifier profile: Identifier = "default" version: str | None = None class CatalogEntry(PluginModel): manifest: PluginManifest release_stage: Literal["stable", "preview"] = "stable" archive_url: str sha256: str = Field(pattern=r"^[a-f0-9]{64}$") verification: str = Field(min_length=1, max_length=4000) @field_validator("archive_url") @classmethod def https_archive(cls, value: str) -> str: parsed = urlsplit(value) if parsed.scheme != "https" or not parsed.hostname or parsed.username or parsed.password: raise ValueError("市场安装包必须使用不含凭据的 HTTPS 地址") return value class PluginCatalog(PluginModel): schema_version: Literal[1] = 1 plugins: list[CatalogEntry] = Field(default_factory=list) class PluginCallResult(PluginModel): status: Literal["success", "error", "uncertain"] content: list[dict[str, Any]] = Field(default_factory=list) structured_content: dict[str, Any] | None = None message: str = "" plugin_id: str = "" version: str = "" tool: str = "" image_names: list[str] = Field(default_factory=list) call_id: str = "" source: EvidenceSource | None = None source_id: str = "" error_code: str = "" execution: Literal["not_started", "completed", "partial", "unknown"] = "unknown" side_effect: Literal["none", "possible", "confirmed"] = "possible" retryable: bool = False requires_observation: bool = False attachments: list[dict[str, Any]] = Field(default_factory=list) pre_image_valid: bool = False recording: dict[str, Any] | None = None def text(self) -> str: import json parts = [str(block["text"]) for block in self.content if block.get("type") == "text"] if self.structured_content is not None: parts.append(json.dumps(self.structured_content, ensure_ascii=False)) if self.message: parts.append(self.message) return "\n".join(parts) or "工具已返回；请根据证据判断业务结果。"