This commit is contained in:
X-iZhang
2026-02-06 22:50:32 +00:00
parent 5518f4b1f9
commit d23acd9f01
5 changed files with 250 additions and 208 deletions
+1 -1
View File
@@ -31,7 +31,7 @@ workspace/
skills/
memory/
.deno_cache/
*test.ipynb
*.ipynb
*CLAUDE.md
*AGENTS.md
*meals/
+53 -28
View File
@@ -26,9 +26,12 @@ from typing import Any, Optional
import typer # type: ignore[import-untyped]
from prompt_toolkit import PromptSession # type: ignore[import-untyped]
from prompt_toolkit.completion import Completer, Completion # type: ignore[import-untyped]
from prompt_toolkit.history import FileHistory # type: ignore[import-untyped]
from prompt_toolkit.auto_suggest import AutoSuggestFromHistory # type: ignore[import-untyped]
from prompt_toolkit.formatted_text import HTML # type: ignore[import-untyped]
from prompt_toolkit.shortcuts import CompleteStyle # type: ignore[import-untyped]
from prompt_toolkit.styles import Style as PtStyle # type: ignore[import-untyped]
from rich.panel import Panel # type: ignore[import-untyped]
from rich.text import Text # type: ignore[import-untyped]
from rich.table import Table # type: ignore[import-untyped]
@@ -200,17 +203,8 @@ def print_banner(
for line, color in zip(EVOSCIENTIST_ASCII_LINES, _GRADIENT_COLORS):
console.print(Text(line, style=f"{color} bold"))
info = Text()
info.append(" Thread: ", style="dim")
info.append(thread_id, style="yellow")
if workspace_dir:
info.append("\n Workspace: ", style="dim")
info.append(_shorten_path(workspace_dir), style="cyan")
if memory_dir:
trimmed = memory_dir.rstrip("/").rstrip("\\")
info.append("\n Memory dir: ", style="dim")
info.append(_shorten_path(trimmed), style="cyan")
if model or provider or mode:
info.append("\n ", style="dim")
info.append(" ", style="dim")
parts = []
if model:
parts.append(("Model: ", model))
@@ -223,20 +217,9 @@ def print_banner(
info.append(" ", style="dim")
info.append(label, style="dim")
info.append(value, style="magenta")
info.append("\n Commands: ", style="dim")
info.append("/exit", style="bold")
info.append(", ", style="dim")
info.append("/new", style="bold")
info.append(", ", style="dim")
info.append("/thread", style="bold")
info.append(", ", style="dim")
info.append("/skills", style="bold")
info.append(", ", style="dim")
info.append("/install-skill", style="bold")
info.append(", ", style="dim")
info.append("/uninstall-skill", style="bold")
info.append(", ", style="dim")
info.append("/mcp", style="bold")
info.append("\n Type ", style="#ffe082")
info.append("/", style="#ffe082 bold")
info.append(" for commands", style="#ffe082")
console.print(info)
console.print()
@@ -254,13 +237,13 @@ def _cmd_list_skills() -> None:
skills = list_skills(include_system=False)
if not skills:
console.print("[dim]No user skills installed.[/dim]")
console.print("[dim]No user-installed skills.[/dim]")
console.print("[dim]Install with:[/dim] /install-skill <path-or-url>")
console.print(f"[dim]Skills directory:[/dim] [cyan]{_shorten_path(str(USER_SKILLS_DIR))}[/cyan]")
console.print()
return
console.print(f"[bold]Installed Skills[/bold] ({len(skills)}):")
console.print(f"[bold]User-Installed Skills[/bold] ({len(skills)}):")
for skill in skills:
console.print(f" [green]{skill.name}[/green] - {skill.description}")
console.print(f"\n[dim]Location:[/dim] [cyan]{_shorten_path(str(USER_SKILLS_DIR))}[/cyan]")
@@ -620,6 +603,45 @@ def _auto_start_channel(agent: Any, thread_id: str, allowed_senders_csv: str) ->
_print_channel_panel([("iMessage", False, str(e))])
_SLASH_COMMANDS = [
("/thread", "Show thread ID, workspace & memory dir"),
("/new", "Start a new session"),
("/skills", "List installed skills"),
("/install-skill", "Add a skill from path or GitHub"),
("/uninstall-skill", "Remove an installed skill"),
("/mcp", "Manage MCP servers"),
("/channel", "Configure messaging channels"),
("/exit", "Quit EvoScientist"),
]
_COMPLETION_STYLE = PtStyle.from_dict({
"completion-menu": "bg:default noreverse nounderline noitalic",
"completion-menu.completion": "bg:default #888888 noreverse",
"completion-menu.completion.current": "bg:default default bold noreverse",
"completion-menu.meta.completion": "bg:default #888888 noreverse",
"completion-menu.meta.completion.current": "bg:default default bold noreverse",
"scrollbar.background": "bg:default",
"scrollbar.button": "bg:default",
})
class SlashCommandCompleter(Completer):
"""Autocomplete for slash commands — triggers when input starts with '/'."""
def get_completions(self, document, complete_event):
text = document.text_before_cursor
if not text.startswith("/"):
return
for cmd, desc in _SLASH_COMMANDS:
if cmd.startswith(text):
yield Completion(
cmd,
start_position=-len(text),
display=f"{cmd:<40}",
display_meta=desc,
)
def cmd_interactive(
agent: Any,
show_thinking: bool = True,
@@ -656,7 +678,10 @@ def cmd_interactive(
session = PromptSession(
history=FileHistory(history_file),
auto_suggest=AutoSuggestFromHistory(),
enable_history_search=True,
completer=SlashCommandCompleter(),
complete_style=CompleteStyle.COLUMN,
complete_while_typing=True,
style=_COMPLETION_STYLE,
)
def _print_separator():
@@ -785,7 +810,7 @@ def cmd_interactive(
while state["running"]:
try:
user_input = await session.prompt_async(
HTML('<ansiblue><b>&gt;</b></ansiblue> ')
HTML('<ansiblue><b>❯</b></ansiblue> ')
)
user_input = user_input.strip()
@@ -1,16 +1,18 @@
#!/usr/bin/env python3
"""Install a skill from GitHub into a local skills directory.
Self-contained installer — no external dependencies beyond git.
Thin CLI wrapper around EvoScientist.skills_manager — all clone, resolve,
and validation logic lives in the core module.
Usage examples:
# Install from a GitHub URL (auto-detects repo, ref, path)
python install_skill.py --url https://github.com/anthropics/skills/tree/main/excel
# Install from repo + path
python install_skill.py --repo anthropics/skills --path excel
# Install from shorthand
python install_skill.py --url anthropics/skills@excel
# Install multiple skills from the same repo
# Install from repo + path(s)
python install_skill.py --repo anthropics/skills --path excel
python install_skill.py --repo anthropics/skills --path excel --path pdf
# Install with a specific git ref
@@ -20,131 +22,20 @@ Usage examples:
from __future__ import annotations
import argparse
import os
import re
import shutil
import subprocess
import sys
import tempfile
from EvoScientist.skills_manager import install_skill, _parse_github_url
def parse_github_url(url: str) -> tuple[str, str | None, str | None]:
"""Parse a GitHub URL into (repo, ref, path).
Supports formats:
https://github.com/owner/repo
https://github.com/owner/repo/tree/main/path/to/skill
github.com/owner/repo/tree/branch/path
owner/repo@skill-name (shorthand from skills.sh)
Returns:
(repo, ref_or_none, path_or_none)
"""
# Shorthand: owner/repo@path
if "@" in url and "://" not in url:
repo, path = url.split("@", 1)
return repo.strip(), None, path.strip()
# Strip protocol and github.com prefix
cleaned = re.sub(r"^https?://", "", url)
cleaned = re.sub(r"^github\.com/", "", cleaned)
cleaned = cleaned.rstrip("/")
# Match: owner/repo/tree/ref/path...
m = re.match(r"^([^/]+/[^/]+)/tree/([^/]+)(?:/(.+))?$", cleaned)
if m:
return m.group(1), m.group(2), m.group(3)
# Match: owner/repo (no tree)
m = re.match(r"^([^/]+/[^/]+)$", cleaned)
if m:
return m.group(1), None, None
raise ValueError(f"Cannot parse GitHub URL: {url}")
def clone_repo(repo: str, ref: str | None, dest: str) -> None:
"""Shallow-clone a GitHub repo."""
clone_url = f"https://github.com/{repo}.git"
cmd = ["git", "clone", "--depth", "1"]
def _build_source(repo: str, ref: str | None, path: str | None) -> str:
"""Build a GitHub URL string that install_skill() can consume."""
if ref and path:
return f"https://github.com/{repo}/tree/{ref}/{path}"
if ref:
cmd += ["--branch", ref]
cmd += [clone_url, dest]
result = subprocess.run(cmd, capture_output=True, text=True)
if result.returncode != 0:
raise RuntimeError(f"git clone failed: {result.stderr.strip()}")
def copy_skill(src: str, dest_dir: str) -> str:
"""Copy a skill directory to the destination.
Returns:
The skill name (directory basename).
"""
skill_name = os.path.basename(src.rstrip("/"))
target = os.path.join(dest_dir, skill_name)
if os.path.exists(target):
shutil.rmtree(target)
print(f" Replaced existing: {skill_name}")
shutil.copytree(src, target)
return skill_name
def validate_skill(path: str) -> bool:
"""Check that a directory looks like a valid skill (has SKILL.md)."""
return os.path.isfile(os.path.join(path, "SKILL.md"))
def install(
repo: str,
paths: list[str],
ref: str | None,
dest: str,
) -> list[str]:
"""Install skill(s) from a GitHub repo.
Returns:
List of installed skill names.
"""
os.makedirs(dest, exist_ok=True)
installed: list[str] = []
with tempfile.TemporaryDirectory(prefix="skill-install-") as tmp:
clone_dir = os.path.join(tmp, "repo")
print(f"Cloning {repo}" + (f" @{ref}" if ref else "") + "...")
clone_repo(repo, ref, clone_dir)
if not paths:
# No path specified — treat entire repo as a single skill
if validate_skill(clone_dir):
name = copy_skill(clone_dir, dest)
installed.append(name)
else:
# List top-level directories that look like skills
for entry in sorted(os.listdir(clone_dir)):
entry_path = os.path.join(clone_dir, entry)
if os.path.isdir(entry_path) and validate_skill(entry_path):
name = copy_skill(entry_path, dest)
installed.append(name)
if not installed:
print("No valid skills found in repository root.", file=sys.stderr)
else:
for p in paths:
skill_path = os.path.join(clone_dir, p.strip("/"))
if not os.path.isdir(skill_path):
print(f" Path not found: {p}", file=sys.stderr)
continue
if not validate_skill(skill_path):
print(f" No SKILL.md in: {p}", file=sys.stderr)
continue
name = copy_skill(skill_path, dest)
installed.append(name)
return installed
return f"https://github.com/{repo}/tree/{ref}"
if path:
return f"{repo}@{path}"
return f"https://github.com/{repo}"
def main() -> int:
@@ -178,11 +69,10 @@ def main() -> int:
)
args = parser.parse_args()
dest = args.dest
# Parse source
if args.url:
repo, ref, path = parse_github_url(args.url)
repo, ref, path = _parse_github_url(args.url)
ref = args.ref or ref
paths = [path] if path else args.path
else:
@@ -190,22 +80,28 @@ def main() -> int:
ref = args.ref
paths = args.path
try:
installed = install(repo, paths, ref, dest)
except RuntimeError as e:
print(f"Error: {e}", file=sys.stderr)
return 1
# Install each path (or the whole repo if no paths given)
sources = [_build_source(repo, ref, p) for p in paths] if paths else [_build_source(repo, ref, None)]
installed = []
for source in sources:
print(f"Installing from: {source}")
result = install_skill(source, dest_dir=args.dest)
if result["success"]:
print(f" Installed: {result['name']} ({result.get('description', '')})")
installed.append(result["name"])
else:
print(f" Failed: {result['error']}", file=sys.stderr)
if installed:
print(f"\nInstalled {len(installed)} skill(s) to {dest}/:")
print(f"\nInstalled {len(installed)} skill(s) to {args.dest}/:")
for name in installed:
print(f" - {name}")
return 0
else:
print("No skills were installed.", file=sys.stderr)
return 1
return 0
if __name__ == "__main__":
raise SystemExit(main())
+93 -22
View File
@@ -77,6 +77,11 @@ def _parse_skill_md(skill_md_path: Path) -> dict[str, str]:
try:
frontmatter = yaml.safe_load(frontmatter_match.group(1))
if not isinstance(frontmatter, dict):
return {
"name": skill_md_path.parent.name,
"description": "(empty frontmatter)",
}
return {
"name": frontmatter.get("name", skill_md_path.parent.name),
"description": frontmatter.get("description", "(no description)"),
@@ -123,6 +128,9 @@ def _parse_github_url(url: str) -> tuple[str, str | None, str | None]:
raise ValueError(f"Cannot parse GitHub URL: {url}")
_CLONE_TIMEOUT = 120 # seconds
def _clone_repo(repo: str, ref: str | None, dest: str) -> None:
"""Shallow-clone a GitHub repo."""
clone_url = f"https://github.com/{repo}.git"
@@ -131,7 +139,10 @@ def _clone_repo(repo: str, ref: str | None, dest: str) -> None:
cmd += ["--branch", ref]
cmd += [clone_url, dest]
result = subprocess.run(cmd, capture_output=True, text=True)
try:
result = subprocess.run(cmd, capture_output=True, text=True, timeout=_CLONE_TIMEOUT)
except subprocess.TimeoutExpired:
raise RuntimeError(f"git clone timed out after {_CLONE_TIMEOUT}s for {repo}")
if result.returncode != 0:
raise RuntimeError(f"git clone failed: {result.stderr.strip()}")
@@ -159,6 +170,42 @@ def _validate_skill_dir(path: Path) -> bool:
return (path / "SKILL.md").is_file()
def _find_skill_in_tree(root: str, skill_name: str) -> Path | None:
"""Walk a directory tree to find a subdirectory named *skill_name* containing SKILL.md.
Skips hidden directories (starting with '.').
Returns:
The absolute Path to the skill directory, or None.
"""
for dirpath, dirnames, _files in os.walk(root):
# Prune hidden directories
dirnames[:] = [d for d in dirnames if not d.startswith(".")]
if os.path.basename(dirpath) == skill_name:
candidate = Path(dirpath)
if _validate_skill_dir(candidate):
return candidate
return None
# Allowed pattern for skill names: alphanumeric, hyphens, underscores
_VALID_SKILL_NAME = re.compile(r"^[a-zA-Z0-9][a-zA-Z0-9._-]*$")
def _sanitize_name(name: str) -> str | None:
"""Validate and sanitize a skill name.
Returns the cleaned name, or None if invalid.
"""
name = name.strip()
if not name or not _VALID_SKILL_NAME.match(name):
return None
# Block path traversal components
if ".." in name or "/" in name or "\\" in name:
return None
return name
def install_skill(source: str, dest_dir: str | None = None) -> dict:
"""Install a skill from a local path or GitHub URL.
@@ -197,10 +244,14 @@ def _install_from_local(source: str, dest_dir: str) -> dict:
# Parse SKILL.md to get the skill name
skill_info = _parse_skill_md(source_path / "SKILL.md")
skill_name = skill_info["name"]
skill_name = _sanitize_name(skill_info["name"])
if not skill_name:
return {"success": False, "error": f"Invalid skill name in SKILL.md: {skill_info['name']!r}"}
# Destination path
target_path = Path(dest_dir) / skill_name
# Destination path — resolve and verify it stays inside dest_dir
target_path = (Path(dest_dir) / skill_name).resolve()
if not str(target_path).startswith(str(Path(dest_dir).resolve())):
return {"success": False, "error": f"Skill name escapes destination: {skill_info['name']!r}"}
# Remove existing if present
if target_path.exists():
@@ -238,34 +289,48 @@ def _install_from_github(source: str, dest_dir: str) -> dict:
else:
skill_source = Path(clone_dir)
if not skill_source.exists():
return {"success": False, "error": f"Path not found in repo: {path or '/'}"}
if not _validate_skill_dir(skill_source):
# Maybe the repo root contains multiple skills?
if not path:
# Try to find skills in repo root
# Validate — if the direct path doesn't have SKILL.md, try auto-resolve
if not skill_source.exists() or not _validate_skill_dir(skill_source):
if path:
# The shorthand path (e.g. "canvas-design") may be nested deeper
# Walk the tree to find a directory with that name + SKILL.md
skill_name_hint = path.rstrip("/").rsplit("/", 1)[-1]
resolved = _find_skill_in_tree(clone_dir, skill_name_hint)
if resolved:
skill_source = resolved
else:
return {"success": False, "error": f"No SKILL.md found at '{path}' (also searched subdirectories) in: {source}"}
else:
# No path specified — list available skills in repo root
found_skills = []
for entry in os.listdir(clone_dir):
entry_path = Path(clone_dir) / entry
if entry_path.is_dir() and _validate_skill_dir(entry_path):
found_skills.append(entry)
if found_skills:
if len(found_skills) == 1:
# Only one skill in repo — just install it
skill_source = Path(clone_dir) / found_skills[0]
elif found_skills:
return {
"success": False,
"error": (
f"Multiple skills found in repo. "
f"Please specify one: {', '.join(found_skills)}"
f"Please specify one: {', '.join(sorted(found_skills))}"
),
}
return {"success": False, "error": f"No SKILL.md found in: {source}"}
else:
return {"success": False, "error": f"No SKILL.md found in: {source}"}
# Parse skill info and copy
skill_info = _parse_skill_md(skill_source / "SKILL.md")
skill_name = skill_info["name"]
target_path = Path(dest_dir) / skill_name
skill_name = _sanitize_name(skill_info["name"])
if not skill_name:
return {"success": False, "error": f"Invalid skill name in SKILL.md: {skill_info['name']!r}"}
target_path = (Path(dest_dir) / skill_name).resolve()
if not str(target_path).startswith(str(Path(dest_dir).resolve())):
return {"success": False, "error": f"Skill name escapes destination: {skill_info['name']!r}"}
if target_path.exists():
shutil.rmtree(target_path)
@@ -348,8 +413,14 @@ def uninstall_skill(name: str) -> dict:
- success: bool
- error: error message (if failed)
"""
user_dir = Path(USER_SKILLS_DIR)
target_path = user_dir / name
user_dir = Path(USER_SKILLS_DIR).resolve()
# Validate name to prevent path traversal
clean_name = _sanitize_name(name)
if not clean_name:
return {"success": False, "error": f"Invalid skill name: {name!r}"}
target_path = (user_dir / clean_name).resolve()
if not target_path.exists():
# Try to find by directory name (in case name differs from dir name)
@@ -358,15 +429,15 @@ def uninstall_skill(name: str) -> dict:
for entry in user_dir.iterdir():
if entry.is_dir() and _validate_skill_dir(entry):
info = _parse_skill_md(entry / "SKILL.md")
if info["name"] == name:
found = entry
if info["name"] == clean_name:
found = entry.resolve()
break
if not found:
return {"success": False, "error": f"Skill not found: {name}"}
target_path = found
# Check if it's a user skill (not system)
# Check resolved path is still inside user_dir
if not str(target_path).startswith(str(user_dir)):
return {"success": False, "error": f"Cannot uninstall system skill: {name}"}
+72 -22
View File
@@ -122,35 +122,51 @@ async def tavily_search(
@tool(parse_docstring=True)
def skill_manager(
action: Literal["install", "list", "uninstall"],
action: Literal["install", "list", "uninstall", "info"],
source: str = "",
name: str = "",
include_system: bool = False,
) -> str:
"""Manage user skills: install, list, or uninstall.
"""Manage user-installable skills: install from GitHub or local path, list available skills, get details, or uninstall.
Use this tool when the user asks to:
- Install a skill (action="install", source required)
- List installed skills (action="list")
- Uninstall a skill (action="uninstall", name required)
Actions and required parameters:
Supported sources for install:
- Local path: "./my-skill" or "/path/to/skill"
- GitHub URL: "https://github.com/owner/repo/tree/main/skill-name"
- GitHub shorthand: "owner/repo@skill-name"
action="install" (requires source):
Install a skill. The source can be:
- GitHub shorthand: "owner/repo@skill-name" (e.g. "anthropics/skills@peft")
- GitHub URL: "https://github.com/owner/repo/tree/main/skill-name"
- Local path: "./my-skill" or "/path/to/skill"
Nested skills are auto-resolved — if the skill is not at the repo root, subdirectories are searched automatically.
action="list":
List installed skills. By default only shows user-installed skills.
Set include_system=True to also show built-in system skills (peft, accelerate, flash-attention, etc.).
action="info" (requires name):
Get details (description, source, path) about a specific skill by name.
Searches both user and system skills.
action="uninstall" (requires name):
Remove a user-installed skill by name. System skills cannot be uninstalled.
Args:
action: One of "install", "list", or "uninstall"
source: For install - local path or GitHub URL/shorthand
name: For uninstall - skill name to remove
action: The operation to perform — "install", "list", "info", or "uninstall"
source: Required for install — GitHub shorthand, GitHub URL, or local directory path
name: Required for info and uninstall — the skill name (e.g. "peft", "my-custom-skill")
include_system: Only for list — set True to include built-in system skills in the output
Returns:
Result message
"""
from .skills_manager import install_skill, list_skills, uninstall_skill
from .skills_manager import install_skill, list_skills, uninstall_skill, get_skill_info
if action == "install":
if not source:
return "Error: 'source' is required for install action"
return (
"Error: 'source' is required for install action. "
"Provide a GitHub shorthand (e.g. source='owner/repo@skill-name'), "
"a GitHub URL, or a local directory path."
)
result = install_skill(source)
if result["success"]:
return (
@@ -163,25 +179,59 @@ def skill_manager(
return f"Failed to install skill: {result['error']}"
elif action == "list":
skills = list_skills(include_system=False)
skills = list_skills(include_system=include_system)
if not skills:
return "No user skills installed. Use action='install' to add skills."
lines = [f"Installed User Skills ({len(skills)}):"]
for skill in skills:
lines.append(f" - {skill.name}: {skill.description}")
if include_system:
return "No skills found."
return "No user skills installed. Use action='install' to add skills, or set include_system=True to see built-in skills."
user_skills = [s for s in skills if s.source == "user"]
system_skills = [s for s in skills if s.source == "system"]
lines = []
if user_skills:
lines.append(f"User Skills ({len(user_skills)}):")
for skill in user_skills:
lines.append(f" - {skill.name}: {skill.description}")
if system_skills:
if lines:
lines.append("")
lines.append(f"System Skills ({len(system_skills)}):")
for skill in system_skills:
lines.append(f" - {skill.name}: {skill.description}")
return "\n".join(lines)
elif action == "uninstall":
if not name:
return "Error: 'name' is required for uninstall action"
return (
"Error: 'name' is required for uninstall action. "
"Use action='list' first to see installed skill names."
)
result = uninstall_skill(name)
if result["success"]:
return f"Successfully uninstalled skill: {name}"
else:
return f"Failed to uninstall skill: {result['error']}"
elif action == "info":
if not name:
return (
"Error: 'name' is required for info action. "
"Use action='list' with include_system=True to see all available skill names."
)
info = get_skill_info(name)
if not info:
return (
f"Skill not found: {name}. "
f"Use action='list' with include_system=True to see all available skills."
)
return (
f"Name: {info.name}\n"
f"Description: {info.description}\n"
f"Source: {info.source}\n"
f"Path: {info.path}"
)
else:
return f"Unknown action: {action}. Use 'install', 'list', or 'uninstall'."
return f"Unknown action: {action}. Use 'install', 'list', 'uninstall', or 'info'."
@tool(parse_docstring=True)