feat: cloud skills manage

agent/protocol/agent.py

@@ -1,4 +1,5 @@
 import json
+import os
 import time
 import threading
 
@@ -61,7 +62,8 @@ class Agent:
                 # Auto-create skill manager
                 try:
                     from agent.skills import SkillManager
-                    self.skill_manager = SkillManager(workspace_dir=workspace_dir)
+                    custom_dir = os.path.join(workspace_dir, "skills") if workspace_dir else None
+                    self.skill_manager = SkillManager(custom_dir=custom_dir)
                     logger.debug(f"Initialized SkillManager with {len(self.skill_manager.skills)} skills")
                 except Exception as e:
                     logger.warning(f"Failed to initialize SkillManager: {e}")

agent/skills/__init__.py

@@ -15,6 +15,7 @@ from agent.skills.types import (
 )
 from agent.skills.loader import SkillLoader
 from agent.skills.manager import SkillManager
+from agent.skills.service import SkillService
 from agent.skills.formatter import format_skills_for_prompt
 
 __all__ = [
@@ -25,5 +26,6 @@ __all__ = [
     "LoadSkillsResult",
     "SkillLoader",
     "SkillManager",
+    "SkillService",
     "format_skills_for_prompt",
 ]

agent/skills/loader.py

@@ -12,25 +12,20 @@ from agent.skills.frontmatter import parse_frontmatter, parse_metadata, parse_bo
 
 class SkillLoader:
     """Loads skills from various directories."""
-    
+
-    def __init__(self, workspace_dir: Optional[str] = None):
+    def __init__(self):
-        """
+        pass
-        Initialize the skill loader.
-        
-        :param workspace_dir: Agent workspace directory (for workspace-specific skills)
-        """
-        self.workspace_dir = workspace_dir
     
     def load_skills_from_dir(self, dir_path: str, source: str) -> LoadSkillsResult:
         """
         Load skills from a directory.
-        
+
         Discovery rules:
         - Direct .md files in the root directory
         - Recursive SKILL.md files under subdirectories
-        
+
         :param dir_path: Directory path to scan
-        :param source: Source identifier (e.g., 'managed', 'workspace', 'bundled')
+        :param source: Source identifier ('builtin' or 'custom')
         :return: LoadSkillsResult with skills and diagnostics
         """
         skills = []
@@ -216,61 +211,49 @@ class SkillLoader:
     
     def load_all_skills(
         self,
-        managed_dir: Optional[str] = None,
+        builtin_dir: Optional[str] = None,
-        workspace_skills_dir: Optional[str] = None,
+        custom_dir: Optional[str] = None,
-        extra_dirs: Optional[List[str]] = None,
     ) -> Dict[str, SkillEntry]:
         """
-        Load skills from all configured locations with precedence.
+        Load skills from builtin and custom directories.
-        
+
         Precedence (lowest to highest):
-        1. Extra directories
+        1. builtin  — project root ``skills/``, shipped with the codebase
-        2. Managed skills directory
+        2. custom   — workspace ``skills/``, installed via cloud console or skill creator
-        3. Workspace skills directory
+
-        
+        Same-name custom skills override builtin ones.
-        :param managed_dir: Managed skills directory (e.g., ~/.cow/skills)
+
-        :param workspace_skills_dir: Workspace skills directory (e.g., workspace/skills)
+        :param builtin_dir: Built-in skills directory
-        :param extra_dirs: Additional directories to load skills from
+        :param custom_dir: Custom skills directory
         :return: Dictionary mapping skill name to SkillEntry
         """
         skill_map: Dict[str, SkillEntry] = {}
         all_diagnostics = []
-        
+
-        # Load from extra directories (lowest precedence)
+        # Load builtin skills (lower precedence)
-        if extra_dirs:
+        if builtin_dir and os.path.exists(builtin_dir):
-            for extra_dir in extra_dirs:
+            result = self.load_skills_from_dir(builtin_dir, source='builtin')
-                if not os.path.exists(extra_dir):
-                    continue
-                result = self.load_skills_from_dir(extra_dir, source='extra')
-                all_diagnostics.extend(result.diagnostics)
-                for skill in result.skills:
-                    entry = self._create_skill_entry(skill)
-                    skill_map[skill.name] = entry
-        
-        # Load from managed directory
-        if managed_dir and os.path.exists(managed_dir):
-            result = self.load_skills_from_dir(managed_dir, source='managed')
             all_diagnostics.extend(result.diagnostics)
             for skill in result.skills:
                 entry = self._create_skill_entry(skill)
                 skill_map[skill.name] = entry
-        
+
-        # Load from workspace directory (highest precedence)
+        # Load custom skills (higher precedence, overrides builtin)
-        if workspace_skills_dir and os.path.exists(workspace_skills_dir):
+        if custom_dir and os.path.exists(custom_dir):
-            result = self.load_skills_from_dir(workspace_skills_dir, source='workspace')
+            result = self.load_skills_from_dir(custom_dir, source='custom')
             all_diagnostics.extend(result.diagnostics)
             for skill in result.skills:
                 entry = self._create_skill_entry(skill)
                 skill_map[skill.name] = entry
-        
+
         # Log diagnostics
         if all_diagnostics:
             logger.debug(f"Skill loading diagnostics: {len(all_diagnostics)} issues")
-            for diag in all_diagnostics[:5]:  # Log first 5
+            for diag in all_diagnostics[:5]:
                 logger.debug(f"  - {diag}")
-        
+
-        logger.debug(f"Loaded {len(skill_map)} skills from all sources")
+        logger.debug(f"Loaded {len(skill_map)} skills total")
-        
+
         return skill_map
     
     def _create_skill_entry(self, skill: Skill) -> SkillEntry:

agent/skills/manager.py

@@ -3,6 +3,7 @@ Skill manager for managing skill lifecycle and operations.
 """
 
 import os
+import json
 from typing import Dict, List, Optional
 from pathlib import Path
 from common.log import logger
@@ -10,56 +11,131 @@ from agent.skills.types import Skill, SkillEntry, SkillSnapshot
 from agent.skills.loader import SkillLoader
 from agent.skills.formatter import format_skill_entries_for_prompt
 
+SKILLS_CONFIG_FILE = "skills_config.json"
+
 
 class SkillManager:
     """Manages skills for an agent."""
-    
+
     def __init__(
         self,
-        workspace_dir: Optional[str] = None,
+        builtin_dir: Optional[str] = None,
-        managed_skills_dir: Optional[str] = None,
+        custom_dir: Optional[str] = None,
-        extra_dirs: Optional[List[str]] = None,
         config: Optional[Dict] = None,
     ):
         """
         Initialize the skill manager.
-        
+
-        :param workspace_dir: Agent workspace directory
+        :param builtin_dir: Built-in skills directory (project root ``skills/``)
-        :param managed_skills_dir: Managed skills directory (e.g., ~/.cow/skills)
+        :param custom_dir: Custom skills directory (workspace ``skills/``)
-        :param extra_dirs: Additional skill directories
         :param config: Configuration dictionary
         """
-        self.workspace_dir = workspace_dir
+        project_root = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
-        self.managed_skills_dir = managed_skills_dir or self._get_default_managed_dir()
+        self.builtin_dir = builtin_dir or os.path.join(project_root, 'skills')
-        self.extra_dirs = extra_dirs or []
+        self.custom_dir = custom_dir or os.path.join(project_root, 'workspace', 'skills')
         self.config = config or {}
-        
+        self._skills_config_path = os.path.join(self.custom_dir, SKILLS_CONFIG_FILE)
-        self.loader = SkillLoader(workspace_dir=workspace_dir)
+
+        # skills_config: full skill metadata keyed by name
+        # { "web-fetch": {"name": ..., "description": ..., "source": ..., "enabled": true}, ... }
+        self.skills_config: Dict[str, dict] = {}
+
+        self.loader = SkillLoader()
         self.skills: Dict[str, SkillEntry] = {}
-        
+
         # Load skills on initialization
         self.refresh_skills()
-    
+
-    def _get_default_managed_dir(self) -> str:
-        """Get the default managed skills directory."""
-        # Use project root skills directory as default
-        import os
-        project_root = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
-        return os.path.join(project_root, 'skills')
-    
     def refresh_skills(self):
-        """Reload all skills from configured directories."""
+        """Reload all skills from builtin and custom directories, then sync config."""
-        workspace_skills_dir = None
-        if self.workspace_dir:
-            workspace_skills_dir = os.path.join(self.workspace_dir, 'skills')
-        
         self.skills = self.loader.load_all_skills(
-            managed_dir=self.managed_skills_dir,
+            builtin_dir=self.builtin_dir,
-            workspace_skills_dir=workspace_skills_dir,
+            custom_dir=self.custom_dir,
-            extra_dirs=self.extra_dirs,
         )
-        
+        self._sync_skills_config()
         logger.debug(f"SkillManager: Loaded {len(self.skills)} skills")
+
+    # ------------------------------------------------------------------
+    # skills_config.json management
+    # ------------------------------------------------------------------
+    def _load_skills_config(self) -> Dict[str, dict]:
+        """Load skills_config.json from custom_dir. Returns empty dict if not found."""
+        if not os.path.exists(self._skills_config_path):
+            return {}
+        try:
+            with open(self._skills_config_path, "r", encoding="utf-8") as f:
+                data = json.load(f)
+            if isinstance(data, dict):
+                return data
+        except Exception as e:
+            logger.warning(f"[SkillManager] Failed to load {SKILLS_CONFIG_FILE}: {e}")
+        return {}
+
+    def _save_skills_config(self):
+        """Persist skills_config to custom_dir/skills_config.json."""
+        os.makedirs(self.custom_dir, exist_ok=True)
+        try:
+            with open(self._skills_config_path, "w", encoding="utf-8") as f:
+                json.dump(self.skills_config, f, indent=4, ensure_ascii=False)
+        except Exception as e:
+            logger.error(f"[SkillManager] Failed to save {SKILLS_CONFIG_FILE}: {e}")
+
+    def _sync_skills_config(self):
+        """
+        Merge directory-scanned skills with the persisted config file.
+
+        - New skills discovered on disk are added with enabled=True.
+        - Skills that no longer exist on disk are removed.
+        - Existing entries preserve their enabled state; name/description/source
+          are refreshed from the latest scan.
+        """
+        saved = self._load_skills_config()
+        merged: Dict[str, dict] = {}
+
+        for name, entry in self.skills.items():
+            skill = entry.skill
+            prev = saved.get(name, {})
+            merged[name] = {
+                "name": name,
+                "description": skill.description,
+                "source": skill.source,
+                "enabled": prev.get("enabled", True),
+            }
+
+        self.skills_config = merged
+        self._save_skills_config()
+
+    def is_skill_enabled(self, name: str) -> bool:
+        """
+        Check if a skill is enabled according to skills_config.
+
+        :param name: skill name
+        :return: True if enabled (default True if not in config)
+        """
+        entry = self.skills_config.get(name)
+        if entry is None:
+            return True
+        return entry.get("enabled", True)
+
+    def set_skill_enabled(self, name: str, enabled: bool):
+        """
+        Set a skill's enabled state and persist.
+
+        :param name: skill name
+        :param enabled: True to enable, False to disable
+        """
+        if name not in self.skills_config:
+            raise ValueError(f"skill '{name}' not found in config")
+        self.skills_config[name]["enabled"] = enabled
+        self._save_skills_config()
+
+    def get_skills_config(self) -> Dict[str, dict]:
+        """
+        Return the full skills_config dict (for query API).
+
+        :return: copy of skills_config
+        """
+        return dict(self.skills_config)
     
     def get_skill(self, name: str) -> Optional[SkillEntry]:
         """
@@ -85,25 +161,24 @@ class SkillManager:
     ) -> List[SkillEntry]:
         """
         Filter skills based on criteria.
-        
+
         Simple rule: Skills are auto-enabled if requirements are met.
-        - Has required API keys → included
+        - Has required API keys -> included
-        - Missing API keys → excluded
+        - Missing API keys -> excluded
-        
+
         :param skill_filter: List of skill names to include (None = all)
-        :param include_disabled: Whether to include skills with disable_model_invocation=True
+        :param include_disabled: Whether to include disabled skills
         :return: Filtered list of skill entries
         """
         from agent.skills.config import should_include_skill
