Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions CLAUDE.md
Original file line number Diff line number Diff line change
Expand Up @@ -256,4 +256,5 @@ kbagent context
kbagent init [--from-global]
kbagent doctor [--fix]
kbagent version
kbagent update
```
1 change: 1 addition & 0 deletions plugins/kbagent/skills/kbagent/SKILL.md
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,7 @@ If kbagent is not installed or you need the full standalone reference, run `kbag
<!-- BEGIN AUTO-GENERATED COMMANDS -->
| Goal | Command |
|------|---------|
| Update kbagent to the latest version | `kbagent update` |
| Add a new Keboola project connection | `kbagent project add --project ALIAS` |
| List all connected Keboola projects | `kbagent project list` |
| Remove a Keboola project connection | `kbagent project remove --project ALIAS` |
Expand Down
3 changes: 2 additions & 1 deletion src/keboola_agent_cli/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,7 @@
from .commands.storage import storage_app
from .commands.sync import sync_app
from .commands.tool import tool_app
from .commands.version import version_command
from .commands.version import update_command, version_command
from .commands.workspace import workspace_app
from .config_store import ConfigStore, resolve_config_dir
from .output import OutputFormatter
Expand Down Expand Up @@ -50,6 +50,7 @@
app.command("init", rich_help_panel=_SETUP)(init_command)
app.command("doctor", rich_help_panel=_SETUP)(doctor_command)
app.command("version", rich_help_panel=_SETUP)(version_command)
app.command("update", rich_help_panel=_SETUP)(update_command)
app.command("context", rich_help_panel=_SETUP)(context_command)
app.command("repl", rich_help_panel=_SETUP)(repl_command)

Expand Down
5 changes: 4 additions & 1 deletion src/keboola_agent_cli/commands/context.py
Original file line number Diff line number Diff line change
Expand Up @@ -235,7 +235,10 @@
Health checks. --fix auto-installs MCP server binary.

kbagent version
Version info and MCP server update check.
Version info, update check for kbagent and MCP server.

kbagent update
Self-update kbagent to latest version (via uv tool install --upgrade).

## Tips for AI Agents

Expand Down
29 changes: 28 additions & 1 deletion src/keboola_agent_cli/commands/version.py
Original file line number Diff line number Diff line change
Expand Up @@ -66,7 +66,15 @@ def _format_dep_auto_update(text: Text, dep: dict) -> None:
def _format_version_panel(console: Console, data: dict) -> None:
"""Render version information as a Rich panel."""
text = Text()
text.append(f"kbagent v{data['kbagent']['version']}", style="bold")
kbagent = data["kbagent"]
text.append(f"kbagent v{kbagent['version']}", style="bold")

if kbagent.get("up_to_date") is False and kbagent.get("latest_version"):
text.append(f" -> v{kbagent['latest_version']} available", style="yellow")
text.append(" (run: kbagent update)", style="dim")
elif kbagent.get("up_to_date") is True:
text.append(" up to date", style="green")

text.append("\n\nDependencies:\n")

for dep in data["dependencies"]:
Expand All @@ -84,3 +92,22 @@ def version_command(ctx: typer.Context) -> None:
version_service = get_service(ctx, "version_service")
result = version_service.get_versions()
formatter.output(result, _format_version_panel)


def update_command(ctx: typer.Context) -> None:
"""Update kbagent to the latest version.

Uses 'uv tool install --upgrade' (preferred) or 'pip install --upgrade'
to install the latest version from the GitHub repository.
"""
formatter = get_formatter(ctx)
version_service = get_service(ctx, "version_service")
result = version_service.self_update()

if formatter.json_mode:
formatter.output(result)
else:
if result["updated"]:
formatter.success(result["message"])
else:
formatter.console.print(result["message"])
2 changes: 2 additions & 0 deletions src/keboola_agent_cli/constants.py
Original file line number Diff line number Diff line change
Expand Up @@ -75,6 +75,8 @@
# --- Version Check ---
VERSION_CHECK_TIMEOUT: float = 4.0 # seconds for fetching latest version from remote
MCP_PYPI_URL: str = "https://pypi.org/pypi/keboola-mcp-server/json"
KBAGENT_GITHUB_REPO: str = "padak/keboola_agent_cli"
KBAGENT_INSTALL_SOURCE: str = "git+https://github.com/padak/keboola_agent_cli"

