diff --git a/app/chain/__init__.py b/app/chain/__init__.py
index 2b9f54c5..c8577c88 100644
--- a/app/chain/__init__.py
+++ b/app/chain/__init__.py
@@ -1407,6 +1407,7 @@ class ChainBase(metaclass=ABCMeta):
             chat_id: Union[str, int],
             text: str,
             title: Optional[str] = None,
+            buttons: Optional[List[List[dict]]] = None,
     ) -> bool:
         """
         编辑已发送的消息
@@ -1416,6 +1417,7 @@ class ChainBase(metaclass=ABCMeta):
         :param chat_id: 聊天ID
         :param text: 新的消息内容
         :param title: 消息标题
+        :param buttons: 更新后的按钮列表
         :return: 编辑是否成功
         """
         return self.run_module(
@@ -1426,6 +1428,7 @@ class ChainBase(metaclass=ABCMeta):
             chat_id=chat_id,
             text=text,
             title=title,
+            buttons=buttons,
         )
 
     def send_direct_message(self, message: Notification) -> Optional[MessageResponse]:
diff --git a/app/chain/message.py b/app/chain/message.py
index da83a479..d8d7ba55 100644
--- a/app/chain/message.py
+++ b/app/chain/message.py
@@ -16,6 +16,7 @@ from app.chain import ChainBase
 from app.chain.download import DownloadChain
 from app.chain.media import MediaChain
 from app.chain.search import SearchChain
+from app.chain.skills import SkillsChain, skills_interaction_manager
 from app.chain.subscribe import SubscribeChain
 from app.chain.transfer import TransferChain
 from app.core.config import settings, global_vars
@@ -246,6 +247,15 @@ class MessageChain(ChainBase):
                     EventType.CommandExcute,
                     {"cmd": text, "user": userid, "channel": channel, "source": source},
                 )
+            elif skills_interaction_manager.get_by_user(userid):
+                if SkillsChain().handle_text_interaction(
+                    channel=channel,
+                    source=source,
+                    userid=userid,
+                    username=username,
+                    text=text,
+                ):
+                    return
             elif text.lower().startswith("/ai"):
                 self._handle_ai_message(
                     text=text,
@@ -785,6 +795,17 @@ class MessageChain(ChainBase):
         ):
             return
 