-        
+
         entries = list(self.skills.values())
-        
+
         # Check requirements (platform, binaries, env vars)
         entries = [e for e in entries if should_include_skill(e, self.config)]
-        
+
         # Apply skill filter
         if skill_filter is not None:
-            # Flatten and normalize skill names (handle both strings and nested lists)
             normalized = []
             for item in skill_filter:
                 if isinstance(item, str):
@@ -111,20 +186,18 @@ class SkillManager:
                     if name:
                         normalized.append(name)
                 elif isinstance(item, list):
-                    # Handle nested lists
                     for subitem in item:
                         if isinstance(subitem, str):
                             name = subitem.strip()
                             if name:
                                 normalized.append(name)
-            
             if normalized:
                 entries = [e for e in entries if e.skill.name in normalized]
-        
+
-        # Filter out disabled skills unless explicitly requested
+        # Filter out disabled skills based on skills_config.json
         if not include_disabled:
-            entries = [e for e in entries if not e.skill.disable_model_invocation]
+            entries = [e for e in entries if self.is_skill_enabled(e.skill.name)]
-        
+
         return entries
     
     def build_skills_prompt(

agent/skills/service.py

@@ -0,0 +1,204 @@
+"""
+Skill service for handling skill CRUD operations.
+
+This service provides a unified interface for managing skills, which can be
+called from the cloud control client (LinkAI), the local web console, or any
+other management entry point.
+"""
+
+import os
+import shutil
+from typing import Dict, List, Optional
+from common.log import logger
+from agent.skills.types import Skill, SkillEntry
+from agent.skills.manager import SkillManager
+
+try:
+    import requests
+except ImportError:
+    requests = None
+
+
+class SkillService:
+    """
+    High-level service for skill lifecycle management.
+    Wraps SkillManager and provides network-aware operations such as
+    downloading skill files from remote URLs.
+    """
+
+    def __init__(self, skill_manager: SkillManager):
+        """
+        :param skill_manager: The SkillManager instance to operate on
+        """
+        self.manager = skill_manager
+
+    # ------------------------------------------------------------------
+    # query
+    # ------------------------------------------------------------------
+    def query(self) -> List[dict]:
+        """
+        Query all skills and return a serialisable list.
+        Reads from skills_config.json (refreshes from disk if needed).
+
+        :return: list of skill info dicts
+        """
+        self.manager.refresh_skills()
+        config = self.manager.get_skills_config()
+        result = list(config.values())
+        logger.info(f"[SkillService] query: {len(result)} skills found")
+        return result
+
+    # ------------------------------------------------------------------
+    # add / install
+    # ------------------------------------------------------------------
+    def add(self, payload: dict) -> None:
+        """
+        Add (install) a skill from a remote payload.
+
+        The payload follows the socket protocol::
+
+            {
+                "name": "web_search",
+                "type": "url",
+                "enabled": true,
+                "files": [
+                    {"url": "https://...", "path": "README.md"},
+                    {"url": "https://...", "path": "scripts/main.py"}
+                ]
+            }
+
+        Files are downloaded and saved under the custom skills directory
+        using *name* as the sub-directory.
+
+        :param payload: skill add payload from server
+        """
+        name = payload.get("name")
+        if not name:
+            raise ValueError("skill name is required")
+
+        files = payload.get("files", [])
+        if not files:
+            raise ValueError("skill files list is empty")
+
+        skill_dir = os.path.join(self.manager.custom_dir, name)
+        os.makedirs(skill_dir, exist_ok=True)
+
+        for file_info in files:
+            url = file_info.get("url")
+            rel_path = file_info.get("path")
+            if not url or not rel_path:
+                logger.warning(f"[SkillService] add: skip invalid file entry {file_info}")
+                continue
+            dest = os.path.join(skill_dir, rel_path)
+            self._download_file(url, dest)
+
+        # Reload to pick up the new skill and sync config
+        self.manager.refresh_skills()
+        logger.info(f"[SkillService] add: skill '{name}' installed ({len(files)} files)")
+
+    # ------------------------------------------------------------------
+    # open / close (enable / disable)
+    # ------------------------------------------------------------------
+    def open(self, payload: dict) -> None:
+        """
+        Enable a skill by name.
+
+        :param payload: {"name": "skill_name"}
+        """
+        name = payload.get("name")
+        if not name:
+            raise ValueError("skill name is required")
+        self.manager.set_skill_enabled(name, enabled=True)
+        logger.info(f"[SkillService] open: skill '{name}' enabled")
+
+    def close(self, payload: dict) -> None:
+        """
+        Disable a skill by name.
+
+        :param payload: {"name": "skill_name"}
+        """
+        name = payload.get("name")
+        if not name:
+            raise ValueError("skill name is required")
+        self.manager.set_skill_enabled(name, enabled=False)
+        logger.info(f"[SkillService] close: skill '{name}' disabled")
+
+    # ------------------------------------------------------------------
+    # delete
+    # ------------------------------------------------------------------
+    def delete(self, payload: dict) -> None:
+        """
+        Delete a skill by removing its directory entirely.
+
+        :param payload: {"name": "skill_name"}
+        """
+        name = payload.get("name")
+        if not name:
+            raise ValueError("skill name is required")
+
+        skill_dir = os.path.join(self.manager.custom_dir, name)
+        if os.path.exists(skill_dir):
+            shutil.rmtree(skill_dir)
+            logger.info(f"[SkillService] delete: removed directory {skill_dir}")
+        else:
+            logger.warning(f"[SkillService] delete: skill directory not found: {skill_dir}")
+
+        # Refresh will remove the deleted skill from config automatically
+        self.manager.refresh_skills()
+        logger.info(f"[SkillService] delete: skill '{name}' deleted")
+
+    # ------------------------------------------------------------------
+    # dispatch - single entry point for protocol messages
+    # ------------------------------------------------------------------
+    def dispatch(self, action: str, payload: Optional[dict] = None) -> dict:
+        """
+        Dispatch a skill management action and return a protocol-compatible
+        response dict.
+
+        :param action: one of query / add / open / close / delete
+        :param payload: action-specific payload (may be None for query)
+        :return: dict with action, code, message, payload
+        """
+        payload = payload or {}
+        try:
+            if action == "query":
+                result_payload = self.query()
+                return {"action": action, "code": 200, "message": "success", "payload": result_payload}
+            elif action == "add":
+                self.add(payload)
+            elif action == "open":
+                self.open(payload)
+            elif action == "close":
+                self.close(payload)
+            elif action == "delete":
+                self.delete(payload)
+            else:
+                return {"action": action, "code": 400, "message": f"unknown action: {action}", "payload": None}
+            return {"action": action, "code": 200, "message": "success", "payload": None}
+        except Exception as e:
+            logger.error(f"[SkillService] dispatch error: action={action}, error={e}")
+            return {"action": action, "code": 500, "message": str(e), "payload": None}
+
+    # ------------------------------------------------------------------
+    # internal helpers
+    # ------------------------------------------------------------------
+    @staticmethod
+    def _download_file(url: str, dest: str):
+        """
+        Download a file from *url* and save to *dest*.
+
+        :param url: remote file URL
+        :param dest: local destination path
+        """
+        if requests is None:
+            raise RuntimeError("requests library is required for downloading skill files")
+
+        dest_dir = os.path.dirname(dest)
+        if dest_dir:
+            os.makedirs(dest_dir, exist_ok=True)
+
+        resp = requests.get(url, timeout=60)
+        resp.raise_for_status()
+        with open(dest, "wb") as f:
+            f.write(resp.content)
+        logger.debug(f"[SkillService] downloaded {url} -> {dest}")

agent/skills/types.py

@@ -45,7 +45,7 @@ class Skill:
     description: str
     file_path: str
     base_dir: str
-    source: str  # managed, workspace, bundled, etc.
+    source: str  # builtin or custom
     content: str  # Full markdown content
     disable_model_invocation: bool = False
     frontmatter: Dict[str, Any] = field(default_factory=dict)

app.py

@@ -54,8 +54,8 @@ class ChannelManager:
 
                 if conf().get("use_linkai"):
                     try:
-                        from common import linkai_client
+                        from common import cloud_client
-                        threading.Thread(target=linkai_client.start, args=(channel, self), daemon=True).start()
+                        threading.Thread(target=cloud_client.start, args=(channel, self), daemon=True).start()
                     except Exception as e:
                         pass
 
@@ -64,7 +64,7 @@ class ChannelManager:
                 target=self._run_channel, args=(channel,), daemon=True
             )
             self._channel_thread.start()
