llamactl shell autocomplete (#521)

This commit is contained in:
Adrian Lyjak
2026-04-10 18:49:54 -04:00
committed by GitHub
parent 1d612a499f
commit e08c17c47f
15 changed files with 1049 additions and 156 deletions
+5
View File
@@ -0,0 +1,5 @@
---
"llamactl": minor
---
Add shell tab-completion support with `llamactl completion generate` and `llamactl completion install`
@@ -713,10 +713,10 @@ def test_validate_detects_circular_resource_dependency() -> None:
class B:
pass
def cyclic_factory_a(b: Annotated[B, "placeholder"]) -> A: # type: ignore
def cyclic_factory_a(b: Annotated[B, "placeholder"]) -> A:
return A()
def cyclic_factory_b(a: Annotated[A, "placeholder"]) -> B: # type: ignore
def cyclic_factory_b(a: Annotated[A, "placeholder"]) -> B:
return B()
cyclic_res_a = Resource(cyclic_factory_a)
@@ -3,6 +3,7 @@ import warnings
from llama_agents.cli.commands.agentcore import agentcore
from llama_agents.cli.commands.auth import auth
from llama_agents.cli.commands.completion import completion
from llama_agents.cli.commands.deployment import deployments
from llama_agents.cli.commands.dev import dev
from llama_agents.cli.commands.env import env_group
@@ -28,6 +29,7 @@ def main() -> None:
__all__ = [
"app",
"completion",
"deployments",
"auth",
"serve",
@@ -10,6 +10,7 @@ from pathlib import Path
from typing import TYPE_CHECKING, Any
import click
from llama_agents.cli.param_types import ProfileType, ProjectType
from llama_agents.cli.styles import (
ACTIVE_INDICATOR,
HEADER_COLOR,
@@ -207,7 +208,7 @@ def config_database() -> None:
@auth.command("switch")
@global_options
@click.argument("name", required=False)
@click.argument("name", required=False, type=ProfileType())
@interactive_option
def switch_profile(name: str | None, interactive: bool) -> None:
"""Switch to a different profile"""
@@ -228,7 +229,7 @@ def switch_profile(name: str | None, interactive: bool) -> None:
@auth.command("logout")
@global_options
@click.argument("name", required=False)
@click.argument("name", required=False, type=ProfileType())
@interactive_option
def delete_profile(name: str | None, interactive: bool) -> None:
"""Logout from a profile and wipe all associated data"""
@@ -282,7 +283,7 @@ def me() -> None:
# Projects commands
@auth.command("project")
@click.argument("project_id", required=False)
@click.argument("project_id", required=False, type=ProjectType())
@interactive_option
@global_options
def change_project(project_id: str | None, interactive: bool) -> None:
@@ -0,0 +1,300 @@
# SPDX-License-Identifier: MIT
# Copyright (c) 2026 LlamaIndex Inc.
from __future__ import annotations
import os
import re
from pathlib import Path
import click
from click.shell_completion import get_completion_class
from llama_agents.cli.app import app
from llama_agents.cli.options import global_options
from llama_agents.cli.paths import (
bash_completion_dir,
bash_rc_path,
fish_completion_dir,
zsh_completion_dir,
zsh_rc_path,
)
from rich import print as rprint
@app.group(help="Shell completion helpers.", no_args_is_help=True)
@global_options
def completion() -> None:
pass
def _completion_source(shell: str) -> str:
"""Build and return the shell completion script for the given shell."""
ctx = click.get_current_context()
root_cmd = ctx.find_root().command
cls = get_completion_class(shell)
if cls is None:
raise click.ClickException(f"Unsupported shell: {shell}")
comp = cls(root_cmd, {}, "llamactl", "_LLAMACTL_COMPLETE")
return comp.source()
@completion.command("generate")
@click.argument("shell", type=click.Choice(["bash", "zsh", "fish"]))
@global_options
def generate(shell: str) -> None:
"""Print shell completion script to stdout.
Example: llamactl completion generate zsh > ~/.zfunc/_llamactl
"""
click.echo(_completion_source(shell))
@completion.command("install")
@click.option(
"--shell",
type=click.Choice(["bash", "zsh", "fish"]),
default=None,
help="Override auto-detected shell.",
)
@click.option(
"--dry-run",
is_flag=True,
help="Show what would be done without modifying anything.",
)
@global_options
def install(shell: str | None, dry_run: bool) -> None:
"""Auto-detect your shell and install completions.
Example: llamactl completion install
"""
if shell is None:
shell = _detect_shell()
source = _completion_source(shell)
if shell == "bash":
_install_bash(source, dry_run)
elif shell == "zsh":
_install_zsh(source, dry_run)
elif shell == "fish":
_install_fish(source, dry_run)
if not dry_run:
rprint(f"\nDetected shell: [bold]{shell}[/bold]")
rprint("Restart your shell or source the config to activate completions.")
# ---------------------------------------------------------------------------
# Shell-specific install helpers
# ---------------------------------------------------------------------------
_MARKER = "# llamactl shell completion"
_ZSH_BLOCK_START = "# >>> llamactl completion >>>"
_ZSH_BLOCK_END = "# <<< llamactl completion <<<"
_ZSH_FPATH_LINE = "fpath=(~/.zfunc $fpath)"
_ZSH_COMPINIT_LINE = "autoload -Uz compinit && compinit"
def _detect_shell() -> str:
shell_env = os.environ.get("SHELL", "")
name = os.path.basename(shell_env)
if name in ("bash", "zsh", "fish"):
return name
return "bash"
def _install_bash(source: str, dry_run: bool) -> None:
comp_dir = bash_completion_dir()
target = comp_dir / "llamactl"
if dry_run:
rprint(f"Would write completion script to [cyan]{target}[/cyan]")
return
comp_dir.mkdir(parents=True, exist_ok=True)
target.write_text(source)
rprint(f"Wrote completions to [cyan]{target}[/cyan]")
# Ensure .bashrc sources the completion dir
bashrc = bash_rc_path()
_ensure_source_line(
bashrc,
f"source {target}",
dry_run,
)
def _install_zsh(source: str, dry_run: bool) -> None:
zfunc = zsh_completion_dir()
target = zfunc / "_llamactl"
if dry_run:
rprint(f"Would write completion script to [cyan]{target}[/cyan]")
rprint(
"Would ensure [cyan]~/.zfunc[/cyan] is in fpath in [cyan]~/.zshrc[/cyan]"
)
return
zfunc.mkdir(parents=True, exist_ok=True)
target.write_text(source)
rprint(f"Wrote completions to [cyan]{target}[/cyan]")
zshrc = zsh_rc_path()
_ensure_zsh_fpath(zshrc, dry_run)
def _install_fish(source: str, dry_run: bool) -> None:
comp_dir = fish_completion_dir()
target = comp_dir / "llamactl.fish"
if dry_run:
rprint(f"Would write completion script to [cyan]{target}[/cyan]")
return
comp_dir.mkdir(parents=True, exist_ok=True)
target.write_text(source)
rprint(f"Wrote completions to [cyan]{target}[/cyan]")
def _ensure_zsh_fpath(zshrc: Path, dry_run: bool) -> None:
"""Ensure llamactl's zsh block appears before the first live compinit."""
if zshrc.exists():
content = zshrc.read_text()
else:
content = ""
updated_content, actions = _plan_zshrc_update(content)
if not actions:
return
if dry_run:
for action in actions:
rprint(f"Would update [cyan]{zshrc}[/cyan]: {action}")
return
zshrc.write_text(updated_content)
for action in actions:
rprint(f"Updated [cyan]{zshrc}[/cyan]: {action}")
def _plan_zshrc_update(content: str) -> tuple[str, list[str]]:
lines = content.splitlines()
had_trailing_newline = content.endswith("\n")
sanitized_lines, removed_block, removed_compinit = _strip_managed_zsh_lines(lines)
first_compinit_index = _first_live_compinit_index(sanitized_lines)
first_fpath_index = _first_live_zfunc_fpath_index(sanitized_lines)
needs_fpath_block = (
first_fpath_index is None
or first_compinit_index is not None
and first_fpath_index > first_compinit_index
)
updated_lines = list(sanitized_lines)
actions: list[str] = []
if needs_fpath_block:
insert_at = (
first_compinit_index
if first_compinit_index is not None
else len(updated_lines)
)
updated_lines[insert_at:insert_at] = _managed_zsh_fpath_block()
actions.append(
"ensure ~/.zfunc is added before compinit via the llamactl block"
)
elif removed_block:
actions.append("remove stale llamactl zsh block")
has_compinit = _first_live_compinit_index(updated_lines) is not None
if not has_compinit:
if updated_lines and updated_lines[-1] != "":
updated_lines.append("")
updated_lines.append(f"{_ZSH_COMPINIT_LINE} {_MARKER}")
actions.append("add compinit because none was configured")
elif removed_compinit:
actions.append("remove stale llamactl-managed compinit line")
updated_content = "\n".join(updated_lines)
if updated_content and had_trailing_newline:
updated_content += "\n"
elif updated_content and not had_trailing_newline:
updated_content += "\n"
return updated_content, actions
def _strip_managed_zsh_lines(
lines: list[str],
) -> tuple[list[str], bool, bool]:
sanitized: list[str] = []
in_block = False
removed_block = False
removed_compinit = False
for line in lines:
stripped = line.strip()
if stripped == _ZSH_BLOCK_START:
in_block = True
removed_block = True
continue
if stripped == _ZSH_BLOCK_END:
in_block = False
continue
if in_block:
continue
if stripped == f"{_ZSH_FPATH_LINE} {_MARKER}":
removed_block = True
continue
if stripped == f"{_ZSH_COMPINIT_LINE} {_MARKER}":
removed_compinit = True
continue
sanitized.append(line)
return sanitized, removed_block, removed_compinit
def _first_live_compinit_index(lines: list[str]) -> int | None:
for index, line in enumerate(lines):
if _is_live_compinit_line(line):
return index
return None
def _first_live_zfunc_fpath_index(lines: list[str]) -> int | None:
for index, line in enumerate(lines):
stripped = line.lstrip()
if stripped.startswith("#"):
continue
if "~/.zfunc" in line and "fpath" in line:
return index
return None
def _is_live_compinit_line(line: str) -> bool:
stripped = line.lstrip()
if stripped.startswith("#"):
return False
return bool(re.search(r"\bcompinit\b", stripped))
def _managed_zsh_fpath_block() -> list[str]:
return [
_ZSH_BLOCK_START,
f"{_ZSH_FPATH_LINE} {_MARKER}",
_ZSH_BLOCK_END,
]
def _ensure_source_line(rc_file: Path, line: str, dry_run: bool) -> None:
"""Append line to rc_file if not already present."""
if rc_file.exists():
content = rc_file.read_text()
if line in content:
return
else:
content = ""
if dry_run:
rprint(f"Would add to [cyan]{rc_file}[/cyan]: {line}")
return
with rc_file.open("a") as f:
f.write(f"\n{line} {_MARKER}\n")
rprint(f"Added to [cyan]{rc_file}[/cyan]: {line}")
@@ -11,6 +11,7 @@ import subprocess
import click
from llama_agents.cli.commands.auth import validate_authenticated_profile
from llama_agents.cli.param_types import DeploymentType, GitShaType
from llama_agents.cli.styles import HEADER_COLOR, MUTED_COL, PRIMARY_COL, WARNING
from llama_agents.core.schema.deployments import (
INTERNAL_CODE_REPO_SCHEME,
@@ -94,7 +95,7 @@ def list_deployments(interactive: bool) -> None:
@deployments.command("get")
@global_options
@click.argument("deployment_id", required=False)
@click.argument("deployment_id", required=False, type=DeploymentType())
@interactive_option
def get_deployment(deployment_id: str | None, interactive: bool) -> None:
"""Get details of a specific deployment"""
@@ -178,7 +179,7 @@ def create_deployment(
@deployments.command("configure-git-remote")
@global_options
@click.argument("deployment_id", required=False)
@click.argument("deployment_id", required=False, type=DeploymentType())
@interactive_option
def configure_git_remote_cmd(deployment_id: str | None, interactive: bool) -> None:
"""Configure a git remote for a deployment.
@@ -227,7 +228,7 @@ def configure_git_remote_cmd(deployment_id: str | None, interactive: bool) -> No
@deployments.command("delete")
@global_options
@click.argument("deployment_id", required=False)
@click.argument("deployment_id", required=False, type=DeploymentType())
@interactive_option
def delete_deployment(deployment_id: str | None, interactive: bool) -> None:
"""Delete a deployment"""
@@ -260,7 +261,7 @@ def delete_deployment(deployment_id: str | None, interactive: bool) -> None:
@deployments.command("edit")
@global_options
@click.argument("deployment_id", required=False)
@click.argument("deployment_id", required=False, type=DeploymentType())
@interactive_option
def edit_deployment(deployment_id: str | None, interactive: bool) -> None:
"""Interactively edit a deployment"""
@@ -342,7 +343,7 @@ def _internal_push_refspec(git_ref: str | None) -> tuple[str, str]:
@deployments.command("update")
@global_options
@click.argument("deployment_id", required=False)
@click.argument("deployment_id", required=False, type=DeploymentType())
@click.option(
"--git-ref",
help="Reference branch, tag, or commit SHA for the deployment. If not provided, the current reference and latest commit on it will be used.",
@@ -402,7 +403,7 @@ def refresh_deployment(
@deployments.command("history")
@global_options
@click.argument("deployment_id", required=False)
@click.argument("deployment_id", required=False, type=DeploymentType())
@interactive_option
def show_history(deployment_id: str | None, interactive: bool) -> None:
"""Show release history for a deployment."""
@@ -442,10 +443,12 @@ def show_history(deployment_id: str | None, interactive: bool) -> None:
raise click.Abort()
@deployments.command("rollback", hidden=True)
@deployments.command("rollback")
@global_options
@click.argument("deployment_id", required=False)
@click.option("--git-sha", required=False, help="Git SHA to roll back to")
@click.argument("deployment_id", required=False, type=DeploymentType())
@click.option(
"--git-sha", required=False, type=GitShaType(), help="Git SHA to roll back to"
)
@interactive_option
def rollback(deployment_id: str | None, git_sha: str | None, interactive: bool) -> None:
"""Rollback a deployment to a previous git sha."""
@@ -5,6 +5,7 @@ from typing import TYPE_CHECKING
import click
from llama_agents.cli.config.schema import Environment
from llama_agents.cli.param_types import EnvironmentType
from llama_agents.cli.styles import (
ACTIVE_INDICATOR,
HEADER_COLOR,
@@ -113,7 +114,7 @@ def add_environment_cmd(api_url: str | None, interactive: bool) -> None:
@env_group.command("delete")
@click.argument("api_url", required=False)
@click.argument("api_url", required=False, type=EnvironmentType())
@interactive_option
@global_options
def delete_environment_cmd(api_url: str | None, interactive: bool) -> None:
@@ -148,7 +149,7 @@ def delete_environment_cmd(api_url: str | None, interactive: bool) -> None:
@env_group.command("switch")
@click.argument("api_url", required=False)
@click.argument("api_url", required=False, type=EnvironmentType())
@interactive_option
@global_options
def switch_environment_cmd(api_url: str | None, interactive: bool) -> None:
@@ -3,7 +3,6 @@ from __future__ import annotations
import os
import shutil
import subprocess
from dataclasses import dataclass
from pathlib import Path
import click
@@ -13,7 +12,14 @@ from llama_agents.cli.options import (
global_options,
interactive_option,
)
from llama_agents.cli.param_types import TemplateType
from llama_agents.cli.styles import HEADER_COLOR_HEX
from llama_agents.cli.templates import (
ALL_TEMPLATES,
HEADLESS_TEMPLATES,
UI_TEMPLATES,
TemplateOption,
)
from rich import print as rprint
from rich.text import Text
@@ -28,6 +34,7 @@ _ClickPath = getattr(click, "Path")
)
@click.option(
"--template",
type=TemplateType(),
help="The template to use for the new app",
)
@click.option(
@@ -67,132 +74,6 @@ def _create(
) -> None:
import questionary
@dataclass
class TemplateOption:
id: str
name: str
description: str
source: GithubTemplateRepo
llama_cloud: bool
@dataclass
class GithubTemplateRepo:
url: str
ui_options = [
TemplateOption(
id="basic-ui",
name="Basic UI",
description="A basic starter workflow with a React Vite UI",
source=GithubTemplateRepo(
url="https://github.com/run-llama/template-workflow-basic-ui"
),
llama_cloud=False,
),
TemplateOption(
id="showcase",
name="Showcase",
description="A collection of workflow and UI patterns to build LlamaDeploy apps",
source=GithubTemplateRepo(
url="https://github.com/run-llama/template-workflow-showcase"
),
llama_cloud=False,
),
TemplateOption(
id="document-qa",
name="Document Question & Answer",
description="Upload documents and run question answering through a React UI",
source=GithubTemplateRepo(
url="https://github.com/run-llama/template-workflow-document-qa"
),
llama_cloud=True,
),
TemplateOption(
id="extraction-review",
name="Extraction Agent with Review UI",
description="Extract data from documents using a custom schema and Llama Cloud. Includes a UI to review and correct the results",
source=GithubTemplateRepo(
url="https://github.com/run-llama/template-workflow-data-extraction"
),
llama_cloud=True,
),
TemplateOption(
id="classify-extract-sec",
name="SEC Insights",
description="Upload SEC filings, classifying them to the appropriate type and extracting key information",
source=GithubTemplateRepo(
url="https://github.com/run-llama/template-workflow-classify-extract-sec"
),
llama_cloud=True,
),
TemplateOption(
id="extract-reconcile-invoice",
name="Invoice Extraction & Reconciliation",
description="Extract and reconcile invoice data against contracts",
source=GithubTemplateRepo(
url="https://github.com/run-llama/template-workflow-extract-reconcile-invoice"
),
llama_cloud=True,
),
]
headless_options = [
TemplateOption(
id="basic",
name="Basic Workflow",
description="A base example that showcases usage patterns for workflows",
source=GithubTemplateRepo(
url="https://github.com/run-llama/template-workflow-basic"
),
llama_cloud=False,
),
TemplateOption(
id="document_parsing",
name="Document Parser",
description="A workflow that, using LlamaParse, parses unstructured documents and returns their raw text content",
source=GithubTemplateRepo(
url="https://github.com/run-llama/template-workflow-document-parsing"
),
llama_cloud=True,
),
TemplateOption(
id="human_in_the_loop",
name="Human in the Loop",
description="A workflow showcasing how to use human in the loop with LlamaIndex workflows",
source=GithubTemplateRepo(
url="https://github.com/run-llama/template-workflow-human-in-the-loop"
),
llama_cloud=False,
),
TemplateOption(
id="invoice_extraction",
name="Invoice Extraction",
description="A workflow that, given an invoice, extracts several key details using LlamaExtract",
source=GithubTemplateRepo(
url="https://github.com/run-llama/template-workflow-invoice-extraction"
),
llama_cloud=True,
),
TemplateOption(
id="rag",
name="RAG",
description="A workflow that embeds, indexes and queries your documents on the fly, providing you with a simple RAG pipeline",
source=GithubTemplateRepo(
url="https://github.com/run-llama/template-workflow-rag"
),
llama_cloud=False,
),
TemplateOption(
id="web_scraping",
name="Web Scraping",
description="A workflow that, given several urls, scrapes and summarizes their content using Google's Gemini API",
source=GithubTemplateRepo(
url="https://github.com/run-llama/template-workflow-web-scraping"
),
llama_cloud=False,
),
]
# Initialize git repository if git is available
has_git = False
git_initialized = False
@@ -218,7 +99,7 @@ def _create(
choices=[questionary.Separator("------------ With UI -------------")]
+ [
questionary.Choice(title=o.name, value=o.id, description=o.description)
for o in ui_options
for o in UI_TEMPLATES
]
+ [
questionary.Separator(" "),
@@ -226,7 +107,7 @@ def _create(
]
+ [
questionary.Choice(title=o.name, value=o.id, description=o.description)
for o in headless_options
for o in HEADLESS_TEMPLATES
],
style=questionary.Style(
[
@@ -235,7 +116,7 @@ def _create(
),
).ask()
if template is None:
options = [o.id for o in ui_options + headless_options]
options = [o.id for o in ALL_TEMPLATES]
rprint(
Text(
f"No template selected. Select a template or pass a template name with --template <{'|'.join(options)}>"
@@ -256,7 +137,7 @@ def _create(
dir = Path(template)
resolved_template: TemplateOption | None = next(
(o for o in ui_options + headless_options if o.id == template), None
(o for o in ALL_TEMPLATES if o.id == template), None
)
if resolved_template is None:
rprint(f"Template {template} not found")
@@ -1,12 +1,13 @@
"""Configuration and profile management for llamactl"""
import functools
import os
import sqlite3
import uuid
from pathlib import Path
from typing import Any
from llama_agents.cli.paths import resolve_llamactl_config_dir
from ._migrations import run_migrations
from .schema import DEFAULT_ENVIRONMENT, Auth, DeviceOIDC, Environment
@@ -58,14 +59,7 @@ class ConfigManager:
Honors LLAMACTL_CONFIG_DIR when set. This helps tests isolate state.
"""
override = os.environ.get("LLAMACTL_CONFIG_DIR")
if override:
return Path(override).expanduser()
if os.name == "nt": # Windows
config_dir = Path(os.environ.get("APPDATA", "~")) / "llamactl"
else: # Unix-like (Linux, macOS)
config_dir = Path.home() / ".config" / "llamactl"
return config_dir.expanduser()
return resolve_llamactl_config_dir()
def _ensure_config_dir(self) -> None:
"""Create configuration directory if it doesn't exist"""
@@ -0,0 +1,148 @@
# SPDX-License-Identifier: MIT
# Copyright (c) 2026 LlamaIndex Inc.
from __future__ import annotations
import asyncio
from concurrent.futures import ThreadPoolExecutor
from typing import Any
import click
from click.shell_completion import CompletionItem
from llama_agents.cli.templates import ALL_TEMPLATES
def _safe_fetch(fn: Any, timeout: float = 2.0) -> list[Any]:
"""Run a fetch function in a thread with a timeout. Returns [] on failure."""
pool = ThreadPoolExecutor(max_workers=1)
try:
future = pool.submit(fn)
return future.result(timeout=timeout)
except Exception:
return []
finally:
pool.shutdown(wait=False)
def _fetch_deployments() -> list[CompletionItem]:
from llama_agents.cli.client import get_project_client
client = get_project_client()
deployments = asyncio.run(client.list_deployments())
return [CompletionItem(d.id) for d in deployments]
def _fetch_projects() -> list[CompletionItem]:
from llama_agents.cli.client import get_control_plane_client
client = get_control_plane_client()
projects = asyncio.run(client.list_projects())
return [
CompletionItem(
p.project_id,
help=f"{p.project_name} ({p.deployment_count} deployments)",
)
for p in projects
]
def _fetch_deployment_history(deployment_id: str) -> list[CompletionItem]:
from llama_agents.cli.client import get_project_client
client = get_project_client()
async def _fetch() -> Any:
return await client.get_deployment_history(deployment_id)
history = asyncio.run(_fetch())
return [
CompletionItem(item.git_sha, help=item.released_at.isoformat())
for item in history.history
]
def _filter(items: list[CompletionItem], incomplete: str) -> list[CompletionItem]:
lower = incomplete.lower()
return [item for item in items if item.value.lower().startswith(lower)]
class DeploymentType(click.ParamType):
name = "deployment"
def shell_complete(
self, ctx: click.Context, param: click.Parameter, incomplete: str
) -> list[CompletionItem]:
return _filter(_safe_fetch(_fetch_deployments), incomplete)
class ProfileType(click.ParamType):
name = "profile"
def shell_complete(
self, ctx: click.Context, param: click.Parameter, incomplete: str
) -> list[CompletionItem]:
def _fetch() -> list[CompletionItem]:
from llama_agents.cli.config.env_service import service
auth_svc = service.current_auth_service()
profiles = auth_svc.list_profiles()
return [CompletionItem(p.name, help=p.api_url) for p in profiles]
return _filter(_safe_fetch(_fetch), incomplete)
class ProjectType(click.ParamType):
name = "project"
def shell_complete(
self, ctx: click.Context, param: click.Parameter, incomplete: str
) -> list[CompletionItem]:
return _filter(_safe_fetch(_fetch_projects), incomplete)
class EnvironmentType(click.ParamType):
name = "environment"
def shell_complete(
self, ctx: click.Context, param: click.Parameter, incomplete: str
) -> list[CompletionItem]:
def _fetch() -> list[CompletionItem]:
from llama_agents.cli.config.env_service import service
envs = service.list_environments()
current = service.get_current_environment()
return [
CompletionItem(
e.api_url,
help="(current)" if e.api_url == current.api_url else "",
)
for e in envs
]
return _filter(_safe_fetch(_fetch), incomplete)
class TemplateType(click.ParamType):
name = "template"
def shell_complete(
self, ctx: click.Context, param: click.Parameter, incomplete: str
) -> list[CompletionItem]:
return _filter(
[CompletionItem(t.id, help=t.description) for t in ALL_TEMPLATES],
incomplete,
)
class GitShaType(click.ParamType):
name = "git_sha"
def shell_complete(
self, ctx: click.Context, param: click.Parameter, incomplete: str
) -> list[CompletionItem]:
deployment_id = ctx.params.get("deployment_id")
if not deployment_id:
return []
return _filter(
_safe_fetch(lambda: _fetch_deployment_history(deployment_id)),
incomplete,
)
@@ -0,0 +1,46 @@
# SPDX-License-Identifier: MIT
# Copyright (c) 2026 LlamaIndex Inc.
from __future__ import annotations
import os
from pathlib import Path
def resolve_llamactl_config_dir() -> Path:
"""Resolve the config directory from the override/env/platform policy."""
override = os.environ.get("LLAMACTL_CONFIG_DIR")
if override:
return Path(override).expanduser()
if os.name == "nt":
return (Path(os.environ.get("APPDATA", "~")) / "llamactl").expanduser()
xdg_config_home = os.environ.get("XDG_CONFIG_HOME")
if xdg_config_home:
return (Path(xdg_config_home).expanduser() / "llamactl").expanduser()
return (Path.home() / ".config" / "llamactl").expanduser()
def bash_completion_dir(home: Path | None = None) -> Path:
"""Return the preferred per-user bash completion directory."""
resolved_home = home or Path.home()
preferred = resolved_home / ".local" / "share" / "bash-completion" / "completions"
if preferred.exists():
return preferred
return resolved_home / ".bash_completion.d"
def bash_rc_path(home: Path | None = None) -> Path:
return (home or Path.home()) / ".bashrc"
def zsh_completion_dir(home: Path | None = None) -> Path:
return (home or Path.home()) / ".zfunc"
def zsh_rc_path(home: Path | None = None) -> Path:
return (home or Path.home()) / ".zshrc"
def fish_completion_dir(home: Path | None = None) -> Path:
return (home or Path.home()) / ".config" / "fish" / "completions"
@@ -0,0 +1,136 @@
# SPDX-License-Identifier: MIT
# Copyright (c) 2026 LlamaIndex Inc.
from __future__ import annotations
from dataclasses import dataclass
@dataclass
class GithubTemplateRepo:
url: str
@dataclass
class TemplateOption:
id: str
name: str
description: str
source: GithubTemplateRepo
llama_cloud: bool
UI_TEMPLATES = [
TemplateOption(
id="basic-ui",
name="Basic UI",
description="A basic starter workflow with a React Vite UI",
source=GithubTemplateRepo(
url="https://github.com/run-llama/template-workflow-basic-ui"
),
llama_cloud=False,
),
TemplateOption(
id="showcase",
name="Showcase",
description="A collection of workflow and UI patterns to build LlamaDeploy apps",
source=GithubTemplateRepo(
url="https://github.com/run-llama/template-workflow-showcase"
),
llama_cloud=False,
),
TemplateOption(
id="document-qa",
name="Document Question & Answer",
description="Upload documents and run question answering through a React UI",
source=GithubTemplateRepo(
url="https://github.com/run-llama/template-workflow-document-qa"
),
llama_cloud=True,
),
TemplateOption(
id="extraction-review",
name="Extraction Agent with Review UI",
description="Extract data from documents using a custom schema and Llama Cloud. Includes a UI to review and correct the results",
source=GithubTemplateRepo(
url="https://github.com/run-llama/template-workflow-data-extraction"
),
llama_cloud=True,
),
TemplateOption(
id="classify-extract-sec",
name="SEC Insights",
description="Upload SEC filings, classifying them to the appropriate type and extracting key information",
source=GithubTemplateRepo(
url="https://github.com/run-llama/template-workflow-classify-extract-sec"
),
llama_cloud=True,
),
TemplateOption(
id="extract-reconcile-invoice",
name="Invoice Extraction & Reconciliation",
description="Extract and reconcile invoice data against contracts",
source=GithubTemplateRepo(
url="https://github.com/run-llama/template-workflow-extract-reconcile-invoice"
),
llama_cloud=True,
),
]
HEADLESS_TEMPLATES = [
TemplateOption(
id="basic",
name="Basic Workflow",
description="A base example that showcases usage patterns for workflows",
source=GithubTemplateRepo(
url="https://github.com/run-llama/template-workflow-basic"
),
llama_cloud=False,
),
TemplateOption(
id="document_parsing",
name="Document Parser",
description="A workflow that, using LlamaParse, parses unstructured documents and returns their raw text content",
source=GithubTemplateRepo(
url="https://github.com/run-llama/template-workflow-document-parsing"
),
llama_cloud=True,
),
TemplateOption(
id="human_in_the_loop",
name="Human in the Loop",
description="A workflow showcasing how to use human in the loop with LlamaIndex workflows",
source=GithubTemplateRepo(
url="https://github.com/run-llama/template-workflow-human-in-the-loop"
),
llama_cloud=False,
),
TemplateOption(
id="invoice_extraction",
name="Invoice Extraction",
description="A workflow that, given an invoice, extracts several key details using LlamaExtract",
source=GithubTemplateRepo(
url="https://github.com/run-llama/template-workflow-invoice-extraction"
),
llama_cloud=True,
),
TemplateOption(
id="rag",
name="RAG",
description="A workflow that embeds, indexes and queries your documents on the fly, providing you with a simple RAG pipeline",
source=GithubTemplateRepo(
url="https://github.com/run-llama/template-workflow-rag"
),
llama_cloud=False,
),
TemplateOption(
id="web_scraping",
name="Web Scraping",
description="A workflow that, given several urls, scrapes and summarizes their content using Google's Gemini API",
source=GithubTemplateRepo(
url="https://github.com/run-llama/template-workflow-web-scraping"
),
llama_cloud=False,
),
]
ALL_TEMPLATES = UI_TEMPLATES + HEADLESS_TEMPLATES
@@ -0,0 +1,131 @@
# SPDX-License-Identifier: MIT
# Copyright (c) 2026 LlamaIndex Inc.
from __future__ import annotations
from pathlib import Path
import pytest
from click.testing import CliRunner
from llama_agents.cli.app import app
def _first_matching_line_index(lines: list[str], predicate: str) -> int:
for index, line in enumerate(lines):
if predicate in line and not line.lstrip().startswith("#"):
return index
raise AssertionError(f"Could not find live line containing {predicate!r}")
def _invoke_zsh_install(home: Path, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(Path, "home", lambda: home)
runner = CliRunner()
result = runner.invoke(app, ["completion", "install", "--shell", "zsh"])
assert result.exit_code == 0, result.output
def test_completion_generate_zsh() -> None:
runner = CliRunner()
result = runner.invoke(app, ["completion", "generate", "zsh"])
assert result.exit_code == 0
assert "_LLAMACTL_COMPLETE" in result.output or "compdef" in result.output
def test_completion_generate_bash() -> None:
runner = CliRunner()
result = runner.invoke(app, ["completion", "generate", "bash"])
assert result.exit_code == 0
assert "_LLAMACTL_COMPLETE" in result.output or "complete" in result.output
def test_completion_generate_fish() -> None:
runner = CliRunner()
result = runner.invoke(app, ["completion", "generate", "fish"])
assert result.exit_code == 0
assert "llamactl" in result.output
def test_completion_generate_invalid_shell() -> None:
runner = CliRunner()
result = runner.invoke(app, ["completion", "generate", "powershell"])
assert result.exit_code != 0
def test_completion_install_dry_run() -> None:
runner = CliRunner()
result = runner.invoke(
app, ["completion", "install", "--shell", "zsh", "--dry-run"]
)
assert result.exit_code == 0
assert "Would write" in result.output
def test_completion_install_dry_run_bash() -> None:
runner = CliRunner()
result = runner.invoke(
app, ["completion", "install", "--shell", "bash", "--dry-run"]
)
assert result.exit_code == 0
assert "Would write" in result.output
def test_completion_install_dry_run_fish() -> None:
runner = CliRunner()
result = runner.invoke(
app, ["completion", "install", "--shell", "fish", "--dry-run"]
)
assert result.exit_code == 0
assert "Would write" in result.output
def test_completion_install_zsh_repairs_ordered_completion_block(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
home = tmp_path / "home"
home.mkdir()
zshrc = home / ".zshrc"
zshrc.write_text(
"autoload -Uz compinit && compinit\n"
"fpath=(~/.zfunc $fpath)\n"
'echo "custom shell setup"\n'
)
_invoke_zsh_install(home, monkeypatch)
lines = zshrc.read_text().splitlines()
fpath_index = _first_matching_line_index(lines, "~/.zfunc")
compinit_index = _first_matching_line_index(lines, "compinit")
assert fpath_index < compinit_index
first_pass = zshrc.read_text()
_invoke_zsh_install(home, monkeypatch)
assert zshrc.read_text() == first_pass
def test_completion_install_zsh_bootstraps_compinit_when_missing(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
home = tmp_path / "home"
home.mkdir()
zshrc = home / ".zshrc"
zshrc.write_text('export PATH="$HOME/bin:$PATH"\n')
_invoke_zsh_install(home, monkeypatch)
lines = zshrc.read_text().splitlines()
fpath_index = _first_matching_line_index(lines, "~/.zfunc")
compinit_index = _first_matching_line_index(lines, "compinit")
assert fpath_index < compinit_index
live_compinit_lines = [
line
for line in lines
if "compinit" in line and not line.lstrip().startswith("#")
]
assert len(live_compinit_lines) == 1
def test_completion_group_help() -> None:
runner = CliRunner()
result = runner.invoke(app, ["completion", "--help"])
assert result.exit_code == 0
assert "generate" in result.output
assert "install" in result.output
+48
View File
@@ -1,5 +1,9 @@
# SPDX-License-Identifier: MIT
# Copyright (c) 2026 LlamaIndex Inc.
"""Tests for config.py - Database operations and profile management"""
from __future__ import annotations
import sqlite3
import tempfile
from collections.abc import Generator
@@ -267,3 +271,47 @@ def test_environment_methods_and_current_behavior(temp_config: ConfigManager) ->
preferred = temp_config.get_current_profile(env_only_url)
assert preferred is not None
assert preferred.name == "only-here"
def test_config_manager_honors_llamactl_config_dir_override(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
override_dir = tmp_path / "override-config"
monkeypatch.setenv("LLAMACTL_CONFIG_DIR", str(override_dir))
cfg = ConfigManager()
assert cfg.config_dir == override_dir
assert cfg.db_path == override_dir / "profiles.db"
assert cfg.db_path.exists()
def test_config_manager_uses_xdg_config_home_on_unix(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
home_dir = tmp_path / "home"
xdg_dir = tmp_path / "xdg-config"
monkeypatch.setenv("HOME", str(home_dir))
monkeypatch.delenv("LLAMACTL_CONFIG_DIR", raising=False)
monkeypatch.setenv("XDG_CONFIG_HOME", str(xdg_dir))
cfg = ConfigManager()
assert cfg.config_dir == xdg_dir / "llamactl"
assert cfg.db_path == xdg_dir / "llamactl" / "profiles.db"
assert cfg.db_path.exists()
def test_config_manager_defaults_to_dot_config_on_unix(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
home_dir = tmp_path / "home"
monkeypatch.setenv("HOME", str(home_dir))
monkeypatch.delenv("LLAMACTL_CONFIG_DIR", raising=False)
monkeypatch.delenv("XDG_CONFIG_HOME", raising=False)
cfg = ConfigManager()
assert cfg.config_dir == home_dir / ".config" / "llamactl"
assert cfg.db_path == home_dir / ".config" / "llamactl" / "profiles.db"
assert cfg.db_path.exists()
+197
View File
@@ -0,0 +1,197 @@
# SPDX-License-Identifier: MIT
# Copyright (c) 2026 LlamaIndex Inc.
from __future__ import annotations
from dataclasses import dataclass
import click
import llama_agents.cli.config.env_service as es
import llama_agents.cli.param_types as pt
import pytest
from click.shell_completion import CompletionItem
from llama_agents.cli.param_types import (
DeploymentType,
EnvironmentType,
GitShaType,
ProfileType,
ProjectType,
TemplateType,
)
@pytest.fixture()
def ctx() -> click.Context:
"""A minimal Click context for shell_complete calls."""
cmd = click.Command("test")
return click.Context(cmd)
@pytest.fixture()
def param() -> click.Parameter:
return click.Argument(["test_arg"])
def test_deployment_type_returns_fetched_items(
ctx: click.Context, param: click.Parameter, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setattr(
pt,
"_fetch_deployments",
lambda: [
CompletionItem("my-app"),
CompletionItem("staging"),
],
)
dt = DeploymentType()
items = dt.shell_complete(ctx, param, "")
assert len(items) == 2
assert items[0].value == "my-app"
# Filter by prefix
items = dt.shell_complete(ctx, param, "my")
assert len(items) == 1
assert items[0].value == "my-app"
def test_deployment_type_fetch_failure(
ctx: click.Context, param: click.Parameter, monkeypatch: pytest.MonkeyPatch
) -> None:
def _boom() -> list[CompletionItem]:
raise RuntimeError("API down")
monkeypatch.setattr(pt, "_fetch_deployments", _boom)
dt = DeploymentType()
items = dt.shell_complete(ctx, param, "")
assert items == []
def test_profile_type_returns_profiles(
ctx: click.Context, param: click.Parameter, monkeypatch: pytest.MonkeyPatch
) -> None:
@dataclass
class FakeAuth:
name: str
api_url: str
class FakeAuthService:
def list_profiles(self) -> list[FakeAuth]:
return [
FakeAuth(name="prod", api_url="https://api.prod.example.com"),
FakeAuth(name="dev", api_url="https://api.dev.example.com"),
]
class FakeService:
def current_auth_service(self) -> FakeAuthService:
return FakeAuthService()
monkeypatch.setattr(es, "service", FakeService())
prof = ProfileType()
items = prof.shell_complete(ctx, param, "")
assert len(items) == 2
assert items[0].value == "prod"
assert items[0].help == "https://api.prod.example.com"
items = prof.shell_complete(ctx, param, "dev")
assert len(items) == 1
assert items[0].value == "dev"
def test_project_type_returns_fetched_items(
ctx: click.Context, param: click.Parameter, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setattr(
pt,
"_fetch_projects",
lambda: [
CompletionItem("proj_abc", help="My Project (3 deployments)"),
CompletionItem("proj_def", help="Staging (1 deployment)"),
],
)
proj = ProjectType()
items = proj.shell_complete(ctx, param, "")
assert len(items) == 2
assert items[0].value == "proj_abc"
def test_environment_type_returns_environments(
ctx: click.Context, param: click.Parameter, monkeypatch: pytest.MonkeyPatch
) -> None:
@dataclass
class FakeEnv:
api_url: str
requires_auth: bool = True
class FakeService:
def list_environments(self) -> list[FakeEnv]:
return [
FakeEnv(api_url="https://api.prod.example.com"),
FakeEnv(api_url="https://api.dev.example.com"),
]
def get_current_environment(self) -> FakeEnv:
return FakeEnv(api_url="https://api.prod.example.com")
monkeypatch.setattr(es, "service", FakeService())
et = EnvironmentType()
items = et.shell_complete(ctx, param, "")
assert len(items) == 2
# Current env should have "(current)" help
assert items[0].help == "(current)"
assert items[1].help == ""
def test_template_type_returns_all_templates(
ctx: click.Context, param: click.Parameter
) -> None:
tt = TemplateType()
items = tt.shell_complete(ctx, param, "")
assert len(items) == 12 # 6 UI + 6 headless
# Filter
items = tt.shell_complete(ctx, param, "basic")
assert len(items) == 2 # basic-ui and basic
def test_template_type_case_insensitive(
ctx: click.Context, param: click.Parameter
) -> None:
tt = TemplateType()
items = tt.shell_complete(ctx, param, "RAG")
assert len(items) == 1
assert items[0].value == "rag"
def test_git_sha_type_no_deployment_id(
ctx: click.Context, param: click.Parameter
) -> None:
ctx.params = {}
gt = GitShaType()
items = gt.shell_complete(ctx, param, "")
assert items == []
def test_git_sha_type_with_deployment_id(
ctx: click.Context, param: click.Parameter, monkeypatch: pytest.MonkeyPatch
) -> None:
ctx.params = {"deployment_id": "my-deploy"}
monkeypatch.setattr(
pt,
"_fetch_deployment_history",
lambda dep_id: [
CompletionItem("abc1234", help="2026-01-01T00:00:00"),
CompletionItem("def5678", help="2026-01-02T00:00:00"),
],
)
gt = GitShaType()
items = gt.shell_complete(ctx, param, "")
assert len(items) == 2
items = gt.shell_complete(ctx, param, "abc")
assert len(items) == 1
assert items[0].value == "abc1234"