# --- AI Service ---
AI_SERVICE_TIMEOUT: httpx.Timeout = httpx.Timeout(connect=5.0, read=15.0, write=5.0, pool=5.0)
Expand Down
102 changes: 102 additions & 0 deletions src/keboola_agent_cli/services/version_service.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,8 @@

from .. import __version__
from ..constants import (
KBAGENT_GITHUB_REPO,
KBAGENT_INSTALL_SOURCE,
MCP_PYPI_URL,
VERSION_CHECK_TIMEOUT,
)
Expand All @@ -27,6 +29,34 @@ def _is_uvx_available() -> bool:
return shutil.which("uvx") is not None


def _fetch_kbagent_latest_version(timeout: float = VERSION_CHECK_TIMEOUT) -> str | None:
"""Fetch latest kbagent version from GitHub releases.

Args:
timeout: HTTP request timeout in seconds.

Returns:
Version string like '0.16.0', or None on failure.
"""
try:
response = httpx.get(
f"https://github.com/ghapi/repos/{KBAGENT_GITHUB_REPO}/releases/latest",
timeout=timeout,
follow_redirects=True,
headers={"Accept": "application/vnd.github.v3+json"},
)
response.raise_for_status()
tag = response.json().get("tag_name", "")
# Strip leading 'v' from tag (e.g. 'v0.16.0' -> '0.16.0')
version = tag.lstrip("v")
if re.match(r"\d+\.\d+\.\d+", version):
return version
return None
except (httpx.HTTPError, KeyError, ValueError):
logger.debug("Failed to fetch latest kbagent version", exc_info=True)
return None


def _fetch_mcp_latest_version(timeout: float = VERSION_CHECK_TIMEOUT) -> str | None:
"""Fetch latest keboola-mcp-server version from PyPI.

Expand Down Expand Up @@ -90,6 +120,8 @@ def get_versions(self) -> dict[str, Any]:
"""
uvx_available = _is_uvx_available()
mcp_latest = _fetch_mcp_latest_version()
kbagent_latest = _fetch_kbagent_latest_version()
kbagent_up_to_date = _is_up_to_date(__version__, kbagent_latest)

mcp_entry: dict[str, Any] = {
"name": "keboola-mcp-server",
Expand All @@ -102,8 +134,78 @@ def get_versions(self) -> dict[str, Any]:
return {
"kbagent": {
"version": __version__,
"latest_version": kbagent_latest,
"up_to_date": kbagent_up_to_date,
"upgrade_command": f"uv tool install --upgrade {KBAGENT_INSTALL_SOURCE}",
},
"dependencies": [
mcp_entry,
],
}

def self_update(self) -> dict[str, Any]:
"""Update kbagent to the latest version via uv tool install.

Returns:
Dict with update result (old version, new version, output).
"""
import subprocess

old_version = __version__
kbagent_latest = _fetch_kbagent_latest_version()
up_to_date = _is_up_to_date(old_version, kbagent_latest)

if up_to_date is True:
return {
"updated": False,
"current_version": old_version,
"latest_version": kbagent_latest,
"message": f"kbagent v{old_version} is already up to date.",
}

# Try uv tool install --upgrade first, fall back to pip
uv_path = shutil.which("uv")
if uv_path:
cmd = [uv_path, "tool", "install", "--upgrade", KBAGENT_INSTALL_SOURCE]
else:
pip_path = shutil.which("pip")
if pip_path is None:
return {
"updated": False,
"current_version": old_version,
"latest_version": kbagent_latest,
"message": "Neither 'uv' nor 'pip' found on PATH. "
f"Install manually: uv tool install --upgrade {KBAGENT_INSTALL_SOURCE}",
}
cmd = [pip_path, "install", "--upgrade", KBAGENT_INSTALL_SOURCE]

try:
result = subprocess.run(
cmd,
capture_output=True,
text=True,
timeout=120,
)
if result.returncode == 0:
return {
"updated": True,
"current_version": old_version,
"latest_version": kbagent_latest,
"message": f"Updated kbagent from v{old_version} to v{kbagent_latest}. "
"Restart your shell to use the new version.",
"output": result.stdout.strip(),
}
return {
"updated": False,
"current_version": old_version,
"latest_version": kbagent_latest,
"message": f"Update failed: {result.stderr.strip()}",
"output": result.stderr.strip(),
}
except subprocess.TimeoutExpired:
return {
"updated": False,
"current_version": old_version,
"latest_version": kbagent_latest,
"message": "Update timed out after 120 seconds.",
}
Loading