-            logger.info(f"[ChannelManager] Channel '{channel_name}' started in sub-thread")
+            logger.debug(f"[ChannelManager] Channel '{channel_name}' started in sub-thread")
 
     def _run_channel(self, channel):
         try:

bridge/agent_bridge.py

@@ -28,7 +28,7 @@ def add_openai_compatible_support(bot_instance):
     """
     if hasattr(bot_instance, 'call_with_tools'):
         # Bot already has tool calling support (e.g., ZHIPUAIBot)
-        logger.info(f"[AgentBridge] {type(bot_instance).__name__} already has native tool calling support")
+        logger.debug(f"[AgentBridge] {type(bot_instance).__name__} already has native tool calling support")
         return bot_instance
 
     # Create a temporary mixin class that combines the bot with OpenAI compatibility

bridge/agent_initializer.py

@@ -291,7 +291,7 @@ class AgentInitializer:
         """Initialize skill manager"""
         try:
             from agent.skills import SkillManager
-            skill_manager = SkillManager(workspace_dir=workspace_root)
+            skill_manager = SkillManager(custom_dir=os.path.join(workspace_root, "skills"))
             return skill_manager
         except Exception as e:
             logger.warning(f"[AgentInitializer] Failed to initialize SkillManager: {e}")

channel/wechat/wechat_channel.py

@@ -151,7 +151,7 @@ class WechatChannel(ChatChannel):
 
     def exitCallback(self):
         try:
-            from common.linkai_client import chat_client
+            from common.cloud_client import chat_client
             if chat_client.client_id and conf().get("use_linkai"):
                 _send_logout()
                 time.sleep(2)
@@ -283,7 +283,7 @@ class WechatChannel(ChatChannel):
 
 def _send_login_success():
     try:
-        from common.linkai_client import chat_client
+        from common.cloud_client import chat_client
         if chat_client.client_id:
             chat_client.send_login_success()
     except Exception as e:
@@ -292,7 +292,7 @@ def _send_login_success():
 
 def _send_logout():
     try:
-        from common.linkai_client import chat_client
+        from common.cloud_client import chat_client
         if chat_client.client_id:
             chat_client.send_logout()
     except Exception as e:
@@ -301,7 +301,7 @@ def _send_logout():
 
 def _send_qr_code(qrcode_list: list):
     try:
-        from common.linkai_client import chat_client
+        from common.cloud_client import chat_client
         if chat_client.client_id:
             chat_client.send_qrcode(qrcode_list)
     except Exception as e:

common/cloud_client.py

@@ -1,3 +1,10 @@
+"""
+Cloud management client for connecting to the LinkAI control console.
+
+Handles remote configuration sync, message push, and skill management
+via the LinkAI socket protocol.
+"""
+
 from bridge.context import Context, ContextType
 from bridge.reply import Reply, ReplyType
 from common.log import logger
@@ -13,13 +20,34 @@ import os
 chat_client: LinkAIClient
 
 
-class ChatClient(LinkAIClient):
+class CloudClient(LinkAIClient):
-    def __init__(self, api_key, host, channel):
+    def __init__(self, api_key: str, channel, host: str = ""):
         super().__init__(api_key, host)
         self.channel = channel
         self.client_type = channel.channel_type
         self.channel_mgr = None
+        self._skill_service = None
 
+    @property
+    def skill_service(self):
+        """Lazy-init SkillService so it is available once SkillManager exists."""
+        if self._skill_service is None:
+            try:
+                from agent.skills.manager import SkillManager
+                from agent.skills.service import SkillService
+                from config import conf
+                from common.utils import expand_path
+                workspace_root = expand_path(conf().get("agent_workspace", "~/cow"))
+                manager = SkillManager(custom_dir=os.path.join(workspace_root, "skills"))
+                self._skill_service = SkillService(manager)
+                logger.debug("[CloudClient] SkillService initialised")
+            except Exception as e:
+                logger.error(f"[CloudClient] Failed to init SkillService: {e}")
+        return self._skill_service
+
+    # ------------------------------------------------------------------
+    # message push callback
+    # ------------------------------------------------------------------
     def on_message(self, push_msg: PushMsg):
         session_id = push_msg.session_id
         msg_content = push_msg.msg_content
@@ -30,21 +58,24 @@ class ChatClient(LinkAIClient):
         context["isgroup"] = push_msg.is_group
         self.channel.send(Reply(ReplyType.TEXT, content=msg_content), context)
 
+    # ------------------------------------------------------------------
+    # config callback
+    # ------------------------------------------------------------------
     def on_config(self, config: dict):
         if not self.client_id:
             return
-        logger.info(f"[LinkAI] 从客户端管理加载远程配置: {config}")
+        logger.info(f"[CloudClient] Loading remote config: {config}")
         if config.get("enabled") != "Y":
             return
 
         local_config = conf()
         need_restart_channel = False
-        
+
         for key in config.keys():
             if key in available_setting and config.get(key) is not None:
                 local_config[key] = config.get(key)
-        
+
-        # 语音配置
+        # Voice settings
         reply_voice_mode = config.get("reply_voice_mode")
         if reply_voice_mode:
             if reply_voice_mode == "voice_reply_voice":
@@ -60,16 +91,16 @@ class ChatClient(LinkAIClient):
         # Model configuration
         if config.get("model"):
             local_config["model"] = config.get("model")
-        
+
         # Channel configuration
         if config.get("channelType"):
             if local_config.get("channel_type") != config.get("channelType"):
                 local_config["channel_type"] = config.get("channelType")
                 need_restart_channel = True
-        
+
         # Channel-specific app credentials
         current_channel_type = local_config.get("channel_type", "")
-        
+
         if config.get("app_id") is not None:
             if current_channel_type == "feishu":
                 if local_config.get("feishu_app_id") != config.get("app_id"):
@@ -79,7 +110,7 @@ class ChatClient(LinkAIClient):
                 if local_config.get("dingtalk_client_id") != config.get("app_id"):
                     local_config["dingtalk_client_id"] = config.get("app_id")
                     need_restart_channel = True
-            elif current_channel_type == "wechatmp" or current_channel_type == "wechatmp_service":
+            elif current_channel_type in ("wechatmp", "wechatmp_service"):
                 if local_config.get("wechatmp_app_id") != config.get("app_id"):
                     local_config["wechatmp_app_id"] = config.get("app_id")
                     need_restart_channel = True
@@ -87,7 +118,7 @@ class ChatClient(LinkAIClient):
                 if local_config.get("wechatcomapp_agent_id") != config.get("app_id"):
                     local_config["wechatcomapp_agent_id"] = config.get("app_id")
                     need_restart_channel = True
-        
+
         if config.get("app_secret"):
             if current_channel_type == "feishu":
                 if local_config.get("feishu_app_secret") != config.get("app_secret"):
@@ -97,7 +128,7 @@ class ChatClient(LinkAIClient):
                 if local_config.get("dingtalk_client_secret") != config.get("app_secret"):
                     local_config["dingtalk_client_secret"] = config.get("app_secret")
                     need_restart_channel = True
-            elif current_channel_type == "wechatmp" or current_channel_type == "wechatmp_service":
+            elif current_channel_type in ("wechatmp", "wechatmp_service"):
                 if local_config.get("wechatmp_app_secret") != config.get("app_secret"):
                     local_config["wechatmp_app_secret"] = config.get("app_secret")
                     need_restart_channel = True
@@ -108,7 +139,7 @@ class ChatClient(LinkAIClient):
 
         if config.get("admin_password"):
             if not pconf("Godcmd"):
-                write_plugin_config({"Godcmd": {"password": config.get("admin_password"), "admin_users": []} })
+                write_plugin_config({"Godcmd": {"password": config.get("admin_password"), "admin_users": []}})
             else:
                 pconf("Godcmd")["password"] = config.get("admin_password")
             PluginManager().instances["GODCMD"].reload()
@@ -127,22 +158,46 @@ class ChatClient(LinkAIClient):
         elif config.get("text_to_image") and config.get("text_to_image") in ["dall-e-2", "dall-e-3"]:
             if pconf("linkai")["midjourney"]:
                 pconf("linkai")["midjourney"]["use_image_create_prefix"] = False
-        
+
         # Save configuration to config.json file
         self._save_config_to_file(local_config)
 
         if need_restart_channel:
             self._restart_channel(local_config.get("channel_type", ""))
-    
+
+    # ------------------------------------------------------------------
+    # skill callback
+    # ------------------------------------------------------------------
+    def on_skill(self, data: dict) -> dict:
+        """
+        Handle SKILL messages from the cloud console.
+        Delegates to SkillService.dispatch for the actual operations.
+
+        :param data: message data with 'action', 'clientId', 'payload'
+        :return: response dict
+        """
+        action = data.get("action", "")
+        payload = data.get("payload")
+        logger.info(f"[CloudClient] on_skill: action={action}")
+
+        svc = self.skill_service
+        if svc is None:
+            return {"action": action, "code": 500, "message": "SkillService not available", "payload": None}
+
+        return svc.dispatch(action, payload)
+
+    # ------------------------------------------------------------------
+    # channel restart helpers
+    # ------------------------------------------------------------------
     def _restart_channel(self, new_channel_type: str):
         """
         Restart the channel via ChannelManager when channel type changes.
         """
         if self.channel_mgr:
-            logger.info(f"[LinkAI] Restarting channel to '{new_channel_type}'...")
+            logger.info(f"[CloudClient] Restarting channel to '{new_channel_type}'...")
             threading.Thread(target=self._do_restart_channel, args=(self.channel_mgr, new_channel_type), daemon=True).start()
         else:
-            logger.warning("[LinkAI] ChannelManager not available, please restart the application manually")
+            logger.warning("[CloudClient] ChannelManager not available, please restart the application manually")
 
     def _do_restart_channel(self, mgr, new_channel_type: str):
         """
