Skills MCP

Instructions

Implementation Reference

• src/skills_mcp/server.py:54-63 (handler)

  The handler function for the 'skills_install' MCP tool. Includes registration via the @mcp.tool() decorator, input schema defined with Pydantic Field annotations, and logic that delegates to the local.install_skill helper.

  @mcp.tool()
  def skills_install(
      name: str = Field(description="Exact name of the skill"),
      force: bool = Field(default=False, description="Overwrite if exists")
  ) -> str:
      """Download and install a skill to the local environment."""
      try:
          return local.install_skill(name, force)
      except Exception as e:
          return f"Installation failed: {str(e)}"

• src/skills_mcp/local.py:46-125 (helper)

  Core helper function implementing the skill installation logic: checks if already installed, downloads ZIP from registry API, performs security validations (zip slip prevention, nested dir handling), extracts files, verifies SKILL.md presence, and handles errors with cleanup.

  def install_skill(name: str, force: bool = False) -> str:
      target_dir = config.root_dir / name
  
      if target_dir.exists():
          if not force:
              return f"Skill '{name}' is already installed at {target_dir}."
          else:
              shutil.rmtree(target_dir)
  
      # Download logic
      url = f"{client.base_url}/download/{name}"
  
      try:
          with httpx.stream("GET", url, headers=client.headers, timeout=30.0) as resp:
              if resp.status_code == 404:
                  raise RuntimeError(f"Skill '{name}' not found in registry.")
              resp.raise_for_status()
  
              # Download full content to memory (assuming zip files are small < 50MB)
              # For larger files, we should use a temporary file.
              data = io.BytesIO()
              for chunk in resp.iter_bytes():
                  data.write(chunk)
  
              with zipfile.ZipFile(data) as zf:
                  # Security Check: Prevent Zip Slip
                  for member in zf.namelist():
                      if ".." in member or member.startswith("/"):
                          raise RuntimeError("Malicious zip file detected.")
  
                  # 1. 检查是否存在顶层目录嵌套
                  # 改进后的判定逻辑：只要 SKILL.md 是在子目录里，就认为是嵌套的
                  skill_md_path = next((f for f in zf.namelist() if f.endswith("SKILL.md")), None)
  
                  if skill_md_path and "/" in skill_md_path:
                      # 例如 "docx/SKILL.md"
                      prefix = skill_md_path.split("/SKILL.md")[0] + "/"
                      is_nested = True
                  else:
                      prefix = ""
                      is_nested = False
  
                  target_dir.mkdir(parents=True, exist_ok=True)
  
                  if is_nested:
                      # 智能解压：去掉第一层目录
                      for member in zf.infolist():
                          if member.filename == prefix: continue # 跳过顶层目录本身
                          
                          # 去掉前缀
                          new_name = member.filename[len(prefix):]
                          if not new_name: continue
                          
                          target_path = target_dir / new_name
                          
                          if member.is_dir():
                              target_path.mkdir(parents=True, exist_ok=True)
                          else:
                              target_path.parent.mkdir(parents=True, exist_ok=True)
                              with zf.open(member, "r") as source, open(target_path, "wb") as target:
                                  shutil.copyfileobj(source, target)
                  else:
                      # 直接解压
                      zf.extractall(target_dir)
                  
                  # 最后的完整性检查
                  if not (target_dir / "SKILL.md").exists():
                       # 回滚
                       shutil.rmtree(target_dir)
                       raise RuntimeError("Invalid skill package: missing SKILL.md after extraction.")
  
      except httpx.HTTPStatusError as e:
          raise RuntimeError(f"Download failed: {e.response.text}")
      except Exception as e:
          # Cleanup partial install
          if target_dir.exists():
              shutil.rmtree(target_dir)
          raise e
  
      return f"Successfully installed '{name}' to {target_dir}."