+        if SkillsChain().handle_callback_interaction(
+            callback_data=callback_data,
+            channel=channel,
+            source=source,
+            userid=userid,
+            username=username,
+            original_message_id=original_message_id,
+            original_chat_id=original_chat_id,
+        ):
+            return
+
         if self._handle_agent_choice_callback(
             callback_data=callback_data,
             channel=channel,
diff --git a/app/chain/skills.py b/app/chain/skills.py
new file mode 100644
index 00000000..503d7796
--- /dev/null
+++ b/app/chain/skills.py
@@ -0,0 +1,868 @@
+import math
+import re
+from dataclasses import dataclass, field
+from datetime import datetime, timedelta
+from threading import Lock
+from typing import Dict, List, Optional, Tuple, Union
+import uuid
+
+from app.chain import ChainBase
+from app.helper.skill import SkillHelper, SkillInfo
+from app.schemas import Notification
+from app.schemas.message import ChannelCapabilityManager
+from app.schemas.types import MessageChannel
+
+
+@dataclass
+class PendingSkillsInteraction:
+    """
+    记录一次 /skills 会话的上下文，便于按钮和文本回复共用同一状态。
+    """
+
+    request_id: str
+    user_id: str
+    channel: Optional[MessageChannel]
+    source: Optional[str]
+    username: Optional[str]
+    view: str = "root"
+    local_page: int = 0
+    market_page: int = 0
+    created_at: datetime = field(default_factory=datetime.now)
+
+
+class SkillsInteractionManager:
+    """
+    管理用户当前的技能交互状态。
+
+    每个用户同一时间只保留一个有效会话，避免旧按钮继续生效。
+    """
+
+    _ttl = timedelta(hours=24)
+
+    def __init__(self):
+        self._by_id: Dict[str, PendingSkillsInteraction] = {}
+        self._by_user: Dict[str, str] = {}
+        self._lock = Lock()
+
+    def _cleanup_locked(self):
+        """
+        清理超时会话，避免按钮回调无限积累。
+        """
+        expire_before = datetime.now() - self._ttl
+        expired = [
+            request_id
+            for request_id, request in self._by_id.items()
+            if request.created_at < expire_before
+        ]
+        for request_id in expired:
+            request = self._by_id.pop(request_id, None)
+            if request:
+                self._by_user.pop(str(request.user_id), None)
+
+    def create_or_replace(
+        self,
+        user_id: Union[str, int],
+        channel: Optional[MessageChannel],
+        source: Optional[str],
+        username: Optional[str],
+    ) -> PendingSkillsInteraction:
+        """
+        为用户创建新会话，并替换掉旧的技能交互状态。
+        """
+        with self._lock:
+            self._cleanup_locked()
+            user_key = str(user_id)
+            old_request_id = self._by_user.get(user_key)
+            if old_request_id:
+                self._by_id.pop(old_request_id, None)
+            request_id = uuid.uuid4().hex[:12]
+            request = PendingSkillsInteraction(
+                request_id=request_id,
+                user_id=user_key,
+                channel=channel,
+                source=source,
+                username=username,
+            )
+            self._by_id[request_id] = request
+            self._by_user[user_key] = request_id
+            return request
+
+    def get_by_user(
+        self, user_id: Union[str, int]
+    ) -> Optional[PendingSkillsInteraction]:
+        """
+        按用户获取当前有效会话，供纯文本回复路由使用。
+        """
+        with self._lock:
+            self._cleanup_locked()
+            request_id = self._by_user.get(str(user_id))
+            if not request_id:
+                return None
+            return self._by_id.get(request_id)
+
+    def get_by_id(
+        self, request_id: str, user_id: Union[str, int]
+    ) -> Optional[PendingSkillsInteraction]:
+        """
+        按请求 ID 获取会话，并校验会话归属用户。
+        """
+        with self._lock:
+            self._cleanup_locked()
+            request = self._by_id.get(request_id)
+            if not request or str(request.user_id) != str(user_id):
+                return None
+            return request
+
+    def remove(self, request_id: str) -> None:
+        """
+        主动结束会话，释放用户和请求 ID 的双向索引。
+        """
+        with self._lock:
+            request = self._by_id.pop(request_id, None)
+            if request:
+                self._by_user.pop(str(request.user_id), None)
+
+    def clear(self):
+        """
+        清空所有会话，主要用于测试场景。
+        """
+        with self._lock:
+            self._by_id.clear()
+            self._by_user.clear()
+
+
+skills_interaction_manager = SkillsInteractionManager()
+
+
+class SkillsChain(ChainBase):
+    """
+    处理 /skills 指令、按钮回调和文本式技能管理交互。
+    """
+
+    _button_page_size = 6
+    _text_page_size = 8
+
+    def __init__(self):
+        super().__init__()
+        self.skillhelper = SkillHelper()
+
+    def remote_manage(
+        self,
+        arg_str: str,
+        channel: MessageChannel,
+        userid: Union[str, int],
+        source: Optional[str] = None,
+    ):
+        """
+        /skills 入口。创建新会话并渲染首屏菜单。
+        """
+        request = skills_interaction_manager.create_or_replace(
+            user_id=userid,
+            channel=channel,
+            source=source,
+            username=None,
+        )
+        force = (arg_str or "").strip().lower() in {"refresh", "刷新"}
+        self._render_interaction(
+            request=request,
+            channel=channel,
+            source=source,
+            userid=userid,
+            username=None,
+            force_market_refresh=force,
+        )
+
+    @staticmethod
+    def parse_callback(callback_data: str) -> Optional[Tuple[str, str, Optional[int]]]:
+        """
+        解析 /skills 按钮回调。
+
+        回调格式：skills:{request_id}:{action}[:index]
+        """
+        if not callback_data.startswith("skills:"):
+            return None
+        parts = callback_data.split(":")
+        if len(parts) < 3:
+            return None
+        request_id = parts[1]
+        action = parts[2]
+        index = None
+        if len(parts) >= 4 and parts[3].isdigit():
+            index = int(parts[3])
+        return request_id, action, index
+
+    def handle_callback_interaction(
+        self,
+        callback_data: str,
+        channel: MessageChannel,
+        source: str,
+        userid: Union[str, int],
+        username: str,
+        original_message_id: Optional[Union[str, int]] = None,
+        original_chat_id: Optional[str] = None,
+    ) -> bool:
+        """
+        处理按钮交互，并在同一条消息上刷新当前视图。
+        """
+        parsed = self.parse_callback(callback_data)
+        if not parsed:
+            return False
+
+        request_id, action, index = parsed
+        request = skills_interaction_manager.get_by_id(request_id, userid)
+        if not request:
+            self.post_message(
+                Notification(
+                    channel=channel,
+                    source=source,
+                    userid=userid,
+                    username=username,
+                    title="技能交互已失效，请重新发送 /skills",
+                )
+            )
+            return True
+
+        request.channel = channel
+        request.source = source
+        request.username = username
+
+        if action == "close":
+            skills_interaction_manager.remove(request.request_id)
+            self._update_or_post_message(
+                channel=channel,
+                source=source,
+                userid=userid,
+                username=username,
+                title="技能管理",
+                text="技能交互已结束",
+                original_message_id=original_message_id,
+                original_chat_id=original_chat_id,
+            )
+            return True
+
+        if action == "root":
+            request.view = "root"
+        elif action == "installed":
+            request.view = "installed"
+            request.local_page = 0
+        elif action == "market":
+            request.view = "market"
+            request.market_page = 0
+        elif action == "refresh":
+            self._render_interaction(
+                request=request,
+                channel=channel,
+                source=source,
+                userid=userid,
+                username=username,
+                original_message_id=original_message_id,
+                original_chat_id=original_chat_id,
+                force_market_refresh=True,
+            )
+            return True
+        elif action == "page-next":
+            if request.view == "installed":
+                request.local_page += 1
+            elif request.view == "market":
+                request.market_page += 1
+        elif action == "page-prev":
+            if request.view == "installed":
+                request.local_page = max(0, request.local_page - 1)
+            elif request.view == "market":
+                request.market_page = max(0, request.market_page - 1)
+        elif action == "install" and index:
+            success, message = self._install_market_skill(request, index)
+            if success:
+                self.post_message(
+                    Notification(
+                        channel=channel,
+                        source=source,
+                        userid=userid,
+                        username=username,
+                        title=message,
+                    )
+                )
+            else:
+                self.post_message(
+                    Notification(
+                        channel=channel,
+                        source=source,
+                        userid=userid,
+                        username=username,
+                        title=message,
+                    )
+                )
+        elif action == "remove" and index:
+            success, message = self._remove_local_skill(request, index)
+            self.post_message(
+                Notification(
+                    channel=channel,
+                    source=source,
+                    userid=userid,
+                    username=username,
+                    title=message,
+                )
+            )
+            if not success:
+                # 保持当前页
+                pass
+
+        self._render_interaction(
+            request=request,
+            channel=channel,
+            source=source,
+            userid=userid,
+            username=username,
+            original_message_id=original_message_id,
+            original_chat_id=original_chat_id,
+        )
+        return True
+
+    def handle_text_interaction(
+        self,
+        channel: MessageChannel,
+        source: str,
+        userid: Union[str, int],
+        username: str,
+        text: str,
+    ) -> bool:
+        """
+        处理不支持按钮渠道上的文本指令，也兼容用户直接回复文字操作。
+        """
+        request = skills_interaction_manager.get_by_user(userid)
+        if not request:
+            return False
+
+        request.channel = channel
+        request.source = source
+        request.username = username
+
+        normalized = (text or "").strip()
+        lowered = normalized.lower()
+        if lowered in {"退出", "关闭", "q", "quit", "exit"}:
+            skills_interaction_manager.remove(request.request_id)
+            self.post_message(
+                Notification(
+                    channel=channel,
+                    source=source,
+                    userid=userid,
+                    username=username,
+                    title="技能交互已结束",
+                )
+            )
+            return True
+
+        if lowered in {"返回", "back"}:
+            request.view = "root"
+            self._render_interaction(
+                request=request,
+                channel=channel,
+                source=source,
+                userid=userid,
+                username=username,
+            )
+            return True
+
+        if lowered in {"刷新", "refresh"}:
+            self._render_interaction(
+                request=request,
+                channel=channel,
+                source=source,
+                userid=userid,
+                username=username,
+                force_market_refresh=True,
+            )
+            return True
+
+        if lowered in {"p", "prev", "上一页"}:
+            if request.view == "installed":
+                request.local_page = max(0, request.local_page - 1)
+            elif request.view == "market":
+                request.market_page = max(0, request.market_page - 1)
+            self._render_interaction(
+                request=request,
+                channel=channel,
+                source=source,
+                userid=userid,
+                username=username,
+            )
+            return True
+
+        if lowered in {"n", "next", "下一页"}:
+            if request.view == "installed":
+                request.local_page += 1
+            elif request.view == "market":
+                request.market_page += 1
+            self._render_interaction(
+                request=request,
+                channel=channel,
+                source=source,
+                userid=userid,
+                username=username,
+            )
+            return True
+
+        if request.view == "root":
+            if lowered in {"1", "已安装", "本地", "local"}:
+                request.view = "installed"
+            elif lowered in {"2", "市场", "market"}:
+                request.view = "market"
+            else:
+                self.post_message(
+                    Notification(
+                        channel=channel,
+                        source=source,
+                        userid=userid,
+                        username=username,
+                        title="请输入 1 查看已安装技能，2 查看技能市场，或回复 刷新/退出",
+                    )
+                )
+                return True
+            self._render_interaction(
+                request=request,
+                channel=channel,
+                source=source,
+                userid=userid,
+                username=username,
+            )
+            return True
+
+        install_match = re.match(r"^(?:安装|装)\s*(\d+)$", normalized)
+        remove_match = re.match(r"^(?:删除|删)\s*(\d+)$", normalized)
+        if request.view == "market" and install_match:
+            success, message = self._install_market_skill(
+                request=request,
+                page_index=int(install_match.group(1)),
+            )
+            self.post_message(
+                Notification(
+                    channel=channel,
+                    source=source,
+                    userid=userid,
+                    username=username,
+                    title=message,
+                )
+            )
+            self._render_interaction(
+                request=request,
+                channel=channel,
+                source=source,
+                userid=userid,
+                username=username,
+            )
+            return True
+        if request.view == "installed" and remove_match:
+            success, message = self._remove_local_skill(
+                request=request,
+                page_index=int(remove_match.group(1)),
+            )
+            self.post_message(
+                Notification(
+                    channel=channel,
+                    source=source,
+                    userid=userid,
+                    username=username,
+                    title=message,
+                )
+            )
+            self._render_interaction(
+                request=request,
+                channel=channel,
+                source=source,
+                userid=userid,
+                username=username,
+            )
+            return True
+
+        self.post_message(
+            Notification(
+                channel=channel,
+                source=source,
+                userid=userid,
+                username=username,
+                title=self._build_usage_hint(request.view),
+            )
+        )
+        return True
+
+    def _install_market_skill(
+        self,
+        request: PendingSkillsInteraction,
+        page_index: int,
+    ) -> Tuple[bool, str]:
+        """
+        按当前市场页的可见序号安装技能，避免跨页序号歧义。
+        """
+        market_skills = [skill for skill in self.skillhelper.list_market_skills() if not skill.installed]
+        page_items, page, _ = self._page_items(
+            items=market_skills,
+            page=request.market_page,
+            page_size=self._page_size(request.channel),
+        )
+        request.market_page = page
+        if page_index < 1 or page_index > len(page_items):
+            return False, "安装序号无效"
+        return self.skillhelper.install_market_skill(page_items[page_index - 1])
+
+    def _remove_local_skill(
+        self,
+        request: PendingSkillsInteraction,
+        page_index: int,
+    ) -> Tuple[bool, str]:
+        """
+        按当前已安装页的可见序号删除技能，并拦截内置技能。
+        """
+        local_skills = self.skillhelper.list_local_skills()
+        page_items, page, _ = self._page_items(
+            items=local_skills,
+            page=request.local_page,
+            page_size=self._page_size(request.channel),
+        )
+        request.local_page = page
+        if page_index < 1 or page_index > len(page_items):
+            return False, "删除序号无效"
+        target = page_items[page_index - 1]
+        if not target.removable:
+            return False, f"技能 {target.id} 是内置技能，不能删除"
+        return self.skillhelper.remove_local_skill(target.id)
+
+    def _render_interaction(
+        self,
+        request: PendingSkillsInteraction,
+        channel: MessageChannel,
+        source: Optional[str],
+        userid: Union[str, int],
+        username: Optional[str],
+        original_message_id: Optional[Union[str, int]] = None,
+        original_chat_id: Optional[str] = None,
+        force_market_refresh: bool = False,
+    ) -> None:
+        """
+        根据当前视图生成内容，并选择编辑原消息或发送新消息。
+        """
+        if request.view == "installed":
+            title, text, buttons = self._build_installed_view(
+                request=request,
+                force_market_refresh=force_market_refresh,
+            )
+        elif request.view == "market":
+            title, text, buttons = self._build_market_view(
+                request=request,
+                force_market_refresh=force_market_refresh,
+            )
+        else:
+            title, text, buttons = self._build_root_view(
+                request=request,
+                force_market_refresh=force_market_refresh,
+            )
+
+        self._update_or_post_message(
+            channel=channel,
+            source=source,
+            userid=userid,
+            username=username,
+            title=title,
+            text=text,
+            buttons=buttons,
+            original_message_id=original_message_id,
+            original_chat_id=original_chat_id,
+        )
+
+    def _build_root_view(
+        self,
+        request: PendingSkillsInteraction,
+        force_market_refresh: bool = False,
+    ) -> Tuple[str, str, Optional[List[List[dict]]]]:
+        """
+        构建根菜单视图，汇总本地技能和市场概览。
+        """
+        local_skills = self.skillhelper.list_local_skills()
+        market_skills = [
+            skill
+            for skill in self.skillhelper.list_market_skills(force=force_market_refresh)
+            if not skill.installed
+        ]
+        sources = self.skillhelper.get_market_sources()
+        source_lines = []
+        for index, source in enumerate(sources, start=1):
+            source_lines.append(f"{index}. {source}")
+
+        text_lines = [
+            f"已安装技能：{len(local_skills)}",
+            f"市场可安装技能：{len(market_skills)}",
+        ]
+        if source_lines:
+            text_lines.extend(["", "公开技能源：", *source_lines])
+        text_lines.extend(
+            [
+                "",
+                "1. 查看已安装技能",
+                "2. 浏览技能市场",
+                "回复 刷新 重新获取市场数据，回复 退出 结束交互",
+            ]
+        )
+
+        buttons = None
+        if self._supports_interactive_buttons(request.channel):
+            buttons = [
+                [{"text": "已安装技能", "callback_data": f"skills:{request.request_id}:installed"}],
+                [{"text": "技能市场", "callback_data": f"skills:{request.request_id}:market"}],
+                [
+                    {"text": "刷新市场", "callback_data": f"skills:{request.request_id}:refresh"},
+                    {"text": "关闭", "callback_data": f"skills:{request.request_id}:close"},
+                ],
+            ]
+        return "技能管理", "\n".join(text_lines), buttons
+
+    def _build_installed_view(
+        self,
+        request: PendingSkillsInteraction,
+        force_market_refresh: bool = False,  # noqa: ARG002
+    ) -> Tuple[str, str, Optional[List[List[dict]]]]:
+        """
+        构建已安装技能视图，列出来源和可删除状态。
+        """
+        local_skills = self.skillhelper.list_local_skills()
+        page_items, page, total_pages = self._page_items(
+            items=local_skills,
+            page=request.local_page,
+            page_size=self._page_size(request.channel),
+        )
+        request.local_page = page
+
+        text_lines = [f"第 {page + 1}/{total_pages} 页，共 {len(local_skills)} 个技能"]
+        if not page_items:
+            text_lines.append("")
+            text_lines.append("当前没有已安装技能")
+        else:
+            for index, skill in enumerate(page_items, start=1):
+                action = "可删除" if skill.removable else "内置不可删"
+                text_lines.extend(
+                    [
+                        "",
+                        f"{index}. {skill.id} ({skill.source_label}，{action})",
+                        self._truncate(skill.description),
+                    ]
+                )
+
+        text_lines.extend(
+            [
+                "",
+                "回复 删除 <序号> 删除技能，回复 n/p 翻页，回复 返回 回到菜单，回复 退出 结束交互",
+            ]
+        )
+
+        buttons = None
+        if self._supports_interactive_buttons(request.channel):
+            buttons = []
+            for index, skill in enumerate(page_items, start=1):
+                if not skill.removable:
+                    continue
+                buttons.append(
+                    [
+                        {
+                            "text": f"删除 {index}",
+                            "callback_data": f"skills:{request.request_id}:remove:{index}",
+                        }
+                    ]
+                )
+            buttons.extend(self._navigation_buttons(request, page, total_pages))
+            buttons.append(
+                [
+                    {"text": "返回", "callback_data": f"skills:{request.request_id}:root"},
+                    {"text": "关闭", "callback_data": f"skills:{request.request_id}:close"},
+                ]
+            )
+        return "已安装技能", "\n".join(text_lines), buttons
+
+    def _build_market_view(
+        self,
+        request: PendingSkillsInteraction,
+        force_market_refresh: bool = False,
+    ) -> Tuple[str, str, Optional[List[List[dict]]]]:
+        """
+        构建技能市场视图，仅展示尚未安装的技能。
+        """
+        market_skills = [
+            skill
+            for skill in self.skillhelper.list_market_skills(force=force_market_refresh)
+            if not skill.installed
+        ]
+        page_items, page, total_pages = self._page_items(
+            items=market_skills,
+            page=request.market_page,
+            page_size=self._page_size(request.channel),
+        )
+        request.market_page = page
+
+        text_lines = [f"第 {page + 1}/{total_pages} 页，共 {len(market_skills)} 个可安装技能"]
+        if not page_items:
+            text_lines.append("")
+            text_lines.append("当前没有可安装的市场技能")
+        else:
+            for index, skill in enumerate(page_items, start=1):
+                text_lines.extend(
+                    [
+                        "",
+                        f"{index}. {skill.id} ({skill.source_label})",
+                        self._truncate(skill.description),
+                    ]
+                )
+
+        text_lines.extend(
+            [
+                "",
+                "回复 安装 <序号> 安装技能，回复 刷新 重新拉取市场，回复 n/p 翻页，回复 返回 回到菜单，回复 退出 结束交互",
+            ]
+        )
+
+        buttons = None
+        if self._supports_interactive_buttons(request.channel):
+            buttons = []
+            for index, _skill in enumerate(page_items, start=1):
+                buttons.append(
+                    [
+                        {
+                            "text": f"安装 {index}",
+                            "callback_data": f"skills:{request.request_id}:install:{index}",
+                        }
+                    ]
+                )
+            buttons.extend(self._navigation_buttons(request, page, total_pages))
+            buttons.append(
+                [
+                    {"text": "刷新", "callback_data": f"skills:{request.request_id}:refresh"},
+                    {"text": "返回", "callback_data": f"skills:{request.request_id}:root"},
+                    {"text": "关闭", "callback_data": f"skills:{request.request_id}:close"},
+                ]
+            )
+        return "技能市场", "\n".join(text_lines), buttons
+
+    @staticmethod
+    def _truncate(text: str, limit: int = 140) -> str:
+        """
+        对技能描述做轻量截断，避免消息过长。
+        """
+        if len(text) <= limit:
+            return text
+        return text[: limit - 3] + "..."
+
+    @staticmethod
+    def _page_items(
+        items: List[SkillInfo],
+        page: int,
+        page_size: int,
+    ) -> Tuple[List[SkillInfo], int, int]:
+        """
+        返回当前页的数据，并把页码钳制到有效范围内。
+        """
+        total_pages = max(1, math.ceil(len(items) / page_size)) if page_size else 1
+        page = min(max(0, page), total_pages - 1)
+        start = page * page_size
+        end = start + page_size
+        return items[start:end], page, total_pages
+
+    def _page_size(self, channel: Optional[MessageChannel]) -> int:
+        """
+        按渠道能力选择分页大小，按钮渠道单页更短，便于直接操作。
+        """
+        return (
+            self._button_page_size
+            if self._supports_interactive_buttons(channel)
+            else self._text_page_size
+        )
+
+    @staticmethod
+    def _supports_interactive_buttons(channel: Optional[MessageChannel]) -> bool:
+        """
+        判断当前渠道是否同时支持按钮展示和回调。
+        """
+        return bool(
+            channel
+            and ChannelCapabilityManager.supports_buttons(channel)
+            and ChannelCapabilityManager.supports_callbacks(channel)
+        )
+
+    @staticmethod
+    def _navigation_buttons(
+        request: PendingSkillsInteraction,
+        page: int,
+        total_pages: int,
+    ) -> List[List[dict]]:
+        """
+        为分页视图生成上一页和下一页按钮。
+        """
+        buttons = []
+        nav_row = []
+        if page > 0:
+            nav_row.append(
+                {
+                    "text": "⬅️ 上一页",
+                    "callback_data": f"skills:{request.request_id}:page-prev",
+                }
+            )
+        if page < total_pages - 1:
+            nav_row.append(
+                {
+                    "text": "下一页 ➡️",
+                    "callback_data": f"skills:{request.request_id}:page-next",
+                }
+            )
+        if nav_row:
+            buttons.append(nav_row)
+        return buttons
+
+    def _update_or_post_message(
+        self,
+        channel: MessageChannel,
+        source: Optional[str],
+        userid: Union[str, int],
+        username: Optional[str],
+        title: str,
+        text: str,
+        buttons: Optional[List[List[dict]]] = None,
+        original_message_id: Optional[Union[str, int]] = None,
+        original_chat_id: Optional[str] = None,
+    ) -> None:
+        """
+        优先编辑原消息，编辑失败时再回退为发送新消息。
+        """
+        if (
+            original_message_id
+            and original_chat_id
+            and ChannelCapabilityManager.supports_editing(channel)
+        ):
+            edited = self.edit_message(
+                channel=channel,
+                source=source,
+                message_id=original_message_id,
+                chat_id=original_chat_id,
+                title=title,
+                text=text,
+                buttons=buttons,
+            )
+            if edited:
+                return
+
+        self.post_message(
+            Notification(
+                channel=channel,
+                source=source,
+                userid=userid,
+                username=username,
+                title=title,
+                text=text,
+                buttons=buttons,
+            )
+        )
+
+    @staticmethod
+    def _build_usage_hint(view: str) -> str:
+        """
+        根据当前视图返回可执行的文本命令提示。
+        """
+        if view == "market":
+            return "请输入 安装 <序号>、刷新、n、p、返回 或 退出"
+        if view == "installed":
+            return "请输入 删除 <序号>、n、p、返回 或 退出"
+        return "请输入 1、2、刷新 或 退出"
diff --git a/app/command.py b/app/command.py
index b0cb2823..717e3c65 100644
--- a/app/command.py
+++ b/app/command.py
@@ -7,6 +7,7 @@ from app.chain import ChainBase
 from app.chain.download import DownloadChain
 from app.chain.message import MessageChain
 from app.chain.site import SiteChain
+from app.chain.skills import SkillsChain
 from app.chain.subscribe import SubscribeChain
 from app.chain.system import SystemChain
 from app.chain.transfer import TransferChain
@@ -154,6 +155,12 @@ class Command(metaclass=Singleton):
                 "category": "管理",
                 "data": {},
             },