@@ -150,49 +205,49 @@ class ChatClient(LinkAIClient):
         """
         try:
             mgr.restart(new_channel_type)
-            # Update the linkai client's channel reference
+            # Update the client's channel reference
             if mgr.channel:
                 self.channel = mgr.channel
                 self.client_type = mgr.channel.channel_type
-                logger.info(f"[LinkAI] Channel reference updated to '{new_channel_type}'")
+                logger.info(f"[CloudClient] Channel reference updated to '{new_channel_type}'")
         except Exception as e:
-            logger.error(f"[LinkAI] Channel restart failed: {e}")
+            logger.error(f"[CloudClient] Channel restart failed: {e}")
 
+    # ------------------------------------------------------------------
+    # config persistence
+    # ------------------------------------------------------------------
     def _save_config_to_file(self, local_config: dict):
         """
-        Save configuration to config.json file
+        Save configuration to config.json file.
         """
         try:
             config_path = os.path.join(get_root(), "config.json")
             if not os.path.exists(config_path):
-                logger.warning(f"[LinkAI] config.json not found at {config_path}, skip saving")
+                logger.warning(f"[CloudClient] config.json not found at {config_path}, skip saving")
                 return
-            
+
-            # Read current config file
             with open(config_path, "r", encoding="utf-8") as f:
                 file_config = json.load(f)
-            
+
-            # Update file config with memory config
             file_config.update(dict(local_config))
-            
+
-            # Write back to file
             with open(config_path, "w", encoding="utf-8") as f:
                 json.dump(file_config, f, indent=4, ensure_ascii=False)
-            
+
-            logger.info("[LinkAI] Configuration saved to config.json successfully")
+            logger.info("[CloudClient] Configuration saved to config.json successfully")
         except Exception as e:
-            logger.error(f"[LinkAI] Failed to save configuration to config.json: {e}")
+            logger.error(f"[CloudClient] Failed to save configuration to config.json: {e}")
 
 
 def start(channel, channel_mgr=None):
     global chat_client
-    chat_client = ChatClient(api_key=conf().get("linkai_api_key"), channel=channel)
+    chat_client = CloudClient(api_key=conf().get("linkai_api_key"), host=conf().get("cloud_host", ""), channel=channel)
     chat_client.channel_mgr = channel_mgr
     chat_client.config = _build_config()
     chat_client.start()
     time.sleep(1.5)
     if chat_client.client_id:
-        logger.info("[LinkAI] 可前往控制台进行线上登录和配置：https://link-ai.tech/console/clients")
+        logger.info("[CloudClient] Console: https://link-ai.tech/console/clients")
 
 
 def _build_config():
@@ -214,20 +269,20 @@ def _build_config():
         "agent_max_context_turns": local_conf.get("agent_max_context_turns"),
         "agent_max_context_tokens": local_conf.get("agent_max_context_tokens"),
         "agent_max_steps": local_conf.get("agent_max_steps"),
-        "channelType": local_conf.get("channel_type")
+        "channelType": local_conf.get("channel_type"),
     }
-    
+
     if local_conf.get("always_reply_voice"):
         config["reply_voice_mode"] = "always_reply_voice"
     elif local_conf.get("voice_reply_voice"):
         config["reply_voice_mode"] = "voice_reply_voice"
-    
+
     if pconf("linkai"):
         config["group_app_map"] = pconf("linkai").get("group_app_map")
-    
+
     if plugin_config.get("Godcmd"):
         config["admin_password"] = plugin_config.get("Godcmd").get("password")
-    
+
     # Add channel-specific app credentials
     current_channel_type = local_conf.get("channel_type", "")
     if current_channel_type == "feishu":
@@ -236,11 +291,11 @@ def _build_config():
     elif current_channel_type == "dingtalk":
         config["app_id"] = local_conf.get("dingtalk_client_id")
         config["app_secret"] = local_conf.get("dingtalk_client_secret")
-    elif current_channel_type == "wechatmp" or current_channel_type == "wechatmp_service":
+    elif current_channel_type in ("wechatmp", "wechatmp_service"):
         config["app_id"] = local_conf.get("wechatmp_app_id")
         config["app_secret"] = local_conf.get("wechatmp_app_secret")
     elif current_channel_type == "wechatcom_app":
         config["app_id"] = local_conf.get("wechatcomapp_agent_id")
         config["app_secret"] = local_conf.get("wechatcomapp_secret")
-    
+
     return config