866 lines
30 KiB
Python
866 lines
30 KiB
Python
"""Tool loading, saving, and management."""
|
|
|
|
import os
|
|
import re
|
|
import shutil
|
|
import stat
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
from typing import Any, 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"
|
|
_TOOL_REF_RE = re.compile(
|
|
r"^(?:[a-z0-9](?:[a-z0-9-]{0,37}[a-z0-9])?/)?[A-Za-z0-9_-]{1,64}$"
|
|
)
|
|
|
|
|
|
def _validate_skill_selection(skills: Optional[List[str]]) -> None:
|
|
"""Validate a step's optional Agent Skills selection."""
|
|
if skills is None:
|
|
return
|
|
if not isinstance(skills, list):
|
|
raise ValueError("skills must be a list of skill names")
|
|
if any(not isinstance(name, str) for name in skills):
|
|
raise ValueError("skills entries must be strings")
|
|
if "*" in skills and skills != ["*"]:
|
|
raise ValueError("'*' must be the only entry when enabling all skills")
|
|
if len(skills) != len(set(skills)):
|
|
raise ValueError("skills must not contain duplicate names")
|
|
|
|
from .skills import _validate_skill_name
|
|
for name in skills:
|
|
if name != "*":
|
|
_validate_skill_name(name)
|
|
|
|
|
|
def _validate_optional_patterns(patterns: Optional[List[str]], field_name: str) -> None:
|
|
if patterns is not None and (
|
|
not isinstance(patterns, list)
|
|
or not all(isinstance(pattern, str) and pattern for pattern in patterns)
|
|
):
|
|
raise ValueError(f"{field_name} must be a list of non-empty strings or null")
|
|
|
|
|
|
def validate_json_schema(schema: Optional[dict], field_name: str) -> None:
|
|
"""Validate an optional tool contract as a real JSON Schema."""
|
|
if schema is None:
|
|
return
|
|
if not isinstance(schema, dict):
|
|
raise ValueError(f"{field_name} must be a JSON Schema object")
|
|
|
|
from jsonschema.exceptions import SchemaError
|
|
from jsonschema.validators import validator_for
|
|
|
|
try:
|
|
validator_for(schema).check_schema(schema)
|
|
except SchemaError as exc:
|
|
raise ValueError(f"Invalid {field_name}: {exc.message}") from exc
|
|
|
|
|
|
@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[Any] = None
|
|
description: str = ""
|
|
type: str = "string"
|
|
enum: Optional[List[Any]] = None
|
|
required: bool = False
|
|
|
|
def __post_init__(self) -> None:
|
|
if self.type not in ("string", "integer", "number", "boolean"):
|
|
raise ValueError(
|
|
"ToolArgument type must be string, integer, number, or boolean"
|
|
)
|
|
if self.enum is not None and (
|
|
not isinstance(self.enum, list) or not self.enum
|
|
):
|
|
raise ValueError("ToolArgument enum must be a non-empty list")
|
|
if not isinstance(self.required, bool):
|
|
raise ValueError("ToolArgument required must be a boolean")
|
|
|
|
def to_dict(self) -> dict:
|
|
d = {"flag": self.flag, "variable": self.variable}
|
|
if self.default is not None:
|
|
d["default"] = self.default
|
|
if self.description:
|
|
d["description"] = self.description
|
|
if self.type != "string":
|
|
d["type"] = self.type
|
|
if self.enum is not None:
|
|
d["enum"] = self.enum
|
|
if self.required:
|
|
d["required"] = True
|
|
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", ""),
|
|
type=data.get("type", "string"),
|
|
enum=data.get("enum"),
|
|
required=data.get("required", False),
|
|
)
|
|
|
|
|
|
@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)
|
|
skills: Optional[List[str]] = None # Skill names to enable for this step
|
|
|
|
def __post_init__(self) -> None:
|
|
_validate_skill_selection(self.skills)
|
|
|
|
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
|
|
if self.skills is not None:
|
|
d["skills"] = self.skills
|
|
return d
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: dict) -> "PromptStep":
|
|
max_tokens = data.get("max_tokens")
|
|
if max_tokens is not None:
|
|
try:
|
|
max_tokens = int(max_tokens)
|
|
except (TypeError, ValueError):
|
|
raise ValueError("max_tokens must be an integer")
|
|
if max_tokens <= 0 or max_tokens > 1_000_000:
|
|
raise ValueError("max_tokens must be between 1 and 1000000")
|
|
|
|
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=max_tokens,
|
|
skills=data.get("skills"),
|
|
)
|
|
|
|
|
|
@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
|
|
profile: Optional[str] = None # AI persona for the nested tool
|
|
skills: Optional[List[str]] = None # Skills to enable in the nested tool
|
|
name: Optional[str] = None # Optional display name for the step
|
|
tools: Optional[List[str]] = None # Tools the delegated agent may call
|
|
|
|
def __post_init__(self) -> None:
|
|
_validate_skill_selection(self.skills)
|
|
_validate_optional_patterns(self.tools, "tools")
|
|
if self.profile is not None and (
|
|
not isinstance(self.profile, str) or not self.profile
|
|
):
|
|
raise ValueError("profile must be a non-empty string or null")
|
|
|
|
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.profile:
|
|
d["profile"] = self.profile
|
|
if self.skills is not None:
|
|
d["skills"] = self.skills
|
|
if self.tools is not None:
|
|
d["tools"] = self.tools
|
|
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"),
|
|
profile=data.get("profile"),
|
|
skills=data.get("skills"),
|
|
tools=data.get("tools"),
|
|
name=data.get("name")
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class McpStep:
|
|
"""A step that calls an MCP server tool."""
|
|
server: str # Server name in ~/.cmdforge/mcp.yaml
|
|
tool: str # Tool name on that server
|
|
arguments: dict = field(default_factory=dict)
|
|
output_var: str = "mcp_result"
|
|
result_mode: str = "auto" # auto | structured | content | text
|
|
name: Optional[str] = None
|
|
|
|
def to_dict(self) -> dict:
|
|
d = {
|
|
"type": "mcp",
|
|
"server": self.server,
|
|
"tool": self.tool,
|
|
"output_var": self.output_var,
|
|
}
|
|
if self.arguments:
|
|
d["arguments"] = self.arguments
|
|
if self.result_mode != "auto":
|
|
d["result_mode"] = self.result_mode
|
|
if self.name:
|
|
d["name"] = self.name
|
|
return d
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: dict) -> "McpStep":
|
|
server = data.get("server")
|
|
tool = data.get("tool")
|
|
arguments = data.get("arguments", {})
|
|
output_var = data.get("output_var", "mcp_result")
|
|
result_mode = data.get("result_mode", "auto")
|
|
if not isinstance(server, str) or not server.strip():
|
|
raise ValueError("MCP step server must be a non-empty string")
|
|
if not isinstance(tool, str) or not tool.strip():
|
|
raise ValueError("MCP step tool must be a non-empty string")
|
|
if not isinstance(arguments, dict):
|
|
raise ValueError("MCP step arguments must be a mapping")
|
|
if not isinstance(output_var, str) or not output_var.strip():
|
|
raise ValueError("MCP step output_var must be a non-empty string")
|
|
if result_mode not in ("auto", "structured", "content", "text"):
|
|
raise ValueError("MCP step result_mode must be auto, structured, content, or text")
|
|
return cls(
|
|
server=server,
|
|
tool=tool,
|
|
arguments=arguments,
|
|
output_var=output_var,
|
|
result_mode=result_mode,
|
|
name=data.get("name"),
|
|
)
|
|
|
|
|
|
Step = PromptStep | CodeStep | ToolStep | McpStep
|
|
|
|
@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
|
|
visibility: str = "public" # "public", "private", or "unlisted"
|
|
input_schema: Optional[dict] = None # JSON Schema for tool input contract
|
|
output_schema: Optional[dict] = None # JSON Schema for tool output contract
|
|
path: Optional[Path] = None # Path to config.yaml (set by load_tool)
|
|
deprecated: bool = False # Tool is deprecated
|
|
deprecated_message: str = "" # Migration guidance for deprecated tools
|
|
replacement: Optional[str] = None # Suggested replacement tool name
|
|
|
|
def __post_init__(self) -> None:
|
|
validate_json_schema(self.input_schema, "input_schema")
|
|
validate_json_schema(self.output_schema, "output_schema")
|
|
if not isinstance(self.deprecated, bool):
|
|
raise ValueError("deprecated must be true or false")
|
|
if not isinstance(self.deprecated_message, str):
|
|
raise ValueError("deprecated_message must be a string")
|
|
if len(self.deprecated_message) > 500:
|
|
raise ValueError("deprecated_message must be at most 500 characters")
|
|
if self.replacement is not None and (
|
|
not isinstance(self.replacement, str)
|
|
or not _TOOL_REF_RE.fullmatch(self.replacement)
|
|
):
|
|
raise ValueError("replacement must be a tool name or owner/tool reference")
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: dict) -> "Tool":
|
|
if not isinstance(data, dict):
|
|
raise ValueError("Tool configuration must be a YAML mapping")
|
|
|
|
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))
|
|
elif step.get("type") == "mcp":
|
|
steps.append(McpStep.from_dict(step))
|
|
|
|
# Parse source attribution if present
|
|
source = None
|
|
if "source" in data:
|
|
source_data = data["source"]
|
|
if isinstance(source_data, dict):
|
|
source = ToolSource.from_dict(source_data)
|
|
elif isinstance(source_data, str) and source_data.strip():
|
|
source = ToolSource(
|
|
type="imported", original_tool=source_data.strip()
|
|
)
|
|
elif source_data is not None:
|
|
raise ValueError("source must be an attribution object or string")
|
|
|
|
# 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", ""),
|
|
visibility=data.get("visibility", "public"),
|
|
deprecated=data.get("deprecated", False),
|
|
deprecated_message=data.get("deprecated_message", ""),
|
|
replacement=data.get("replacement"),
|
|
input_schema=data.get("input_schema"),
|
|
output_schema=data.get("output_schema"),
|
|
)
|
|
|
|
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
|
|
# Only include visibility if it's not the default
|
|
if self.visibility and self.visibility != "public":
|
|
d["visibility"] = self.visibility
|
|
if self.deprecated:
|
|
d["deprecated"] = True
|
|
if self.deprecated_message:
|
|
d["deprecated_message"] = self.deprecated_message
|
|
if self.replacement:
|
|
d["replacement"] = self.replacement
|
|
if self.input_schema is not None:
|
|
d["input_schema"] = self.input_schema
|
|
if self.output_schema is not None:
|
|
d["output_schema"] = self.output_schema
|
|
# 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}",
|
|
"visibility": data.get("visibility", "public"),
|
|
}
|
|
|
|
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 deprecation_warning(tool: Tool) -> Optional[str]:
|
|
"""Return consistent migration guidance for a deprecated tool."""
|
|
if not tool.deprecated:
|
|
return None
|
|
message = tool.deprecated_message or "This tool is no longer maintained."
|
|
if tool.replacement:
|
|
message += f" Use '{tool.replacement}' instead."
|
|
return f"Warning: '{tool.name}' is deprecated. {message}"
|
|
|
|
|
|
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, ""
|