+            "/skills": {
+                "func": SkillsChain().remote_manage,
+                "description": "管理技能",
+                "category": "智能体",
+                "data": {},
+            },
         }
         # 插件命令集合
         self._plugin_commands = {}
diff --git a/app/core/config.py b/app/core/config.py
index 804c8eb5..8344cb0f 100644
--- a/app/core/config.py
+++ b/app/core/config.py
@@ -420,6 +420,14 @@ class ConfigModel(BaseModel):
     # 本地插件仓库目录，多个地址使用,分隔
     PLUGIN_LOCAL_REPO_PATHS: Optional[str] = None
 
+    # ==================== 技能配置 ====================
+    # 技能市场仓库地址，多个地址使用,分隔
+    SKILL_MARKET: str = (
+        "https://github.com/openai/skills,"
+        "https://github.com/anthropics/skills,"
+        "https://github.com/vercel-labs/agent-skills"
+    )
+
     # ==================== Github & PIP ====================
     # Github token，提高请求api限流阈值 ghp_****
     GITHUB_TOKEN: Optional[str] = None
diff --git a/app/helper/skill.py b/app/helper/skill.py
new file mode 100644
index 00000000..1cba16c2
--- /dev/null
+++ b/app/helper/skill.py
@@ -0,0 +1,466 @@
+import io
+import json
+import shutil
+import zipfile
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Dict, List, Optional, Tuple
+from urllib.parse import urlparse
+
+from app.agent.middleware.skills import _parse_skill_metadata
+from app.core.cache import cached, fresh
+from app.core.config import settings
+from app.log import logger
+from app.utils.http import RequestUtils
+from app.utils.singleton import WeakSingleton
+from app.utils.url import UrlUtils
+
+_SOURCE_META_FILENAME = ".moviepilot-skill-source.json"
+_DEFAULT_BRANCHES = ("main", "master")
+_MARKET_CACHE_TTL = 60 * 30
+
+
+@dataclass
+class SkillInfo:
+    id: str
+    name: str
+    description: str
+    version: int = 0
+    path: str = ""
+    source_type: str = "local"
+    source_label: str = "本地"
+    repo_url: Optional[str] = None
+    repo_name: Optional[str] = None
+    skill_path: Optional[str] = None
+    installed: bool = False
+    removable: bool = False
+
+
+class SkillHelper(metaclass=WeakSingleton):
+    """
+    技能市场与本地技能管理
+    """
+
+    @staticmethod
+    def get_user_skills_dir() -> Path:
+        """
+        返回用户技能目录，所有市场安装的技能都落在这里。
+        """
+        return settings.CONFIG_PATH / "agent" / "skills"
+
+    @staticmethod
+    def get_bundled_skills_dir() -> Path:
+        """
+        返回仓库内置技能目录。
+        """
+        return settings.ROOT_PATH / "skills"
+
+    @staticmethod
+    def get_market_sources() -> List[str]:
+        """
+        解析配置中的技能市场列表。
+        """
+        if not settings.SKILL_MARKET:
+            return []
+        return [item.strip() for item in settings.SKILL_MARKET.split(",") if item.strip()]
+
+    @staticmethod
+    def _ensure_user_skills_dir() -> Path:
+        """
+        确保用户技能目录存在，供安装和扫描复用。
+        """
+        skill_dir = SkillHelper.get_user_skills_dir()
+        skill_dir.mkdir(parents=True, exist_ok=True)
+        return skill_dir
+
+    @staticmethod
+    def _normalize_repo_url(repo_url: str) -> Optional[str]:
+        """
+        将技能市场配置统一归一为 GitHub HTTPS 地址。
+        """
+        repo_url = (repo_url or "").strip()
+        if not repo_url:
+            return None
+        if repo_url.startswith(("http://", "https://")):
+            return repo_url.rstrip("/")
+        return f"https://github.com/{repo_url.strip('/')}"
+
+    @staticmethod
+    def _parse_market_repo(repo_url: str) -> Optional[dict]:
+        """
+        解析市场仓库地址，提取仓库、分支和技能根目录信息。
+        """
+        normalized = SkillHelper._normalize_repo_url(repo_url)
+        if not normalized:
+            return None
+        parsed = urlparse(normalized)
+        if parsed.netloc.lower() not in {"github.com", "www.github.com"}:
+            logger.warning("暂不支持的技能市场地址：%s", repo_url)
+            return None
+
+        parts = [part for part in parsed.path.strip("/").split("/") if part]
+        if len(parts) < 2:
+            return None
+
+        owner = parts[0]
+        repo = parts[1].removesuffix(".git")
+        branch = None
+        root_path = "skills"
+        if len(parts) >= 4 and parts[2] == "tree":
+            branch = parts[3]
+            if len(parts) > 4:
+                root_path = "/".join(parts[4:])
+
+        return {
+            "repo_url": f"https://github.com/{owner}/{repo}",
+            "repo_name": f"{owner}/{repo}",
+            "owner": owner,
+            "repo": repo,
+            "branch": branch,
+            "root_path": root_path.strip("/") or "skills",
+        }
+
+    @staticmethod
+    def _read_source_meta(skill_dir: Path) -> dict:
+        """
+        读取技能来源元数据，用于区分本地、市场和内置技能。
+        """
+        meta_path = skill_dir / _SOURCE_META_FILENAME
+        if not meta_path.exists():
+            return {}
+        try:
+            payload = json.loads(meta_path.read_text(encoding="utf-8"))
+            return payload if isinstance(payload, dict) else {}
+        except Exception as e:
+            logger.warning("读取技能来源元数据失败：%s - %s", meta_path, e)
+            return {}
+
+    @staticmethod
+    def _write_source_meta(skill_dir: Path, payload: dict) -> None:
+        """
+        写入技能来源元数据，便于后续展示来源和追踪安装来源。
+        """
+        meta_path = skill_dir / _SOURCE_META_FILENAME
+        meta_path.write_text(
+            json.dumps(payload, ensure_ascii=False, indent=2),
+            encoding="utf-8",
+        )
+
+    @staticmethod
+    def _is_bundled_skill(skill_id: str) -> bool:
+        """
+        判断技能是否来自仓库内置目录。
+        """
+        return (SkillHelper.get_bundled_skills_dir() / skill_id / "SKILL.md").exists()
+
+    def list_local_skills(self) -> List[SkillInfo]:
+        """
+        扫描本地已安装技能，并补充来源和是否可删除等展示信息。
+        """
+        skill_root = self._ensure_user_skills_dir()
+        results: List[SkillInfo] = []
+
+        for path in sorted(skill_root.iterdir(), key=lambda item: item.name.lower()):
+            if not path.is_dir():
+                continue
+            skill_md = path / "SKILL.md"
+            if not skill_md.exists():
+                continue
+            try:
+                content = skill_md.read_text(encoding="utf-8")
+            except Exception as e:
+                logger.warning("读取技能文件失败：%s - %s", skill_md, e)
+                continue
+
+            metadata = _parse_skill_metadata(content, str(skill_md), path.name)
+            if not metadata:
+                continue
+
+            bundled = self._is_bundled_skill(path.name)
+            source_meta = self._read_source_meta(path)
+            source_type = "bundled" if bundled else source_meta.get("source", "local")
+            if source_type == "market":
+                repo_name = source_meta.get("repo_name") or source_meta.get("repo_url")
+                source_label = f"市场 · {repo_name}" if repo_name else "市场"
+            elif source_type == "bundled":
+                source_label = "内置"
+            else:
+                source_label = "本地"
+
+            results.append(
+                SkillInfo(
+                    id=path.name,
+                    name=metadata["name"],
+                    description=metadata["description"],
+                    version=metadata["version"],
+                    path=str(skill_md),
+                    source_type=source_type,
+                    source_label=source_label,
+                    repo_url=source_meta.get("repo_url"),
+                    repo_name=source_meta.get("repo_name"),
+                    skill_path=source_meta.get("skill_path"),
+                    installed=True,
+                    removable=not bundled,
+                )
+            )
+
+        return results
+
+    def list_market_skills(self, force: bool = False) -> List[SkillInfo]:
+        """
+        聚合所有市场源的技能，并用本地技能状态标记已安装项。
+        """
+        local_skills = self.list_local_skills()
+        local_ids = {skill.id for skill in local_skills}
+        local_names = {skill.name for skill in local_skills}
+
+        deduped: Dict[str, SkillInfo] = {}
+        for source in self.get_market_sources():
+            with fresh(force):
+                market_skills = self._list_market_repo_skills(source)
+            for skill in market_skills:
+                key = skill.name or skill.id
+                if key in deduped:
+                    continue
+                skill.installed = skill.id in local_ids or skill.name in local_names
+                deduped[key] = skill
+
+        return sorted(
+            deduped.values(),
+            key=lambda item: (
+                item.installed,
+                (item.repo_name or "").lower(),
+                item.name.lower(),
+            ),
+        )
+
+    @cached(maxsize=16, ttl=_MARKET_CACHE_TTL, skip_empty=True)
+    def _list_market_repo_skills(self, repo_url: str) -> List[SkillInfo]:
+        """
+        读取单个市场仓库中的技能列表。
+
+        仓库按 zip 方式拉取后直接在压缩包内解析，避免落地整个仓库。
+        """
+        repo = self._parse_market_repo(repo_url)
+        if not repo:
+            return []
+
+        repo_bytes = self._download_repo_archive(repo)
+        if not repo_bytes:
+            return []
+
+        try:
+            with zipfile.ZipFile(io.BytesIO(repo_bytes)) as zf:
+                names = zf.namelist()
+                if not names:
+                    return []
+                root_prefix = names[0].split("/", 1)[0] + "/"
+                results: List[SkillInfo] = []
+                seen_paths = set()
+                for archive_name in names:
+                    if not archive_name.endswith("/SKILL.md"):
+                        continue
+                    if not archive_name.startswith(root_prefix):
+                        continue
+
+                    rel_path = archive_name[len(root_prefix):].strip("/")
+                    if not rel_path.startswith(f"{repo['root_path'].strip('/')}/"):
+                        continue
+                    if "/.system/" in f"/{rel_path}/":
+                        continue
+                    if rel_path in seen_paths:
+                        continue
+                    seen_paths.add(rel_path)
+
+                    skill_dir = rel_path[: -len("/SKILL.md")]
+                    skill_id = Path(skill_dir).name
+                    try:
+                        content = zf.read(archive_name).decode("utf-8")
+                    except Exception as e:
+                        logger.warning("读取市场技能失败：%s - %s", archive_name, e)
+                        continue
+
+                    metadata = _parse_skill_metadata(
+                        content,
+                        f"{repo['repo_url']}:{rel_path}",
+                        skill_id,
+                    )
+                    if not metadata:
+                        continue
+
+                    results.append(
+                        SkillInfo(
+                            id=skill_id,
+                            name=metadata["name"],
+                            description=metadata["description"],
+                            version=metadata["version"],
+                            path=f"{repo['repo_url']}/tree/{repo['branch']}/{skill_dir}",
+                            source_type="market",
+                            source_label=f"市场 · {repo['repo_name']}",
+                            repo_url=repo["repo_url"],
+                            repo_name=repo["repo_name"],
+                            skill_path=skill_dir,
+                            installed=False,
+                            removable=False,
+                        )
+                    )
+                return results
+        except Exception as e:
+            logger.error("解析技能市场压缩包失败：%s", e)
+            return []
+
+    def install_market_skill(self, skill: SkillInfo) -> Tuple[bool, str]:
+        """
+        将市场技能安装到用户技能目录，并记录来源元数据。
+        """
+        if not skill.repo_url or not skill.skill_path:
+            return False, "技能来源信息不完整，无法安装"
+
+        target_root = self._ensure_user_skills_dir()
+        target_dir = target_root / skill.id
+        if target_dir.exists():
+            return False, f"技能 {skill.id} 已存在"
+        if self._is_bundled_skill(skill.id):
+            return False, f"技能 {skill.id} 是 MoviePilot 内置技能，不能覆盖安装"
+
+        repo = self._parse_market_repo(skill.repo_url)
+        if not repo:
+            return False, "技能市场地址无效"
+
+        repo_bytes = self._download_repo_archive(repo)
+        if not repo_bytes:
+            return False, "下载技能仓库失败，请检查网络连接或 GitHub 配置"
+
+        try:
+            with zipfile.ZipFile(io.BytesIO(repo_bytes)) as zf:
+                names = zf.namelist()
+                if not names:
+                    return False, "技能仓库为空"
+                root_prefix = names[0].split("/", 1)[0] + "/"
+                skill_prefix = f"{root_prefix}{skill.skill_path.strip('/')}/"
+                matched = [name for name in names if name.startswith(skill_prefix)]
+                if not matched:
+                    return False, f"未找到技能目录：{skill.skill_path}"
+
+                target_dir.mkdir(parents=True, exist_ok=False)
+                try:
+                    wrote = False
+                    for archive_name in matched:
+                        rel_name = archive_name[len(skill_prefix):]
+                        if not rel_name:
+                            continue
+                        output_path = target_dir / rel_name
+                        if archive_name.endswith("/"):
+                            output_path.mkdir(parents=True, exist_ok=True)
+                            continue
+                        output_path.parent.mkdir(parents=True, exist_ok=True)
+                        with zf.open(archive_name, "r") as src, open(output_path, "wb") as dst:
+                            shutil.copyfileobj(src, dst)
+                        wrote = True
+
+                    if not wrote or not (target_dir / "SKILL.md").exists():
+                        shutil.rmtree(target_dir, ignore_errors=True)
+                        return False, "技能目录内容不完整，安装失败"
+
+                    self._write_source_meta(
+                        target_dir,
+                        {
+                            "source": "market",
+                            "repo_url": repo["repo_url"],
+                            "repo_name": repo["repo_name"],
+                            "branch": repo["branch"],
+                            "skill_path": skill.skill_path,
+                            "installed_at": datetime.now(timezone.utc).isoformat(),
+                        },
+                    )
+                    return True, f"技能 {skill.id} 已安装到 {target_dir}"
+                except Exception:
+                    shutil.rmtree(target_dir, ignore_errors=True)
+                    raise
+        except Exception as e:
+            logger.error("安装市场技能失败：%s", e)
+            return False, f"安装技能失败：{e}"
+
+    def remove_local_skill(self, skill_id: str) -> Tuple[bool, str]:
+        """
+        删除一个本地技能目录，内置技能会被显式拦截。
+        """
+        if not skill_id:
+            return False, "技能ID不能为空"
+        if self._is_bundled_skill(skill_id):
+            return False, f"技能 {skill_id} 是 MoviePilot 内置技能，不能删除"
+
+        skill_dir = self._ensure_user_skills_dir() / skill_id
+        if not skill_dir.exists():
+            return False, f"技能 {skill_id} 不存在"
+        if not (skill_dir / "SKILL.md").exists():
+            return False, f"{skill_id} 不是有效的技能目录"
+
+        try:
+            shutil.rmtree(skill_dir)
+            return True, f"技能 {skill_id} 已删除"
+        except Exception as e:
+            logger.error("删除技能失败：%s", e)
+            return False, f"删除技能失败：{e}"
+
+    def _download_repo_archive(self, repo: dict) -> Optional[bytes]:
+        """
+        下载市场仓库压缩包，并在缺省分支之间做回退尝试。
+        """
+        branches = [repo.get("branch")] if repo.get("branch") else []
+        branches.extend([branch for branch in _DEFAULT_BRANCHES if branch not in branches])
+        for branch in branches:
+            archive_url = (
+                f"https://codeload.github.com/{repo['owner']}/{repo['repo']}/zip/refs/heads/{branch}"
+            )
+            response = self._request_github(
+                url=archive_url,
+                repo_name=repo["repo_name"],
+                is_api=False,
+            )
+            if response is not None and response.status_code == 200:
+                repo["branch"] = branch
+                return response.content
+        logger.warning("下载技能市场仓库失败：%s", repo["repo_url"])
+        return None
+
+    @staticmethod
+    def _request_github(
+        url: str,
+        repo_name: str,
+        is_api: bool = False,
+        timeout: int = 30,
+    ):
+        """
+        按代理优先级顺序请求 GitHub 资源，兼容代理和直连场景。
+        """
+        strategies = []
+        headers = settings.REPO_GITHUB_HEADERS(repo=repo_name)
+        if not is_api and settings.GITHUB_PROXY:
+            proxy_url = f"{UrlUtils.standardize_base_url(settings.GITHUB_PROXY)}{url}"
+            strategies.append((proxy_url, {"headers": headers, "timeout": timeout}))
+        if settings.PROXY_HOST:
+            strategies.append(
+                (
+                    url,
+                    {
+                        "headers": headers,
+                        "proxies": settings.PROXY,
+                        "timeout": timeout,
+                    },
+                )
+            )
+        strategies.append((url, {"headers": headers, "timeout": timeout}))
+
+        for target_url, kwargs in strategies:
+            try:
+                response = RequestUtils(**kwargs).get_res(
+                    url=target_url,
+                    raise_exception=True,
+                )
+                if response is not None and response.status_code == 200:
+                    return response
+            except Exception as e:
+                logger.warning("请求技能市场失败：%s - %s", target_url, e)
+        return None
diff --git a/app/modules/discord/__init__.py b/app/modules/discord/__init__.py
index 8b7bbe89..578383e1 100644
--- a/app/modules/discord/__init__.py
+++ b/app/modules/discord/__init__.py
@@ -439,6 +439,7 @@ class DiscordModule(_ModuleBase, _MessageBase[Discord]):
         chat_id: Union[str, int],
         text: str,
         title: Optional[str] = None,
