diff --git a/backend/package/yuxi/channel/plugins/__init__.py b/backend/package/yuxi/channel/plugins/__init__.py
new file mode 100644
index 00000000..8b137891
--- /dev/null
+++ b/backend/package/yuxi/channel/plugins/__init__.py
@@ -0,0 +1 @@
+
diff --git a/backend/package/yuxi/channel/plugins/catalog.py b/backend/package/yuxi/channel/plugins/catalog.py
new file mode 100644
index 00000000..900a4113
--- /dev/null
+++ b/backend/package/yuxi/channel/plugins/catalog.py
@@ -0,0 +1,87 @@
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from .discovery import ChannelPluginDiscovery
+from .install import install_channel_plugin
+from .manifest import ChannelPluginManifest, load_manifests_from_directory
+from .registry import ChannelPluginRegistry
+
+logger = logging.getLogger(__name__)
+
+
+class ChannelPluginCatalog:
+    """渠道插件目录 — 管理可用插件的注册、安装与查询。"""
+
+    def __init__(self, manifest_dir: str | None = None):
+        self._manifest_dir = manifest_dir
+        self._manifests: dict[str, ChannelPluginManifest] = {}
+        self._installed: set[str] = set()
+
+    def load_manifests(self) -> None:
+        if not self._manifest_dir:
+            return
+        manifests = load_manifests_from_directory(self._manifest_dir)
+        for manifest in manifests:
+            self._manifests[manifest.id] = manifest
+        logger.info("Loaded %d plugin manifests", len(self._manifests))
+
+    def get_manifest(self, plugin_id: str) -> ChannelPluginManifest | None:
+        return self._manifests.get(plugin_id)
+
+    def list_manifests(self) -> list[ChannelPluginManifest]:
+        return list(self._manifests.values())
+
+    def list_available(self) -> list[str]:
+        return list(self._manifests.keys())
+
+    def list_installed(self) -> list[str]:
+        return list(self._installed)
+
+    def is_installed(self, plugin_id: str) -> bool:
+        return plugin_id in self._installed
+
+    def install(self, plugin_id: str) -> bool:
+        manifest = self._manifests.get(plugin_id)
+        if not manifest:
+            logger.warning("Cannot install plugin '%s': manifest not found", plugin_id)
+            return False
+        try:
+            install_channel_plugin(manifest)
+            self._installed.add(plugin_id)
+            logger.info("Installed channel plugin: %s", plugin_id)
+            return True
+        except Exception:
+            logger.exception("Failed to install plugin '%s'", plugin_id)
+            return False
+
+    def uninstall(self, plugin_id: str) -> bool:
+        if plugin_id not in self._installed:
+            return False
+        self._installed.discard(plugin_id)
+
+        manifest = self._manifests.get(plugin_id)
+        target_dir = manifest.target_path if manifest else None
+        dependencies = manifest.dependencies if manifest else None
+
+        from .install import uninstall_channel_plugin
+
+        uninstall_channel_plugin(
+            plugin_id=plugin_id,
+            target_dir=target_dir,
+            dependencies=dependencies,
+        )
+        return True
+
+    def discover_builtin(self, package_name: str = "yuxi.channels") -> list[Any]:
+        discovered = ChannelPluginDiscovery.discover_builtin(package_name)
+        for plugin in discovered:
+            self._installed.add(getattr(plugin, "id"))
+        return discovered
+
+    def discover_from_path(self, plugin_path: str) -> list[Any]:
+        discovered = ChannelPluginDiscovery.discover_from_path(plugin_path)
+        for plugin in discovered:
+            self._installed.add(getattr(plugin, "id"))
+        return discovered
diff --git a/backend/package/yuxi/channel/plugins/dependency.py b/backend/package/yuxi/channel/plugins/dependency.py
new file mode 100644
index 00000000..1152ec72
--- /dev/null
+++ b/backend/package/yuxi/channel/plugins/dependency.py
@@ -0,0 +1,131 @@
+from __future__ import annotations
+
+import logging
+from collections import defaultdict, deque
+from typing import Any
+
+from .manifest import ChannelPluginManifest
+
+logger = logging.getLogger(__name__)
+
+
+class DependencyGraph:
+    """渠道插件依赖图 — 检测循环依赖并计算拓扑排序安装顺序。"""
+
+    def __init__(self):
+        self._nodes: dict[str, ChannelPluginManifest] = {}
+        self._edges: dict[str, set[str]] = defaultdict(set)
+        self._reverse_edges: dict[str, set[str]] = defaultdict(set)
+
+    def add(self, manifest: ChannelPluginManifest) -> None:
+        self._nodes[manifest.id] = manifest
+        for dep in manifest.dependencies:
+            dep_id = dep if isinstance(dep, str) else dep.get("id", "")
+            if dep_id:
+                self._edges[manifest.id].add(dep_id)
+                self._reverse_edges[dep_id].add(manifest.id)
+
+    def remove(self, plugin_id: str) -> None:
+        if plugin_id not in self._nodes:
+            return
+        del self._nodes[plugin_id]
+        for dep_id in list(self._edges.get(plugin_id, [])):
+            self._reverse_edges[dep_id].discard(plugin_id)
+        self._edges.pop(plugin_id, None)
+        for other_id, deps in list(self._edges.items()):
+            if plugin_id in deps:
+                deps.discard(plugin_id)
+                self._reverse_edges[plugin_id].discard(other_id)
+
+    def get_dependencies(self, plugin_id: str) -> set[str]:
+        return set(self._edges.get(plugin_id, []))
+
+    def get_dependents(self, plugin_id: str) -> set[str]:
+        return set(self._reverse_edges.get(plugin_id, []))
+
+    def has_cycle(self) -> tuple[bool, list[str]]:
+        """检测是否存在循环依赖，返回 (是否有环, 环中的节点列表)。"""
+        visited: set[str] = set()
+        rec_stack: set[str] = set()
+        cycle: list[str] = []
+
+        def dfs(node: str, path: list[str]) -> bool:
+            visited.add(node)
+            rec_stack.add(node)
+            path.append(node)
+            for neighbor in self._edges.get(node, set()):
+                if neighbor not in visited:
+                    if dfs(neighbor, path):
+                        return True
+                elif neighbor in rec_stack:
+                    cycle_start = path.index(neighbor)
+                    cycle.extend(path[cycle_start:])
+                    return True
+            path.pop()
+            rec_stack.remove(node)
+            return False
+
+        for node in self._nodes:
+            if node not in visited:
+                if dfs(node, []):
+                    return True, cycle
+        return False, []
+
+    def topological_sort(self) -> list[str]:
+        """返回拓扑排序后的插件 ID 列表。如果存在循环依赖，抛出 ValueError。"""
+        has_cycle, cycle = self.has_cycle()
+        if has_cycle:
+            raise ValueError(f"Circular dependency detected: {' -> '.join(cycle)}")
+
+        in_degree = {node: 0 for node in self._nodes}
+        for node, deps in self._edges.items():
+            for dep in deps:
+                if dep in in_degree:
+                    in_degree[node] += 1
+
+        queue = deque(node for node, degree in in_degree.items() if degree == 0)
+        result = []
+
+        while queue:
+            node = queue.popleft()
+            result.append(node)
+            for dependent in self._reverse_edges.get(node, set()):
+                if dependent in in_degree:
+                    in_degree[dependent] -= 1
+                    if in_degree[dependent] == 0:
+                        queue.append(dependent)
+
+        if len(result) != len(self._nodes):
+            raise ValueError("Dependency graph has unresolved dependencies")
+
+        return result
+
+    def install_order(self) -> list[str]:
+        """获取安装顺序（依赖优先）。"""
+        return self.topological_sort()
+
+    def uninstall_order(self) -> list[str]:
+        """获取卸载顺序（反向拓扑排序，被依赖者优先）。"""
+        return list(reversed(self.topological_sort()))
+
+    def get_subgraph(self, plugin_id: str) -> "DependencyGraph":
+        """获取包含指定插件及其所有依赖的子图。"""
+        subgraph = DependencyGraph()
+        visited: set[str] = set()
+
+        def visit(node: str) -> None:
+            if node in visited or node not in self._nodes:
+                return
+            visited.add(node)
+            subgraph.add(self._nodes[node])
+            for dep in self._edges.get(node, set()):
+                visit(dep)
+
+        visit(plugin_id)
+        return subgraph
+
+    def to_dict(self) -> dict[str, Any]:
+        return {
+            "nodes": list(self._nodes.keys()),
+            "edges": {k: list(v) for k, v in self._edges.items()},
+        }
diff --git a/backend/package/yuxi/channel/plugins/dependency_validator.py b/backend/package/yuxi/channel/plugins/dependency_validator.py
new file mode 100644
index 00000000..7c21b768
--- /dev/null
+++ b/backend/package/yuxi/channel/plugins/dependency_validator.py
@@ -0,0 +1,287 @@
+from __future__ import annotations
+
+import logging
+import os
+import re
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass(frozen=True)
+class AllowedPackage:
+    """白名单中的允许安装包配置。"""
+
+    name: str
+    min_version: str | None = None
+    max_version: str | None = None
+    allowed_versions: list[str] | None = None
+    blocked_versions: list[str] | None = None
+
+
+_BUILTIN_ALLOWED = {
+    "aiofiles": AllowedPackage(name="aiofiles", min_version="0.8.0"),
+    "aiohttp": AllowedPackage(name="aiohttp", min_version="3.8.0"),
+    "aiosignal": AllowedPackage(name="aiosignal", min_version="1.2.0"),
+    "asyncpg": AllowedPackage(name="asyncpg", min_version="0.28.0"),
+    "cachetools": AllowedPackage(name="cachetools", min_version="5.0.0"),
+    "cryptography": AllowedPackage(name="cryptography", min_version="41.0.0"),
+    "fastapi": AllowedPackage(name="fastapi", min_version="0.100.0"),
+    "frozenlist": AllowedPackage(name="frozenlist", min_version="1.3.0"),
+    "httpx": AllowedPackage(name="httpx", min_version="0.24.0"),
+    "lxml": AllowedPackage(name="lxml", min_version="4.9.0"),
+    "multidict": AllowedPackage(name="multidict", min_version="6.0.0"),
+    "numpy": AllowedPackage(name="numpy", min_version="1.24.0"),
+    "orjson": AllowedPackage(name="orjson", min_version="3.9.0"),
+    "pillow": AllowedPackage(name="pillow", min_version="10.0.0"),
+    "psycopg2-binary": AllowedPackage(name="psycopg2-binary", min_version="2.9.0"),
+    "pydantic": AllowedPackage(name="pydantic", min_version="2.0.0"),
+    "pydantic-core": AllowedPackage(name="pydantic-core", min_version="2.0.0"),
+    "pydantic-settings": AllowedPackage(name="pydantic-settings", min_version="2.0.0"),
+    "python-dotenv": AllowedPackage(name="python-dotenv", min_version="1.0.0"),
+    "pyyaml": AllowedPackage(name="pyyaml", min_version="6.0"),
+    "redis": AllowedPackage(name="redis", min_version="4.5.0"),
+    "requests": AllowedPackage(name="requests", min_version="2.28.0"),
+    "sqlalchemy": AllowedPackage(name="sqlalchemy", min_version="2.0.0"),
+    "starlette": AllowedPackage(name="starlette", min_version="0.27.0"),
+    "structlog": AllowedPackage(name="structlog", min_version="23.0.0"),
+    "tenacity": AllowedPackage(name="tenacity", min_version="8.0.0"),
+    "urllib3": AllowedPackage(name="urllib3", min_version="1.26.0"),
+    "uvicorn": AllowedPackage(name="uvicorn", min_version="0.22.0"),
+    "websockets": AllowedPackage(name="websockets", min_version="10.0"),
+    "yarl": AllowedPackage(name="yarl", min_version="1.8.0"),
+}
+
+
+@dataclass
+class ValidationResult:
+    """依赖校验结果。"""
+
+    is_valid: bool
+    package: str
+    requested_spec: str
+    errors: list[str] = field(default_factory=list)
+
+
+class DependencyValidator:
+    """渠道插件依赖校验器 —— 防止恶意包安装。
+
+    功能：
+    1. 白名单校验：只允许安装预审批的包
+    2. 版本约束校验：确保版本符合安全策略
+    3. 包名规范化：防止依赖混淆攻击
+    """
+
+    _PEP508_NAME_RE = re.compile(r"^[A-Za-z0-9][-A-Za-z0-9._]*$")
+    _VERSION_SPEC_RE = re.compile(
+        r"^[A-Za-z0-9][-A-Za-z0-9._]*"  # 包名
+        r"(\s*\[\s*[A-Za-z0-9_,\s-]+\s*\])?"  # extras
+        r"(\s*(==|!=|<=|>=|<|>|~=|===)\s*[^\s;]+)*"  # 版本约束
+        r"(\s*;.*)?$"  # 环境标记
+    )
+
+    def __init__(self, allowed_packages: dict[str, AllowedPackage] | None = None):
+        self._allowed: dict[str, AllowedPackage] = {}
+        if allowed_packages:
+            for name, cfg in allowed_packages.items():
+                self._allowed[self._normalize_name(name)] = cfg
+
+    @classmethod
+    def from_config_file(cls, config_path: str | Path) -> "DependencyValidator":
+        """从 JSON 配置文件加载白名单。"""
+        import json
+
+        path = Path(config_path)
+        if not path.exists():
+            logger.warning("Dependency whitelist config not found: %s", path)
+            return cls()
+
+        try:
+            data = json.loads(path.read_text(encoding="utf-8"))
+            allowed = {}
+            for item in data.get("allowed_packages", []):
+                name = item["name"]
+                allowed[name] = AllowedPackage(
+                    name=name,
+                    min_version=item.get("min_version"),
+                    max_version=item.get("max_version"),
+                    allowed_versions=item.get("allowed_versions"),
+                    blocked_versions=item.get("blocked_versions"),
+                )
+            return cls(allowed)
+        except Exception:
+            logger.exception("Failed to load dependency whitelist from %s", path)
+            return cls()
+
+    @classmethod
+    def default_validator(cls) -> "DependencyValidator":
+        """获取默认校验器。
+
+        加载优先级：
+        1. 环境变量 YUXI_DEPENDENCY_WHITELIST 指定的文件
+        2. 统一配置 yuxi.config.config.dependency_whitelist
+        3. 内置硬编码白名单（降级）
+        """
+        whitelist_file = os.getenv("YUXI_DEPENDENCY_WHITELIST")
+        if whitelist_file:
+            path = Path(whitelist_file)
+            if path.exists():
+                logger.info("Loading dependency whitelist from env: %s", path)
+                return cls.from_config_file(path)
+
+        try:
+            from yuxi.config import config
+
+            allowed: dict[str, AllowedPackage] = {}
+            for entry in config.dependency_whitelist:
+                allowed[entry.name] = AllowedPackage(
+                    name=entry.name,
+                    min_version=entry.min_version,
+                    max_version=entry.max_version,
+                    allowed_versions=entry.allowed_versions,
+                    blocked_versions=entry.blocked_versions,
+                )
+            logger.info("Loaded dependency whitelist from unified config (%d packages)", len(allowed))
+            return cls(allowed)
+        except Exception:
+            logger.exception("Failed to load dependency whitelist from unified config, using builtin defaults")
+
+        logger.info("Using builtin dependency whitelist defaults (%d packages)", len(_BUILTIN_ALLOWED))
+        return cls(dict(_BUILTIN_ALLOWED))
+
+    def validate(self, dependency_spec: str) -> ValidationResult:
+        """校验单个依赖规格是否允许安装。"""
+        dependency_spec = dependency_spec.strip()
+        if not dependency_spec:
+            return ValidationResult(
+                is_valid=False,
+                package="",
+                requested_spec=dependency_spec,
+                errors=["Empty dependency specification"],
+            )
+
+        if not self._VERSION_SPEC_RE.match(dependency_spec):
+            return ValidationResult(
+                is_valid=False,
+                package="",
+                requested_spec=dependency_spec,
+                errors=["Invalid dependency specification format"],
+            )
+
+        package_name = self._extract_package_name(dependency_spec)
+        normalized = self._normalize_name(package_name)
+
+        if normalized not in self._allowed:
+            return ValidationResult(
+                is_valid=False,
+                package=package_name,
+                requested_spec=dependency_spec,
+                errors=[f"Package '{package_name}' is not in the allowed whitelist"],
+            )
+
+        allowed = self._allowed[normalized]
+        version_constraints = self._extract_version_constraints(dependency_spec)
+
+        errors = []
+        for op, ver in version_constraints:
+            if not self._is_version_allowed(allowed, op, ver):
+                errors.append(
+                    f"Version constraint '{op}{ver}' violates policy for '{package_name}'"
+                )
+
+        if allowed.blocked_versions:
+            for _, ver in version_constraints:
+                if ver in allowed.blocked_versions:
+                    errors.append(
+                        f"Version '{ver}' is explicitly blocked for '{package_name}'"
+                    )
+
+        return ValidationResult(
+            is_valid=len(errors) == 0,
+            package=package_name,
+            requested_spec=dependency_spec,
+            errors=errors,
+        )
+
+    def validate_all(self, dependencies: list[str]) -> list[ValidationResult]:
+        """批量校验依赖列表。"""
+        return [self.validate(dep) for dep in dependencies]
+
+    def is_allowed(self, dependency_spec: str) -> bool:
+        """快速判断单个依赖是否允许。"""
+        return self.validate(dependency_spec).is_valid
+
+    @staticmethod
+    def _normalize_name(name: str) -> str:
+        """规范化包名（PEP 503）：转小写，替换 _ 和 . 为 -。"""
+        return name.lower().replace("_", "-").replace(".", "-")
+
+    @staticmethod
+    def _extract_package_name(spec: str) -> str:
+        """从依赖规格中提取包名。"""
+        spec = spec.strip()
+        match = re.match(r"^([A-Za-z0-9][-A-Za-z0-9._]*)", spec)
+        if match:
+            return match.group(1)
+        return spec.split("[")[0].split(";")[0].strip()
+
+    @staticmethod
+    def _extract_version_constraints(spec: str) -> list[tuple[str, str]]:
+        """从依赖规格中提取版本约束列表 [(operator, version), ...]。"""
+        constraints = []
+        pattern = re.compile(r"(==|!=|<=|>=|<|>|~=|===)\s*([^\s;,]+)")
+        for match in pattern.finditer(spec):
+            constraints.append((match.group(1), match.group(2)))
+        return constraints
+
+    @staticmethod
+    def _is_version_allowed(allowed: AllowedPackage, op: str, version: str) -> bool:
+        """检查版本约束是否符合白名单策略。"""
+        if allowed.allowed_versions and version not in allowed.allowed_versions:
+            return False
+
+        if allowed.min_version and op in (">=", ">", "=="):
+            if not DependencyValidator._version_gte(version, allowed.min_version):
+                return False
+
+        if allowed.max_version and op in ("<=", "<", "=="):
+            if not DependencyValidator._version_lte(version, allowed.max_version):
+                return False
+
+        return True
+
+    @staticmethod
+    def _version_gte(v1: str, v2: str) -> bool:
+        """简化版本比较：v1 >= v2。"""
+        try:
+            return DependencyValidator._parse_version(v1) >= DependencyValidator._parse_version(v2)
+        except ValueError:
+            return True
+
+    @staticmethod
+    def _version_lte(v1: str, v2: str) -> bool:
+        """简化版本比较：v1 <= v2。"""
+        try:
+            return DependencyValidator._parse_version(v1) <= DependencyValidator._parse_version(v2)
+        except ValueError:
+            return True
+
+    @staticmethod
+    def _parse_version(version: str) -> tuple[int, ...]:
+        """解析版本号为可比较的元组。"""
+        version = version.strip()
+        parts = re.split(r"[.-]", version)
+        result = []
+        for part in parts:
+            if part.isdigit():
+                result.append(int(part))
+            else:
+                match = re.match(r"(\d+)", part)
+                if match:
+                    result.append(int(match.group(1)))
+                break
+        if not result:
+            raise ValueError(f"Invalid version: {version}")
+        return tuple(result)
diff --git a/backend/package/yuxi/channel/plugins/discovery.py b/backend/package/yuxi/channel/plugins/discovery.py
new file mode 100644
index 00000000..059fd2cb
--- /dev/null
+++ b/backend/package/yuxi/channel/plugins/discovery.py
@@ -0,0 +1,291 @@
+from __future__ import annotations
+
+import importlib
+import logging
+import os
+import pkgutil
+import threading
+from pathlib import Path
+from typing import Any
+
+from yuxi.channel.protocols import MetaProtocol
+
+from .manifest import ChannelPluginManifest
+from .registry import ChannelPluginRegistry
+from .sandbox import PluginSandbox, SandboxViolationError
+
+logger = logging.getLogger(__name__)
+
+_CacheEntry = tuple[float, float, list[Path]]
+
+
+class ChannelPluginDiscovery:
+    """自动发现项目中的渠道插件。"""
+
+    _path_allowlist: list[str] | None = None
+    _path_allowlist_loaded: bool = False
+    _discovery_cache: dict[str, _CacheEntry] = {}
+    _discovery_cache_lock: threading.Lock | None = None
+
+    @classmethod
+    def _ensure_path_allowlist_loaded(cls) -> None:
+        if cls._path_allowlist_loaded:
+            return
+        cls._path_allowlist_loaded = True
+        cls._load_path_allowlist_from_env()
+
+    @classmethod
+    def _load_path_allowlist_from_env(cls) -> None:
+        allowed = []
+
+        env_val = os.getenv("YUXI_PLUGIN_ALLOWLIST")
+        if env_val:
+            for entry in env_val.split(os.pathsep):
+                entry = entry.strip()
+                if entry:
+                    allowed.append(str(Path(entry).resolve()))
+
+        extensions_dir = Path(__file__).resolve().parent.parent / "extensions"
+        if extensions_dir.exists():
+            allowed.append(str(extensions_dir))
+
+        cls._path_allowlist = allowed
+
+    @classmethod
+    def set_path_allowlist(cls, paths: list[str]) -> None:
+        cls._path_allowlist = [str(Path(p).resolve()) for p in paths]
+        cls._path_allowlist_loaded = True
+
+    @classmethod
+    def get_path_allowlist(cls) -> list[str]:
+        cls._ensure_path_allowlist_loaded()
+        return list(cls._path_allowlist) if cls._path_allowlist else []
+
+    @classmethod
+    def clear_path_allowlist(cls) -> None:
+        cls._path_allowlist = None
+        cls._path_allowlist_loaded = False
+
+    @classmethod
+    def _ensure_cache_lock(cls) -> threading.Lock:
+        if cls._discovery_cache_lock is None:
+            cls._discovery_cache_lock = threading.Lock()
+        return cls._discovery_cache_lock
+
+    @classmethod
+    def clear_discovery_cache(cls) -> None:
+        with cls._ensure_cache_lock():
+            cls._discovery_cache.clear()
+
+    @classmethod
+    def _is_path_allowed(cls, path: Path) -> bool:
+        cls._ensure_path_allowlist_loaded()
+        if not cls._path_allowlist:
+            return False
+        resolved = str(path.resolve())
+        return any(resolved == allowed or resolved.startswith(allowed + os.sep) for allowed in cls._path_allowlist)
+
+    @classmethod
+    def validate_plugin_path(cls, path: str | Path, purpose: str = "load") -> Path:
+        """验证插件路径安全性并返回解析后的路径。
+
+        校验规则：
+        1. 路径不能包含 ``..`` 组件（防御深度，防止目录遍历）
+        2. 路径必须在白名单目录内
+        3. 白名单为空时拒绝所有路径（安全默认）
+
+        Args:
+            path: 要验证的路径
+            purpose: 操作目的描述（用于日志和错误消息）
+
+        Returns:
+            解析后的绝对路径
+
+        Raises:
+            PermissionError: 路径校验失败，存在安全风险
+        """
+        cls._ensure_path_allowlist_loaded()
+
+        raw_path = Path(path)
+        if ".." in raw_path.parts:
+            logger.warning(
+                "Path traversal attempt blocked for %s: '%s' contains '..' component",
+                purpose, path,
+            )
+            raise PermissionError(
+                f"Path traversal not allowed for plugin {purpose}: '{path}'"
+            )
+
+        resolved = raw_path.resolve()
+
+        if not cls._path_allowlist:
+            logger.warning(
+                "Plugin path rejected for %s (allowlist is empty): %s",
+                purpose, resolved,
+            )
+            raise PermissionError(
+                f"Plugin {purpose} rejected: no allowed directories configured"
+            )
+
+        if not cls._is_path_allowed(resolved):
+            logger.warning(
+                "Plugin path not in allowlist for %s: '%s' (resolved: %s)",
+                purpose, path, resolved,
+            )
+            raise PermissionError(
+                f"Plugin path not in allowlist for {purpose}: '{resolved}'"
+            )
+
+        return resolved
+
+    @classmethod
+    def discover_builtin(cls, package_name: str = "yuxi.channels") -> list[Any]:
+        """扫描指定 Python 包下的所有模块，注册实现了 ConfigProtocol 的类。
+
+        导入前对每个模块源文件执行 AST 静态安全分析，
+        拦截包含 ``__import__()`` 调用、``from ... import *`` 等危险语法结构的模块。
+        """
+        discovered = []
+        try:
+            package = importlib.import_module(package_name)
+        except ImportError:
+            logger.warning("Cannot import package '%s' for plugin discovery", package_name)
+            return discovered
+
+        package_path = getattr(package, "__path__", None)
+        if not package_path:
+            logger.warning("Package '%s' has no __path__", package_name)
+            return discovered
+
+        for _, module_name, is_pkg in pkgutil.iter_modules(package_path):
+            if is_pkg:
+                continue
+            full_name = f"{package_name}.{module_name}"
+
+            module_file = Path(package_path[0]) / f"{module_name}.py"
+            if module_file.exists():
+                try:
+                    source = module_file.read_text(encoding="utf-8")
+                    PluginSandbox.validate_source(source, str(module_file))
+                except SandboxViolationError:
+                    logger.warning(
+                        "Builtin plugin '%s' failed AST security validation: %s",
+                        full_name, module_file,
+                    )
+                    continue
+
+            try:
+                module = importlib.import_module(full_name)
+            except Exception:
+                logger.exception("Failed to import module %s", full_name)
+                continue
+
+            for attr_name in dir(module):
+                obj = getattr(module, attr_name)
+                if not isinstance(obj, type):
+                    continue
+                if getattr(obj, "__module__", None) != full_name:
+                    continue
+                if not isinstance(obj, MetaProtocol):
+                    continue
+                plugin_id = getattr(obj, "id", None)
+                if not plugin_id or not isinstance(plugin_id, str) or not plugin_id.strip():
+                    continue
+                try:
+                    ChannelPluginRegistry.register(obj)
+                    discovered.append(obj)
+                except ValueError:
+                    continue
+
+        return discovered
+
+    @classmethod
+    def discover_from_path(cls, plugin_path: str | Path) -> list[Any]:
+        """从文件系统路径加载插件模块（带路径白名单校验、扫描缓存和沙箱执行）。"""
+        discovered = []
+        plugin_path = Path(plugin_path)
+        if not plugin_path.exists():
+            logger.warning("Plugin path does not exist: %s", plugin_path)
+            return discovered
+
+        if not cls._is_path_allowed(plugin_path):
+            logger.warning("Plugin path not in allowlist: %s", plugin_path)
+            return discovered
+
+        if plugin_path.is_file() and plugin_path.suffix == ".py":
+            module_files = [plugin_path]
+        elif plugin_path.is_dir():
+            module_files = cls._scan_with_cache(plugin_path)
+        else:
+            return discovered
+
+        sandbox = PluginSandbox()
+
+        for file_path in module_files:
+            try:
+                module = sandbox.load_module_from_file(file_path, file_path.stem)
+            except Exception:
+                logger.exception("Failed to load plugin from %s", file_path)
+                continue
+
+            for attr_name in dir(module):
+                obj = getattr(module, attr_name)
+                if not isinstance(obj, type):
+                    continue
+                if getattr(obj, "__module__", None) != module.__name__:
+                    continue
+                if not isinstance(obj, MetaProtocol):
+                    continue
+                plugin_id = getattr(obj, "id", None)
+                if not plugin_id or not isinstance(plugin_id, str) or not plugin_id.strip():
+                    continue
+                try:
+                    ChannelPluginRegistry.register(obj)
+                    discovered.append(obj)
+                except ValueError:
+                    continue
+
+        return discovered
+
+    @classmethod
+    def _scan_with_cache(cls, plugin_path: Path) -> list[Path]:
+        resolved = str(plugin_path.resolve())
+        with cls._ensure_cache_lock():
+            cached = cls._discovery_cache.get(resolved)
+            if cached is not None:
+                _max_mtime, dir_mtime, files = cached
+                try:
+                    if plugin_path.stat().st_mtime <= dir_mtime and cls._cache_still_valid(files, _max_mtime):
+                        return list(files)
+                except OSError:
+                    pass
+
+        module_files = list(plugin_path.rglob("*.py"))
+        max_mtime = max(
+            (f.stat().st_mtime for f in module_files),
+            default=0.0,
+        )
+        try:
+            dir_mtime = plugin_path.stat().st_mtime
+        except OSError:
+            dir_mtime = 0.0
+        with cls._ensure_cache_lock():
+            cls._discovery_cache[resolved] = (max_mtime, dir_mtime, module_files)
+        return module_files
+
+    @staticmethod
+    def _cache_still_valid(files: list[Path], cached_max_mtime: float) -> bool:
+        for f in files:
+            try:
+                if f.stat().st_mtime > cached_max_mtime:
+                    return False
+            except OSError:
+                return False
+        return True
+
+    @classmethod
+    def discover_manifests(cls, manifest_dir: str | Path) -> list[ChannelPluginManifest]:
+        """从清单目录加载插件元数据。"""
+        from .manifest import load_manifests_from_directory
+
+        return load_manifests_from_directory(manifest_dir)
diff --git a/backend/package/yuxi/channel/plugins/install.py b/backend/package/yuxi/channel/plugins/install.py
new file mode 100644
index 00000000..423f3acc
--- /dev/null
+++ b/backend/package/yuxi/channel/plugins/install.py
@@ -0,0 +1,169 @@
+from __future__ import annotations
+
+import logging
+import shutil
+import subprocess
+import sys
+from pathlib import Path
+from typing import Any
+
+from .dependency_validator import DependencyValidator
+from .discovery import ChannelPluginDiscovery
+from .manifest import ChannelPluginManifest
+
+logger = logging.getLogger(__name__)
+
+
+class SecurityError(Exception):
+    """安全校验错误 —— 阻断不安全的插件安装。"""
+
+
+def install_channel_plugin(
+    manifest: ChannelPluginManifest,
+    validator: DependencyValidator | None = None,
+) -> None:
+    """安装一个渠道插件：
+    1. 安装依赖 (pip install) —— 带白名单校验
+    2. 复制/链接插件代码到目标目录
+    3. 发现并注册插件类
+
+    Args:
+        manifest: 插件清单
+        validator: 依赖校验器，为 None 时使用默认校验器
+    """
+    logger.info("Installing channel plugin: %s", manifest.id)
+
+    if manifest.dependencies:
+        logger.info("Installing dependencies for %s: %s", manifest.id, manifest.dependencies)
+        _install_dependencies_with_validation(manifest.dependencies, validator)
+
+    if manifest.source_path:
+        try:
+            source = ChannelPluginDiscovery.validate_plugin_path(manifest.source_path, purpose="source copy")
+        except PermissionError as e:
+            raise SecurityError(str(e)) from e
+        target = Path(manifest.target_path or f"yuxi/channels/{manifest.id}")
+        if source.exists():
+            if target.exists():
+                shutil.rmtree(target)
+            if source.is_dir():
+                shutil.copytree(source, target)
+            else:
+                target.parent.mkdir(parents=True, exist_ok=True)
+                shutil.copy2(source, target)
+            logger.info("Copied plugin source from %s to %s", source, target)
+
+    if manifest.entry_point:
+        try:
+            entry_path = ChannelPluginDiscovery.validate_plugin_path(manifest.entry_point, purpose="entry point")
+        except PermissionError as e:
+            raise SecurityError(str(e)) from e
+        ChannelPluginDiscovery.discover_from_path(entry_path)
+
+    logger.info("Channel plugin '%s' installed successfully", manifest.id)
+
+
+def _install_dependencies_with_validation(
+    dependencies: list[str],
+    validator: DependencyValidator | None = None,
+) -> None:
+    """带白名单校验的依赖安装。
+
+    校验流程：
+    1. 使用 DependencyValidator 校验每个依赖
+    2. 收集所有校验错误
+    3. 如有错误，抛出 SecurityError 阻断安装
+    4. 校验通过后执行 pip install
+    """
+    validator = validator or DependencyValidator.default_validator()
+    results = validator.validate_all(dependencies)
+
+    errors = [err for r in results for err in r.errors]
+    if errors:
+        for err in errors:
+            logger.error("Dependency validation failed: %s", err)
+        raise SecurityError(f"Dependency validation failed: {'; '.join(errors)}")
+
+    allowed_deps = [r.requested_spec for r in results if r.is_valid]
+    if not allowed_deps:
+        logger.info("No valid dependencies to install after validation")
+        return
+
+    logger.info("Installing validated dependencies: %s", allowed_deps)
+    try:
+        pip_cmd = [
+            sys.executable, "-m", "pip", "install",
+            "--no-build-isolation",
+            *allowed_deps,
+        ]
+        subprocess.check_output(
+            pip_cmd,
+            stderr=subprocess.STDOUT,
+            text=True,
+        )
+    except subprocess.CalledProcessError as e:
+        logger.error("pip install failed: %s", e.output)
+        raise SecurityError(f"Failed to install dependencies: {e.output}") from e
+
+
+def uninstall_channel_plugin(
+    plugin_id: str,
+    target_dir: str | Path | None = None,
+    dependencies: list[str] | None = None,
+) -> bool:
+    """卸载一个渠道插件：
+
+    1. 从 Registry 中注销
+    2. 清理文件系统中的插件文件
+    3. 卸载 pip 安装的依赖（尽力而为，不阻断卸载流程）
+
+    Args:
+        plugin_id: 插件 ID
+        target_dir: 插件文件所在目录（从 manifest.target_path 提取）
+        dependencies: 插件依赖列表（从 manifest.dependencies 提取）
+    """
+    from .registry import ChannelPluginRegistry
+
+    ChannelPluginRegistry.unregister(plugin_id)
+
+    if target_dir:
+        target = Path(target_dir)
+        if target.exists():
+            shutil.rmtree(target)
+            logger.info("Removed plugin directory: %s", target)
+
+    if dependencies:
+        _uninstall_dependencies(plugin_id, dependencies)
+
+    logger.info("Channel plugin '%s' fully uninstalled", plugin_id)
+    return True
+
+
+def _uninstall_dependencies(plugin_id: str, dependencies: list[str]) -> None:
+    """卸载插件依赖（尽力而为：失败仅记录告警，不抛异常）。"""
+    if not dependencies:
+        return
+
+    package_names = []
+    for dep in dependencies:
+        name = dep.strip().split("=")[0].split(">")[0].split("<")[0].split("~")[0].split("!")[0].strip()
+        name = name.split("[")[0].split(";")[0].strip()
+        if name:
+            package_names.append(name)
+
+    if not package_names:
+        return
+
+    logger.info("Uninstalling dependencies for '%s': %s", plugin_id, package_names)
+    try:
+        subprocess.check_output(
+            [sys.executable, "-m", "pip", "uninstall", "-y", *package_names],
+            stderr=subprocess.STDOUT,
+            text=True,
+        )
+        logger.info("Successfully uninstalled dependencies for '%s'", plugin_id)
+    except subprocess.CalledProcessError as e:
+        logger.warning(
+            "Failed to uninstall dependencies for '%s': %s (plugin already unregistered)",
+            plugin_id, e.output.strip() if e.output else str(e),
+        )
\ No newline at end of file
diff --git a/backend/package/yuxi/channel/plugins/manifest.py b/backend/package/yuxi/channel/plugins/manifest.py
new file mode 100644
index 00000000..22b2625d
--- /dev/null
+++ b/backend/package/yuxi/channel/plugins/manifest.py
@@ -0,0 +1,89 @@
+from __future__ import annotations
+
+import json
+import logging
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class ChannelPluginManifest:
+    id: str
+    name: str
+    version: str
+    description: str = ""
+    author: str = ""
+    dependencies: list[str] = field(default_factory=list)
+    entry_point: str | None = None
+    source_path: str | None = None
+    target_path: str | None = None
+    config_schema: dict[str, Any] = field(default_factory=dict)
+    capabilities: list[str] = field(default_factory=list)
+    tags: list[str] = field(default_factory=list)
+
+    @classmethod
+    def from_dict(cls, data: dict[str, Any]) -> "ChannelPluginManifest":
+        return cls(
+            id=data["id"],
+            name=data.get("name", data["id"]),
+            version=data.get("version", "0.0.1"),
+            description=data.get("description", ""),
+            author=data.get("author", ""),
+            dependencies=data.get("dependencies", []),
+            entry_point=data.get("entry_point"),
+            source_path=data.get("source_path"),
+            target_path=data.get("target_path"),
+            config_schema=data.get("config_schema", {}),
+            capabilities=data.get("capabilities", []),
+            tags=data.get("tags", []),
+        )
+
+    def to_dict(self) -> dict[str, Any]:
+        return {
+            "id": self.id,
+            "name": self.name,
+            "version": self.version,
+            "description": self.description,
+            "author": self.author,
+            "dependencies": self.dependencies,
+            "entry_point": self.entry_point,
+            "source_path": self.source_path,
+            "target_path": self.target_path,
+            "config_schema": self.config_schema,
+            "capabilities": self.capabilities,
+            "tags": self.tags,
+        }
+
+
+def load_manifests_from_directory(manifest_dir: str | Path) -> list[ChannelPluginManifest]:
+    """从目录加载所有插件清单文件 (.json)。"""
+    manifest_dir = Path(manifest_dir)
+    if not manifest_dir.exists():
+        logger.warning("Manifest directory does not exist: %s", manifest_dir)
+        return []
+
+    manifests = []
+    for file_path in manifest_dir.glob("*.json"):
+        try:
+            data = json.loads(file_path.read_text(encoding="utf-8"))
+            if isinstance(data, list):
+                for item in data:
+                    manifests.append(ChannelPluginManifest.from_dict(item))
+            else:
+                manifests.append(ChannelPluginManifest.from_dict(data))
+        except Exception:
+            logger.exception("Failed to load manifest from %s", file_path)
+    return manifests
+
+
+def save_manifest(manifest: ChannelPluginManifest, manifest_dir: str | Path) -> Path:
+    """保存插件清单到指定目录。"""
+    manifest_dir = Path(manifest_dir)
+    manifest_dir.mkdir(parents=True, exist_ok=True)
+    file_path = manifest_dir / f"{manifest.id}.json"
+    file_path.write_text(json.dumps(manifest.to_dict(), indent=2, ensure_ascii=False), encoding="utf-8")
+    logger.info("Saved manifest to %s", file_path)
+    return file_path
diff --git a/backend/package/yuxi/channel/plugins/registry.py b/backend/package/yuxi/channel/plugins/registry.py
new file mode 100644
index 00000000..8056ef8b
--- /dev/null
+++ b/backend/package/yuxi/channel/plugins/registry.py
@@ -0,0 +1,106 @@
+from __future__ import annotations
+
+import logging
+import threading
+from typing import Any, Iterator
+
+from yuxi.channel.protocols import MetaProtocol
+
+logger = logging.getLogger(__name__)
+
+
+class _RegistryMeta(type):
+    def __len__(cls) -> int:
+        return len(cls._plugins)
+
+    def __iter__(cls) -> Iterator[Any]:
+        return iter(
+            sorted(
+                cls._plugins.values(),
+                key=lambda p: (getattr(p, "order", 0), getattr(p, "id", "")),
+            )
+        )
+
+    def __contains__(cls, plugin_id: str) -> bool:
+        return plugin_id in cls._plugins
+
+    def __repr__(cls) -> str:
+        if not cls._plugins:
+            return "<ChannelPluginRegistry empty>"
+        ids = sorted(cls._plugins.keys())
+        preview = ", ".join(ids[:5])
+        tail = "..." if len(ids) > 5 else ""
+        return f"<ChannelPluginRegistry {len(cls._plugins)} plugins: {preview}{tail}>"
+
+
+class ChannelPluginRegistry(metaclass=_RegistryMeta):
+    _plugins: dict[str, Any] = {}
+    _lock: threading.Lock | None = None
+
+    @classmethod
+    def _ensure_lock(cls) -> threading.Lock:
+        if cls._lock is None:
+            cls._lock = threading.Lock()
+        return cls._lock
+
+    @classmethod
+    def register(cls, plugin: Any) -> Any:
+        if not isinstance(plugin, MetaProtocol):
+            raise ValueError("Channel plugin must satisfy MetaProtocol (id, name, order)")
+        plugin_id = getattr(plugin, "id", None)
+        if not plugin_id or not isinstance(plugin_id, str) or not plugin_id.strip():
+            raise ValueError("Channel plugin must have a non-empty string 'id' attribute")
+        with cls._ensure_lock():
+            if plugin_id in cls._plugins:
+                raise ValueError(f"Channel plugin '{plugin_id}' is already registered")
+            cls._plugins[plugin_id] = plugin
+        logger.info("Registered channel plugin: %s", plugin_id)
+        return plugin
+
+    @classmethod
+    def get(cls, plugin_id: str) -> Any:
+        return cls._plugins.get(plugin_id)
+
+    @classmethod
+    def list_plugins(cls) -> list[str]:
+        return cls.list_ids()
+
+    @classmethod
+    def list_all(cls) -> list[Any]:
+        return sorted(
+            cls._plugins.values(),
+            key=lambda p: (getattr(p, "order", 0), getattr(p, "id", "")),
+        )
+
+    @classmethod
+    def list_ids(cls) -> list[str]:
+        return sorted(cls._plugins.keys())
+
+    @classmethod
+    def all(cls) -> list[Any]:
+        return cls.list_all()
+
+    @classmethod
+    def unregister(cls, plugin_id: str) -> Any | None:
+        with cls._ensure_lock():
+            plugin = cls._plugins.pop(plugin_id, None)
+            if plugin is not None:
+                logger.info("Unregistered channel plugin: %s", plugin_id)
+            return plugin
+
+    @classmethod
+    def retire(cls, plugin_id: str) -> Any | None:
+        return cls.unregister(plugin_id)
+
+    @classmethod
+    def clear(cls) -> None:
+        with cls._ensure_lock():
+            cls._plugins.clear()
+
+    @classmethod
+    def has(cls, plugin_id: str) -> bool:
+        return plugin_id in cls._plugins
+
+    @classmethod
+    def is_registered(cls, plugin_id: str) -> bool:
+        return cls.has(plugin_id)
\ No newline at end of file
diff --git a/backend/package/yuxi/channel/plugins/sandbox.py b/backend/package/yuxi/channel/plugins/sandbox.py
new file mode 100644
index 00000000..328bc9da
--- /dev/null
+++ b/backend/package/yuxi/channel/plugins/sandbox.py
@@ -0,0 +1,552 @@
+from __future__ import annotations
+
+import ast
+import builtins
+import importlib.util
+import logging
+import types
+from pathlib import Path
+from typing import Any
+
+logger = logging.getLogger(__name__)
+
+
+class SandboxViolationError(Exception):
+    """插件代码违反沙箱安全策略时抛出。"""
+
+
+class PluginSandbox:
+    """插件代码沙箱 — 限制动态加载的插件代码只能访问安全的内置函数和模块。
+
+    通过 AST 静态分析和运行时全局变量限制，防止插件执行危险操作：
+    - 禁止文件系统访问（open, file 等）
+    - 禁止系统命令执行（os.system, subprocess 等）
+    - 禁止网络访问（socket, urllib 等）
+    - 禁止导入未授权模块
+    - 禁止访问私有属性（__import__ 等）
+    """
+
+    # 允许的内置函数白名单
+    ALLOWED_BUILTINS: set[str] = {
+        # 基础类型
+        "bool",
+        "bytearray",
+        "bytes",
+        "complex",
+        "dict",
+        "enumerate",
+        "filter",
+        "float",
+        "frozenset",
+        "int",
+        "list",
+        "map",
+        "memoryview",
+        "object",
+        "range",
+        "set",
+        "slice",
+        "str",
+        "tuple",
+        "zip",
+        # 常用函数
+        "abs",
+        "all",
+        "any",
+        "ascii",
+        "bin",
+        "callable",
+        "chr",
+        "classmethod",
+        "compile",
+        "delattr",
+        "dir",
+        "divmod",
+        "format",
+        "getattr",
+        "hasattr",
+        "hash",
+        "hex",
+        "id",
+        "isinstance",
+        "issubclass",
+        "iter",
+        "len",
+        "max",
+        "min",
+        "next",
+        "oct",
+        "ord",
+        "pow",
+        "property",
+        "repr",
+        "reversed",
+        "round",
+        "sorted",
+        "staticmethod",
+        "sum",
+        "vars",
+        # 异常处理
+        "BaseException",
+        "Exception",
+        "TypeError",
+        "ValueError",
+        "AttributeError",
+        "KeyError",
+        "IndexError",
+        "RuntimeError",
+        "NotImplementedError",
+        "StopIteration",
+        "GeneratorExit",
+        "SystemExit",
+        "ArithmeticError",
+        "LookupError",
+        "AssertionError",
+        "BufferError",
+        "EOFError",
+        "ImportError",
+        "ModuleNotFoundError",
+        "NameError",
+        "OSError",
+        "OverflowError",
+        "RecursionError",
+        "ReferenceError",
+        "SyntaxError",
+        "TimeoutError",
+        "ZeroDivisionError",
+        # 其他安全函数
+        "help",
+        "print",
+        "input",
+        "open",  # 允许但会在模块级别重写为安全版本
+        "__build_class__",
+        "__name__",
+        "__doc__",
+        "__package__",
+        "__spec__",
+        "__annotations__",
+        "__dict__",
+        "__class__",
+        "__bases__",
+        "__mro__",
+        "__subclasses__",
+        "__init__",
+        "__new__",
+        "__setattr__",
+        "__getattribute__",
+        "__delattr__",
+        "__str__",
+        "__repr__",
+        "__eq__",
+        "__ne__",
+        "__lt__",
+        "__le__",
+        "__gt__",
+        "__ge__",
+        "__hash__",
+        "__bool__",
+        "__len__",
+        "__iter__",
+        "__next__",
+        "__contains__",
+        "__getitem__",
+        "__setitem__",
+        "__delitem__",
+        "__enter__",
+        "__exit__",
+        "__await__",
+        "__aiter__",
+        "__anext__",
+        "__call__",
+    }
+
+    # 禁止导入的模块黑名单
+    FORBIDDEN_MODULES: set[str] = {
+        "os",
+        "sys",
+        "subprocess",
+        "socket",
+        "urllib",
+        "urllib.request",
+        "urllib.parse",
+        "http",
+        "http.client",
+        "ftplib",
+        "smtplib",
+        "poplib",
+        "imaplib",
+        "telnetlib",
+        "ctypes",
+        "mmap",
+        "resource",
+        "signal",
+        "pty",
+        "pickle",
+        "cPickle",
+        "marshal",
+        "imp",
+        "builtins",
+        "importlib",
+        "importlib.util",
+        "importlib.machinery",
+        "types",
+        "inspect",
+        "code",
+        "codeop",
+        "compileall",
+        "py_compile",
+        "shutil",
+        "pathlib",
+        "tempfile",
+        "glob",
+        "fnmatch",
+        "linecache",
+        "traceback",
+        "warnings",
+        "contextlib",
+        "atexit",
+        "gc",
+        "sysconfig",
+        "platform",
+        "pwd",
+        "grp",
+        "spwd",
+        "crypt",
+        "termios",
+        "tty",
+        "fcntl",
+        "msvcrt",
+        "winreg",
+        "winsound",
+        "posix",
+        "nt",
+        "java",
+        "org",
+        "sun",
+        "com",
+        "net",
+    }
+
+    # 允许导入的模块白名单（如果为空则使用黑名单模式）
+    ALLOWED_MODULES: set[str] = {
+        "abc",
+        "array",
+        "base64",
+        "binascii",
+        "bisect",
+        "calendar",
+        "collections",
+        "collections.abc",
+        "copy",
+        "csv",
+        "dataclasses",
+        "datetime",
+        "decimal",
+        "enum",
+        "fractions",
+        "functools",
+        "heapq",
+        "html",
+        "html.entities",
+        "html.parser",
+        "io",
+        "itertools",
+        "json",
+        "keyword",
+        "math",
+        "numbers",
+        "operator",
+        "random",
+        "re",
+        "string",
+        "struct",
+        "textwrap",
+        "time",
+        "typing",
+        "unicodedata",
+        "uuid",
+        "weakref",
+        "xml.etree.ElementTree",
+        "xml.dom.minidom",
+        "xml.sax",
+        "xml.parsers.expat",
+        "zoneinfo",
+        "hashlib",
+        "hmac",
+        "secrets",
+        "statistics",
+        "stringprep",
+        "tokenize",
+        "trace",
+        "types",
+        "warnings",
+        "contextlib",
+        "copyreg",
+        "dis",
+        "opcode",
+        "pickletools",
+        "quopri",
+        "reprlib",
+        "shelve",
+        "symtable",
+        "tabnanny",
+        "timeit",
+        "token",
+        "turtle",
+        "webbrowser",
+        "xxlimited",
+        "xxlimited_35",
+        "_thread",
+        "_warnings",
+        "_weakref",
+        "_weakrefset",
+        "_io",
+        "_collections",
+        "_collections_abc",
+        "_functools",
+        "_operator",
+        "_stat",
+        "_string",
+        "_symtable",
+        "_typing",
+    }
+
+    def __init__(self) -> None:
+        self._violation_log: list[str] = []
+
+    @property
+    def violations(self) -> list[str]:
+        return list(self._violation_log)
+
+    def _safe_import(self, name: str, globals_: dict | None = None, locals_: dict | None = None, fromlist: tuple = (), level: int = 0) -> Any:
+        """安全的 __import__ 替代函数，限制模块导入。"""
+        # 解析完整模块名
+        if level > 0 and globals_ is not None:
+            package = globals_.get("__package__", "")
+            if package:
+                if name:
+                    full_name = f"{package}.{name}"
+                else:
+                    full_name = package
+            else:
+                full_name = name
+        else:
+            full_name = name
+
+        # 检查是否命中黑名单
+        for forbidden in self.FORBIDDEN_MODULES:
+            if full_name == forbidden or full_name.startswith(forbidden + "."):
+                msg = f"Import of forbidden module '{full_name}' is not allowed in sandbox"
+                self._violation_log.append(msg)
+                raise SandboxViolationError(msg)
+
+        # 检查是否在白名单中（如果白名单非空）
+        if self.ALLOWED_MODULES:
+            allowed = any(
+                full_name == mod or full_name.startswith(mod + ".")
+                for mod in self.ALLOWED_MODULES
+            )
+            if not allowed:
+                msg = f"Import of module '{full_name}' is not allowed in sandbox"
+                self._violation_log.append(msg)
+                raise SandboxViolationError(msg)
+
+        # 使用原始 import 但限制在允许范围内
+        return __import__(name, globals_, locals_, fromlist, level)
+
+    def _safe_open(self, *args: Any, **kwargs: Any) -> Any:
+        """安全的 open 替代函数，禁止文件系统访问。"""
+        msg = "File operations (open) are not allowed in sandbox"
+        self._violation_log.append(msg)
+        raise SandboxViolationError(msg)
+
+    def _safe_eval(self, *args: Any, **kwargs: Any) -> Any:
+        """禁止 eval。"""
+        msg = "eval() is not allowed in sandbox"
+        self._violation_log.append(msg)
+        raise SandboxViolationError(msg)
+
+    def _safe_exec(self, *args: Any, **kwargs: Any) -> Any:
+        """禁止 exec。"""
+        msg = "exec() is not allowed in sandbox"
+        self._violation_log.append(msg)
+        raise SandboxViolationError(msg)
+
+    def _safe_compile(self, *args: Any, **kwargs: Any) -> Any:
+        """禁止 compile。"""
+        msg = "compile() is not allowed in sandbox"
+        self._violation_log.append(msg)
+        raise SandboxViolationError(msg)
+
+    @staticmethod
+    def _is_private_attr_name(name: str) -> bool:
+        """检查属性名是否为私有属性（双下划线前缀或 Python name mangling 模式）。"""
+        if name.startswith("__") and not name.endswith("__"):
+            return True
+        if "__" in name and not name.startswith("__") and not name.endswith("__"):
+            return True
+        return False
+
+    def _safe_getattr(self, obj: Any, name: str, default: Any = None) -> Any:
+        """安全的 getattr，禁止访问双下划线私有属性及 name mangling 绕过。"""
+        if self._is_private_attr_name(name):
+            msg = f"Access to private attribute '{name}' is not allowed in sandbox"
+            self._violation_log.append(msg)
+            raise SandboxViolationError(msg)
+        return getattr(obj, name, default)
+
+    def _safe_setattr(self, obj: Any, name: str, value: Any) -> None:
+        """安全的 setattr，禁止修改双下划线私有属性及 name mangling 绕过。"""
+        if self._is_private_attr_name(name):
+            msg = f"Modification of private attribute '{name}' is not allowed in sandbox"
+            self._violation_log.append(msg)
+            raise SandboxViolationError(msg)
+        setattr(obj, name, value)
+
+    def _safe_delattr(self, obj: Any, name: str) -> None:
+        """安全的 delattr，禁止删除双下划线私有属性及 name mangling 绕过。"""
+        if self._is_private_attr_name(name):
+            msg = f"Deletion of private attribute '{name}' is not allowed in sandbox"
+            self._violation_log.append(msg)
+            raise SandboxViolationError(msg)
+        delattr(obj, name)
+
+    def _create_restricted_globals(self) -> dict[str, Any]:
+        """创建受限的全局命名空间。"""
+        restricted_builtins = {}
+        for name in self.ALLOWED_BUILTINS:
+            if hasattr(builtins, name):
+                restricted_builtins[name] = getattr(builtins, name)
+
+        # 替换危险函数
+        restricted_builtins["__import__"] = self._safe_import
+        restricted_builtins["open"] = self._safe_open
+        restricted_builtins["eval"] = self._safe_eval
+        restricted_builtins["exec"] = self._safe_exec
+        restricted_builtins["compile"] = self._safe_compile
+        restricted_builtins["getattr"] = self._safe_getattr
+        restricted_builtins["setattr"] = self._safe_setattr
+        restricted_builtins["delattr"] = self._safe_delattr
+
+        return {"__builtins__": restricted_builtins}
+
+    @staticmethod
+    def validate_source(source: str, filename: str = "<sandbox>") -> ast.AST:
+        """静态分析源代码 AST，检测危险语法结构（不依赖沙箱实例）。
+
+        可用于在任何动态代码执行前进行安全预检，无需创建沙箱实例。
+        仅执行 AST 级别的静态检查，不涉及运行时全局变量限制。
+
+        Args:
+            source: 要验证的 Python 源代码
+            filename: 用于错误报告的源文件名
+
+        Returns:
+            解析后的 AST 树
+
+        Raises:
+            SandboxViolationError: 代码包含危险语法结构时
+        """
+        try:
+            tree = ast.parse(source, filename=filename)
+        except SyntaxError as e:
+            raise SandboxViolationError(f"Syntax error in source code: {e}") from e
+
+        for node in ast.walk(tree):
+            if isinstance(node, ast.Call):
+                if isinstance(node.func, ast.Name) and node.func.id == "__import__":
+                    raise SandboxViolationError(
+                        "Direct __import__() calls are not allowed in sandbox"
+                    )
+
+            if isinstance(node, ast.ImportFrom):
+                if node.names[0].name == "*":
+                    raise SandboxViolationError(
+                        "'from ... import *' is not allowed in sandbox"
+                    )
+
+        return tree
+
+    def _validate_ast(self, source: str, filename: str = "<sandbox>") -> ast.AST:
+        """静态分析 AST，检测危险语法结构（实例方法，会记录违规日志）。"""
+        try:
+            tree = self.validate_source(source, filename)
+        except SandboxViolationError as e:
+            self._violation_log.append(str(e))
+            raise
+        return tree
+
+    def execute(self, source: str, filename: str = "<sandbox>", extra_globals: dict[str, Any] | None = None) -> dict[str, Any]:
+        """在沙箱中执行 Python 源代码，返回执行后的全局命名空间。
+
+        Args:
+            source: 要执行的 Python 源代码
+            filename: 用于错误报告的源文件名
+            extra_globals: 额外注入到全局命名空间的安全对象
+
+        Returns:
+            执行后的全局命名空间字典
+
+        Raises:
+            SandboxViolationError: 代码违反安全策略时
+        """
+        # AST 静态分析
+        tree = self._validate_ast(source, filename)
+
+        # 编译为代码对象
+        code = compile(tree, filename, "exec")
+
+        # 创建受限全局环境
+        restricted_globals = self._create_restricted_globals()
+        if extra_globals:
+            # 只允许注入安全的额外对象
+            for key, value in extra_globals.items():
+                if not key.startswith("__") or key in ("__name__", "__file__", "__doc__", "__package__", "__spec__"):
+                    restricted_globals[key] = value
+
+        # 执行代码
+        exec(code, restricted_globals)  # noqa: S102
+
+        return restricted_globals
+
+    def load_module_from_source(self, source: str, module_name: str, filename: str = "<sandbox>") -> types.ModuleType:
+        """在沙箱中从源代码创建模块。
+
+        Args:
+            source: 模块源代码
+            module_name: 模块名称
+            filename: 用于错误报告的源文件名
+
+        Returns:
+            创建的模块对象
+        """
+        tree = self._validate_ast(source, filename)
+        code = compile(tree, filename, "exec")
+
+        module = types.ModuleType(module_name)
+        module.__dict__.update(self._create_restricted_globals())
+        module.__name__ = module_name
+        module.__file__ = filename
+
+        exec(code, module.__dict__)  # noqa: S102
+
+        return module
+
+    def load_module_from_file(self, file_path: str | Path, module_name: str | None = None) -> types.ModuleType:
+        """在沙箱中从文件加载模块。
+
+        Args:
+            file_path: 文件路径
+            module_name: 模块名称（默认为文件名）
+
+        Returns:
+            创建的模块对象
+        """
+        file_path = Path(file_path)
+        if not file_path.exists():
+            raise FileNotFoundError(f"Plugin file not found: {file_path}")
+
+        source = file_path.read_text(encoding="utf-8")
+        name = module_name or file_path.stem
+
+        return self.load_module_from_source(source, name, str(file_path))