644 lines
21 KiB
Python
644 lines
21 KiB
Python
"""Tool loading, saving, and management."""
|
|
|
|
import os
|
|
import shutil
|
|
import stat
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
from typing import Optional, List, Literal
|
|
|
|
import yaml
|
|
|
|
|
|
# Default tools directory
|
|
TOOLS_DIR = Path.home() / ".cmdforge"
|
|
|
|
# Default bin directory for wrapper scripts
|
|
BIN_DIR = Path.home() / ".local" / "bin"
|
|
|
|
|
|
@dataclass
|
|
class SystemDependency:
|
|
"""A system-level package dependency (apt, brew, pacman, etc.)."""
|
|
name: str
|
|
description: str = ""
|
|
binaries: List[str] = field(default_factory=list) # Executables to check for
|
|
packages: dict = field(default_factory=dict) # {apt: pkg, brew: pkg, ...}
|
|
_original_format: str = field(default="long", repr=False) # Track for round-trip serialization
|
|
|
|
def to_dict(self):
|
|
"""Serialize to dict or string (short form)."""
|
|
# Short form: if only name is set, return just the string
|
|
if self._original_format == "short" or (
|
|
not self.description and not self.binaries and not self.packages
|
|
):
|
|
return self.name
|
|
|
|
d = {"name": self.name}
|
|
if self.description:
|
|
d["description"] = self.description
|
|
if self.binaries:
|
|
d["binaries"] = self.binaries
|
|
if self.packages:
|
|
d["packages"] = self.packages
|
|
return d
|
|
|
|
@classmethod
|
|
def from_dict(cls, data) -> "SystemDependency":
|
|
"""Create from dict or string (short form)."""
|
|
if isinstance(data, str):
|
|
return cls(name=data, _original_format="short")
|
|
return cls(
|
|
name=data.get("name", ""),
|
|
description=data.get("description", ""),
|
|
binaries=data.get("binaries", []),
|
|
packages=data.get("packages", {}),
|
|
_original_format="long"
|
|
)
|
|
|
|
def get_binaries_to_check(self) -> List[str]:
|
|
"""Return binaries to check, defaulting to [name]."""
|
|
return self.binaries if self.binaries else [self.name]
|
|
|
|
def get_package_name(self, pkg_manager: str) -> str:
|
|
"""Return package name for a manager, defaulting to self.name."""
|
|
return self.packages.get(pkg_manager, self.name)
|
|
|
|
|
|
@dataclass
|
|
class ToolArgument:
|
|
"""Definition of a custom input argument."""
|
|
flag: str # e.g., "--max-size"
|
|
variable: str # e.g., "max_size"
|
|
default: Optional[str] = None
|
|
description: str = ""
|
|
|
|
def to_dict(self) -> dict:
|
|
d = {"flag": self.flag, "variable": self.variable}
|
|
if self.default:
|
|
d["default"] = self.default
|
|
if self.description:
|
|
d["description"] = self.description
|
|
return d
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: dict) -> "ToolArgument":
|
|
return cls(
|
|
flag=data["flag"],
|
|
variable=data["variable"],
|
|
default=data.get("default"),
|
|
description=data.get("description", "")
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class PromptStep:
|
|
"""A prompt step that calls an AI provider."""
|
|
prompt: str # The prompt template
|
|
provider: str # Provider name
|
|
output_var: str # Variable to store output
|
|
prompt_file: Optional[str] = None # Optional filename for external prompt
|
|
profile: Optional[str] = None # Optional AI persona profile name
|
|
name: Optional[str] = None # Optional display name for the step
|
|
strip_fences: bool = False # Strip markdown code fences from output
|
|
# Structured output fields
|
|
output_schema: Optional[dict] = None # JSON schema for validation (None = use default)
|
|
max_retries: int = 1 # Retry count on validation failure
|
|
plain_text: bool = False # Bypass structured output enforcement
|
|
max_tokens: Optional[int] = None # Max output tokens (provider-dependent)
|
|
|
|
def to_dict(self) -> dict:
|
|
d = {
|
|
"type": "prompt",
|
|
"prompt": self.prompt,
|
|
"provider": self.provider,
|
|
"output_var": self.output_var
|
|
}
|
|
if self.prompt_file:
|
|
d["prompt_file"] = self.prompt_file
|
|
if self.profile:
|
|
d["profile"] = self.profile
|
|
if self.name:
|
|
d["name"] = self.name
|
|
if self.strip_fences:
|
|
d["strip_fences"] = self.strip_fences
|
|
if self.output_schema:
|
|
d["output_schema"] = self.output_schema
|
|
if self.max_retries != 1:
|
|
d["max_retries"] = self.max_retries
|
|
if self.plain_text:
|
|
d["plain_text"] = self.plain_text
|
|
if self.max_tokens:
|
|
d["max_tokens"] = self.max_tokens
|
|
return d
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: dict) -> "PromptStep":
|
|
return cls(
|
|
prompt=data["prompt"],
|
|
provider=data["provider"],
|
|
output_var=data["output_var"],
|
|
prompt_file=data.get("prompt_file"),
|
|
profile=data.get("profile"),
|
|
name=data.get("name"),
|
|
strip_fences=data.get("strip_fences", False),
|
|
output_schema=data.get("output_schema"),
|
|
max_retries=data.get("max_retries", 1),
|
|
plain_text=data.get("plain_text", False),
|
|
max_tokens=data.get("max_tokens")
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class CodeStep:
|
|
"""A code step that runs Python code."""
|
|
code: str # Python code (inline or loaded from file)
|
|
output_var: str # Variable name(s) to capture (comma-separated for multiple)
|
|
code_file: Optional[str] = None # Optional filename for external code
|
|
name: Optional[str] = None # Optional display name for the step
|
|
|
|
def to_dict(self) -> dict:
|
|
d = {
|
|
"type": "code",
|
|
"code": self.code,
|
|
"output_var": self.output_var
|
|
}
|
|
if self.code_file:
|
|
d["code_file"] = self.code_file
|
|
if self.name:
|
|
d["name"] = self.name
|
|
return d
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: dict) -> "CodeStep":
|
|
return cls(
|
|
code=data.get("code", ""),
|
|
output_var=data["output_var"],
|
|
code_file=data.get("code_file"),
|
|
name=data.get("name")
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class ToolStep:
|
|
"""A step that calls another tool."""
|
|
tool: str # Tool reference (owner/name or just name)
|
|
output_var: str # Variable to store output
|
|
input_template: str = "{input}" # Input template (supports variable substitution)
|
|
args: dict = field(default_factory=dict) # Arguments to pass to the tool
|
|
provider: Optional[str] = None # Provider override for the called tool
|
|
name: Optional[str] = None # Optional display name for the step
|
|
|
|
def to_dict(self) -> dict:
|
|
d = {
|
|
"type": "tool",
|
|
"tool": self.tool,
|
|
"output_var": self.output_var,
|
|
}
|
|
if self.input_template != "{input}":
|
|
d["input"] = self.input_template
|
|
if self.args:
|
|
d["args"] = self.args
|
|
if self.provider:
|
|
d["provider"] = self.provider
|
|
if self.name:
|
|
d["name"] = self.name
|
|
return d
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: dict) -> "ToolStep":
|
|
return cls(
|
|
tool=data["tool"],
|
|
output_var=data["output_var"],
|
|
input_template=data.get("input", "{input}"),
|
|
args=data.get("args", {}),
|
|
provider=data.get("provider"),
|
|
name=data.get("name")
|
|
)
|
|
|
|
|
|
Step = PromptStep | CodeStep | ToolStep
|
|
|
|
|
|
@dataclass
|
|
class ToolSource:
|
|
"""Attribution and source information for imported/external tools."""
|
|
type: str = "original" # "original", "imported", "forked"
|
|
license: Optional[str] = None
|
|
url: Optional[str] = None
|
|
author: Optional[str] = None
|
|
original_tool: Optional[str] = None # e.g., "fabric/patterns/extract_wisdom"
|
|
|
|
def to_dict(self) -> dict:
|
|
d = {"type": self.type}
|
|
if self.license:
|
|
d["license"] = self.license
|
|
if self.url:
|
|
d["url"] = self.url
|
|
if self.author:
|
|
d["author"] = self.author
|
|
if self.original_tool:
|
|
d["original_tool"] = self.original_tool
|
|
return d
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: dict) -> "ToolSource":
|
|
return cls(
|
|
type=data.get("type", "original"),
|
|
license=data.get("license"),
|
|
url=data.get("url"),
|
|
author=data.get("author"),
|
|
original_tool=data.get("original_tool"),
|
|
)
|
|
|
|
|
|
# Default categories for organizing tools
|
|
DEFAULT_CATEGORIES = ["Text", "Developer", "Data", "Other"]
|
|
|
|
|
|
def get_all_categories() -> list:
|
|
"""
|
|
Get all unique categories from default list and local tools.
|
|
|
|
Returns sorted list with defaults first, then custom categories.
|
|
"""
|
|
categories = set(DEFAULT_CATEGORIES)
|
|
|
|
# Add categories from local tools
|
|
try:
|
|
for name in list_tools():
|
|
tool = load_tool(name)
|
|
if tool and tool.category:
|
|
categories.add(tool.category)
|
|
except Exception:
|
|
pass # If tool loading fails, just use defaults
|
|
|
|
# Sort: defaults first (in order), then others alphabetically
|
|
defaults = [c for c in DEFAULT_CATEGORIES if c in categories]
|
|
others = sorted([c for c in categories if c not in DEFAULT_CATEGORIES])
|
|
return defaults + others
|
|
|
|
|
|
@dataclass
|
|
class Tool:
|
|
"""A CmdForge tool definition."""
|
|
name: str
|
|
description: str = ""
|
|
category: str = "Other" # Tool category for organization
|
|
arguments: List[ToolArgument] = field(default_factory=list)
|
|
steps: List[Step] = field(default_factory=list)
|
|
output: str = "{input}" # Output template
|
|
dependencies: List[str] = field(default_factory=list) # Required tools for meta-tools
|
|
system_dependencies: List[SystemDependency] = field(default_factory=list) # System packages (apt, brew, etc.)
|
|
source: Optional[ToolSource] = None # Attribution for imported/external tools
|
|
version: str = "" # Tool version
|
|
path: Optional[Path] = None # Path to config.yaml (set by load_tool)
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: dict) -> "Tool":
|
|
arguments = []
|
|
for arg in data.get("arguments", []):
|
|
arguments.append(ToolArgument.from_dict(arg))
|
|
|
|
steps = []
|
|
for step in data.get("steps", []):
|
|
if step.get("type") == "prompt":
|
|
steps.append(PromptStep.from_dict(step))
|
|
elif step.get("type") == "code":
|
|
steps.append(CodeStep.from_dict(step))
|
|
elif step.get("type") == "tool":
|
|
steps.append(ToolStep.from_dict(step))
|
|
|
|
# Parse source attribution if present
|
|
source = None
|
|
if "source" in data:
|
|
source = ToolSource.from_dict(data["source"])
|
|
|
|
# Normalize dependencies - can be strings or dicts with name/version
|
|
raw_deps = data.get("dependencies", [])
|
|
dependencies = []
|
|
for dep in raw_deps:
|
|
if isinstance(dep, str):
|
|
dependencies.append(dep)
|
|
elif isinstance(dep, dict) and "name" in dep:
|
|
dependencies.append(dep["name"])
|
|
# Skip invalid entries
|
|
|
|
# Parse system dependencies
|
|
raw_sys_deps = data.get("system_dependencies", [])
|
|
system_dependencies = []
|
|
for sd in raw_sys_deps:
|
|
system_dependencies.append(SystemDependency.from_dict(sd))
|
|
|
|
return cls(
|
|
name=data["name"],
|
|
description=data.get("description", ""),
|
|
category=data.get("category", "Other"),
|
|
arguments=arguments,
|
|
steps=steps,
|
|
output=data.get("output", "{input}"),
|
|
dependencies=dependencies,
|
|
system_dependencies=system_dependencies,
|
|
source=source,
|
|
version=data.get("version", ""),
|
|
)
|
|
|
|
def to_dict(self) -> dict:
|
|
d = {
|
|
"name": self.name,
|
|
"description": self.description,
|
|
}
|
|
if self.version:
|
|
d["version"] = self.version
|
|
# Only include category if it's not the default
|
|
if self.category and self.category != "Other":
|
|
d["category"] = self.category
|
|
# Include source attribution if present
|
|
if self.source:
|
|
d["source"] = self.source.to_dict()
|
|
if self.dependencies:
|
|
d["dependencies"] = self.dependencies
|
|
if self.system_dependencies:
|
|
d["system_dependencies"] = [sd.to_dict() for sd in self.system_dependencies]
|
|
d["arguments"] = [arg.to_dict() for arg in self.arguments]
|
|
d["steps"] = [step.to_dict() for step in self.steps]
|
|
d["output"] = self.output
|
|
return d
|
|
|
|
def get_available_variables(self) -> List[str]:
|
|
"""Get all variables available for use in templates."""
|
|
variables = ["input"] # Always available
|
|
|
|
# Add argument variables
|
|
for arg in self.arguments:
|
|
variables.append(arg.variable)
|
|
|
|
# Add step output variables (handle comma-separated output_var)
|
|
for step in self.steps:
|
|
for var in step.output_var.split(','):
|
|
variables.append(var.strip())
|
|
|
|
return variables
|
|
|
|
|
|
def get_tools_dir() -> Path:
|
|
"""Get the tools directory, creating it if needed."""
|
|
TOOLS_DIR.mkdir(parents=True, exist_ok=True)
|
|
return TOOLS_DIR
|
|
|
|
|
|
def ensure_settings(tool_dir: Path) -> Optional[Path]:
|
|
"""Ensure settings.yaml exists if defaults.yaml exists.
|
|
|
|
Called on tool load, save, and install to ensure consistency
|
|
across all tool creation paths (registry, local, GUI).
|
|
|
|
Args:
|
|
tool_dir: Path to the tool directory (e.g., ~/.cmdforge/my-tool/)
|
|
|
|
Returns:
|
|
Path to settings.yaml if created/exists, None otherwise.
|
|
"""
|
|
defaults_path = tool_dir / "defaults.yaml"
|
|
settings_path = tool_dir / "settings.yaml"
|
|
|
|
if defaults_path.exists() and not settings_path.exists():
|
|
shutil.copy(defaults_path, settings_path)
|
|
return settings_path
|
|
elif settings_path.exists():
|
|
return settings_path
|
|
return None
|
|
|
|
|
|
def get_bin_dir() -> Path:
|
|
"""Get the bin directory for wrapper scripts, creating it if needed."""
|
|
BIN_DIR.mkdir(parents=True, exist_ok=True)
|
|
return BIN_DIR
|
|
|
|
|
|
def list_tools() -> list[str]:
|
|
"""List all available tools.
|
|
|
|
Returns tools from:
|
|
- Direct children: ~/.cmdforge/<name>/config.yaml
|
|
- Owner subdirectories: ~/.cmdforge/<owner>/<name>/config.yaml
|
|
"""
|
|
tools_dir = get_tools_dir()
|
|
tools = []
|
|
|
|
if not tools_dir.exists():
|
|
return tools
|
|
|
|
for item in tools_dir.iterdir():
|
|
if item.is_dir():
|
|
config = item / "config.yaml"
|
|
if config.exists():
|
|
# Direct tool (e.g., ~/.cmdforge/my-tool/)
|
|
tools.append(item.name)
|
|
else:
|
|
# Check if this is an owner directory with nested tools
|
|
# (e.g., ~/.cmdforge/official/summarize/)
|
|
for subitem in item.iterdir():
|
|
if subitem.is_dir():
|
|
subconfig = subitem / "config.yaml"
|
|
if subconfig.exists():
|
|
# Qualified name: owner/name
|
|
tools.append(f"{item.name}/{subitem.name}")
|
|
|
|
return sorted(tools)
|
|
|
|
|
|
def load_tool(name: str) -> Optional[Tool]:
|
|
"""Load a tool by name.
|
|
|
|
Args:
|
|
name: Tool name, can be:
|
|
- Simple: "my-tool" (looks in ~/.cmdforge/my-tool/)
|
|
- Qualified: "official/summarize" (looks in ~/.cmdforge/official/summarize/)
|
|
"""
|
|
tools_dir = get_tools_dir()
|
|
|
|
# Try direct path first (handles both simple and qualified names)
|
|
config_path = tools_dir / name / "config.yaml"
|
|
|
|
if not config_path.exists():
|
|
return None
|
|
|
|
try:
|
|
data = yaml.safe_load(config_path.read_text())
|
|
|
|
# Handle legacy format (prompt/provider/provider_args/inputs)
|
|
if "prompt" in data and "steps" not in data:
|
|
# Convert to new format
|
|
steps = []
|
|
if data.get("prompt"):
|
|
steps.append({
|
|
"type": "prompt",
|
|
"prompt": data["prompt"],
|
|
"provider": data.get("provider", "mock"),
|
|
"output_var": "response"
|
|
})
|
|
|
|
arguments = []
|
|
for inp in data.get("inputs", []):
|
|
arguments.append({
|
|
"flag": inp.get("flag", f"--{inp['name']}"),
|
|
"variable": inp["name"],
|
|
"default": inp.get("default"),
|
|
"description": inp.get("description", "")
|
|
})
|
|
|
|
data = {
|
|
"name": data["name"],
|
|
"description": data.get("description", ""),
|
|
"arguments": arguments,
|
|
"steps": steps,
|
|
"output": "{response}" if steps else "{input}"
|
|
}
|
|
|
|
tool = Tool.from_dict(data)
|
|
tool.path = config_path
|
|
# Ensure settings.yaml exists if defaults.yaml exists
|
|
ensure_settings(config_path.parent)
|
|
return tool
|
|
except yaml.YAMLError as e:
|
|
import sys
|
|
print(f"Error loading tool '{name}': YAML syntax error", file=sys.stderr)
|
|
if hasattr(e, 'problem_mark') and e.problem_mark:
|
|
mark = e.problem_mark
|
|
print(f" Line {mark.line + 1}, column {mark.column + 1}", file=sys.stderr)
|
|
# Show the problematic line with context
|
|
try:
|
|
lines = config_path.read_text().split('\n')
|
|
if mark.line < len(lines):
|
|
print(file=sys.stderr)
|
|
# Show line before for context
|
|
if mark.line > 0:
|
|
print(f" {mark.line}: {lines[mark.line - 1]}", file=sys.stderr)
|
|
print(f" > {mark.line + 1}: {lines[mark.line]}", file=sys.stderr)
|
|
print(f" {' ' * (mark.column + 4)}^", file=sys.stderr)
|
|
except Exception:
|
|
pass
|
|
if hasattr(e, 'problem') and e.problem:
|
|
print(f"\n Problem: {e.problem}", file=sys.stderr)
|
|
return None
|
|
except KeyError as e:
|
|
import sys
|
|
print(f"Error loading tool '{name}': Missing required field {e}", file=sys.stderr)
|
|
return None
|
|
except Exception as e:
|
|
import sys
|
|
print(f"Error loading tool '{name}': {e}", file=sys.stderr)
|
|
return None
|
|
|
|
|
|
_REGISTRY_FIELDS = ("registry_hash", "registry_status", "registry_owner", "registry_feedback")
|
|
|
|
|
|
def save_tool(tool: Tool) -> Path:
|
|
"""Save a tool to disk, preserving registry metadata from existing config."""
|
|
tool_dir = get_tools_dir() / tool.name
|
|
tool_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
config_path = tool_dir / "config.yaml"
|
|
|
|
# Preserve registry fields from existing config (not part of Tool model)
|
|
preserved = {}
|
|
if config_path.exists():
|
|
try:
|
|
existing = yaml.safe_load(config_path.read_text()) or {}
|
|
for key in _REGISTRY_FIELDS:
|
|
if key in existing:
|
|
preserved[key] = existing[key]
|
|
except Exception:
|
|
pass
|
|
|
|
new_data = tool.to_dict()
|
|
new_data.update(preserved)
|
|
config_path.write_text(yaml.dump(new_data, default_flow_style=False, sort_keys=False))
|
|
|
|
# Create wrapper script
|
|
create_wrapper_script(tool.name)
|
|
|
|
# Ensure settings.yaml exists if defaults.yaml exists
|
|
ensure_settings(tool_dir)
|
|
|
|
return config_path
|
|
|
|
|
|
def delete_tool(name: str) -> bool:
|
|
"""Delete a tool."""
|
|
tool_dir = get_tools_dir() / name
|
|
|
|
if not tool_dir.exists():
|
|
return False
|
|
|
|
# Remove wrapper script
|
|
wrapper = get_bin_dir() / name
|
|
if wrapper.exists():
|
|
wrapper.unlink()
|
|
|
|
# Remove tool directory
|
|
import shutil
|
|
shutil.rmtree(tool_dir)
|
|
|
|
return True
|
|
|
|
|
|
def create_wrapper_script(name: str) -> Path:
|
|
"""Create a wrapper script for a tool in ~/.local/bin."""
|
|
import sys
|
|
|
|
bin_dir = get_bin_dir()
|
|
wrapper_path = bin_dir / name
|
|
|
|
# Use the current Python interpreter to ensure cmdforge is available
|
|
python_path = sys.executable
|
|
|
|
script = f"""#!/bin/bash
|
|
# CmdForge wrapper for '{name}'
|
|
# Auto-generated - do not edit
|
|
exec "{python_path}" -m cmdforge.runner "{name}" "$@"
|
|
"""
|
|
|
|
wrapper_path.write_text(script)
|
|
wrapper_path.chmod(wrapper_path.stat().st_mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH)
|
|
|
|
return wrapper_path
|
|
|
|
|
|
def tool_exists(name: str) -> bool:
|
|
"""Check if a tool exists."""
|
|
return (get_tools_dir() / name / "config.yaml").exists()
|
|
|
|
|
|
def validate_tool_name(name: str) -> tuple[bool, str]:
|
|
"""
|
|
Validate a tool name.
|
|
|
|
Returns:
|
|
(is_valid, error_message) - error_message is empty if valid
|
|
"""
|
|
if not name:
|
|
return False, "Tool name cannot be empty"
|
|
|
|
if ' ' in name:
|
|
return False, "Tool name cannot contain spaces"
|
|
|
|
# Check for shell-problematic characters
|
|
bad_chars = set('/\\|&;$`"\'<>(){}[]!?*#~')
|
|
found = [c for c in name if c in bad_chars]
|
|
if found:
|
|
return False, f"Tool name cannot contain: {' '.join(found)}"
|
|
|
|
# Must start with letter or underscore
|
|
if not (name[0].isalpha() or name[0] == '_'):
|
|
return False, "Tool name must start with a letter or underscore"
|
|
|
|
# Check it's a valid identifier-ish (alphanumeric, underscore, dash)
|
|
for c in name:
|
|
if not (c.isalnum() or c in '_-'):
|
|
return False, f"Tool name can only contain letters, numbers, underscore, and dash"
|
|
|
|
return True, ""
|