+        buttons: Optional[List[List[dict]]] = None,
     ) -> bool:
         """
         编辑消息
@@ -448,6 +449,7 @@ class DiscordModule(_ModuleBase, _MessageBase[Discord]):
         :param chat_id: 聊天ID
         :param text: 新的消息内容
         :param title: 消息标题
+        :param buttons: 新的按钮列表
         :return: 编辑是否成功
         """
         if channel != self._channel:
@@ -460,6 +462,7 @@ class DiscordModule(_ModuleBase, _MessageBase[Discord]):
                 result = client.send_msg(
                     title=title or "",
                     text=text,
+                    buttons=buttons,
                     original_message_id=message_id,
                     original_chat_id=str(chat_id),
                 )
diff --git a/app/modules/slack/__init__.py b/app/modules/slack/__init__.py
index 1f089c90..25c33182 100644
--- a/app/modules/slack/__init__.py
+++ b/app/modules/slack/__init__.py
@@ -557,6 +557,7 @@ class SlackModule(_ModuleBase, _MessageBase[Slack]):
         chat_id: Union[str, int],
         text: str,
         title: Optional[str] = None,
+        buttons: Optional[List[List[dict]]] = None,
     ) -> bool:
         """
         编辑消息
@@ -566,6 +567,7 @@ class SlackModule(_ModuleBase, _MessageBase[Slack]):
         :param chat_id: 聊天ID
         :param text: 新的消息内容
         :param title: 消息标题
+        :param buttons: 新的按钮列表
         :return: 编辑是否成功
         """
         if channel != self._channel:
