diff --git a/cliffy/ai.py b/cliffy/ai.py index 15950f1..d1edcb0 100644 --- a/cliffy/ai.py +++ b/cliffy/ai.py @@ -1,37 +1,351 @@ +"""AI-powered commands: ``cli ai generate`` and ``cli ai ask``. + +pydantic-ai is an optional dependency (``cliffy[ai]``) and slow to import, so it is only imported +when one of the commands actually talks to a model. Importing this module is cheap and always safe. +""" + +from __future__ import annotations + import json -from io import TextIOWrapper +import os +import re from pathlib import Path +from typing import IO, TYPE_CHECKING, Any -from pydantic_ai import Agent -from pydantic_ai.models import KnownModelName -from pydantic_ai.settings import ModelSettings -from pydantic_ai.usage import UsageLimits +import yaml +from click import Context, Group +from pydantic import ValidationError -from cliffy.helper import ManifestOrCLI, out +from cliffy.helper import ManifestOrCLI, exit_err, out from cliffy.homer import get_metadata from cliffy.manifest import CLIManifest from cliffy.rich import click -from cliffy.transformer import Transformer + +if TYPE_CHECKING: + from pydantic_ai.models import Model + from pydantic_ai.settings import ModelSettings + from pydantic_ai.usage import RunUsage + +MODEL_ENV_VAR = "CLIFFY_AI_MODEL" +INSTALL_HINT = "pip install 'cliffy[ai]'" + +# Used when neither --model nor CLIFFY_AI_MODEL is given: the first provider with an API key in the +# environment wins. Where a provider publishes a rolling alias it is preferred over a dated model id. +PROVIDER_DEFAULTS: tuple[tuple[tuple[str, ...], str], ...] = ( + (("GOOGLE_API_KEY", "GEMINI_API_KEY"), "google:gemini-flash-latest"), + (("ANTHROPIC_API_KEY",), "anthropic:claude-sonnet-5-5"), + (("OPENAI_API_KEY",), "openai:gpt-5.4-mini"), +) +FALLBACK_MODEL = PROVIDER_DEFAULTS[0][1] + +CLI_NAME_PATTERN = re.compile(r"[A-Za-z0-9][A-Za-z0-9_-]*") +FENCED_BLOCK_PATTERN = re.compile(r"```[A-Za-z]*[ \t]*\n(.*?)```", re.DOTALL) + +MANIFEST_RULES = """\ +Rules for writing a manifest: +- `manifestVersion` is `v3`. `name`, `version` and `commands` are required. +- Typer is the CLI framework of the generated CLI, so `typer` is always available in run blocks. +- A command's `run` block is Python. A line starting with `$` is run as a shell command. +- Parameters are available in run blocks as variables named after the parameter. +- Parameter shorthand: `name: str!` is a required argument, `--name: str = "x"` is an option with a default, \ +`--name|-n: str` adds a short flag. A Typer expression such as `name: str = typer.Option(...)` also works. +- Nested subcommands are joined with `.`, as in `db.create`. A command that has subcommands cannot run on its \ +own, so never define a `run` block for a parent command. +- Anything the run blocks import goes in `imports`. Third-party packages also go in `requires`. +- Add an `examples` section listing example invocations of the CLI.""" + +GENERATE_OUTPUT_RULES = """\ +Reply with the YAML manifest only: no explanations, no surrounding prose. +The manifest must be self-contained: do not use `includes`, and do not use the `types` section. +Your reply is validated against the schema. If it is rejected you will be told why; fix it and reply with the \ +complete corrected manifest.""" + +HOW_IT_WORKS = """\ +How cliffy works: +1. A CLI is defined in a YAML manifest. +2. `cli load ` parses the manifest, generates a Typer (or Click, with `use_click: true`) CLI and \ +installs it as a script, so it can be run straight from the terminal. +3. Code in a run block that starts with `$` is translated to subprocess calls by PyBash. +4. `cli build` packages a CLI into a portable zipapp built with Shiv.""" + + +class ManifestOutputError(ValueError): + """A model reply that is not a usable CLI manifest. The message is sent back to the model.""" + + +def default_model() -> str: + """Pick the model to use when the user did not choose one.""" + for env_vars, model in PROVIDER_DEFAULTS: + if any(os.environ.get(var) for var in env_vars): + return model + return FALLBACK_MODEL + + +def require_ai_extra() -> None: + """Import pydantic-ai, or exit with an install hint when the ``ai`` extra is missing.""" + try: + import pydantic_ai + except ModuleNotFoundError as e: + # only a missing pydantic-ai is an install problem; anything else is a real bug and must surface + if e.name is None or e.name.partition(".")[0] != "pydantic_ai": + raise + exit_err(f"~ AI commands need the `ai` extra, which is not installed. Install it with: {INSTALL_HINT}") + + # keep pydantic-ai's first-run banner out of the command output + pydantic_ai.BANNER_ENABLED = False + + +def resolve_model(model: str | None) -> Model: + """Turn a ``provider:model`` name into a pydantic-ai model, exiting with a readable error if that fails.""" + require_ai_extra() + from pydantic_ai.exceptions import UserError + from pydantic_ai.models import infer_model + + name = model or default_model() + api_keys = ", ".join(env_vars[0] for env_vars, _ in PROVIDER_DEFAULTS) + hint = ( + f"Set the API key of a provider ({api_keys}, ...) or choose a model with --model or {MODEL_ENV_VAR}, " + f"as `provider:model` (for example `{FALLBACK_MODEL}`)." + ) + try: + return infer_model(name) + except UserError as e: + # unknown provider or model name, or the provider's API key is not set + reason = str(e).partition(" To try Pydantic AI")[0] # drop pydantic-ai's pointer to its test model + exit_err(f"~ cannot use AI model `{name}`: {reason}\n{hint}") + except ImportError as e: + # the provider's SDK is not installed (pydantic-ai-slim without that provider's extra) + exit_err(f"~ cannot use AI model `{name}`: {e}\n{hint}") + + +def model_settings(max_tokens: int | None) -> ModelSettings | None: + from pydantic_ai.settings import ModelSettings + + return ModelSettings(max_tokens=max_tokens) if max_tokens else None + + +def command_reference() -> str: + """Describe cliffy's own commands, read from the live click group so it cannot go stale.""" + from cliffy.cli import ALIASES, cli + + lines: list[str] = [] + + def walk(group: Group, prefix: str) -> None: + for name in sorted(group.commands): + command = group.commands[name] + if command.hidden: + continue + if isinstance(command, Group): + walk(command, f"{prefix} {name}") + continue + usage = " ".join(command.collect_usage_pieces(Context(command, info_name=name))) + lines.append(f"- `{prefix} {name} {usage}`: {command.get_short_help_str(limit=200)}") + + walk(cli, "cli") + aliases = ", ".join(f"`{alias}` = `{command}`" for alias, command in ALIASES.items()) + return "\n".join([*lines, f"Aliases: {aliases}"]) + + +def manifest_reference() -> str: + """The manifest documentation that ships with cliffy: annotated template and JSON schema.""" + return ( + "Annotated example manifest (the output of `cli init`). The comments describe each section:\n" + f"```yaml\n{CLIManifest.get_template('mycli', json_schema=False)}```\n\n" + "JSON schema of a manifest:\n" + f"```json\n{json.dumps(CLIManifest.model_json_schema())}\n```" + ) + + +def generate_instructions(cli_name: str) -> str: + return ( + "You write YAML manifests for `cliffy`, a tool that generates Python CLIs from YAML manifests.\n" + f"Write a manifest for a CLI named `{cli_name}` that fulfils the description given by the user.\n\n" + f"{MANIFEST_RULES}\n\n{GENERATE_OUTPUT_RULES}\n\n{manifest_reference()}" + ) + + +def ask_instructions() -> str: + return ( + "You are an expert on `cliffy`, a tool that generates Python CLIs from YAML manifests.\n" + "Answer the user's question using the reference below. If the reference does not cover it, say so " + "instead of guessing. Show manifest snippets as YAML.\n\n" + f"{HOW_IT_WORKS}\n\ncliffy commands:\n{command_reference()}\n\n{MANIFEST_RULES}\n\n{manifest_reference()}" + ) + + +def cli_reference(cli: IO[Any] | str | None) -> str: + """Render the manifest given with ``--cli`` (a manifest file or the name of a loaded CLI) for the prompt.""" + if cli is None: + return "" + + if isinstance(cli, str): + metadata = get_metadata(cli) or exit_err(f"~ {cli} is not a loaded CLI or a manifest path") + manifest = metadata.manifest + else: + content = cli.read() + manifest = content.decode() if isinstance(content, bytes) else content + + return f"\n\nThe question is about this CLI manifest:\n```yaml\n{manifest.strip()}\n```" -@click.group(help="AI-powered commands") +def extract_yaml(text: str) -> str: + """Strip the markdown code fence models tend to wrap YAML in.""" + if fenced := FENCED_BLOCK_PATTERN.search(text): + text = fenced.group(1) + return text.strip() + "\n" + + +def validate_manifest_yaml(text: str, cli_name: str) -> str: + """Check that a model reply is a manifest cliffy can load and return it as clean YAML. + + Raises: + ManifestOutputError: describing what is wrong, phrased so the model can fix it. + """ + manifest_yaml = extract_yaml(text) + try: + data = yaml.safe_load(manifest_yaml) + except yaml.YAMLError as e: + raise ManifestOutputError(f"The reply is not valid YAML: {e}") from e + + if not isinstance(data, dict): + raise ManifestOutputError("The reply must be a YAML mapping containing the manifest and nothing else.") + + try: + manifest = CLIManifest.model_validate(data) + except ValidationError as e: + problems = "\n".join( + f"- {'.'.join(str(part) for part in error['loc']) or '(manifest)'}: {error['msg']}" + for error in e.errors(include_url=False) + ) + raise ManifestOutputError(f"The manifest does not match the schema:\n{problems}") from e + + if manifest.name != cli_name: + raise ManifestOutputError(f"The manifest `name` must be `{cli_name}`, not `{manifest.name}`.") + if manifest.includes: + raise ManifestOutputError("The manifest must be self-contained: remove `includes`.") + + # Jinja expressions are only rendered when the manifest is loaded (with the user's environment), + # so the generated code can only be checked here for manifests that do not use them. + if "{{" not in manifest_yaml and "{%" not in manifest_yaml: + check_generated_code(manifest) + + return manifest_yaml + + +def check_generated_code(manifest: CLIManifest) -> None: + """Generate the CLI for a manifest and make sure the result is valid Python. Nothing is executed.""" + from cliffy.commander import generate_cli + from cliffy.commanders.click import ClickCommander + from cliffy.commanders.typer import TyperCommander + + try: + cli = generate_cli(manifest, commander_cls=ClickCommander if manifest.use_click else TyperCommander) + except Exception as e: # the commanders raise a variety of errors on manifests they cannot handle + raise ManifestOutputError(f"cliffy could not generate a CLI from the manifest: {e}") from e + + try: + compile(cli.code, f"{manifest.name}.py", "exec") + except SyntaxError as e: + raise ManifestOutputError( + f"The CLI generated from the manifest is not valid Python: {e.msg} near `{(e.text or '').strip()}`. " + "Check the run blocks, functions and imports." + ) from e + + +def generate_manifest( + description: str, cli_name: str, model: Model, *, max_tokens: int | None = None, retries: int = 2 +) -> tuple[str, RunUsage]: + """Ask a model for a manifest and return it once it validates. + + A reply that fails validation is sent back to the model with the reason, up to ``retries`` times. + + Raises: + ManifestOutputError: if the model never produced a valid manifest. + pydantic_ai.exceptions.AgentRunError: if the request itself failed. + """ + from pydantic_ai import Agent, ModelRetry + from pydantic_ai.exceptions import UnexpectedModelBehavior + + agent = Agent( + model, + output_type=str, + instructions=generate_instructions(cli_name), + model_settings=model_settings(max_tokens), + retries=retries, + ) + rejections: list[ManifestOutputError] = [] + + @agent.output_validator + def validate_output(output: str) -> str: + try: + return validate_manifest_yaml(output, cli_name) + except ManifestOutputError as e: + rejections.append(e) + raise ModelRetry(str(e)) from e + + try: + result = agent.run_sync(description) + except UnexpectedModelBehavior as e: + if rejections: + raise ManifestOutputError( + f"the model did not produce a valid manifest ({len(rejections)} rejected). " + f"Last problem:\n{rejections[-1]}" + ) from e + raise + return result.output, result.usage + + +def ask_question(question: str, model: Model, *, max_tokens: int | None = None) -> tuple[str, RunUsage]: + """Ask a model a question about cliffy, grounded in the reference built from the installed package.""" + from pydantic_ai import Agent + + agent = Agent(model, output_type=str, instructions=ask_instructions(), model_settings=model_settings(max_tokens)) + result = agent.run_sync(question) + return result.output, result.usage + + +def report_usage(usage: RunUsage) -> None: + # on stderr, so that redirecting stdout captures the answer only + out(f"\ntoken usage: {usage.input_tokens} in, {usage.output_tokens} out, {usage.requests} request(s)", err=True) + + +model_option = click.option( + "--model", + "-m", + envvar=MODEL_ENV_VAR, + show_envvar=True, + default=None, + help="Model to use, as `provider:model`. See https://pydantic.dev/docs/ai/models/overview/. " + f"Defaults to a model of the first provider with an API key set, else {FALLBACK_MODEL}.", +) +max_tokens_option = click.option( + "--max-tokens", type=click.IntRange(min=1), default=None, help="Maximum number of tokens in a model response." +) +preview_option = click.option( + "--preview", is_flag=True, default=False, help="Print the prompt instead of sending it. Needs no model." +) + + +@click.group(help="AI-powered commands. Requires the `ai` extra: " + INSTALL_HINT) def ai() -> None: pass -@click.option("--preview", is_flag=True, help="Display the generated prompt before sending the request.", default=False) -@click.option("--max-tokens", type=int, help="The maximum number of tokens to generate before stopping.", default=None) +@preview_option +@max_tokens_option +@model_option @click.option( - "--model", - "-m", - help="AI model to use. See https://pydantic.dev/docs/ai/models/overview/ for supported models.", - default="google:gemini-2.0-flash", + "--retries", + type=click.IntRange(min=0), + default=2, show_default=True, + help="How often to send an invalid manifest back to the model to be fixed.", ) +@click.option("--force", "-f", is_flag=True, default=False, help="Overwrite the manifest file if it exists.") @click.option( "--output-dir", "-o", - type=click.Path(exists=True, dir_okay=True, file_okay=False), + type=click.Path(exists=True, dir_okay=True, file_okay=False, path_type=Path), default=Path(), show_default=True, help="Output directory", @@ -39,119 +353,69 @@ def ai() -> None: @click.argument("cli_name", required=True) @click.argument("description", required=True) def generate( - cli_name: str, description: str, model: KnownModelName, max_tokens: int, output_dir: Path, preview: bool + cli_name: str, + description: str, + output_dir: Path, + force: bool, + retries: int, + model: str | None, + max_tokens: int | None, + preview: bool, ) -> None: - SYSTEM_PROMPT = f"""You are a YAML manifest generator for CLIs. -Here is the json schema for the YAML to generate: -```json{json.dumps(CLIManifest.model_json_schema())}``` -Typer is the CLI framework used. -Nested commands are joined by a "." -Due to a feature limitation, parent command cannot be triggered if they have subcommands. -Do not write a group command definition for the parent if it has a subcommand. -Do not use types section at all. -Parameters can be used in the run scripts by referencing the parameter name directly. -Always provide `examples` section to list example commands for the CLI. -Use Python and Typer features and imports as needed to craft the best CLI for the following listed CLI requirements. - -""" + if not CLI_NAME_PATTERN.fullmatch(cli_name): + exit_err(f"~ invalid CLI name `{cli_name}`: use letters, digits, `-` and `_` only") if preview: - out(SYSTEM_PROMPT) - out(description) - exit() + out(generate_instructions(cli_name)) + out(f"\n{description}") + return - model_settings = ModelSettings(max_tokens=max_tokens) if max_tokens else None - agent = Agent(model, system_prompt=SYSTEM_PROMPT, model_settings=model_settings) + manifest_path = output_dir / f"{cli_name}.yaml" + if manifest_path.exists() and not force: + exit_err(f"~ {manifest_path} already exists, pass --force to overwrite it") - usage_limits = UsageLimits(total_tokens_limit=max_tokens) if max_tokens else None + ai_model = resolve_model(model) + from pydantic_ai.exceptions import AgentRunError - result = agent.run_sync(description, usage_limits=usage_limits) + try: + manifest, usage = generate_manifest(description, cli_name, ai_model, max_tokens=max_tokens, retries=retries) + except ManifestOutputError as e: + exit_err(f"~ no manifest written: {e}") + except AgentRunError as e: + exit_err(f"~ request to `{ai_model.model_name}` failed: {e}") - manifest = result.output.strip().removeprefix("```yaml").removesuffix("```").strip() - (output_dir / Path(f"{cli_name}.yaml")).write_text(manifest) - out(f"✨ {cli_name}.yaml") - out("\ntoken usage:") - out("------------") - out(f"request: {result.usage.input_tokens}") - out(f"response: {result.usage.output_tokens}") + manifest_path.write_text(manifest) + out(f"✨ {manifest_path}", fg="green") + out("$", fg="magenta", nl=False) + out(f" cli load {manifest_path}") + report_usage(usage) -@click.option("--preview", is_flag=True, help="Display the generated prompt before sending the request.", default=False) -@click.option("--max-tokens", type=int, help="The maximum number of tokens to generate before stopping.", default=None) -@click.option("--model", "-m", help="LLM model to use.", default="google:gemini-2.0-flash", show_default=True) +@preview_option +@max_tokens_option +@model_option @click.option( - "--cli", type=ManifestOrCLI(), help="Loaded CLI or manifest to include in prompt as reference.", default=None + "--cli", type=ManifestOrCLI(), default=None, help="Loaded CLI or manifest to include in the prompt as reference." ) @click.argument("prompt", required=True) -def ask(cli: ManifestOrCLI | None, prompt: str, model: KnownModelName, max_tokens: int, preview: bool = False) -> None: - SYSTEM_PROMPT = f"""You are an expert of `cliffy`- a YAML manifest to Typer CLI generator. - -## Cliffy Usage -`cli ` - -| Command | Description | -|---|---| -| `init `| Generate a template CLI manifest for a new CLI | -| `load ` | Add a new CLI based on the manifest | -| `render ` | View generated CLI script for a manifest | -| `list, ls` | Output a list of loaded CLIs | -| `update `| Reload a loaded CLI | -| `remove , rm ` | Remove a loaded CLI | -| `run -- `| Runs a CLI manifest command in isolation| -| `build ` | Build a CLI manifest or a loaded CLI into a self-contained zipapp | -| `info ` | Display CLI metadata | -| `dev ` | Start hot-reloader for a manifest for active development | -| `test ` | Run tests defined in a manifest | -| `validate ` | Validate the syntax and structure of a CLI manifest | -| `docs ` | Generate documentation for a CLI | -| `ai generate ` | Generate a CLI manifest based on a description. | -| `ai ask ` | Ask a question about cliffy or a specific CLI manifest. | - -## How it works -1. Define CLI manifests in YAML files -2. Run `cli` commands to load, build, and manage CLIs -3. When loaded, cliffy parses the manifest and generates a Typer CLI that is deployed directly as a script -4. Any code starting with `$` will translate to subprocess calls via PyBash -5. Run loaded CLIs straight from the terminal -6. When ready to share, run `build` to generate portable zipapps built with [Shiv](https://github.com/linkedin/shiv) - -Here is the model json schema for CLI manifest: -```json{json.dumps(CLIManifest.model_json_schema())}``` -Typer is the CLI framework used. -Due to a feature limitation, parent command cannot be triggered if they have subcommands. -Do not write a group command definition for the parent if it has a subcommand. -Parameters can be used in the run scripts by referencing the parameter name directly. -Always provide `examples` section to list example commands for the CLI. -Use Python and Typer features and imports as needed to craft the best CLI for the following listed CLI requirements. - -""" - - reference = "" - if cli: - reference += "Here is the CLI manifest to use as reference:" - reference += "```yaml" - if isinstance(cli, TextIOWrapper): - reference += str(Transformer(cli).manifest.model_dump(mode="json")) - elif isinstance(cli, str): - metadata = get_metadata(cli) - reference += metadata.manifest if metadata else "" - reference += "```" +def ask(prompt: str, cli: IO[Any] | str | None, model: str | None, max_tokens: int | None, preview: bool) -> None: + question = prompt + cli_reference(cli) if preview: - out(SYSTEM_PROMPT + reference + prompt) - exit() + out(ask_instructions()) + out(f"\n{question}") + return - model_settings = ModelSettings(max_tokens=max_tokens) if max_tokens else None - agent = Agent(model, system_prompt=SYSTEM_PROMPT, model_settings=model_settings) + ai_model = resolve_model(model) + from pydantic_ai.exceptions import AgentRunError - usage_limits = UsageLimits(total_tokens_limit=max_tokens) if max_tokens else None + try: + answer, usage = ask_question(question, ai_model, max_tokens=max_tokens) + except AgentRunError as e: + exit_err(f"~ request to `{ai_model.model_name}` failed: {e}") - result = agent.run_sync(reference + prompt, usage_limits=usage_limits) - out(result.output) - out("\ntoken usage:") - out("------------") - out(f"request: {result.usage.input_tokens}") - out(f"response: {result.usage.output_tokens}") + out(answer) + report_usage(usage) ai.command("generate", help="Generate a CLI manifest based on a description.")(generate) diff --git a/cliffy/cli.py b/cliffy/cli.py index aafa571..2d97fe8 100644 --- a/cliffy/cli.py +++ b/cliffy/cli.py @@ -3,6 +3,7 @@ from io import TextIOWrapper from typing import Any, TextIO +from cliffy.ai import ai from cliffy.builder import build_cli, build_cli_from_manifest, run_cli from cliffy.doc import DocGenerator from cliffy.helper import ( @@ -372,6 +373,7 @@ def docs(cli_or_manifest: str, format: str, output_dir: str) -> None: test_command = cli.command("test")(test) validate_command = cli.command("validate")(validate) docs_command = cli.command("docs")(docs) +cli.add_command(ai) # register aliases cli.command("add", hidden=True, epilog="Alias for load")( diff --git a/cliffy/run.py b/cliffy/run.py index 6a121a7..78b08c5 100644 --- a/cliffy/run.py +++ b/cliffy/run.py @@ -1,12 +1,4 @@ def run() -> None: from cliffy.cli import cli - try: - from cliffy.ai import ai - - cli.add_command(ai) - except Exception: - # print(e) - ... - cli() # type: ignore diff --git a/docs/features.md b/docs/features.md index 0339c01..40c510e 100644 --- a/docs/features.md +++ b/docs/features.md @@ -218,19 +218,39 @@ Install the [YAML embedded languages extension](https://marketplace.visualstudio Cliffy includes AI-powered commands to help you generate and understand CLI manifests. Access these features through the `cli ai` command group. !!! note - Requires `cliffy[ai]` dependency group to be installed. See [Installation](getting-started.md#installation) + Requires the `ai` extra: `pip install "cliffy[ai]"`. See [Installation](getting-started.md#installation). + Without it, `cli ai` commands exit with an install hint; the rest of cliffy is unaffected. -!!! tip - For a list of the models supported, see [https://ai.pydantic.dev/models/](https://ai.pydantic.dev/models/). You can specify the model to use for the AI commands with `--model` or `-m` option. +### Choosing a model + +The AI commands are built on [Pydantic AI](https://pydantic.dev/docs/ai/models/overview/) and work with any model it supports. Models are named `provider:model`, and each provider reads its API key from its usual environment variable. + +The model is picked in this order: + +1. the `--model` / `-m` option +2. the `CLIFFY_AI_MODEL` environment variable +3. a default for the first provider that has an API key set: + + | API key | Default model | + |---|---| + | `GOOGLE_API_KEY` (or `GEMINI_API_KEY`) | `google:gemini-flash-latest` | + | `ANTHROPIC_API_KEY` | `anthropic:claude-sonnet-5-5` | + | `OPENAI_API_KEY` | `openai:gpt-5.4-mini` | + +```bash +# use the default model of the provider whose key is set +export ANTHROPIC_API_KEY='your-api-key' +cli ai generate mycli "Create a todo app" - ```bash - # Example using a specific model - export OPENAI_API_KEY='your-api-key' - cli ai generate mycli "Create a todo app" --model gpt-4o - ``` +# pick a model for one command +cli ai generate mycli "Create a todo app" --model openai:gpt-5.5 + +# or for the whole shell session +export CLIFFY_AI_MODEL=google:gemini-2.5-pro +``` !!! tip - Use `--preview` to output the fully constructed prompt for debugging or review. + Use `--preview` to print the fully constructed prompt instead of sending it. This needs neither a model nor an API key. ### Manifest Generation @@ -238,8 +258,16 @@ Generate complete CLI manifests from natural language descriptions: ```bash cli ai generate mycli "Create a CLI for managing docker containers with commands to list, start, stop and remove containers" +cli load mycli.yaml ``` +The model's reply is validated before anything is written: it has to be valid YAML, match the manifest schema, and produce a CLI that is valid Python. A reply that fails is sent back to the model together with the reason, up to `--retries` times (default 2). If no valid manifest comes back, the command fails and writes nothing. + +`mycli.yaml` is written to the current directory, or to `--output-dir`. An existing file is only replaced with `--force`. + +!!! warning + Validation checks that the manifest loads, not that the commands do what you asked. Read the generated run blocks before running the CLI. + ### Interactive Help Get AI assistance for understanding and working with cliffy: @@ -248,8 +276,12 @@ Get AI assistance for understanding and working with cliffy: cli ai ask "How do I create nested subcommands?" ``` -You can also get targeted help about specific CLIs by referencing them: +The question is answered from a reference built from your installed cliffy version: its command list, the annotated `cli init` template and the manifest JSON schema. + +You can also get targeted help about specific CLIs by referencing a manifest or a loaded CLI: ```bash cli ai ask --cli mycli.yaml "How do I add input validation to the start command?" ``` + +The answer is printed to stdout and token usage to stderr, so the answer can be redirected to a file. diff --git a/pyproject.toml b/pyproject.toml index ab871cb..1de4e7e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -106,8 +106,7 @@ select = [ [tool.ruff.lint.per-file-ignores] # long pydantic Field descriptions feed the published JSON schema verbatim "cliffy/manifest.py" = ["E501"] -# trailing whitespace lives inside prompt / HTML template string literals -"cliffy/ai.py" = ["W291"] +# trailing whitespace lives inside HTML template string literals "cliffy/doc.py" = ["W293"] [tool.ruff.lint.pydocstyle] @@ -124,6 +123,9 @@ filterwarnings = [ "ignore:datetime.datetime.utcfromtimestamp:DeprecationWarning", # pydantic-ai 2.x run_sync() calls asyncio.get_event_loop() without a running loop (warns on 3.12/3.13) "ignore:There is no current event loop:DeprecationWarning", + # google-genai (the SDK behind pydantic-ai's `google:` models) compares typing._UnionGenericAlias at import + # (deprecated since Python 3.14) + "ignore:'_UnionGenericAlias' is deprecated:DeprecationWarning", ] [tool.mypy] diff --git a/tests/test_ai.py b/tests/test_ai.py index b01e227..757f86e 100644 --- a/tests/test_ai.py +++ b/tests/test_ai.py @@ -1,7 +1,60 @@ +"""Tests for ``cli ai``. Every model here is a pydantic-ai test or function model: nothing touches the network.""" + +import subprocess +import sys + import pytest +import yaml from click.testing import CliRunner +from pydantic_ai.exceptions import ModelHTTPError +from pydantic_ai.messages import ModelMessage, ModelResponse, RetryPromptPart, TextPart, UserPromptPart +from pydantic_ai.models.function import AgentInfo, FunctionModel +from pydantic_ai.models.test import TestModel + +from cliffy import ai as cliffy_ai +from cliffy.ai import ManifestOutputError, ai, default_model, validate_manifest_yaml +from cliffy.cli import cli +from cliffy.manifest import CLIManifest + +API_KEY_VARS = [var for env_vars, _ in cliffy_ai.PROVIDER_DEFAULTS for var in env_vars] + +VALID_MANIFEST = """\ +manifestVersion: v3 +name: greeter +version: 0.1.0 +help: Greets people +commands: + hello: + help: Say hello + params: [name: str!] + run: | + print(f"Hello, {name}!") +examples: + - greeter hello World +""" + +BROKEN_PYTHON_MANIFEST = VALID_MANIFEST.replace('print(f"Hello, {name}!")', "print(f'Hello, {name}!'") + + +class ScriptedModel: + """Function model that replies with the given texts in order and records what it was sent.""" + + def __init__(self, *replies: str) -> None: + self.replies = list(replies) + self.requests: list[list[ModelMessage]] = [] + self.instructions: list[str | None] = [] + self.model = FunctionModel(self._respond) -from cliffy.ai import ai + def _respond(self, messages: list[ModelMessage], info: AgentInfo) -> ModelResponse: + self.requests.append(messages) + self.instructions.append(info.instructions) + return ModelResponse(parts=[TextPart(self.replies[min(len(self.requests), len(self.replies)) - 1])]) + + def parts(self, part_type: type[UserPromptPart] | type[RetryPromptPart]) -> list[str]: + """Content of all parts of a type in the last request, i.e. the whole conversation so far.""" + return [ + str(part.content) for message in self.requests[-1] for part in message.parts if isinstance(part, part_type) + ] @pytest.fixture @@ -9,14 +62,419 @@ def runner(): return CliRunner() -def test_generate_command(runner, tmp_path_factory): - tmp_dir = tmp_path_factory.mktemp("data") - result = runner.invoke(ai, ["generate", "test_cli", "This is a test CLI description.", "-m", "test", "-o", tmp_dir]) +@pytest.fixture(autouse=True) +def no_ambient_ai_config(monkeypatch): + """Keep the developer's own API keys and model choice out of the tests.""" + for var in [*API_KEY_VARS, cliffy_ai.MODEL_ENV_VAR]: + monkeypatch.delenv(var, raising=False) + + +@pytest.fixture +def scripted(monkeypatch): + """Make the commands use a scripted model, whatever model name they are given.""" + + def install(*replies: str) -> ScriptedModel: + model = ScriptedModel(*replies) + monkeypatch.setattr(cliffy_ai, "resolve_model", lambda name: model.model) + return model + + return install + + +# --- import cost and the optional extra --------------------------------------------------------- + + +def test_importing_cliffy_does_not_import_pydantic_ai(): + code = ( + "import sys, cliffy, cliffy.ai, cliffy.cli, cliffy.run\n" + "loaded = sorted(name for name in sys.modules if name.partition('.')[0] == 'pydantic_ai')\n" + "assert not loaded, loaded\n" + ) + result = subprocess.run([sys.executable, "-c", code], capture_output=True, text=True) + assert result.returncode == 0, result.stderr + + +def test_ai_group_is_registered_on_the_main_cli(runner): + result = runner.invoke(cli, ["ai", "--help"]) + assert result.exit_code == 0 + assert "generate" in result.output + assert "ask" in result.output + + +@pytest.mark.parametrize("args", [["ask", "What is cliffy?"], ["generate", "greeter", "Greets people"]]) +def test_missing_extra_is_reported(runner, monkeypatch, isolated_fs, args): + monkeypatch.setitem(sys.modules, "pydantic_ai", None) # makes `import pydantic_ai` fail + result = runner.invoke(ai, args) + assert result.exit_code == 1 + assert "cliffy[ai]" in result.output + assert not (isolated_fs / "greeter.yaml").exists() + + +def test_unrelated_import_errors_are_not_mistaken_for_a_missing_extra(monkeypatch): + monkeypatch.setitem(sys.modules, "pydantic_ai", None) + monkeypatch.setattr( + "builtins.__import__", lambda *args, **kwargs: (_ for _ in ()).throw(ModuleNotFoundError(name="httpx")) + ) + with pytest.raises(ModuleNotFoundError): + cliffy_ai.require_ai_extra() + + +def test_preview_needs_neither_the_extra_nor_a_model(runner, monkeypatch): + monkeypatch.setitem(sys.modules, "pydantic_ai", None) + result = runner.invoke(ai, ["generate", "greeter", "Greets people", "--preview"]) + assert result.exit_code == 0 + assert "CLI named `greeter`" in result.output + assert "Greets people" in result.output + + result = runner.invoke(ai, ["ask", "How do aliases work?", "--preview"]) + assert result.exit_code == 0 + assert "How do aliases work?" in result.output + + +# --- model selection ---------------------------------------------------------------------------- + + +def test_default_model_without_any_api_key(): + assert default_model() == cliffy_ai.FALLBACK_MODEL + + +@pytest.mark.parametrize(("env_vars", "model"), cliffy_ai.PROVIDER_DEFAULTS) +def test_default_model_follows_the_api_key_that_is_set(monkeypatch, env_vars, model): + for var in env_vars: + monkeypatch.setenv(var, "dummy") + assert default_model() == model + monkeypatch.delenv(var) + + +def test_default_models_have_a_provider_prefix(): + for _, model in cliffy_ai.PROVIDER_DEFAULTS: + provider, _, name = model.partition(":") + assert provider and name + + +def test_missing_api_key_is_reported(runner): + result = runner.invoke(ai, ["ask", "What is cliffy?"]) + assert result.exit_code == 1 + assert cliffy_ai.FALLBACK_MODEL in result.output + assert "GOOGLE_API_KEY" in result.output + assert cliffy_ai.MODEL_ENV_VAR in result.output + assert "Traceback" not in result.output + + +def test_unknown_model_is_reported(runner): + result = runner.invoke(ai, ["ask", "What is cliffy?", "-m", "not-a-model"]) + assert result.exit_code == 1 + assert "cannot use AI model `not-a-model`" in result.output + + +def test_model_flag_selects_the_model(runner): + result = runner.invoke(ai, ["ask", "What is cliffy?", "-m", "test"]) + assert result.exit_code == 0 + assert "success (no tool calls)" in result.output + + +def test_model_env_var_selects_the_model(runner, monkeypatch): + monkeypatch.setenv(cliffy_ai.MODEL_ENV_VAR, "test") + result = runner.invoke(ai, ["ask", "What is cliffy?"]) assert result.exit_code == 0 - assert "test_cli.yaml" in result.output + assert "success (no tool calls)" in result.output -def test_ask_command(runner): +def test_model_flag_wins_over_env_var(runner, monkeypatch): + monkeypatch.setenv(cliffy_ai.MODEL_ENV_VAR, "not-a-model") result = runner.invoke(ai, ["ask", "What is cliffy?", "-m", "test"]) assert result.exit_code == 0 + + +def test_missing_provider_sdk_is_reported(runner, monkeypatch): + def no_sdk(model: str) -> None: + raise ImportError("Please install `groq` to use the Groq provider") + + monkeypatch.setattr("pydantic_ai.models.infer_model", no_sdk) + result = runner.invoke(ai, ["ask", "What is cliffy?", "-m", "groq:some-model"]) + assert result.exit_code == 1 + assert "cannot use AI model `groq:some-model`: Please install `groq`" in result.output + assert "Traceback" not in result.output + + +def test_resolve_model_returns_a_pydantic_ai_model(): + assert isinstance(cliffy_ai.resolve_model("test"), TestModel) + + +# --- ai generate -------------------------------------------------------------------------------- + + +def test_generate_writes_a_validated_manifest(runner, scripted, isolated_fs): + model = scripted(VALID_MANIFEST) + result = runner.invoke(ai, ["generate", "greeter", "Greets people"]) + + assert result.exit_code == 0, result.output + assert "greeter.yaml" in result.output assert "token usage:" in result.output + written = (isolated_fs / "greeter.yaml").read_text() + assert written == VALID_MANIFEST + assert CLIManifest.model_validate(yaml.safe_load(written)).name == "greeter" + + assert len(model.requests) == 1 + assert model.parts(UserPromptPart) == ["Greets people"] + instructions = model.instructions[0] + assert instructions is not None + assert "CLI named `greeter`" in instructions + assert '"title": "CLIManifest"' in instructions + + +def test_generated_manifest_loads_as_a_cli(runner, scripted, isolated_fs): + scripted(VALID_MANIFEST) + assert runner.invoke(ai, ["generate", "greeter", "Greets people"]).exit_code == 0 + + result = runner.invoke(cli, ["run", "greeter.yaml", "--", "hello", "World"]) + assert result.exit_code == 0, result.output + assert "Hello, World!" in result.output + + +def test_generate_strips_markdown_fences(runner, scripted, isolated_fs): + scripted(f"Here is your manifest:\n\n```yaml\n{VALID_MANIFEST}```\n\nEnjoy!") + result = runner.invoke(ai, ["generate", "greeter", "Greets people"]) + + assert result.exit_code == 0, result.output + assert (isolated_fs / "greeter.yaml").read_text() == VALID_MANIFEST + + +def test_generate_writes_to_output_dir(runner, scripted, tmp_path): + scripted(VALID_MANIFEST) + result = runner.invoke(ai, ["generate", "greeter", "Greets people", "-o", str(tmp_path)]) + + assert result.exit_code == 0, result.output + assert (tmp_path / "greeter.yaml").read_text() == VALID_MANIFEST + + +def test_generate_sends_validation_errors_back_to_the_model(runner, scripted, isolated_fs): + model = scripted("commands: {}", VALID_MANIFEST) + result = runner.invoke(ai, ["generate", "greeter", "Greets people"]) + + assert result.exit_code == 0, result.output + assert (isolated_fs / "greeter.yaml").read_text() == VALID_MANIFEST + assert len(model.requests) == 2 + (feedback,) = model.parts(RetryPromptPart) + assert "does not match the schema" in feedback + assert "- name: Field required" in feedback + assert "- version: Field required" in feedback + + +def test_generate_gives_up_after_retries_and_writes_nothing(runner, scripted, isolated_fs): + model = scripted("this is not a manifest") + result = runner.invoke(ai, ["generate", "greeter", "Greets people", "--retries", "1"]) + + assert result.exit_code == 1 + assert "no manifest written" in result.output + assert "must be a YAML mapping" in result.output + assert "Traceback" not in result.output + assert len(model.requests) == 2 + assert not (isolated_fs / "greeter.yaml").exists() + + +def test_generate_without_retries_asks_once(runner, scripted, isolated_fs): + model = scripted("this is not a manifest") + result = runner.invoke(ai, ["generate", "greeter", "Greets people", "--retries", "0"]) + + assert result.exit_code == 1 + assert len(model.requests) == 1 + assert not (isolated_fs / "greeter.yaml").exists() + + +def test_generate_with_the_builtin_test_model_fails_cleanly(runner, isolated_fs): + result = runner.invoke(ai, ["generate", "greeter", "Greets people", "-m", "test"]) + + assert result.exit_code == 1 + assert "no manifest written" in result.output + assert not (isolated_fs / "greeter.yaml").exists() + + +def test_generate_refuses_to_overwrite_without_force(runner, scripted, isolated_fs): + model = scripted(VALID_MANIFEST) + existing = isolated_fs / "greeter.yaml" + existing.write_text("mine") + + result = runner.invoke(ai, ["generate", "greeter", "Greets people"]) + assert result.exit_code == 1 + assert "already exists" in result.output + assert existing.read_text() == "mine" + assert model.requests == [] # refused before spending any tokens + + result = runner.invoke(ai, ["generate", "greeter", "Greets people", "--force"]) + assert result.exit_code == 0, result.output + assert existing.read_text() == VALID_MANIFEST + + +@pytest.mark.parametrize("name", ["../escape", "has space", "", "-leading-dash", "a/b"]) +def test_generate_rejects_unsafe_cli_names(runner, scripted, isolated_fs, name): + model = scripted(VALID_MANIFEST) + result = runner.invoke(ai, ["generate", "--", name, "Greets people"]) + + assert result.exit_code == 1 + assert "invalid CLI name" in result.output + assert model.requests == [] + assert list(isolated_fs.iterdir()) == [] + + +def test_generate_reports_a_failed_request(runner, monkeypatch, isolated_fs): + def unauthorized(messages: list[ModelMessage], info: AgentInfo) -> ModelResponse: + raise ModelHTTPError(status_code=401, model_name="function", body={"error": "invalid api key"}) + + monkeypatch.setattr(cliffy_ai, "resolve_model", lambda name: FunctionModel(unauthorized)) + result = runner.invoke(ai, ["generate", "greeter", "Greets people"]) + + assert result.exit_code == 1 + assert "failed: status_code: 401" in result.output + assert "Traceback" not in result.output + assert not (isolated_fs / "greeter.yaml").exists() + + +def test_generate_reports_an_empty_reply(runner, monkeypatch, isolated_fs): + def silent(messages: list[ModelMessage], info: AgentInfo) -> ModelResponse: + return ModelResponse(parts=[]) + + monkeypatch.setattr(cliffy_ai, "resolve_model", lambda name: FunctionModel(silent)) + result = runner.invoke(ai, ["generate", "greeter", "Greets people", "--retries", "0"]) + + assert result.exit_code == 1 + assert "failed" in result.output + assert "Traceback" not in result.output + assert not (isolated_fs / "greeter.yaml").exists() + + +def test_generate_passes_max_tokens_to_the_model(runner, monkeypatch, isolated_fs): + seen: list[object] = [] + + def respond(messages: list[ModelMessage], info: AgentInfo) -> ModelResponse: + seen.append(info.model_settings) + return ModelResponse(parts=[TextPart(VALID_MANIFEST)]) + + monkeypatch.setattr(cliffy_ai, "resolve_model", lambda name: FunctionModel(respond)) + result = runner.invoke(ai, ["generate", "greeter", "Greets people", "--max-tokens", "1234"]) + + assert result.exit_code == 0, result.output + assert seen == [{"max_tokens": 1234}] + + +# --- manifest validation ------------------------------------------------------------------------ + + +def test_validate_accepts_a_valid_manifest(): + assert validate_manifest_yaml(VALID_MANIFEST, "greeter") == VALID_MANIFEST + + +@pytest.mark.parametrize( + ("reply", "problem"), + [ + ("key: [unclosed", "not valid YAML"), + ("- just\n- a list", "must be a YAML mapping"), + ("name: greeter\nversion: 0.1.0", "- commands: Field required"), + (VALID_MANIFEST.replace("manifestVersion: v3", "manifestVersion: v2"), "v2 schema is deprecated"), + (VALID_MANIFEST.replace("name: greeter", "name: other"), "`name` must be `greeter`"), + (VALID_MANIFEST + "includes: [other.yaml]\n", "remove `includes`"), + (BROKEN_PYTHON_MANIFEST, "not valid Python"), + ], +) +def test_validate_explains_what_is_wrong(reply, problem): + with pytest.raises(ManifestOutputError, match=problem): + validate_manifest_yaml(reply, "greeter") + + +def test_validate_reports_manifests_the_commander_cannot_handle(monkeypatch): + def broken(*args: object, **kwargs: object) -> None: + raise KeyError("no such template") + + monkeypatch.setattr("cliffy.commander.generate_cli", broken) + with pytest.raises(ManifestOutputError, match="could not generate a CLI from the manifest: 'no such template'"): + validate_manifest_yaml(VALID_MANIFEST, "greeter") + + +def test_validate_does_not_render_or_compile_jinja_manifests(): + manifest = VALID_MANIFEST.replace('print(f"Hello, {name}!")', "{% if x %}print(name){% endif %}") + assert validate_manifest_yaml(manifest, "greeter") == manifest + + +def test_validate_accepts_the_init_template(): + template = CLIManifest.get_raw_template("greeter", json_schema=False) + assert validate_manifest_yaml(template, "greeter").strip() == template.strip() + + +# --- ai ask ------------------------------------------------------------------------------------- + + +def test_ask_prints_the_answer(runner, scripted): + model = scripted("Use a `.` between the group and the subcommand.") + result = runner.invoke(ai, ["ask", "How do I nest commands?"]) + + assert result.exit_code == 0, result.output + assert "Use a `.` between the group and the subcommand." in result.stdout + assert "token usage:" in result.stderr + assert "token usage:" not in result.stdout + assert model.parts(UserPromptPart) == ["How do I nest commands?"] + + +def test_ask_is_grounded_in_the_shipped_schema_and_commands(runner, scripted): + model = scripted("answer") + assert runner.invoke(ai, ["ask", "How do I nest commands?"]).exit_code == 0 + + instructions = model.instructions[0] + assert instructions is not None + # the real JSON schema, field by field + for field in CLIManifest.model_fields: + assert f'"{field}"' in instructions + # the annotated template that `cli init` writes + assert CLIManifest.get_template("mycli", json_schema=False) in instructions + # every public cliffy command, read from the click group + for name, command in cli.commands.items(): + if not command.hidden and name != "ai": + assert f"`cli {name} " in instructions + assert "`cli ai generate [OPTIONS] CLI_NAME DESCRIPTION`" in instructions + assert "`rm` = `remove`" in instructions + + +def test_ask_includes_a_manifest_file_as_reference(runner, scripted, isolated_fs): + model = scripted("answer") + (isolated_fs / "greeter.yaml").write_text(VALID_MANIFEST) + result = runner.invoke(ai, ["ask", "What does hello do?", "--cli", "greeter.yaml"]) + + assert result.exit_code == 0, result.output + (prompt,) = model.parts(UserPromptPart) + assert prompt.startswith("What does hello do?") + assert VALID_MANIFEST.strip() in prompt + + +def test_ask_includes_a_loaded_cli_as_reference(runner, scripted, isolated_fs): + manifest = VALID_MANIFEST.replace("greeter", "aiasktestcli") + (isolated_fs / "aiasktestcli.yaml").write_text(manifest) + assert runner.invoke(cli, ["load", "aiasktestcli.yaml"]).exit_code == 0 + try: + model = scripted("answer") + result = runner.invoke(ai, ["ask", "What does hello do?", "--cli", "aiasktestcli"]) + finally: + runner.invoke(cli, ["rm", "aiasktestcli"]) + + assert result.exit_code == 0, result.output + (prompt,) = model.parts(UserPromptPart) + assert manifest.strip() in prompt + + +def test_ask_rejects_a_cli_that_is_not_loaded(runner, scripted): + model = scripted("answer") + result = runner.invoke(ai, ["ask", "What does hello do?", "--cli", "no-such-cli-loaded"]) + + assert result.exit_code == 1 + assert "no-such-cli-loaded is not a loaded CLI" in result.output + assert model.requests == [] + + +def test_ask_reports_a_failed_request(runner, monkeypatch): + def overloaded(messages: list[ModelMessage], info: AgentInfo) -> ModelResponse: + raise ModelHTTPError(status_code=529, model_name="function", body={"error": "overloaded"}) + + monkeypatch.setattr(cliffy_ai, "resolve_model", lambda name: FunctionModel(overloaded)) + result = runner.invoke(ai, ["ask", "What is cliffy?"]) + + assert result.exit_code == 1 + assert "failed: status_code: 529" in result.output + assert "Traceback" not in result.output