Skip to content
Open
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
100 changes: 75 additions & 25 deletions src/smallestai/cli/agent_crew.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@

import asyncio
import base64
import json as _json
import sys
from io import BytesIO
from pathlib import Path
Expand Down Expand Up @@ -36,6 +37,17 @@
def initialise_agent_crew_app(project_config: ProjectConfig, auth_client: AuthClient, atoms_client: AtomsAPIClient):
app = typer.Typer(name="agent-crew")

def _resolve_agent_id(arg: Optional[str]) -> str:
"""Explicit --agent-id wins; otherwise fall back to the linked project agent."""
agent_id = arg or project_config.get_agent_id()
if not agent_id:
console.print(
"[red]No agent linked. Run [bold]smallestai agent-crew init[/bold] in this "
"directory, or pass [bold]--agent-id <id>[/bold].[/red]"
)
raise typer.Exit(1)
return agent_id

@app.command()
def init(
agent_id: Optional[str] = typer.Option(
Expand Down Expand Up @@ -128,21 +140,20 @@ def deploy(
"-e",
help="Entry point file name (e.g., server.py)",
),
agent_id: Optional[str] = typer.Option(
None, "--agent-id", help="Agent id to deploy to (defaults to the linked project agent)."
),
):
"""
Deploy an agent crew to the Atoms platform.

Packages the agent-crew code directory into a zip file and deploys it to the backend.
"""
asyncio.run(async_deploy(".", entry_point))
asyncio.run(async_deploy(".", entry_point, agent_id))

async def async_deploy(directory: str, entry_point: str):
async def async_deploy(directory: str, entry_point: str, agent_id_arg: Optional[str] = None):
"""Deploy an agent crew asynchronously."""
agent_id = project_config.get_agent_id()

if not agent_id:
console.print("[red]Agent not initialized. Run 'smallestai agent init' first.[/red]")
return
agent_id = _resolve_agent_id(agent_id_arg)

# Check if user is logged in
credentials = auth_client.get_credentials()
Expand Down Expand Up @@ -315,23 +326,26 @@ async def run():
@app.command("builds")
def list_builds(
build_id: str = typer.Argument(None, help="Optional build ID to manage directly"),
agent_id: Optional[str] = typer.Option(
None, "--agent-id", help="Agent id (defaults to the linked project agent)."
),
limit: int = typer.Option(50, "--limit", "-l", help="Number of builds to fetch"),
offset: int = typer.Option(0, "--offset", "-o", help="Offset for pagination"),
offset: int = typer.Option(0, "--offset", help="Offset for pagination"),
as_json: bool = typer.Option(False, "--json", help="Emit raw JSON (non-interactive)"),
):
"""
List all builds for the current agent crew and manage them interactively.

If a build_id is provided, directly manage that specific build.
If a build_id is provided, directly manage that specific build. With --json
or in a non-interactive terminal, list and exit without the picker.
"""
asyncio.run(async_list_builds(build_id, limit, offset))
asyncio.run(async_list_builds(build_id, agent_id, limit, offset, as_json))

async def async_list_builds(build_id: str | None, limit: int, offset: int):
async def async_list_builds(
build_id: str | None, agent_id_arg: Optional[str], limit: int, offset: int, as_json: bool
):
"""Async implementation of list builds command."""
agent_id = project_config.get_agent_id()

if not agent_id:
console.print("[red]Agent not initialized. Run 'smallestai agent init' first.[/red]")
return
agent_id = _resolve_agent_id(agent_id_arg)

credentials = auth_client.get_credentials()
if not credentials or not credentials.get("access_token"):
Expand All @@ -342,24 +356,46 @@ async def async_list_builds(build_id: str | None, limit: int, offset: int):

try:
if build_id:
console.print(f"[dim]Fetching build: {build_id}[/dim]")
build = await atoms_client.get_agent_build(
agent_id=agent_id,
build_id=build_id,
api_key=access_token,
)
if as_json:
console.print_json(
_json.dumps(
{
"id": build.id,
"status": build.status.value,
"is_live": build.is_live,
"created_at": build.created_at,
},
default=str,
)
)
return
await _manage_build(agent_id, build, access_token)
return

console.print(f"[dim]Fetching builds for agent: {agent_id}[/dim]")

result = await atoms_client.list_agent_builds(
agent_id=agent_id,
api_key=access_token,
limit=limit,
offset=offset,
)

if as_json:
console.print_json(
_json.dumps(
[
{"id": b.id, "status": b.status.value, "is_live": b.is_live, "created_at": b.created_at}
for b in result.builds
],
default=str,
)
)
return

if not result.builds:
console.print("[yellow]No builds found for this agent.[/yellow]")
return
Expand All @@ -386,6 +422,14 @@ async def async_list_builds(build_id: str | None, limit: int, offset: int):
console.print(table)
console.print(f"[dim]Showing {len(result.builds)} of {result.pagination.total} builds[/dim]\n")

# No interactive picker outside a real terminal (would hang in CI).
if not sys.stdin.isatty():
console.print(
"[dim]Manage a build directly: [bold]smallestai agent-crew builds <build-id>[/bold] "
"(add [bold]--agent-id[/bold] outside a project dir).[/dim]"
)
return

choices = [
questionary.Choice(
title=f"{build.id[:12]}... | {build.status.value} | {'LIVE' if build.is_live else '-'} | {build.created_at}",
Expand Down Expand Up @@ -467,6 +511,12 @@ async def _manage_build(agent_id: str, build, access_token: str):
)
console.print(f"[bold green]✓ Build {build.id[:12]}... is now LIVE![/bold green]")
elif selected_action == "take_down":
confirmed = await questionary.confirm(
f"Take build {build.id[:12]}... offline? This stops the agent serving calls."
).ask_async()
if not confirmed:
console.print("[dim]Left the build live.[/dim]")
return
console.print("[yellow]Taking down build...[/yellow]")
await atoms_client.update_agent_build(
agent_id=agent_id,
Expand All @@ -479,19 +529,19 @@ async def _manage_build(agent_id: str, build, access_token: str):
@app.command("logs")
def build_logs(
build_id: str = typer.Argument(None, help="Build ID to stream logs for (defaults to the latest build)"),
agent_id: Optional[str] = typer.Option(
None, "--agent-id", help="Agent id (defaults to the linked project agent)."
),
):
"""Stream a build's logs (compile + deploy) in real time.

With no build ID, streams the most recent build for the current agent.
Use this to debug a deploy that failed or to watch one in progress.
"""
asyncio.run(async_build_logs(build_id))
asyncio.run(async_build_logs(build_id, agent_id))

async def async_build_logs(build_id: str | None):
agent_id = project_config.get_agent_id()
if not agent_id:
console.print("[red]Agent not initialized. Run 'smallestai agent-crew init' first.[/red]")
raise typer.Exit(1)
async def async_build_logs(build_id: str | None, agent_id_arg: Optional[str] = None):
agent_id = _resolve_agent_id(agent_id_arg)

credentials = auth_client.get_credentials()
if not credentials or not credentials.get("access_token"):
Expand Down
20 changes: 18 additions & 2 deletions src/smallestai/cli/agents.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,8 @@
overrides the endpoint (dev rig).
"""

import json as _json

import typer
from rich.console import Console
from rich.table import Table
Expand All @@ -26,20 +28,34 @@ def initialise_agents_app(auth_client: AuthClient):
agents_app = typer.Typer(name="agents", help="Create, inspect, and call Atoms agents.")

@agents_app.command("list")
def list_agents():
def list_agents(as_json: bool = typer.Option(False, "--json", help="Emit raw JSON")):
"""List agents in your org."""
from smallestai.atoms.helpers import as_page

pg = as_page(_client(auth_client).atoms.agents.list_agents())
if as_json:
console.print_json(
_json.dumps(
[
{"id": getattr(a, "id", None) or getattr(a, "_id", None), "name": getattr(a, "name", None)}
for a in pg.items
],
default=str,
)
)
return
table = Table("ID", "Name", title=f"Agents ({len(pg.items)})")
for a in pg.items:
table.add_row(getattr(a, "id", None) or getattr(a, "_id", "?"), getattr(a, "name", "—"))
console.print(table)

@agents_app.command("get")
def get_agent(agent_id: str):
def get_agent(agent_id: str, as_json: bool = typer.Option(False, "--json", help="Emit raw JSON")):
"""Show one agent's config."""
a = _client(auth_client).atoms.agents.get_agent(id=agent_id).data
if as_json:
console.print_json(a.json() if hasattr(a, "json") else _json.dumps(a, default=str))
return
console.print(f"[bold]{getattr(a, 'name', '—')}[/bold] [dim]{agent_id}[/dim]")
console.print(f" first message : {getattr(a, 'first_message', None)!r}")
console.print(f" language : {getattr(a, 'language', None)}")
Expand Down
Loading
Loading