@@ -578,6 +580,7 @@ class SlackModule(_ModuleBase, _MessageBase[Slack]):
                 result = client.send_msg(
                     title=title or "",
                     text=text,
+                    buttons=buttons,
                     original_message_id=str(message_id),
                     original_chat_id=str(chat_id),
                 )
diff --git a/app/modules/telegram/__init__.py b/app/modules/telegram/__init__.py
index 001669e9..4ea01a35 100644
--- a/app/modules/telegram/__init__.py
+++ b/app/modules/telegram/__init__.py
@@ -564,6 +564,7 @@ class TelegramModule(_ModuleBase, _MessageBase[Telegram]):
         chat_id: Union[str, int],
         text: str,
         title: Optional[str] = None,
+        buttons: Optional[List[List[dict]]] = None,
     ) -> bool:
         """
         编辑消息
@@ -573,6 +574,7 @@ class TelegramModule(_ModuleBase, _MessageBase[Telegram]):
         :param chat_id: 聊天ID
         :param text: 新的消息内容
         :param title: 消息标题
+        :param buttons: 新的按钮列表
         :return: 编辑是否成功
         """
         if channel != self._channel:
@@ -587,6 +589,7 @@ class TelegramModule(_ModuleBase, _MessageBase[Telegram]):
                     message_id=message_id,
                     text=text,
                     title=title,
