feat(agent): YAML tool specs as the LLM-facing semantic layer
Introduce a first-class semantic layer for tool descriptions, separated
from Python signatures (which stay the source of truth for types and
required-ness).
New
- alfred/agent/tools/spec.py — ToolSpec / ParameterSpec / ReturnsSpec
dataclasses with strict YAML validation (ToolSpecError on malformed
or inconsistent specs). compile_description() builds the rich text
passed to the LLM as Tool.description, with sections for summary,
description, when_to_use, when_not_to_use, next_steps, and returns.
compile_parameter_description() injects the 'why_needed' field next
to each parameter so the LLM sees the *intent* of each argument.
- alfred/agent/tools/spec_loader.py — discovers tools/specs/*.yaml,
enforces filename ↔ spec.name match, rejects duplicates.
- alfred/agent/tools/specs/ — one YAML per tool:
* resolve_season_destination.yaml
* resolve_episode_destination.yaml
* resolve_movie_destination.yaml
* resolve_series_destination.yaml
* move_to_destination.yaml
Refactor
- alfred/agent/registry.py
* _create_tool_from_function now takes an optional ToolSpec.
When provided, the long description + per-parameter descriptions
come from the spec; types and required-ness still come from the
Python signature.
* Cross-validates spec.parameters against the function signature —
crashes on missing or extra entries.
* make_tools() loads all specs at startup and hands the right one
to each tool. Tools without a spec fall back to the old
docstring-only behaviour, so the 14 not-yet-migrated tools keep
working unchanged.
* Adds 'array' and 'object' to the Python→JSON type mapping and
handles Optional[X] / X | None annotations.
- alfred/agent/tools/filesystem.py
* Drops the '_tool' suffix on the 4 resolve_* wrappers (option 1:
alias the use-case imports as _resolve_*). Tool names exposed to
the LLM now match the underlying use case verbatim.
* Wrapper docstrings shrink to a one-liner pointing to the YAML
spec — no more duplicated when_to_use/Args/Returns in Python.
Verified
- make_tools() loads 19 tools (5 with YAML spec, 14 doc-only).
- Compiled descriptions render cleanly with all sections.
This commit is contained in:
+97
-54
@@ -1,4 +1,4 @@
|
||||
"""Tool registry - defines and registers all available tools for the agent."""
|
||||
"""Tool registry — defines and registers all available tools for the agent."""
|
||||
|
||||
import inspect
|
||||
import logging
|
||||
@@ -6,6 +6,9 @@ from collections.abc import Callable
|
||||
from dataclasses import dataclass
|
||||
from typing import Any
|
||||
|
||||
from .tools.spec import ToolSpec, ToolSpecError
|
||||
from .tools.spec_loader import load_tool_specs
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@@ -19,49 +22,60 @@ class Tool:
|
||||
parameters: dict[str, Any]
|
||||
|
||||
|
||||
def _create_tool_from_function(func: Callable) -> Tool:
|
||||
"""
|
||||
Create a Tool object from a function.
|
||||
|
||||
Args:
|
||||
func: Function to convert to a tool
|
||||
|
||||
Returns:
|
||||
Tool object with metadata extracted from function
|
||||
"""
|
||||
sig = inspect.signature(func)
|
||||
doc = inspect.getdoc(func)
|
||||
|
||||
# Extract description from docstring (first line)
|
||||
description = doc.strip().split("\n")[0] if doc else func.__name__
|
||||
|
||||
# Build JSON schema from function signature
|
||||
properties = {}
|
||||
required = []
|
||||
|
||||
for param_name, param in sig.parameters.items():
|
||||
if param_name == "self":
|
||||
continue
|
||||
|
||||
# Map Python types to JSON schema types
|
||||
param_type = "string" # default
|
||||
if param.annotation != inspect.Parameter.empty:
|
||||
if param.annotation is str:
|
||||
param_type = "string"
|
||||
elif param.annotation is int:
|
||||
param_type = "integer"
|
||||
elif param.annotation is float:
|
||||
param_type = "number"
|
||||
elif param.annotation is bool:
|
||||
param_type = "boolean"
|
||||
|
||||
properties[param_name] = {
|
||||
"type": param_type,
|
||||
"description": f"Parameter {param_name}",
|
||||
_PY_TYPE_TO_JSON = {
|
||||
str: "string",
|
||||
int: "integer",
|
||||
float: "number",
|
||||
bool: "boolean",
|
||||
list: "array",
|
||||
dict: "object",
|
||||
}
|
||||
|
||||
# Add to required if no default value
|
||||
if param.default == inspect.Parameter.empty:
|
||||
|
||||
def _json_type_for(annotation) -> str:
|
||||
"""Map a Python type annotation to a JSON Schema 'type' string."""
|
||||
if annotation is inspect.Parameter.empty:
|
||||
return "string"
|
||||
# Strip Optional[X] / X | None to X.
|
||||
args = getattr(annotation, "__args__", None)
|
||||
if args:
|
||||
non_none = [a for a in args if a is not type(None)]
|
||||
if len(non_none) == 1:
|
||||
annotation = non_none[0]
|
||||
return _PY_TYPE_TO_JSON.get(annotation, "string")
|
||||
|
||||
|
||||
def _create_tool_from_function(func: Callable, spec: ToolSpec | None = None) -> Tool:
|
||||
"""
|
||||
Create a Tool object from a function, optionally enriched with a spec.
|
||||
|
||||
Types and required-ness always come from the Python signature (source of
|
||||
truth for the API contract). When a spec is provided, the description
|
||||
and per-parameter docs come from the YAML spec instead of the docstring.
|
||||
"""
|
||||
sig = inspect.signature(func)
|
||||
sig_params = {name: p for name, p in sig.parameters.items() if name != "self"}
|
||||
|
||||
if spec is not None:
|
||||
_validate_spec_matches_signature(func.__name__, sig_params, spec)
|
||||
description = spec.compile_description()
|
||||
param_descriptions = {
|
||||
name: spec.compile_parameter_description(name) for name in sig_params
|
||||
}
|
||||
else:
|
||||
doc = inspect.getdoc(func)
|
||||
description = doc.strip().split("\n")[0] if doc else func.__name__
|
||||
param_descriptions = {name: f"Parameter {name}" for name in sig_params}
|
||||
|
||||
properties: dict[str, dict[str, Any]] = {}
|
||||
required: list[str] = []
|
||||
|
||||
for param_name, param in sig_params.items():
|
||||
properties[param_name] = {
|
||||
"type": _json_type_for(param.annotation),
|
||||
"description": param_descriptions[param_name],
|
||||
}
|
||||
if param.default is inspect.Parameter.empty:
|
||||
required.append(param_name)
|
||||
|
||||
parameters = {
|
||||
@@ -78,31 +92,53 @@ def _create_tool_from_function(func: Callable) -> Tool:
|
||||
)
|
||||
|
||||
|
||||
def _validate_spec_matches_signature(
|
||||
func_name: str,
|
||||
sig_params: dict[str, inspect.Parameter],
|
||||
spec: ToolSpec,
|
||||
) -> None:
|
||||
"""Ensure every signature param has a spec entry and vice versa."""
|
||||
sig_names = set(sig_params.keys())
|
||||
spec_names = set(spec.parameters.keys())
|
||||
|
||||
missing_in_spec = sig_names - spec_names
|
||||
if missing_in_spec:
|
||||
raise ToolSpecError(
|
||||
f"tool '{func_name}': spec is missing entries for parameter(s) "
|
||||
f"{sorted(missing_in_spec)}"
|
||||
)
|
||||
|
||||
extra_in_spec = spec_names - sig_names
|
||||
if extra_in_spec:
|
||||
raise ToolSpecError(
|
||||
f"tool '{func_name}': spec has entries for unknown parameter(s) "
|
||||
f"{sorted(extra_in_spec)} (not in function signature)"
|
||||
)
|
||||
|
||||
|
||||
def make_tools(settings) -> dict[str, Tool]:
|
||||
"""
|
||||
Create and register all available tools.
|
||||
|
||||
Args:
|
||||
settings: Application settings instance
|
||||
settings: Application settings instance.
|
||||
|
||||
Returns:
|
||||
Dictionary mapping tool names to Tool objects
|
||||
Dictionary mapping tool names to Tool objects.
|
||||
"""
|
||||
# Import tools here to avoid circular dependencies
|
||||
from .tools import api as api_tools # noqa: PLC0415
|
||||
from .tools import filesystem as fs_tools # noqa: PLC0415
|
||||
from .tools import language as lang_tools # noqa: PLC0415
|
||||
|
||||
# List of all tool functions
|
||||
tool_functions = [
|
||||
fs_tools.set_path_for_folder,
|
||||
fs_tools.list_folder,
|
||||
fs_tools.analyze_release,
|
||||
fs_tools.probe_media,
|
||||
fs_tools.resolve_season_destination_tool,
|
||||
fs_tools.resolve_episode_destination_tool,
|
||||
fs_tools.resolve_movie_destination_tool,
|
||||
fs_tools.resolve_series_destination_tool,
|
||||
fs_tools.resolve_season_destination,
|
||||
fs_tools.resolve_episode_destination,
|
||||
fs_tools.resolve_movie_destination,
|
||||
fs_tools.resolve_series_destination,
|
||||
fs_tools.move_media,
|
||||
fs_tools.move_to_destination,
|
||||
fs_tools.manage_subtitles,
|
||||
@@ -116,11 +152,18 @@ def make_tools(settings) -> dict[str, Tool]:
|
||||
lang_tools.set_language,
|
||||
]
|
||||
|
||||
# Create Tool objects from functions
|
||||
tools = {}
|
||||
specs = load_tool_specs()
|
||||
|
||||
tools: dict[str, Tool] = {}
|
||||
for func in tool_functions:
|
||||
tool = _create_tool_from_function(func)
|
||||
spec = specs.get(func.__name__)
|
||||
tool = _create_tool_from_function(func, spec=spec)
|
||||
tools[tool.name] = tool
|
||||
|
||||
logger.info(f"Registered {len(tools)} tools: {list(tools.keys())}")
|
||||
with_spec = sum(1 for fn in tool_functions if fn.__name__ in specs)
|
||||
logger.info(
|
||||
f"Registered {len(tools)} tools "
|
||||
f"({with_spec} with YAML spec, {len(tools) - with_spec} doc-only): "
|
||||
f"{list(tools.keys())}"
|
||||
)
|
||||
return tools
|
||||
|
||||
@@ -16,10 +16,10 @@ from alfred.application.filesystem import (
|
||||
from alfred.application.filesystem.detect_media_type import detect_media_type
|
||||
from alfred.application.filesystem.enrich_from_probe import enrich_from_probe
|
||||
from alfred.application.filesystem.resolve_destination import (
|
||||
resolve_episode_destination,
|
||||
resolve_movie_destination,
|
||||
resolve_season_destination,
|
||||
resolve_series_destination,
|
||||
resolve_episode_destination as _resolve_episode_destination,
|
||||
resolve_movie_destination as _resolve_movie_destination,
|
||||
resolve_season_destination as _resolve_season_destination,
|
||||
resolve_series_destination as _resolve_series_destination,
|
||||
)
|
||||
from alfred.infrastructure.filesystem import FileManager, create_folder, move
|
||||
from alfred.infrastructure.filesystem.ffprobe import probe
|
||||
@@ -48,19 +48,7 @@ def move_media(source: str, destination: str) -> dict[str, Any]:
|
||||
|
||||
|
||||
def move_to_destination(source: str, destination: str) -> dict[str, Any]:
|
||||
"""
|
||||
Move a file or folder to a destination, creating parent directories if needed.
|
||||
|
||||
Use this after resolve_*_destination to perform the actual move.
|
||||
The destination parent is created automatically if it doesn't exist.
|
||||
|
||||
Args:
|
||||
source: Absolute path to the source file or folder.
|
||||
destination: Absolute path to the destination.
|
||||
|
||||
Returns:
|
||||
Dict with status, source, destination — or error details.
|
||||
"""
|
||||
"""Thin tool wrapper — semantics live in alfred/agent/tools/specs/move_to_destination.yaml."""
|
||||
parent = str(Path(destination).parent)
|
||||
result = create_folder(parent)
|
||||
if result["status"] != "ok":
|
||||
@@ -68,36 +56,19 @@ def move_to_destination(source: str, destination: str) -> dict[str, Any]:
|
||||
return move(source, destination)
|
||||
|
||||
|
||||
def resolve_season_destination_tool(
|
||||
def resolve_season_destination(
|
||||
release_name: str,
|
||||
tmdb_title: str,
|
||||
tmdb_year: int,
|
||||
confirmed_folder: str | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""
|
||||
Compute destination paths for a season pack (folder move).
|
||||
|
||||
Returns series_folder + season_folder. No file paths — the whole
|
||||
source folder is moved as-is into season_folder.
|
||||
|
||||
Args:
|
||||
release_name: Raw release folder name (e.g. "Oz.S03.1080p.WEBRip.x265-KONTRAST").
|
||||
tmdb_title: Canonical show title from TMDB (e.g. "Oz").
|
||||
tmdb_year: Show start year from TMDB (e.g. 1997).
|
||||
confirmed_folder: If needs_clarification was returned, pass the chosen folder name.
|
||||
|
||||
Returns:
|
||||
On success: dict with status, series_folder, season_folder, series_folder_name,
|
||||
season_folder_name, is_new_series_folder.
|
||||
On ambiguity: dict with status="needs_clarification", question, options.
|
||||
On error: dict with status="error", error, message.
|
||||
"""
|
||||
return resolve_season_destination(
|
||||
"""Thin tool wrapper — semantics live in alfred/agent/tools/specs/resolve_season_destination.yaml."""
|
||||
return _resolve_season_destination(
|
||||
release_name, tmdb_title, tmdb_year, confirmed_folder
|
||||
).to_dict()
|
||||
|
||||
|
||||
def resolve_episode_destination_tool(
|
||||
def resolve_episode_destination(
|
||||
release_name: str,
|
||||
source_file: str,
|
||||
tmdb_title: str,
|
||||
@@ -105,26 +76,8 @@ def resolve_episode_destination_tool(
|
||||
tmdb_episode_title: str | None = None,
|
||||
confirmed_folder: str | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""
|
||||
Compute destination paths for a single episode file.
|
||||
|
||||
Returns series_folder + season_folder + library_file (full path to destination .mkv).
|
||||
|
||||
Args:
|
||||
release_name: Raw release file name (e.g. "Oz.S03E01.1080p.WEBRip.x265-KONTRAST.mkv").
|
||||
source_file: Absolute path to the source video file (used for extension).
|
||||
tmdb_title: Canonical show title from TMDB (e.g. "Oz").
|
||||
tmdb_year: Show start year from TMDB (e.g. 1997).
|
||||
tmdb_episode_title: Episode title from TMDB (e.g. "The Routine"). Optional.
|
||||
confirmed_folder: If needs_clarification was returned, pass the chosen folder name.
|
||||
|
||||
Returns:
|
||||
On success: dict with status, series_folder, season_folder, library_file,
|
||||
series_folder_name, season_folder_name, filename, is_new_series_folder.
|
||||
On ambiguity: dict with status="needs_clarification", question, options.
|
||||
On error: dict with status="error", error, message.
|
||||
"""
|
||||
return resolve_episode_destination(
|
||||
"""Thin tool wrapper — semantics live in alfred/agent/tools/specs/resolve_episode_destination.yaml."""
|
||||
return _resolve_episode_destination(
|
||||
release_name,
|
||||
source_file,
|
||||
tmdb_title,
|
||||
@@ -134,56 +87,26 @@ def resolve_episode_destination_tool(
|
||||
).to_dict()
|
||||
|
||||
|
||||
def resolve_movie_destination_tool(
|
||||
def resolve_movie_destination(
|
||||
release_name: str,
|
||||
source_file: str,
|
||||
tmdb_title: str,
|
||||
tmdb_year: int,
|
||||
) -> dict[str, Any]:
|
||||
"""
|
||||
Compute destination paths for a movie file.
|
||||
|
||||
Returns movie_folder + library_file (full path to destination .mkv).
|
||||
|
||||
Args:
|
||||
release_name: Raw release folder/file name (e.g. "Inception.2010.1080p.BluRay.x265-GROUP").
|
||||
source_file: Absolute path to the source video file (used for extension).
|
||||
tmdb_title: Canonical movie title from TMDB (e.g. "Inception").
|
||||
tmdb_year: Movie release year from TMDB (e.g. 2010).
|
||||
|
||||
Returns:
|
||||
On success: dict with status, movie_folder, library_file, movie_folder_name,
|
||||
filename, is_new_folder.
|
||||
On error: dict with status="error", error, message.
|
||||
"""
|
||||
return resolve_movie_destination(
|
||||
"""Thin tool wrapper — semantics live in alfred/agent/tools/specs/resolve_movie_destination.yaml."""
|
||||
return _resolve_movie_destination(
|
||||
release_name, source_file, tmdb_title, tmdb_year
|
||||
).to_dict()
|
||||
|
||||
|
||||
def resolve_series_destination_tool(
|
||||
def resolve_series_destination(
|
||||
release_name: str,
|
||||
tmdb_title: str,
|
||||
tmdb_year: int,
|
||||
confirmed_folder: str | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""
|
||||
Compute destination path for a complete multi-season series pack (folder move).
|
||||
|
||||
Returns only series_folder — the whole pack lands directly inside it.
|
||||
|
||||
Args:
|
||||
release_name: Raw release folder name.
|
||||
tmdb_title: Canonical show title from TMDB.
|
||||
tmdb_year: Show start year from TMDB.
|
||||
confirmed_folder: If needs_clarification was returned, pass the chosen folder name.
|
||||
|
||||
Returns:
|
||||
On success: dict with status, series_folder, series_folder_name, is_new_series_folder.
|
||||
On ambiguity: dict with status="needs_clarification", question, options.
|
||||
On error: dict with status="error", error, message.
|
||||
"""
|
||||
return resolve_series_destination(
|
||||
"""Thin tool wrapper — semantics live in alfred/agent/tools/specs/resolve_series_destination.yaml."""
|
||||
return _resolve_series_destination(
|
||||
release_name, tmdb_title, tmdb_year, confirmed_folder
|
||||
).to_dict()
|
||||
|
||||
|
||||
@@ -0,0 +1,191 @@
|
||||
"""
|
||||
ToolSpec — semantic description of a tool, loaded from YAML.
|
||||
|
||||
Each tool exposed to the agent has a matching YAML spec under
|
||||
alfred/agent/tools/specs/{tool_name}.yaml. The spec carries everything the
|
||||
LLM needs to decide *when* and *why* to call the tool — separated from the
|
||||
Python signature, which remains the source of truth for *how* (types,
|
||||
required-ness).
|
||||
|
||||
The YAML structure is documented in the dataclasses below. Loading a spec
|
||||
validates its shape; missing or unexpected fields raise ToolSpecError.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
from pathlib import Path
|
||||
|
||||
import yaml
|
||||
|
||||
|
||||
class ToolSpecError(ValueError):
|
||||
"""Raised when a YAML tool spec is malformed or inconsistent."""
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ParameterSpec:
|
||||
"""Semantic description of a single tool parameter."""
|
||||
|
||||
description: str # Short: what the value represents.
|
||||
why_needed: str # Why the tool needs this — drives LLM reasoning.
|
||||
example: str | None = None # Concrete example value, shown to the LLM.
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, name: str, data: dict) -> ParameterSpec:
|
||||
_require(data, "description", f"parameter '{name}'")
|
||||
_require(data, "why_needed", f"parameter '{name}'")
|
||||
return cls(
|
||||
description=str(data["description"]).strip(),
|
||||
why_needed=str(data["why_needed"]).strip(),
|
||||
example=str(data["example"]).strip() if data.get("example") is not None else None,
|
||||
)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ReturnsSpec:
|
||||
"""Description of one possible return shape (ok / needs_clarification / error / ...)."""
|
||||
|
||||
description: str
|
||||
fields: dict[str, str] = field(default_factory=dict)
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, key: str, data: dict) -> ReturnsSpec:
|
||||
_require(data, "description", f"returns.{key}")
|
||||
fields = data.get("fields") or {}
|
||||
if not isinstance(fields, dict):
|
||||
raise ToolSpecError(f"returns.{key}.fields must be a dict, got {type(fields).__name__}")
|
||||
return cls(
|
||||
description=str(data["description"]).strip(),
|
||||
fields={str(k): str(v).strip() for k, v in fields.items()},
|
||||
)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ToolSpec:
|
||||
"""Full semantic spec for one tool."""
|
||||
|
||||
name: str
|
||||
summary: str # One-liner — becomes Tool.description.
|
||||
description: str # Longer paragraph.
|
||||
when_to_use: str
|
||||
when_not_to_use: str | None
|
||||
next_steps: str | None
|
||||
parameters: dict[str, ParameterSpec] # name -> ParameterSpec
|
||||
returns: dict[str, ReturnsSpec] # status_key -> ReturnsSpec
|
||||
|
||||
@classmethod
|
||||
def from_yaml_path(cls, path: Path) -> ToolSpec:
|
||||
with open(path, encoding="utf-8") as f:
|
||||
data = yaml.safe_load(f) or {}
|
||||
if not isinstance(data, dict):
|
||||
raise ToolSpecError(f"{path}: top-level must be a mapping")
|
||||
try:
|
||||
return cls.from_dict(data)
|
||||
except ToolSpecError as e:
|
||||
raise ToolSpecError(f"{path}: {e}") from e
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, data: dict) -> ToolSpec:
|
||||
_require(data, "name", "spec")
|
||||
_require(data, "summary", "spec")
|
||||
_require(data, "description", "spec")
|
||||
_require(data, "when_to_use", "spec")
|
||||
|
||||
params_raw = data.get("parameters") or {}
|
||||
if not isinstance(params_raw, dict):
|
||||
raise ToolSpecError("parameters must be a mapping")
|
||||
parameters = {
|
||||
pname: ParameterSpec.from_dict(pname, pdata or {})
|
||||
for pname, pdata in params_raw.items()
|
||||
}
|
||||
|
||||
returns_raw = data.get("returns") or {}
|
||||
if not isinstance(returns_raw, dict):
|
||||
raise ToolSpecError("returns must be a mapping")
|
||||
returns = {
|
||||
rkey: ReturnsSpec.from_dict(rkey, rdata or {})
|
||||
for rkey, rdata in returns_raw.items()
|
||||
}
|
||||
|
||||
return cls(
|
||||
name=str(data["name"]).strip(),
|
||||
summary=str(data["summary"]).strip(),
|
||||
description=str(data["description"]).strip(),
|
||||
when_to_use=str(data["when_to_use"]).strip(),
|
||||
when_not_to_use=_strip_or_none(data.get("when_not_to_use")),
|
||||
next_steps=_strip_or_none(data.get("next_steps")),
|
||||
parameters=parameters,
|
||||
returns=returns,
|
||||
)
|
||||
|
||||
def compile_description(self) -> str:
|
||||
"""
|
||||
Build the long description text passed to the LLM as Tool.description.
|
||||
|
||||
Layout:
|
||||
<summary>
|
||||
|
||||
<description>
|
||||
|
||||
When to use:
|
||||
<when_to_use>
|
||||
|
||||
When NOT to use: (if present)
|
||||
<when_not_to_use>
|
||||
|
||||
Next steps: (if present)
|
||||
<next_steps>
|
||||
|
||||
Returns:
|
||||
<status>: <description>
|
||||
· <field>: <desc>
|
||||
"""
|
||||
parts = [self.summary, "", self.description]
|
||||
|
||||
parts += ["", "When to use:", _indent(self.when_to_use)]
|
||||
|
||||
if self.when_not_to_use:
|
||||
parts += ["", "When NOT to use:", _indent(self.when_not_to_use)]
|
||||
|
||||
if self.next_steps:
|
||||
parts += ["", "Next steps:", _indent(self.next_steps)]
|
||||
|
||||
if self.returns:
|
||||
parts += ["", "Returns:"]
|
||||
for status, ret in self.returns.items():
|
||||
parts.append(f" {status}: {ret.description}")
|
||||
for fname, fdesc in ret.fields.items():
|
||||
parts.append(f" · {fname}: {fdesc}")
|
||||
|
||||
return "\n".join(parts)
|
||||
|
||||
def compile_parameter_description(self, name: str) -> str:
|
||||
"""Build the JSON Schema 'description' field for one parameter."""
|
||||
p = self.parameters.get(name)
|
||||
if p is None:
|
||||
raise ToolSpecError(f"tool '{self.name}': no spec for parameter '{name}'")
|
||||
text = f"{p.description} (Why: {p.why_needed})"
|
||||
if p.example:
|
||||
text += f" Example: {p.example}"
|
||||
return text
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _require(data: dict, key: str, where: str) -> None:
|
||||
if data.get(key) is None or (isinstance(data[key], str) and not data[key].strip()):
|
||||
raise ToolSpecError(f"{where}: missing required field '{key}'")
|
||||
|
||||
|
||||
def _strip_or_none(value) -> str | None:
|
||||
if value is None:
|
||||
return None
|
||||
s = str(value).strip()
|
||||
return s or None
|
||||
|
||||
|
||||
def _indent(text: str, prefix: str = " ") -> str:
|
||||
return "\n".join(prefix + line for line in text.splitlines())
|
||||
@@ -0,0 +1,53 @@
|
||||
"""
|
||||
ToolSpecLoader — discover and load all YAML tool specs from a directory.
|
||||
|
||||
Convention: one YAML file per tool, named exactly like the Python function
|
||||
that implements it (e.g. resolve_season_destination.yaml).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from pathlib import Path
|
||||
|
||||
from .spec import ToolSpec, ToolSpecError
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
_DEFAULT_SPECS_DIR = Path(__file__).parent / "specs"
|
||||
|
||||
|
||||
def load_tool_specs(specs_dir: Path | None = None) -> dict[str, ToolSpec]:
|
||||
"""
|
||||
Load every {tool}.yaml under specs_dir into a {name -> ToolSpec} mapping.
|
||||
|
||||
Args:
|
||||
specs_dir: Directory to scan. Defaults to alfred/agent/tools/specs/.
|
||||
|
||||
Returns:
|
||||
Mapping from tool name to its parsed ToolSpec.
|
||||
|
||||
Raises:
|
||||
ToolSpecError: if a spec is malformed, or if the filename doesn't
|
||||
match the 'name' field inside the YAML.
|
||||
"""
|
||||
root = specs_dir or _DEFAULT_SPECS_DIR
|
||||
if not root.exists():
|
||||
logger.warning(f"Tool specs directory not found: {root}")
|
||||
return {}
|
||||
|
||||
specs: dict[str, ToolSpec] = {}
|
||||
for path in sorted(root.glob("*.yaml")):
|
||||
spec = ToolSpec.from_yaml_path(path)
|
||||
expected_name = path.stem
|
||||
if spec.name != expected_name:
|
||||
raise ToolSpecError(
|
||||
f"{path}: filename stem '{expected_name}' "
|
||||
f"does not match spec.name '{spec.name}'"
|
||||
)
|
||||
if spec.name in specs:
|
||||
raise ToolSpecError(f"duplicate tool spec name: '{spec.name}'")
|
||||
specs[spec.name] = spec
|
||||
|
||||
logger.info(f"Loaded {len(specs)} tool spec(s) from {root}")
|
||||
return specs
|
||||
@@ -0,0 +1,55 @@
|
||||
name: move_to_destination
|
||||
|
||||
summary: >
|
||||
Move a file or folder to a destination, creating parent directories as needed.
|
||||
|
||||
description: |
|
||||
Performs an actual move on disk. Uses the system 'mv' command, so on the
|
||||
same filesystem (e.g. ZFS) this is an instant rename. Creates the parent
|
||||
directory of the destination if it doesn't exist yet, then moves. Returns
|
||||
before/after paths on success, or an error if the destination already
|
||||
exists or the source can't be moved.
|
||||
|
||||
when_to_use: |
|
||||
Use after one of the resolve_*_destination tools returned status=ok, to
|
||||
perform the move it described. The 'source' and 'destination' arguments
|
||||
come directly from the resolved paths.
|
||||
|
||||
when_not_to_use: |
|
||||
- Never move when status was not 'ok' (clarification still pending or
|
||||
error happened) — that would leave the library in a half-broken state.
|
||||
- Don't use this for the seed-link step; use create_seed_links for that.
|
||||
|
||||
next_steps: |
|
||||
- After a successful move: call manage_subtitles to place any subtitle
|
||||
tracks, then create_seed_links to keep qBittorrent seeding.
|
||||
- On error: surface the message; do not retry blindly — check whether
|
||||
the destination already exists or the source path is correct.
|
||||
|
||||
parameters:
|
||||
source:
|
||||
description: Absolute path to the source file or folder to move.
|
||||
why_needed: |
|
||||
The thing being moved. Comes from the user's download folder or from
|
||||
a previous tool's output.
|
||||
example: /downloads/Oz.S03.1080p.WEBRip.x265-KONTRAST
|
||||
|
||||
destination:
|
||||
description: Absolute path of the destination — must not already exist.
|
||||
why_needed: |
|
||||
Where to put the source. Comes from a resolve_*_destination call so
|
||||
that the path matches the library's naming convention.
|
||||
example: /tv_shows/Oz.1997.1080p.WEBRip.x265-KONTRAST/Oz.S03.1080p.WEBRip.x265-KONTRAST
|
||||
|
||||
returns:
|
||||
ok:
|
||||
description: Move succeeded.
|
||||
fields:
|
||||
source: Absolute path of the source (now gone).
|
||||
destination: Absolute path of the destination (now in place).
|
||||
|
||||
error:
|
||||
description: Move failed.
|
||||
fields:
|
||||
error: Short error code (source_not_found, destination_exists, mkdir_failed, move_failed).
|
||||
message: Human-readable explanation of what went wrong.
|
||||
@@ -0,0 +1,93 @@
|
||||
name: resolve_episode_destination
|
||||
|
||||
summary: >
|
||||
Compute destination paths for a single TV episode file (file move).
|
||||
|
||||
description: |
|
||||
Resolves the target series folder, season subfolder, and full destination
|
||||
filename for a single-episode release. Returns paths only — does not move
|
||||
anything. If a series folder with a different name already exists, returns
|
||||
needs_clarification.
|
||||
|
||||
when_to_use: |
|
||||
Use after analyze_release has identified the release as a single episode
|
||||
(media_type=tv_show, season AND episode both set). TMDB must already be
|
||||
queried for the canonical title/year, and optionally the episode title.
|
||||
|
||||
when_not_to_use: |
|
||||
- Season packs (folder containing many episodes): use resolve_season_destination.
|
||||
- Multi-season packs: use resolve_series_destination.
|
||||
- Movies: use resolve_movie_destination.
|
||||
|
||||
next_steps: |
|
||||
- On status=ok: call move_to_destination with the source video file and
|
||||
destination=library_file.
|
||||
- On status=needs_clarification: present question/options to the user,
|
||||
then re-call with confirmed_folder set.
|
||||
- On status=error: surface the message; do not move.
|
||||
|
||||
parameters:
|
||||
release_name:
|
||||
description: Raw release file name (with extension).
|
||||
why_needed: |
|
||||
Drives extraction of quality/source/codec/group, which become part of
|
||||
the destination filename so each file is self-describing.
|
||||
example: Oz.S03E01.1080p.WEBRip.x265-KONTRAST.mkv
|
||||
|
||||
source_file:
|
||||
description: Absolute path to the source video file on disk.
|
||||
why_needed: |
|
||||
Used to read the source file extension (.mkv, .mp4, .avi…) for the
|
||||
destination filename — release names don't always carry the extension.
|
||||
example: /downloads/Oz.S03E01.1080p.WEBRip.x265-KONTRAST/file.mkv
|
||||
|
||||
tmdb_title:
|
||||
description: Canonical show title from TMDB.
|
||||
why_needed: |
|
||||
Title prefix for both the series folder and the destination filename;
|
||||
ensures consistent naming across all episodes of the show.
|
||||
example: Oz
|
||||
|
||||
tmdb_year:
|
||||
description: Show start year from TMDB.
|
||||
why_needed: |
|
||||
Disambiguates remakes/reboots sharing a title; year is part of the
|
||||
series folder identity.
|
||||
example: "1997"
|
||||
|
||||
tmdb_episode_title:
|
||||
description: Episode title from TMDB. Optional.
|
||||
why_needed: |
|
||||
When present, the destination filename embeds the episode title for
|
||||
human-readability (e.g. Oz.S01E01.The.Routine...).
|
||||
example: The Routine
|
||||
|
||||
confirmed_folder:
|
||||
description: Folder name the user picked after needs_clarification.
|
||||
why_needed: |
|
||||
Forces the use case to skip detection and use this exact folder name.
|
||||
example: Oz.1997.1080p.WEBRip.x265-KONTRAST
|
||||
|
||||
returns:
|
||||
ok:
|
||||
description: Paths resolved; ready to move the episode file.
|
||||
fields:
|
||||
series_folder: Absolute path to the series root folder.
|
||||
season_folder: Absolute path to the season subfolder.
|
||||
library_file: Absolute path to the destination .mkv file (move target).
|
||||
series_folder_name: Series folder name for display.
|
||||
season_folder_name: Season folder name for display.
|
||||
filename: Destination filename for display.
|
||||
is_new_series_folder: True if the series folder doesn't exist yet.
|
||||
|
||||
needs_clarification:
|
||||
description: A folder exists with a different name; user must choose.
|
||||
fields:
|
||||
question: Human-readable question.
|
||||
options: List of folder names to pick from.
|
||||
|
||||
error:
|
||||
description: Resolution failed.
|
||||
fields:
|
||||
error: Short error code.
|
||||
message: Human-readable explanation.
|
||||
@@ -0,0 +1,72 @@
|
||||
name: resolve_movie_destination
|
||||
|
||||
summary: >
|
||||
Compute destination paths for a movie file (file move).
|
||||
|
||||
description: |
|
||||
Resolves the target movie folder and full destination filename for a movie
|
||||
release. Returns paths only — does not move anything. Movies do not have
|
||||
the existing-folder disambiguation problem that TV shows have (each
|
||||
release lands in its own folder named after the canonical title + year +
|
||||
tech).
|
||||
|
||||
when_to_use: |
|
||||
Use after analyze_release has identified the release as a movie
|
||||
(media_type=movie). TMDB must already be queried for the canonical title
|
||||
and release year.
|
||||
|
||||
when_not_to_use: |
|
||||
- TV shows in any form: use resolve_season_destination /
|
||||
resolve_episode_destination / resolve_series_destination.
|
||||
- Documentaries when they're treated as series rather than standalone
|
||||
films: route them through the TV-show resolvers.
|
||||
|
||||
next_steps: |
|
||||
- On status=ok: call move_to_destination with the source video file and
|
||||
destination=library_file.
|
||||
- On status=error: surface the message; do not move.
|
||||
|
||||
parameters:
|
||||
release_name:
|
||||
description: Raw release folder or file name.
|
||||
why_needed: |
|
||||
Drives extraction of quality/source/codec/group/edition tokens, which
|
||||
become part of both the movie folder and filename so each release is
|
||||
self-describing on disk.
|
||||
example: Inception.2010.1080p.BluRay.x265-GROUP
|
||||
|
||||
source_file:
|
||||
description: Absolute path to the source video file on disk.
|
||||
why_needed: |
|
||||
Used to read the file extension for the destination filename.
|
||||
example: /downloads/Inception.2010.1080p.BluRay.x265-GROUP/movie.mkv
|
||||
|
||||
tmdb_title:
|
||||
description: Canonical movie title from TMDB.
|
||||
why_needed: |
|
||||
Title prefix for the destination folder/file; ensures the library
|
||||
uses the canonical title and not a sanitized release-name title.
|
||||
example: Inception
|
||||
|
||||
tmdb_year:
|
||||
description: Movie release year from TMDB.
|
||||
why_needed: |
|
||||
Disambiguates remakes that share a title (Dune 1984 vs Dune 2021)
|
||||
and locks the folder identity in time.
|
||||
example: "2010"
|
||||
|
||||
returns:
|
||||
ok:
|
||||
description: Paths resolved; ready to move.
|
||||
fields:
|
||||
movie_folder: Absolute path to the movie folder.
|
||||
library_file: Absolute path to the destination .mkv file (move target).
|
||||
movie_folder_name: Folder name for display.
|
||||
filename: Destination filename for display.
|
||||
is_new_folder: True if the movie folder doesn't exist yet.
|
||||
|
||||
error:
|
||||
description: Resolution failed.
|
||||
fields:
|
||||
error: Short error code (e.g. library_not_set).
|
||||
message: Human-readable explanation.
|
||||
@@ -0,0 +1,84 @@
|
||||
name: resolve_season_destination
|
||||
|
||||
summary: >
|
||||
Compute destination paths for a season pack (folder move) in the TV library.
|
||||
|
||||
description: |
|
||||
Resolves the target series folder and season subfolder for a complete-season
|
||||
download. Returns the paths only — does not perform any move. If a series
|
||||
folder for this show already exists in the library with a different name
|
||||
(different group/quality/source), returns needs_clarification so the user
|
||||
can decide whether to merge into the existing folder or create a new one.
|
||||
|
||||
when_to_use: |
|
||||
Use after analyze_release has identified the release as a season pack
|
||||
(media_type=tv_show, season set, episode unset). TMDB must already be
|
||||
queried so tmdb_title and tmdb_year are canonical values, not raw tokens
|
||||
from the release name.
|
||||
|
||||
when_not_to_use: |
|
||||
- Single-episode files: use resolve_episode_destination instead.
|
||||
- Multi-season packs (S01-S05 etc.): use resolve_series_destination.
|
||||
- Movies: use resolve_movie_destination.
|
||||
|
||||
next_steps: |
|
||||
- On status=ok: call move_to_destination with source=<download folder> and
|
||||
destination=season_folder.
|
||||
- On status=needs_clarification: present the question and options to the
|
||||
user, then re-call this tool with confirmed_folder set to the user's pick.
|
||||
- On status=error: surface the message to the user; do not move anything.
|
||||
|
||||
parameters:
|
||||
release_name:
|
||||
description: Raw release folder name as it appears on disk.
|
||||
why_needed: |
|
||||
Drives extraction of quality/source/codec/group tokens — these are
|
||||
embedded in the target folder name (Title.Year.Quality.Source.Codec-GROUP)
|
||||
to make releases self-describing on the filesystem.
|
||||
example: Oz.S03.1080p.WEBRip.x265-KONTRAST
|
||||
|
||||
tmdb_title:
|
||||
description: Canonical show title from TMDB.
|
||||
why_needed: |
|
||||
Builds the title prefix of the folder name. Must come from TMDB to
|
||||
avoid typos and variant spellings present in the raw release name.
|
||||
example: Oz
|
||||
|
||||
tmdb_year:
|
||||
description: Show start year from TMDB.
|
||||
why_needed: |
|
||||
Disambiguates shows that share a title across decades (e.g. multiple
|
||||
remakes of "The Office") and locks the folder identity.
|
||||
example: "1997"
|
||||
|
||||
confirmed_folder:
|
||||
description: |
|
||||
Folder name chosen by the user after a previous needs_clarification
|
||||
response.
|
||||
why_needed: |
|
||||
Short-circuits the existing-folder detection and forces the use case
|
||||
to use this exact folder name, even if it doesn't match the computed
|
||||
one.
|
||||
example: Oz.1997.1080p.WEBRip.x265-KONTRAST
|
||||
|
||||
returns:
|
||||
ok:
|
||||
description: Paths resolved unambiguously; ready to move.
|
||||
fields:
|
||||
series_folder: Absolute path to the series root folder.
|
||||
season_folder: Absolute path to the season subfolder (move target).
|
||||
series_folder_name: Just the series folder name, for display.
|
||||
season_folder_name: Just the season folder name, for display.
|
||||
is_new_series_folder: True if the series folder doesn't exist yet.
|
||||
|
||||
needs_clarification:
|
||||
description: A folder already exists with a different name; ask the user.
|
||||
fields:
|
||||
question: Human-readable question for the user.
|
||||
options: List of folder names the user can pick from.
|
||||
|
||||
error:
|
||||
description: Resolution failed (config missing, invalid release name, etc.).
|
||||
fields:
|
||||
error: Short error code (e.g. library_not_set).
|
||||
message: Human-readable explanation.
|
||||
@@ -0,0 +1,77 @@
|
||||
name: resolve_series_destination
|
||||
|
||||
summary: >
|
||||
Compute the destination path for a complete multi-season series pack (folder move).
|
||||
|
||||
description: |
|
||||
Resolves the target series folder for a pack that contains multiple seasons
|
||||
(e.g. S01-S05 in a single release). Returns only the series folder — the
|
||||
whole source folder is moved as-is into the library, no per-season
|
||||
restructuring. If a folder with a different name already exists for this
|
||||
show, returns needs_clarification.
|
||||
|
||||
when_to_use: |
|
||||
Use after analyze_release has identified the release as a complete-series
|
||||
pack (media_type=tv_complete, or multi-season indicators). TMDB must
|
||||
already be queried for canonical title/year.
|
||||
|
||||
when_not_to_use: |
|
||||
- Single-season packs: use resolve_season_destination.
|
||||
- Single episodes: use resolve_episode_destination.
|
||||
- Movies: use resolve_movie_destination.
|
||||
|
||||
next_steps: |
|
||||
- On status=ok: call move_to_destination with source=<download folder> and
|
||||
destination=series_folder.
|
||||
- On status=needs_clarification: ask the user, re-call with
|
||||
confirmed_folder set.
|
||||
- On status=error: surface the message; do not move.
|
||||
|
||||
parameters:
|
||||
release_name:
|
||||
description: Raw release folder name as it appears on disk.
|
||||
why_needed: |
|
||||
Drives extraction of quality/source/codec/group tokens for the target
|
||||
folder name, even though the multi-season structure inside is kept
|
||||
as-is.
|
||||
example: The.Wire.S01-S05.1080p.BluRay.x265-GROUP
|
||||
|
||||
tmdb_title:
|
||||
description: Canonical show title from TMDB.
|
||||
why_needed: |
|
||||
Title prefix of the series folder; comes from TMDB to avoid raw
|
||||
release-name spellings.
|
||||
example: The Wire
|
||||
|
||||
tmdb_year:
|
||||
description: Show start year from TMDB.
|
||||
why_needed: |
|
||||
Disambiguates shows that share a title across eras and locks the
|
||||
folder identity.
|
||||
example: "2002"
|
||||
|
||||
confirmed_folder:
|
||||
description: Folder name chosen by the user after needs_clarification.
|
||||
why_needed: |
|
||||
Forces the use case to use this exact folder name and skip detection.
|
||||
example: The.Wire.2002.1080p.BluRay.x265-GROUP
|
||||
|
||||
returns:
|
||||
ok:
|
||||
description: Path resolved; ready to move the pack.
|
||||
fields:
|
||||
series_folder: Absolute path to the destination series folder.
|
||||
series_folder_name: Folder name for display.
|
||||
is_new_series_folder: True if the folder doesn't exist yet.
|
||||
|
||||
needs_clarification:
|
||||
description: A folder exists with a different name; ask the user.
|
||||
fields:
|
||||
question: Human-readable question.
|
||||
options: List of folder names to pick from.
|
||||
|
||||
error:
|
||||
description: Resolution failed.
|
||||
fields:
|
||||
error: Short error code.
|
||||
message: Human-readable explanation.
|
||||
Reference in New Issue
Block a user