Restruct-Corporate-Advantage/agos-cli
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: initial agos-cli standalone repo

Browse Source 
Extracted from the agos monorepo. Contains the full AGOS CLI:
- cli/agos.py — main entry point with login, logout, whoami, ask, db commands
- cli/command_surface.py — API-backed commands (agent, task, workflow, plugin, policy, integration)
- cli/manual.py — rich interactive manual and help system

Uses clean package imports (no sys.path hacks).
Packaged as agos-cli with `agos` console script entry point.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
mohiit1502
2026-08-21 17:05:52 +05:30
commit 5484d1f5d1
7 changed files with 2815 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
.gitignore vendored Normal file
View File

@@ -0,0 +1,15 @@
__pycache__/
*.py[cod]
*.pyo
*.egg
*.egg-info/
dist/
build/
.eggs/
.venv/
venv/
env/
.env
.env.local
*.log
.DS_Store

78
README.md Normal file
View File

@@ -0,0 +1,78 @@
# agos-cli
The official command-line interface for [AGOS](https://agos.armco.dev) — an agent orchestrator and runtime platform.
## Installation
```bash
pip install -e .
```
This registers the `agos` console script globally.
## Quick Start
```bash
# Authenticate with Armco IAM
agos login
# Check who you are
agos whoami
# Ask the platform a question
agos ask "What agents are currently running?"
# List agents
agos agent list
# Run a workflow
agos workflow run wf_123 --agent-id agent_123
```
## Commands
| Command | Description |
|---------|-------------|
| `agos login` | Authenticate via Armco IAM (PKCE browser flow) |
| `agos logout` | Clear the local session |
| `agos whoami` | Show current identity and tenant |
| `agos ask <prompt>` | Send a natural-language prompt to AGOS chat |
| `agos agent <subcommand>` | Manage agents (list, get, create, run, stop, delete) |
| `agos task <subcommand>` | Manage tasks (list, get, create, cancel, delete) |
| `agos workflow <subcommand>` | Manage workflows (list, get, create, run, status) |
| `agos plugin <subcommand>` | Manage plugins (list, search, install, enable, disable, uninstall) |
| `agos policy <subcommand>` | Manage policies (list, create, delete) |
| `agos integration <subcommand>` | Manage integrations (catalog, providers, instances, create, test) |
| `agos db <subcommand>` | Local database ops (bootstrap, seed, status) |
| `agos system <subcommand>` | System health and logs |
| `agos help [topic]` | Focused help for any command |
| `agos man [topic]` | Full interactive manual |
## Configuration
The CLI reads API URL from the following env vars in order:
- `AGOS_CLI_API_URL`
- `AGOS_API_URL`
- `VITE_API_URL`
- `AGOS_BASE_URL`
- Default: `http://localhost:2000`
Auth tokens are cached at `~/.agos/auth.json` (mode `0600`).
## Development
```bash
# Clone and install in editable mode
git clone <repo>
cd agos-cli
pip install -e .
# Run directly
agos --help
```
## Related Repos
- [agos](https://gitea.armco.dev/agos/agos) — backend runtime
- [agos-client](https://gitea.armco.dev/agos/agos-client) — React frontend

0
cli/__init__.py Normal file
View File

772
cli/agos.py Normal file
View File

@@ -0,0 +1,772 @@
#!/usr/bin/env python3
"""AGOS CLI - Main entry point."""
import click
from rich.console import Console
from rich.table import Table
import sys
import os
import base64
import hashlib
import json
import secrets
import threading
import time
import uuid
import webbrowser
from datetime import datetime, timezone
from http.server import BaseHTTPRequestHandler, HTTPServer
from pathlib import Path
from urllib.parse import urlencode, urlparse, parse_qs
import requests
from dotenv import load_dotenv
from cli.command_surface import install_api_commands
from cli.manual import install_manual_commands
# `agos` is installed as a first-class console script (see pyproject.toml
# project.scripts). A non-editable `pip install` copies cli/agos.py into
# site-packages, so __file__-relative paths never reach the actual AGOS
# checkout. Resolve .env files from the canonical checkout locations that
# install.sh itself creates (AGOS_DIR / AGOS_CLIENT_DIR), plus the
# in-repo dev path (this file's own repo, for `python -m cli.agos` usage)
# and cwd (for running the CLI from inside either checkout).
_PACKAGE_REPO_DIR = Path(__file__).resolve().parent.parent
_AGOS_DIR = Path(os.environ.get('AGOS_DIR', Path.home() / '.agos'))
_AGOS_CLIENT_DIR = Path(os.environ.get(
'AGOS_CLIENT_DIR', _AGOS_DIR.parent / 'agos-client',
))
_CWD = Path.cwd()
_ENV_CANDIDATE_DIRS = [
_AGOS_DIR,
_AGOS_CLIENT_DIR,
_PACKAGE_REPO_DIR,
_PACKAGE_REPO_DIR.parent / 'agos-client',
_CWD,
_CWD.parent / 'agos-client',
]
for _env_dir in _ENV_CANDIDATE_DIRS:
load_dotenv(_env_dir / '.env', override=False)
for _env_dir in _ENV_CANDIDATE_DIRS:
load_dotenv(_env_dir / '.env.local', override=True)
console = Console()
_AUTH_DIR = Path.home() / '.agos'
_AUTH_FILE = _AUTH_DIR / 'auth.json'
def _utc_now_iso() -> str:
return datetime.now(timezone.utc).isoformat()
def _emit_log(
level,
operation,
correlation_id,
entity_id='agos_cli',
metadata=None,
component='agos.cli.auth',
):
payload = {
'timestamp': _utc_now_iso(),
'level': level,
'component': component,
'operation': operation,
'entity_id': entity_id,
'correlation_id': correlation_id,
'metadata': metadata or {},
}
click.echo(json.dumps(payload), err=True)
def _first_env(*keys):
for key in keys:
value = os.getenv(key, '').strip()
if value:
return value
return ''
def _parse_scopes(raw_scopes):
normalized = (raw_scopes or '').replace(',', ' ')
scopes = [scope.strip() for scope in normalized.split() if scope.strip()]
return scopes or ['openid', 'profile', 'email', 'offline_access']
def _pkce_verifier():
return secrets.token_urlsafe(64)
def _pkce_challenge(verifier):
digest = hashlib.sha256(verifier.encode('utf-8')).digest()
return base64.urlsafe_b64encode(digest).rstrip(b'=').decode('utf-8')
def _fetch_discovery(issuer, correlation_id):
issuer_base = issuer.rstrip('/')
discovery_url = f'{issuer_base}/.well-known/openid-configuration'
_emit_log('INFO', 'fetch_discovery_start', correlation_id, metadata={'issuer': issuer_base})
response = requests.get(discovery_url, timeout=15)
response.raise_for_status()
discovery = response.json()
_emit_log(
'INFO',
'fetch_discovery_ok',
correlation_id,
metadata={
'issuer': issuer_base,
'authorization_endpoint': discovery.get('authorization_endpoint'),
'token_endpoint': discovery.get('token_endpoint'),
},
)
return discovery
def _write_auth_state(payload):
_AUTH_DIR.mkdir(parents=True, exist_ok=True)
_AUTH_FILE.write_text(json.dumps(payload, indent=2), encoding='utf-8')
os.chmod(_AUTH_FILE, 0o600)
def _start_callback_server(host, port, expected_state, timeout_seconds, correlation_id):
callback_event = threading.Event()
callback_payload = {}
class CallbackHandler(BaseHTTPRequestHandler):
def do_GET(self):
parsed = urlparse(self.path)
params = parse_qs(parsed.query)
callback_payload['path'] = parsed.path
callback_payload['code'] = params.get('code', [None])[0]
callback_payload['state'] = params.get('state', [None])[0]
callback_payload['error'] = params.get('error', [None])[0]
callback_payload['error_description'] = params.get('error_description', [None])[0]
callback_payload['received_at'] = _utc_now_iso()
callback_payload['valid_state'] = callback_payload.get('state') == expected_state
body = (
'<html><body><h2>Agos CLI login complete.</h2>'
'<p>You can return to the terminal.</p></body></html>'
if callback_payload['valid_state'] and not callback_payload.get('error')
else '<html><body><h2>Agos CLI login failed.</h2>'
'<p>Return to the terminal for details.</p></body></html>'
)
status = 200 if callback_payload['valid_state'] else 400
self.send_response(status)
self.send_header('Content-Type', 'text/html; charset=utf-8')
self.end_headers()
self.wfile.write(body.encode('utf-8'))
callback_event.set()
def log_message(self, format, *args):
return
server = HTTPServer((host, port), CallbackHandler)
server.timeout = 0.5
def serve():
deadline = time.time() + timeout_seconds
while time.time() < deadline and not callback_event.is_set():
server.handle_request()
server.server_close()
thread = threading.Thread(target=serve, daemon=True)
thread.start()
_emit_log(
'INFO',
'callback_server_started',
correlation_id,
metadata={'host': host, 'port': port, 'timeout_seconds': timeout_seconds},
)
return callback_event, callback_payload, thread
def _exchange_code_for_tokens(discovery, client_id, code, redirect_uri, code_verifier, correlation_id):
payload = {
'grant_type': 'authorization_code',
'client_id': client_id,
'code': code,
'redirect_uri': redirect_uri,
'code_verifier': code_verifier,
}
_emit_log(
'INFO',
'token_exchange_start',
correlation_id,
metadata={'token_endpoint': discovery.get('token_endpoint'), 'client_id': client_id},
)
response = requests.post(
discovery['token_endpoint'],
data=payload,
headers={'Content-Type': 'application/x-www-form-urlencoded'},
timeout=15,
)
if not response.ok:
try:
error_payload = response.json()
except Exception:
error_payload = {'error': response.text[:200]}
raise click.ClickException(
error_payload.get('error_description')
or error_payload.get('error')
or f'Token exchange failed with status {response.status_code}'
)
tokens = response.json()
_emit_log(
'INFO',
'token_exchange_ok',
correlation_id,
metadata={'expires_in': tokens.get('expires_in'), 'has_refresh_token': bool(tokens.get('refresh_token'))},
)
return tokens
def _fetch_userinfo(discovery, access_token, correlation_id):
userinfo_endpoint = discovery.get('userinfo_endpoint')
if not userinfo_endpoint:
return None
_emit_log('INFO', 'userinfo_fetch_start', correlation_id, metadata={'userinfo_endpoint': userinfo_endpoint})
response = requests.get(
userinfo_endpoint,
headers={'Authorization': f'Bearer {access_token}'},
timeout=15,
)
if not response.ok:
_emit_log('WARN', 'userinfo_fetch_failed', correlation_id, metadata={'status_code': response.status_code})
return None
userinfo = response.json()
_emit_log(
'INFO',
'userinfo_fetch_ok',
correlation_id,
metadata={'subject': userinfo.get('sub'), 'email': userinfo.get('email')},
)
return userinfo
@click.group()
@click.version_option(version='1.0.0', prog_name='AGOS')
def cli():
"""AGOS - Agent Runtime Platform CLI"""
pass
@cli.command()
@click.option('--issuer', default=None, help='Armco IAM issuer URL')
@click.option('--client-id', default=None, help='IAM OAuth client ID for the CLI/native app')
@click.option(
'--host',
default='127.0.0.1',
show_default=True,
help='Loopback host for the local callback listener',
)
@click.option(
'--port',
default=8976,
show_default=True,
type=int,
help='Loopback port for the local callback listener',
)
@click.option(
'--timeout',
'timeout_seconds',
default=180,
show_default=True,
type=int,
help='Seconds to wait for the browser login callback',
)
@click.option('--scopes', default=None, help='Space or comma separated OAuth scopes')
def login(issuer, client_id, host, port, timeout_seconds, scopes):
"""Login with Armco IAM for Agos CLI."""
correlation_id = f'cli_login_{uuid.uuid4().hex[:12]}'
resolved_issuer = issuer or _first_env('AGOS_CLI_IAM_ISSUER', 'VITE_IAM_ISSUER', 'IAM_ISSUER')
resolved_client_id = client_id or _first_env('AGOS_CLI_IAM_CLIENT_ID', 'VITE_IAM_CLIENT_ID')
resolved_scopes = _parse_scopes(scopes or _first_env('AGOS_CLI_IAM_SCOPES'))
redirect_uri = f'http://{host}:{port}/callback'
if not resolved_issuer:
raise click.ClickException('Missing IAM issuer. Set AGOS_CLI_IAM_ISSUER, VITE_IAM_ISSUER, or pass --issuer.')
if not resolved_client_id:
raise click.ClickException('Missing CLI IAM client ID. Set AGOS_CLI_IAM_CLIENT_ID or pass --client-id.')
if os.getenv('AGOS_CLI_IAM_CLIENT_ID', '').strip() == '' and os.getenv('VITE_IAM_CLIENT_ID', '').strip():
console.print(
'[yellow]Using VITE_IAM_CLIENT_ID for CLI login. '
'Ensure IAM allows the loopback redirect URI for this client.[/yellow]'
)
_emit_log(
'INFO',
'login_start',
correlation_id,
metadata={
'issuer': resolved_issuer,
'client_id': resolved_client_id,
'redirect_uri': redirect_uri,
'scopes': resolved_scopes,
},
)
try:
discovery = _fetch_discovery(resolved_issuer, correlation_id)
state = secrets.token_urlsafe(24)
nonce = secrets.token_urlsafe(24)
code_verifier = _pkce_verifier()
code_challenge = _pkce_challenge(code_verifier)
callback_event, callback_payload, thread = _start_callback_server(
host=host,
port=port,
expected_state=state,
timeout_seconds=timeout_seconds,
correlation_id=correlation_id,
)
authorize_params = {
'client_id': resolved_client_id,
'redirect_uri': redirect_uri,
'response_type': 'code',
'scope': ' '.join(resolved_scopes),
'state': state,
'nonce': nonce,
'code_challenge': code_challenge,
'code_challenge_method': 'S256',
}
authorize_url = f"{discovery['authorization_endpoint']}?{urlencode(authorize_params)}"
browser_opened = webbrowser.open(authorize_url)
_emit_log(
'INFO',
'browser_opened',
correlation_id,
metadata={'browser_opened': browser_opened, 'authorization_endpoint': discovery['authorization_endpoint']},
)
console.print('[bold blue]Opening Armco IAM login in your browser...[/bold blue]')
console.print(f'[dim]If the browser does not open, use this URL:[/dim]\n{authorize_url}')
if not callback_event.wait(timeout=timeout_seconds):
raise click.ClickException('Timed out waiting for the IAM login callback.')
thread.join(timeout=1)
if callback_payload.get('error'):
raise click.ClickException(
callback_payload.get('error_description') or callback_payload['error']
)
if not callback_payload.get('valid_state'):
raise click.ClickException('IAM callback state validation failed.')
code = callback_payload.get('code')
if not code:
raise click.ClickException('IAM callback did not include an authorization code.')
tokens = _exchange_code_for_tokens(
discovery=discovery,
client_id=resolved_client_id,
code=code,
redirect_uri=redirect_uri,
code_verifier=code_verifier,
correlation_id=correlation_id,
)
userinfo = _fetch_userinfo(discovery, tokens['access_token'], correlation_id)
auth_state = {
'timestamp': _utc_now_iso(),
'issuer': resolved_issuer.rstrip('/'),
'client_id': resolved_client_id,
'redirect_uri': redirect_uri,
'scopes': resolved_scopes,
'access_token': tokens.get('access_token'),
'refresh_token': tokens.get('refresh_token'),
'id_token': tokens.get('id_token'),
'token_type': tokens.get('token_type'),
'expires_in': tokens.get('expires_in'),
'expires_at': int(time.time()) + int(tokens.get('expires_in', 0) or 0),
'user': userinfo,
}
_write_auth_state(auth_state)
_emit_log(
'INFO',
'login_completed',
correlation_id,
entity_id=(userinfo or {}).get('sub', 'agos_cli'),
metadata={
'auth_file': str(_AUTH_FILE),
'email': (userinfo or {}).get('email'),
'has_refresh_token': bool(tokens.get('refresh_token')),
},
)
console.print('[bold green]✓ Login successful[/bold green]')
if userinfo:
identity_label = userinfo.get('email') or userinfo.get('username') or userinfo.get('sub')
console.print(f'Authenticated as: {identity_label}')
console.print(f'Token cache: {_AUTH_FILE}')
except Exception as exc:
_emit_log('ERROR', 'login_failed', correlation_id, metadata={'error': str(exc)})
if isinstance(exc, click.ClickException):
raise
raise click.ClickException(str(exc)) from exc
@cli.group()
def db():
"""Database management commands"""
pass
@db.command()
@click.option('--host', default='localhost', help='PostgreSQL host')
@click.option('--port', default=5432, help='PostgreSQL port')
@click.option('--database', default='agos', help='Database name')
@click.option('--user', default='postgres', help='Database user')
@click.password_option('--password', help='Database password')
def db_bootstrap_command(host, port, database, user, password):
"""Bootstrap database schema (idempotent)"""
import asyncpg
import asyncio
async def run_bootstrap():
try:
console.print("[bold blue]Bootstrapping AGOS database...[/bold blue]")
conn = await asyncpg.connect(
host=host,
port=port,
database=database,
user=user,
password=password
)
bootstrap_path = os.path.join(
os.path.dirname(__file__), '..', 'seeders', 'bootstrap.sql'
)
with open(bootstrap_path, 'r') as f:
sql = f.read()
await conn.execute(sql)
await conn.close()
console.print("[bold green]✓ Database bootstrap completed successfully![/bold green]")
except Exception as e:
console.print(f"[bold red]✗ Bootstrap failed: {e}[/bold red]")
sys.exit(1)
asyncio.run(run_bootstrap())
@db.command()
@click.option('--host', default='localhost', help='PostgreSQL host')
@click.option('--port', default=5432, help='PostgreSQL port')
@click.option('--database', default='agos', help='Database name')
@click.option('--user', default='postgres', help='Database user')
@click.password_option('--password', help='Database password')
def db_seed_command(host, port, database, user, password):
"""Seed database with test data"""
import asyncpg
import asyncio
async def run_seed():
try:
console.print("[bold blue]Seeding AGOS database...[/bold blue]")
conn = await asyncpg.connect(
host=host,
port=port,
database=database,
user=user,
password=password
)
seed_path = os.path.join(
os.path.dirname(__file__), '..', 'seeders', 'seed.sql'
)
with open(seed_path, 'r') as f:
sql = f.read()
await conn.execute(sql)
await conn.close()
console.print("[bold green]✓ Database seeded successfully![/bold green]")
except Exception as e:
console.print(f"[bold red]✗ Seed failed: {e}[/bold red]")
sys.exit(1)
asyncio.run(run_seed())
@db.command()
@click.option('--host', default='localhost', help='PostgreSQL host')
@click.option('--port', default=5432, help='PostgreSQL port')
@click.option('--database', default='agos', help='Database name')
@click.option('--user', default='postgres', help='Database user')
@click.password_option('--password', help='Database password')
def db_status_command(host, port, database, user, password):
"""Check database status"""
import asyncpg
import asyncio
async def check_status():
try:
console.print("[bold blue]Checking database status...[/bold blue]")
conn = await asyncpg.connect(
host=host,
port=port,
database=database,
user=user,
password=password
)
# Check tables
tables = await conn.fetch("""
SELECT table_name
FROM information_schema.tables
WHERE table_schema = 'public'
ORDER BY table_name
""")
# Check row counts
agent_count = await conn.fetchval("SELECT COUNT(*) FROM agents")
task_count = await conn.fetchval("SELECT COUNT(*) FROM tasks")
plugin_count = await conn.fetchval("SELECT COUNT(*) FROM plugins")
await conn.close()
table = Table(title="Database Status")
table.add_column("Metric", style="cyan")
table.add_column("Value", style="green")
table.add_row("Tables", str(len(tables)))
table.add_row("Agents", str(agent_count))
table.add_row("Tasks", str(task_count))
table.add_row("Plugins", str(plugin_count))
console.print(table)
console.print("[bold green]✓ Database is healthy[/bold green]")
except Exception as e:
console.print(f"[bold red]✗ Status check failed: {e}[/bold red]")
sys.exit(1)
asyncio.run(check_status())
@cli.group()
def agent():
"""Agent management commands"""
pass
@agent.command()
@click.option('--name', required=True, help='Agent name')
@click.option('--description', help='Agent description')
@click.option('--model', default='gpt-4', help='LLM model')
def agent_create_stub(name, description, model):
"""Create a new agent"""
console.print(f"[bold blue]Creating agent: {name}[/bold blue]")
console.print("[yellow]Note: Full agent creation requires API implementation[/yellow]")
console.print(f" Name: {name}")
console.print(f" Description: {description or 'N/A'}")
console.print(f" Model: {model}")
@agent.command()
def agent_list_stub():
"""List all agents"""
console.print("[bold blue]Listing agents...[/bold blue]")
console.print("[yellow]Note: Full agent listing requires API implementation[/yellow]")
@agent.command()
@click.argument('agent_id')
def agent_start_stub(agent_id):
"""Start an agent"""
console.print(f"[bold blue]Starting agent: {agent_id}[/bold blue]")
console.print("[yellow]Note: Full agent start requires runtime implementation[/yellow]")
@agent.command()
@click.argument('agent_id')
def agent_stop_stub(agent_id):
"""Stop an agent"""
console.print(f"[bold blue]Stopping agent: {agent_id}[/bold blue]")
console.print("[yellow]Note: Full agent stop requires runtime implementation[/yellow]")
@agent.command()
@click.argument('agent_id')
@click.confirmation_option(prompt='Are you sure you want to delete this agent?')
def agent_delete_stub(agent_id):
"""Delete an agent"""
console.print(f"[bold red]Deleting agent: {agent_id}[/bold red]")
console.print("[yellow]Note: Full agent deletion requires API implementation[/yellow]")
@cli.group()
def plugin():
"""Plugin management commands"""
pass
@plugin.command()
@click.argument('plugin_name')
def plugin_install_stub(plugin_name):
"""Install a plugin"""
console.print(f"[bold blue]Installing plugin: {plugin_name}[/bold blue]")
console.print("[yellow]Note: Full plugin install requires implementation[/yellow]")
@plugin.command()
def plugin_list_stub():
"""List installed plugins"""
console.print("[bold blue]Listing plugins...[/bold blue]")
console.print("[yellow]Note: Full plugin listing requires API implementation[/yellow]")
@plugin.command()
@click.argument('query')
def plugin_search_stub(query):
"""Search marketplace for plugins"""
console.print(f"[bold blue]Searching for: {query}[/bold blue]")
console.print("[yellow]Note: Full search requires marketplace API[/yellow]")
@plugin.command()
@click.argument('plugin_id')
@click.confirmation_option(prompt='Are you sure you want to uninstall this plugin?')
def plugin_uninstall_stub(plugin_id):
"""Uninstall a plugin"""
console.print(f"[bold red]Uninstalling plugin: {plugin_id}[/bold red]")
console.print("[yellow]Note: Full uninstall requires implementation[/yellow]")
@cli.group()
def workflow():
"""Workflow management commands"""
pass
@workflow.command()
@click.option('--name', required=True, help='Workflow name')
@click.option('--file', type=click.Path(exists=True), help='Workflow definition file')
def workflow_create_stub(name, file):
"""Create a new workflow"""
console.print(f"[bold blue]Creating workflow: {name}[/bold blue]")
if file:
console.print(f" From file: {file}")
console.print("[yellow]Note: Full workflow creation requires API implementation[/yellow]")
@workflow.command()
@click.argument('workflow_id')
def workflow_run_stub(workflow_id):
"""Execute a workflow"""
console.print(f"[bold blue]Running workflow: {workflow_id}[/bold blue]")
console.print("[yellow]Note: Full workflow execution requires runtime[/yellow]")
@workflow.command()
@click.argument('workflow_id')
def workflow_status_stub(workflow_id):
"""Check workflow status"""
console.print(f"[bold blue]Checking status of workflow: {workflow_id}[/bold blue]")
console.print("[yellow]Note: Full status check requires API implementation[/yellow]")
@cli.group()
def policy():
"""Policy management commands"""
pass
@policy.command()
@click.option('--name', required=True, help='Policy name')
@click.option('--type', 'policy_type', required=True,
type=click.Choice(['filesystem', 'network', 'resource']))
@click.option('--effect', required=True,
type=click.Choice(['allow', 'deny', 'require_approval']))
def policy_create_stub(name, policy_type, effect):
"""Create a new policy"""
console.print(f"[bold blue]Creating policy: {name}[/bold blue]")
console.print(f" Type: {policy_type}")
console.print(f" Effect: {effect}")
console.print("[yellow]Note: Full policy creation requires API implementation[/yellow]")
@policy.command()
def policy_list_stub():
"""List all policies"""
console.print("[bold blue]Listing policies...[/bold blue]")
console.print("[yellow]Note: Full policy listing requires API implementation[/yellow]")
@policy.command()
@click.argument('policy_id')
@click.confirmation_option(prompt='Are you sure you want to delete this policy?')
def policy_delete_stub(policy_id):
"""Delete a policy"""
console.print(f"[bold red]Deleting policy: {policy_id}[/bold red]")
console.print("[yellow]Note: Full policy deletion requires API implementation[/yellow]")
@cli.group()
def system():
"""System operations"""
pass
@system.command()
def system_status_stub():
"""Show system status"""
console.print("[bold blue]System Status[/bold blue]")
table = Table(title="AGOS Status")
table.add_column("Component", style="cyan")
table.add_column("Status", style="green")
table.add_row("Database", "✓ Connected")
table.add_row("Runtime", "⚠ Not Started")
table.add_row("API Server", "⚠ Not Started")
table.add_row("Plugin System", "✓ Ready")
console.print(table)
@system.command()
@click.option('--lines', default=50, help='Number of log lines to show')
def system_logs_stub(lines):
"""Show system logs"""
console.print(f"[bold blue]Showing last {lines} log lines...[/bold blue]")
console.print("[yellow]Note: Full log viewing requires implementation[/yellow]")
@system.command()
def system_metrics_stub():
"""Show system metrics"""
console.print("[bold blue]System Metrics[/bold blue]")
console.print("[yellow]Note: Full metrics require observability implementation[/yellow]")
install_api_commands(
cli=cli,
agent_group=agent,
plugin_group=plugin,
workflow_group=workflow,
policy_group=policy,
system_group=system,
console=console,
emit_log=_emit_log,
auth_file=_AUTH_FILE,
)
install_manual_commands(cli=cli, console=console)
if __name__ == '__main__':
cli()

1367
cli/command_surface.py Normal file
View File

File diff suppressed because it is too large Load Diff

562
cli/manual.py Normal file
View File

@@ -0,0 +1,562 @@
from typing import Dict, Iterable, Sequence
import click
from rich.console import Console
from rich.panel import Panel
from rich.table import Table
_MANUAL: Dict[str, Dict[str, object]] = {
'agos': {
'summary': 'AGOS CLI for authentication, runtime operations, and API-backed platform management.',
'examples': [
'agos login',
'agos ask Hi',
'agos ask "Summarize today\'s failed tasks"',
'agos whoami',
'agos agent list',
'agos task list --status running',
'agos workflow run wf_autoblogger_v1 --agent-id agent_123',
'agos help agent',
'agos man workflow run',
],
'notes': [
'Most API-backed commands honor `--api-url` and default to `http://localhost:2000`.',
'Many read commands also accept `--json-output` for scripting.',
'Run `agos help <topic>` for focused usage or `agos man` for the full manual.',
],
},
'login': {
'summary': 'Authenticate the CLI with Armco IAM and cache tokens in `~/.agos/auth.json`.',
'examples': [
'agos login',
'agos login --issuer https://iam.armco.dev --client-id client_abc123',
'agos login --port 8976 --timeout 240',
],
},
'whoami': {
'summary': 'Show the identity, tenant, scopes, and admin status for the current CLI session.',
'examples': [
'agos whoami',
'agos whoami --json-output',
'agos whoami --api-url https://agos.armco.dev',
],
},
'quota': {
'summary': 'Inspect your current quota status or recent quota consumption history.',
'examples': [
'agos quota',
'agos quota --history',
'agos quota --json-output',
],
},
'logout': {
'summary': 'Clear the locally cached Agos session from disk.',
'examples': ['agos logout'],
},
'help': {
'summary': 'Show focused help for a top-level command or nested action.',
'examples': [
'agos help',
'agos help agent',
'agos help workflow run',
],
},
'clear': {
'summary': 'Clear the visible terminal screen for the current CLI session.',
'examples': ['agos clear'],
},
'ask': {
'summary': 'Send a natural-language prompt to Agos chat using the authenticated CLI session.',
'examples': [
'agos ask Hi',
'agos ask "Plan a launch checklist for AGOS"',
'agos ask --json-output "Summarize current quota usage"',
],
},
'man': {
'summary': 'Show the expanded Agos manual with detailed actions and examples.',
'examples': [
'agos man',
'agos man task create',
'agos man installable',
],
},
'db': {
'summary': 'Run local database bootstrap, seed, and status operations.',
'examples': [
'agos db status --host localhost --database agos',
'agos db bootstrap --database agos --user postgres',
],
},
'db bootstrap': {
'summary': 'Apply the Agos bootstrap schema to a PostgreSQL database.',
'examples': [
'agos db bootstrap --database agos --user postgres',
],
},
'db seed': {
'summary': 'Seed the Agos database with local development data.',
'examples': [
'agos db seed --database agos --user postgres',
],
},
'db status': {
'summary': 'Check database connectivity and basic table counts.',
'examples': [
'agos db status --database agos --user postgres',
],
},
'installable': {
'summary': 'Inspect and manage installable Agos features for the current instance.',
'examples': [
'agos installable list',
'agos installable install integrations',
],
},
'installable list': {
'summary': 'List installable features and whether they are enabled.',
'examples': [
'agos installable list',
'agos installable list --json-output',
],
},
'installable install': {
'summary': 'Enable a named installable feature.',
'examples': ['agos installable install integrations'],
},
'installable uninstall': {
'summary': 'Disable a named installable feature.',
'examples': ['agos installable uninstall integrations'],
},
'agent': {
'summary': 'Create, inspect, run, and control Agos agents.',
'examples': [
'agos agent list',
'agos agent get agent_123',
'agos agent run agent_123 --goal "Summarize today\'s issues"',
],
},
'agent list': {
'summary': 'List agents with optional filtering by page, status, or ephemeral flag.',
'examples': [
'agos agent list',
'agos agent list --status running',
'agos agent list --page 2 --page-size 50',
],
},
'agent get': {
'summary': 'Fetch the full details for a single agent.',
'examples': ['agos agent get agent_123'],
},
'agent create': {
'summary': 'Create a new agent with model, capabilities, and optional JSON configuration.',
'examples': [
'agos agent create --name "Ops Agent" --model gpt-4.1',
(
'agos agent create --name "RAG Agent" --model gpt-4.1 '
'--capability search --config-json "{\"temperature\":0.2}"'
),
],
},
'agent start': {
'summary': 'Resume a paused agent.',
'examples': ['agos agent start agent_123'],
},
'agent stop': {
'summary': 'Pause a running agent.',
'examples': ['agos agent stop agent_123'],
},
'agent delete': {
'summary': 'Delete an agent after interactive confirmation.',
'examples': ['agos agent delete agent_123'],
},
'agent run': {
'summary': 'Queue a goal-driven run for an agent.',
'examples': [
'agos agent run agent_123 --goal "Write the daily operations brief"',
'agos agent run agent_123 --goal "Triage open incidents" --max-reasoning-turns 6',
],
},
'task': {
'summary': 'List, create, inspect, cancel, and delete tasks.',
'examples': [
'agos task list',
'agos task get task_123',
'agos task create --agent-id agent_123 --description "Run daily sync"',
],
},
'task list': {
'summary': 'List tasks with optional agent and status filters.',
'examples': [
'agos task list',
'agos task list --status failed',
'agos task list --agent-id agent_123',
],
},
'task get': {
'summary': 'Fetch the full details for a single task.',
'examples': ['agos task get task_123'],
},
'task create': {
'summary': 'Create a task for an agent with structured input and retry controls.',
'examples': [
'agos task create --agent-id agent_123 --description "Analyze churn risk"',
(
'agos task create --agent-id agent_123 --description '
'"Import backlog" --input-json "{\"board_id\":\"b1\"}"'
),
],
},
'task cancel': {
'summary': 'Cancel an in-flight task.',
'examples': ['agos task cancel task_123'],
},
'task delete': {
'summary': 'Delete a task after interactive confirmation.',
'examples': ['agos task delete task_123'],
},
'workflow': {
'summary': 'Inspect, create, run, and monitor workflows.',
'examples': [
'agos workflow list',
'agos workflow create --name "Daily Brief" --definition-file workflow.json',
'agos workflow run wf_123 --agent-id agent_123',
],
},
'workflow list': {
'summary': 'List workflows with optional status and text search filters.',
'examples': [
'agos workflow list',
'agos workflow list --status active',
'agos workflow list --search autoblogger',
],
},
'workflow get': {
'summary': 'Fetch the full details for a workflow.',
'examples': ['agos workflow get wf_123'],
},
'workflow create': {
'summary': 'Create a workflow from inline JSON or a definition file.',
'examples': [
'agos workflow create --name "Daily Brief" --definition-file workflow.json',
'agos workflow create --name "Quick Flow" --definition-json "{\"nodes\":[]}"',
],
},
'workflow run': {
'summary': 'Start a workflow execution with an agent and optional runtime parameters.',
'examples': [
'agos workflow run wf_123 --agent-id agent_123',
'agos workflow run wf_123 --agent-id agent_123 --parameters-json "{\"region\":\"us\"}"',
],
},
'workflow status': {
'summary': 'List recent executions for a workflow.',
'examples': [
'agos workflow status wf_123',
'agos workflow status wf_123 --limit 50',
],
},
'plugin': {
'summary': 'List, search, install, inspect, enable, disable, and uninstall plugins.',
'examples': [
'agos plugin list',
'agos plugin search slack',
'agos plugin enable plugin_123',
],
},
'plugin list': {
'summary': 'List installed plugins with trust and verification filters.',
'examples': [
'agos plugin list',
'agos plugin list --verified-only',
'agos plugin list --trust-level official',
],
},
'plugin get': {
'summary': 'Fetch the full details for a plugin.',
'examples': ['agos plugin get plugin_123'],
},
'plugin search': {
'summary': 'Search the marketplace plugin catalog.',
'examples': [
'agos plugin search slack',
'agos plugin search rag',
],
},
'plugin install': {
'summary': 'Install a plugin from manifest metadata and local manifest JSON.',
'examples': [
'agos plugin install agos-chat-launcher --version 1.0.0 --author armco --manifest-file manifest.json',
],
},
'plugin enable': {
'summary': 'Enable a previously installed plugin.',
'examples': ['agos plugin enable plugin_123'],
},
'plugin disable': {
'summary': 'Disable an installed plugin.',
'examples': ['agos plugin disable plugin_123'],
},
'plugin uninstall': {
'summary': 'Uninstall a plugin after interactive confirmation.',
'examples': ['agos plugin uninstall plugin_123'],
},
'policy': {
'summary': 'List, create, and delete platform policies.',
'examples': [
'agos policy list',
'agos policy create --name "No Public Internet" --type network --conditions-json "{\"egress\":\"deny\"}"',
],
},
'policy list': {
'summary': 'List policies with optional type, agent, and enabled filters.',
'examples': [
'agos policy list',
'agos policy list --enabled-only',
'agos policy list --policy-type network',
],
},
'policy create': {
'summary': 'Create a new policy with structured conditions.',
'examples': [
(
'agos policy create --name "Filesystem Guard" --type '
'filesystem --conditions-json "{\"paths\":[\"/tmp\"]}"'
),
],
},
'policy delete': {
'summary': 'Delete a policy after interactive confirmation.',
'examples': ['agos policy delete policy_123'],
},
'integration': {
'summary': 'Inspect the integration catalog, providers, instances, and connectivity.',
'examples': [
'agos integration catalog --search slack',
'agos integration providers',
'agos integration create --provider-id slack --name "Ops Slack"',
],
},
'integration catalog': {
'summary': 'List catalog entries for supported integrations.',
'examples': [
'agos integration catalog',
'agos integration catalog --search slack',
'agos integration catalog --category communication',
],
},
'integration providers': {
'summary': 'List configured integration providers.',
'examples': [
'agos integration providers',
'agos integration providers --category communication',
],
},
'integration instances': {
'summary': 'List created integration instances.',
'examples': [
'agos integration instances',
'agos integration instances --category communication',
],
},
'integration create': {
'summary': 'Create a new integration instance with JSON config.',
'examples': [
(
'agos integration create --provider-id slack --name '
'"Ops Slack" --config-json "{\"workspace\":\"armco\"}"'
),
],
},
'integration test': {
'summary': 'Run a test against an integration instance.',
'examples': ['agos integration test instance_123'],
},
'system': {
'summary': 'Inspect high-level system health and local runtime views.',
'examples': [
'agos system status',
'agos system logs --lines 100',
],
},
'system status': {
'summary': 'Show a system summary using the authenticated API surface.',
'examples': [
'agos system status',
'agos system status --json-output',
],
},
'system logs': {
'summary': 'Show the legacy local log view.',
'examples': ['agos system logs --lines 100'],
},
'system metrics': {
'summary': 'Show the legacy local metrics view.',
'examples': ['agos system metrics'],
},
}
def _visible_commands(command: click.MultiCommand) -> Iterable[tuple[str, click.Command]]:
for name in command.list_commands(click.Context(command)):
if name.endswith('-stub') or name.endswith('-command'):
continue
yield name, command.get_command(click.Context(command), name)
def _find_command(root: click.MultiCommand, topic: str) -> tuple[str, click.Command]:
normalized = ' '.join(topic.strip().split())
if normalized in {'', 'agos'}:
return 'agos', root
if normalized.startswith('agos '):
normalized = normalized[len('agos '):]
parts = normalized.split()
current: click.Command = root
path_parts = ['agos']
for part in parts:
if not isinstance(current, click.MultiCommand):
raise click.ClickException(f'`{normalized}` is not a valid Agos command path.')
next_command = current.get_command(click.Context(current), part)
if next_command is None:
raise click.ClickException(f'Unknown Agos help topic: `{normalized}`.')
current = next_command
path_parts.append(part)
return ' '.join(path_parts), current
def _entry_for(path: str) -> Dict[str, object]:
normalized = path.replace('agos ', '').strip()
if path == 'agos':
return _MANUAL.get('agos', {})
return _MANUAL.get(normalized, {})
def _usage_for(path: str, command: click.Command) -> str:
ctx = click.Context(command, info_name=path)
return command.get_usage(ctx).strip()
def _examples_for(path: str) -> Sequence[str]:
return _entry_for(path).get('examples', []) # type: ignore[return-value]
def _notes_for(path: str) -> Sequence[str]:
return _entry_for(path).get('notes', []) # type: ignore[return-value]
def _summary_for(path: str, command: click.Command) -> str:
entry = _entry_for(path)
if entry.get('summary'):
return str(entry['summary'])
return command.help or command.short_help or 'No description available.'
def _render_subcommands(console: Console, path: str, command: click.Command) -> None:
if not isinstance(command, click.MultiCommand):
return
rows = list(_visible_commands(command))
if not rows:
return
table = Table(title=f'{path} actions')
table.add_column('Action', style='cyan')
table.add_column('Description', style='green')
for name, subcommand in rows:
sub_path = f'{path} {name}'
table.add_row(sub_path, _summary_for(sub_path, subcommand))
console.print(table)
def _render_params(console: Console, command: click.Command) -> None:
if not command.params:
return
table = Table(title='Arguments and Options')
table.add_column('Name', style='cyan')
table.add_column('Required', style='magenta')
table.add_column('Type', style='yellow')
table.add_column('Details', style='green')
for param in command.params:
if isinstance(param, click.Option):
name = ', '.join([*param.opts, *param.secondary_opts])
details = param.help or ''
if param.default not in (None, (), []):
details = f'{details} Default: {param.default}'.strip()
table.add_row(
name,
'yes' if param.required else 'no',
'option',
details or '—',
)
else:
table.add_row(
param.human_readable_name,
'yes',
'argument',
'Positional argument.',
)
console.print(table)
def _render_examples(console: Console, path: str) -> None:
examples = list(_examples_for(path))
if not examples:
return
body = '\n'.join(f'$ {example}' for example in examples)
console.print(Panel.fit(body, title='Examples'))
def _render_notes(console: Console, path: str) -> None:
notes = list(_notes_for(path))
if not notes:
return
body = '\n'.join(f'- {note}' for note in notes)
console.print(Panel.fit(body, title='Notes'))
def _render_topic(console: Console, root: click.MultiCommand, topic: str) -> None:
path, command = _find_command(root, topic)
console.rule(f'[bold blue]{path}[/bold blue]')
console.print(Panel.fit(_summary_for(path, command), title='Summary'))
console.print(Panel.fit(_usage_for(path, command), title='Usage'))
_render_subcommands(console, path, command)
_render_params(console, command)
_render_examples(console, path)
_render_notes(console, path)
def _render_root_help(console: Console, root: click.MultiCommand) -> None:
console.print(Panel.fit(_summary_for('agos', root), title='Agos Help'))
_render_subcommands(console, 'agos', root)
_render_examples(console, 'agos')
_render_notes(console, 'agos')
def _all_topics(root: click.MultiCommand) -> Sequence[str]:
topics = ['agos']
for name, command in _visible_commands(root):
path = f'agos {name}'
topics.append(path)
if isinstance(command, click.MultiCommand):
for sub_name, _ in _visible_commands(command):
topics.append(f'{path} {sub_name}')
return topics
def install_manual_commands(cli: click.MultiCommand, console: Console) -> None:
@cli.command(name='help')
@click.argument('topic', nargs=-1)
def help_command(topic: Sequence[str]) -> None:
if not topic:
_render_root_help(console, cli)
return
_render_topic(console, cli, ' '.join(topic))
@cli.command(name='man')
@click.argument('topic', nargs=-1)
def man_command(topic: Sequence[str]) -> None:
if topic:
_render_topic(console, cli, ' '.join(topic))
return
console.print(Panel.fit(_summary_for('agos', cli), title='Agos Manual'))
for manual_topic in _all_topics(cli):
_render_topic(console, cli, manual_topic)

21
pyproject.toml Normal file
View File

@@ -0,0 +1,21 @@
[build-system]
requires = ["setuptools>=68", "wheel"]
build-backend = "setuptools.build_meta"
[project]
name = "agos-cli"
version = "0.1.0"
description = "AGOS command line interface"
requires-python = ">=3.11"
dependencies = [
"click==8.1.7",
"rich==13.7.0",
"requests==2.32.3",
"python-dotenv==1.0.1",
]
[project.scripts]
agos = "cli.agos:cli"
[tool.setuptools]
packages = ["cli"]