+                    buttons=buttons,
                 )
                 if result:
                     return True
diff --git a/app/modules/telegram/telegram.py b/app/modules/telegram/telegram.py
index da1e3909..ca872b82 100644
--- a/app/modules/telegram/telegram.py
+++ b/app/modules/telegram/telegram.py
@@ -835,6 +835,7 @@ class Telegram:
             message_id: Union[str, int],
             text: str,
             title: Optional[str] = None,
+            buttons: Optional[List[List[dict]]] = None,
     ) -> Optional[bool]:
         """
         编辑Telegram消息（公开方法）
@@ -842,6 +843,7 @@ class Telegram:
         :param message_id: 消息ID
         :param text: 新的消息内容
         :param title: 消息标题
+        :param buttons: 新的按钮列表
         :return: 编辑是否成功
         """
         if not self._bot:
@@ -861,6 +863,7 @@ class Telegram:
                 chat_id=str(chat_id),
                 message_id=int(message_id),
                 text=caption,
+                buttons=buttons,
             )
         except Exception as e:
             logger.error(f"编辑Telegram消息异常: {str(e)}")
diff --git a/skills/create-moviepilot-skill/SKILL.md b/skills/create-moviepilot-skill/SKILL.md
new file mode 100644
index 00000000..a9bc3590
--- /dev/null
+++ b/skills/create-moviepilot-skill/SKILL.md
@@ -0,0 +1,164 @@
+---
+name: create-moviepilot-skill
+version: 1
+description: >-
+  Use this skill when the user asks to create, scaffold, update, or review a
+  MoviePilot agent skill. This includes adding a new built-in skill under the
+  repository `skills/` directory, editing an existing built-in skill, writing
+  `SKILL.md` frontmatter and workflow instructions, choosing `allowed-tools`,
+  adding helper scripts when needed, and bumping the built-in skill `version`
+  so changes can sync into `config/agent/skills`.
+allowed-tools: list_directory read_file write_file edit_file execute_command
+---
+
+# Create MoviePilot Skill
+
+This skill guides you through creating or updating a built-in MoviePilot agent
+skill in this repository.
+
+## Scope
+
+Use this workflow for repository built-in skills:
+
+- Create or update files under `skills/<skill-id>/`
+- Commit the skill as part of the MoviePilot repository
+- Do not place the implementation only in `config/agent/skills` unless the user
+  explicitly asks for a local override instead of a built-in skill
+
+## MoviePilot-Specific Rules
+
+- The repository root `skills/` directory is the bundled source of truth for
+  built-in skills.
+- On agent startup, bundled skills are synced into `config/agent/skills`.
+- Sync overwrite depends on the `version` field in `SKILL.md`. If you update an
+  existing built-in skill, increment `version`, or users may continue using an
+  older copied version.
+- Keep the folder name and frontmatter `name` identical. Use lowercase letters,
+  digits, and hyphens only.
+- Prefer extending an existing skill instead of creating an overlapping
+  duplicate.
+
+## Workflow
+
+### Step 1: Understand the Request
+
+- Determine whether the user wants a new skill or a change to an existing one.
+- Extract the target task, likely trigger phrases, needed tools, and whether
+  helper scripts are necessary.
+- If the goal is still ambiguous after reading the request and local context,
+  ask one focused clarification question. Otherwise proceed with a reasonable
+  default.
+
+### Step 2: Check Existing Skills First
+
+- Inspect the repository `skills/` directory before creating anything new.
+- If an existing skill already covers most of the workflow, update it instead of
+  adding a near-duplicate.
+- Reuse the repository style: concise YAML frontmatter, trigger-rich
+  description, and procedural body sections.
+
+### Step 3: Choose the Skill ID and Path
+
+- New built-in skill path: `skills/<skill-id>/SKILL.md`
+- Keep `<skill-id>` short, hyphen-case, and under 64 characters.
+- Use a verb-led or domain-led name that makes the trigger obvious, such as
+  `transfer-failed-retry`, `moviepilot-api`, or `create-moviepilot-skill`.
+
+### Step 4: Write Frontmatter Correctly
+
+Use this shape:
+
+```markdown
+---
+name: create-moviepilot-skill
+version: 1
+description: >-
+  Explain what the skill does and exactly when to use it.
+allowed-tools: list_directory read_file write_file edit_file execute_command
+---
+```
+
+Rules:
+
+- `description` is the primary trigger surface. Put concrete "when to use"
+  scenarios there.
+- Include `version` for built-in skills. Increment it whenever you ship a new
+  built-in revision.
+- Add `allowed-tools` when the workflow depends on a small, well-defined tool
+  set.
+- Add `compatibility` only when environment constraints actually matter.
+
+### Step 5: Write the Body
+
+The body should contain:
+
+- A short purpose statement
+- MoviePilot-specific rules or guardrails
+- A step-by-step workflow
+- Concrete examples of matching user requests
+- References to supporting files when they exist
+
+Prefer:
+
+- Imperative instructions
+- Concrete file paths
+- Examples aligned with actual MoviePilot conventions
+
+Avoid:
+
+- Generic theory that does not change execution
+- Large duplicated documentation
+- Extra files like `README.md` or `CHANGELOG.md` inside the skill directory
+
+### Step 6: Add Supporting Files Only When They Help
+
+- Add `scripts/` only when the same deterministic work would otherwise be
+  rewritten repeatedly.
+- Keep helper files inside the same skill directory.
+- Reference helper paths explicitly from `SKILL.md`.
+- If the skill is instructions-only, keep it to a single `SKILL.md`.
+
+### Step 7: Implement the Skill
+
+For a new built-in skill:
+
+1. Create `skills/<skill-id>/`
+2. Create `SKILL.md`
+3. Add helper scripts only if they are justified
+
+For an existing built-in skill:
+
+1. Edit `skills/<skill-id>/SKILL.md`
+2. Increment `version`
+3. Update helper files in the same directory if needed
+
+### Step 8: Validate Before Finishing
+
+- Re-read the frontmatter and confirm `name` matches the directory name.
+- Confirm `description` mentions real trigger scenarios.
+- If you changed an existing built-in skill, confirm `version` increased.
+- If possible, validate the file can be parsed by the MoviePilot skills loader.
+- Report the final path and note whether the agent needs a restart to sync the
+  latest built-in skill into `config/agent/skills`.
+
+## Minimal Example
+
+User request:
+
+`给 MoviePilot agent 加一个处理站点 Cookie 更新的内置技能`
+
+Expected outcome:
+
+- Create or update a directory such as `skills/update-site-cookie/`
+- Write `SKILL.md` with a trigger-rich `description`
+- Include only the tools needed for that workflow
+- Increment `version` when revising an existing built-in skill
+
+## Final Checklist
+
+- Is the skill under the repository `skills/` directory?
+- Does the folder name equal frontmatter `name`?
+- Does `description` clearly say when the skill should trigger?
+- Did you avoid duplicating an existing skill unnecessarily?
+- Did you increment `version` for built-in skill updates?
+- Did you keep the skill lean and procedural?
diff --git a/tests/test_skills_command.py b/tests/test_skills_command.py
new file mode 100644
index 00000000..212fec0d
--- /dev/null
+++ b/tests/test_skills_command.py
@@ -0,0 +1,183 @@
+import io
+import sys
+import tempfile
+import unittest
+import zipfile
+from pathlib import Path
+from types import ModuleType
+from unittest.mock import patch
+
+sys.modules.setdefault("qbittorrentapi", ModuleType("qbittorrentapi"))
+setattr(sys.modules["qbittorrentapi"], "TorrentFilesList", list)
+sys.modules.setdefault("transmission_rpc", ModuleType("transmission_rpc"))
+setattr(sys.modules["transmission_rpc"], "File", object)
+sys.modules.setdefault("psutil", ModuleType("psutil"))
+
+from app.chain.message import MessageChain
+from app.chain.skills import SkillsChain, skills_interaction_manager
+from app.helper.skill import SkillHelper, SkillInfo
+from app.schemas.types import MessageChannel
+
+
+def _build_skill_zip(skill_dir: str, skill_name: str) -> bytes:
+    buf = io.BytesIO()
+    with zipfile.ZipFile(buf, "w") as zf:
+        zf.writestr(
+            f"demo-main/{skill_dir}/SKILL.md",
+            (
+                f"---\n"
+                f"name: {skill_name}\n"
+                f"version: 1\n"
+                f"description: demo skill\n"
+                f"---\n\n"
+                f"# {skill_name}\n"
+            ),
+        )
+        zf.writestr(f"demo-main/{skill_dir}/scripts/example.py", "print('ok')\n")
+    return buf.getvalue()
+
+
+class TestSkillsCommand(unittest.TestCase):
+    def tearDown(self):
+        skills_interaction_manager.clear()
+
+    def test_message_routes_text_reply_to_skills_interaction_before_ai(self):
+        chain = MessageChain()
+        skills_interaction_manager.create_or_replace(
+            user_id="10001",
+            channel=MessageChannel.Wechat,
+            source="wechat-test",
+            username="tester",
+        )
+
+        with patch.object(chain, "_record_user_message"), patch(
+            "app.chain.message.SkillsChain.handle_text_interaction",
+            return_value=True,
+        ) as handle_text, patch.object(chain, "_handle_ai_message") as handle_ai:
+            chain.handle_message(
+                channel=MessageChannel.Wechat,
+                source="wechat-test",
+                userid="10001",
+                username="tester",
+                text="2",
+            )
+
+        handle_text.assert_called_once()
+        handle_ai.assert_not_called()
+
+    def test_callback_routes_to_skills_chain(self):
+        chain = MessageChain()
+        request = skills_interaction_manager.create_or_replace(
+            user_id="10001",
+            channel=MessageChannel.Telegram,
+            source="telegram-test",
+            username="tester",
+        )
+
+        with patch(
+            "app.chain.message.SkillsChain.handle_callback_interaction",
+            return_value=True,
+        ) as handle_callback:
+            chain._handle_callback(
+                text=f"CALLBACK:skills:{request.request_id}:market",
+                channel=MessageChannel.Telegram,
+                source="telegram-test",
+                userid="10001",
+                username="tester",
+            )
+
+        handle_callback.assert_called_once()
+
+    def test_skillhelper_install_and_remove_market_skill(self):
+        helper = SkillHelper()
+        skill = SkillInfo(
+            id="demo-skill",
+            name="demo-skill",
+            description="demo",
+            source_type="market",
+            source_label="市场 · acme/demo",
+            repo_url="https://github.com/acme/demo",
+            repo_name="acme/demo",
+            skill_path="skills/demo-skill",
+        )
+        zip_bytes = _build_skill_zip("skills/demo-skill", "demo-skill")
+
+        with tempfile.TemporaryDirectory() as tempdir:
+            user_root = Path(tempdir) / "user-skills"
+            bundled_root = Path(tempdir) / "bundled-skills"
+            user_root.mkdir(parents=True, exist_ok=True)
+            bundled_root.mkdir(parents=True, exist_ok=True)
+
+            with patch.object(
+                SkillHelper, "get_user_skills_dir", return_value=user_root
+            ), patch.object(
+                SkillHelper, "get_bundled_skills_dir", return_value=bundled_root
+            ), patch.object(
+                helper, "_download_repo_archive", return_value=zip_bytes
+            ):
+                success, message = helper.install_market_skill(skill)
+                self.assertTrue(success, message)
+                self.assertTrue((user_root / "demo-skill" / "SKILL.md").exists())
+                self.assertTrue(
+                    (user_root / "demo-skill" / ".moviepilot-skill-source.json").exists()
+                )
+
+                local_skills = helper.list_local_skills()
+                self.assertEqual(len(local_skills), 1)
+                self.assertEqual(local_skills[0].source_type, "market")
+                self.assertTrue(local_skills[0].removable)
+
+                removed, remove_message = helper.remove_local_skill("demo-skill")
+                self.assertTrue(removed, remove_message)
+                self.assertFalse((user_root / "demo-skill").exists())
+
+                bundled_skill_dir = bundled_root / "builtin-skill"
+                bundled_skill_dir.mkdir(parents=True, exist_ok=True)
+                (bundled_skill_dir / "SKILL.md").write_text(
+                    "---\nname: builtin-skill\ndescription: builtin\n---\n",
+                    encoding="utf-8",
+                )
+                installed_builtin = user_root / "builtin-skill"
+                installed_builtin.mkdir(parents=True, exist_ok=True)
+                (installed_builtin / "SKILL.md").write_text(
+                    "---\nname: builtin-skill\ndescription: builtin\n---\n",
+                    encoding="utf-8",
+                )
+
+                removed, remove_message = helper.remove_local_skill("builtin-skill")
+                self.assertFalse(removed)
+                self.assertIn("内置技能", remove_message)
+
+    def test_skills_chain_updates_buttons_via_edit_message(self):
+        chain = SkillsChain()
+        buttons = [[{"text": "安装 1", "callback_data": "skills:req:install:1"}]]
+
+        with patch.object(chain, "edit_message", return_value=True) as edit_message, patch.object(
+            chain, "post_message"
+        ) as post_message:
+            chain._update_or_post_message(
+                channel=MessageChannel.Telegram,
+                source="telegram-test",
+                userid="10001",
+                username="tester",
+                title="技能市场",
+                text="请选择技能",
+                buttons=buttons,
+                original_message_id=123,
+                original_chat_id="456",
+            )
+
+        edit_message.assert_called_once_with(
+            channel=MessageChannel.Telegram,
+            source="telegram-test",
+            message_id=123,
+            chat_id="456",
+            title="技能市场",
+            text="请选择技能",
+            buttons=buttons,
+        )
+        post_message.assert_not_called()
+
+
+if __name__ == "__main__":
+    unittest.main()