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
295 changes: 275 additions & 20 deletions call_use/cli.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
"""call-use CLI agent-native interface for making outbound calls."""
"""call-use CLI \u2014 agent-native interface for making outbound calls."""

import asyncio
import json
Expand All @@ -10,18 +10,40 @@

from call_use.models import CallError, CallErrorCode, CallEvent

_BASE_ENV_VARS = {
"LIVEKIT_URL": "LiveKit server URL (wss://...)",
"LIVEKIT_API_KEY": "LiveKit API key",
"LIVEKIT_API_SECRET": "LiveKit API secret",
"SIP_TRUNK_ID": "Twilio SIP trunk ID in LiveKit",
"DEEPGRAM_API_KEY": "Deepgram API key (for speech-to-text)",
}

_PROVIDER_ENV_VARS: dict[str, dict[str, str]] = {
"openai": {"OPENAI_API_KEY": "OpenAI API key (LLM + TTS)"},
"openrouter": {
"OPENROUTER_API_KEY": "OpenRouter API key (LLM)",
"OPENAI_API_KEY": "OpenAI API key (for TTS)",
},
"google": {"GOOGLE_API_KEY": "Google API key (LLM + TTS)"},
"grok": {
"XAI_API_KEY": "xAI API key (LLM)",
"OPENAI_API_KEY": "OpenAI API key (for TTS)",
},
}


def _get_env_vars_for_provider(provider: str) -> dict[str, str]:
"""Return the full set of required env vars for the given LLM provider."""
result = dict(_BASE_ENV_VARS)
result.update(_PROVIDER_ENV_VARS.get(provider, _PROVIDER_ENV_VARS["openai"]))
return result


def _check_env():
"""Check required environment variables before attempting a call."""
required = {
"LIVEKIT_URL": "LiveKit server URL (wss://...)",
"LIVEKIT_API_KEY": "LiveKit API key",
"LIVEKIT_API_SECRET": "LiveKit API secret",
"SIP_TRUNK_ID": "Twilio SIP trunk ID in LiveKit",
"OPENAI_API_KEY": "OpenAI API key (for LLM reasoning and text-to-speech)",
"DEEPGRAM_API_KEY": "Deepgram API key (for speech-to-text)",
}
missing = [f" {k} — {v}" for k, v in required.items() if not os.environ.get(k)]
provider = os.environ.get("CALL_USE_LLM_PROVIDER", "openai")
required = _get_env_vars_for_provider(provider)
missing = [f" {k} \u2014 {v}" for k, v in required.items() if not os.environ.get(k)]
if missing:
msg = "Missing required environment variables:\n" + "\n".join(missing)
msg += "\n\nSee: https://github.com/agent-next/call-use#configure"
Expand All @@ -43,7 +65,7 @@ def _event_printer(event: CallEvent):


def _stdin_approval_handler(data: dict) -> str:
"""Interactive approval handler prompts user on stdin."""
"""Interactive approval handler \u2014 prompts user on stdin."""
details = data.get("details", str(data)) if isinstance(data, dict) else str(data)
click.echo(f"\n APPROVAL NEEDED: {details}", err=True)
response = click.prompt(" Approve? [y/n]", type=click.Choice(["y", "n"]), err=True)
Expand Down Expand Up @@ -179,14 +201,11 @@ def dial(phone, instructions, user_info, caller_id, voice_id, timeout, approval_
# doctor command
# ---------------------------------------------------------------------------

_DOCTOR_ENV_VARS = {
"LIVEKIT_URL": "LiveKit server URL",
"LIVEKIT_API_KEY": "LiveKit API key",
"LIVEKIT_API_SECRET": "LiveKit API secret",
"SIP_TRUNK_ID": "Twilio SIP trunk ID in LiveKit",
"OPENAI_API_KEY": "OpenAI API key",
"DEEPGRAM_API_KEY": "Deepgram API key",
}

def _doctor_env_vars() -> dict[str, str]:
"""Return env vars the doctor command should check (provider-aware)."""
provider = os.environ.get("CALL_USE_LLM_PROVIDER", "openai")
return _get_env_vars_for_provider(provider)


def _check_livekit_connectivity() -> tuple[bool, str]:
Expand Down Expand Up @@ -215,7 +234,7 @@ def doctor():
failed = 0

# 1. Environment variables
for var, description in _DOCTOR_ENV_VARS.items():
for var, description in _doctor_env_vars().items():
if os.environ.get(var):
click.echo(click.style(f" \u2713 {var} set", fg="green"))
passed += 1
Expand All @@ -241,3 +260,239 @@ def doctor():
click.echo()
click.echo(f" {passed} passed, {failed} failed")
sys.exit(0 if failed == 0 else 1)


# ---------------------------------------------------------------------------
# setup command
# ---------------------------------------------------------------------------

_INFRA_KEYS: list[dict[str, object]] = [
{
"name": "LIVEKIT_URL",
"hint": "wss://...",
"hide": False,
"validate": lambda v: v.startswith(("wss://", "ws://")),
"error": "Must start with wss:// or ws://",
},
{"name": "LIVEKIT_API_KEY", "hint": None, "hide": False, "validate": None, "error": None},
{"name": "LIVEKIT_API_SECRET", "hint": None, "hide": True, "validate": None, "error": None},
{"name": "SIP_TRUNK_ID", "hint": None, "hide": False, "validate": None, "error": None},
]

_LLM_PROVIDERS: dict[str, dict[str, object]] = {
"1": {
"name": "OpenAI",
"value": "openai",
"keys": [
{
"name": "OPENAI_API_KEY",
"hint": "LLM + TTS",
"hide": True,
"validate": lambda v: v.startswith("sk-"),
"error": "Must start with sk-",
},
],
},
"2": {
"name": "OpenRouter",
"value": "openrouter",
"keys": [
{
"name": "OPENROUTER_API_KEY",
"hint": "LLM",
"hide": True,
"validate": None,
"error": None,
},
{
"name": "OPENAI_API_KEY",
"hint": "for TTS",
"hide": True,
"validate": lambda v: v.startswith("sk-"),
"error": "Must start with sk-",
},
],
},
"3": {
"name": "Google Gemini",
"value": "google",
"keys": [
{
"name": "GOOGLE_API_KEY",
"hint": "LLM + TTS",
"hide": True,
"validate": None,
"error": None,
},
],
},
"4": {
"name": "Grok (xAI)",
"value": "grok",
"keys": [
{
"name": "XAI_API_KEY",
"hint": "LLM",
"hide": True,
"validate": None,
"error": None,
},
{
"name": "OPENAI_API_KEY",
"hint": "for TTS",
"hide": True,
"validate": lambda v: v.startswith("sk-"),
"error": "Must start with sk-",
},
],
},
}

_STT_KEYS: list[dict[str, object]] = [
{"name": "DEEPGRAM_API_KEY", "hint": None, "hide": True, "validate": None, "error": None},
]

_OPTIONAL_KEYS: list[dict[str, object]] = [
{
"name": "API_KEY",
"hint": "for REST API auth",
"hide": False,
"validate": None,
"error": None,
},
]


def _prompt_key(
key_def: dict[str, object], values: dict[str, str], *, required: bool = True
) -> None:
"""Prompt for a single key, validate, and store in *values*.

When *required* is ``False`` an empty value skips the key instead of
re-prompting.
"""
name: str = key_def["name"] # type: ignore[assignment]
hint = f" ({key_def['hint']})" if key_def["hint"] else ""
default = os.environ.get(name, "")

while True:
prompt_display = f" {name}{hint} [{default}]" if default else f" {name}{hint}"
value = click.prompt(
prompt_display,
default=default or "",
hide_input=bool(key_def.get("hide", False)),
show_default=False,
)
value = value.strip().replace("\r", "")

if not value:
if required:
click.echo(click.style(f" \u2717 {name} is required", fg="red"))
continue
click.echo(click.style(f" \u23ed {name} skipped", fg="yellow"))
click.echo()
return

validator = key_def.get("validate")
if validator and callable(validator) and not validator(value):
click.echo(click.style(f" \u2717 {key_def['error']}", fg="red"))
continue

values[name] = value
click.echo(click.style(f" \u2713 {name}", fg="green"))
click.echo()
break


@main.command()
def setup():
"""Interactive first-time configuration wizard."""
from pathlib import Path

env_path = Path(".env")

click.echo()
click.echo(click.style(" call-use setup", bold=True) + " \u2014 first-time configuration")
click.echo(" " + "\u2500" * 38)
click.echo()
click.echo(" This wizard will create a .env file with your API keys.")
click.echo()

# Check for existing .env
if env_path.exists():
overwrite = click.confirm(" Overwrite existing .env?", default=False)
if not overwrite:
click.echo(" Aborted.")
return

values: dict[str, str] = {}

# --- Infrastructure keys ---
_sep = "\u2500"
click.echo(click.style(f" {_sep * 3} Required {_sep * 28}", bold=True))
click.echo()

for key_def in _INFRA_KEYS:
_prompt_key(key_def, values)

# --- LLM provider selection ---
click.echo(" LLM Provider:")
for num, prov in _LLM_PROVIDERS.items():
suffix = " (default)" if num == "1" else ""
click.echo(f" {num}. {prov['name']}{suffix}")

choice = click.prompt(" Select", default="1")
while choice not in _LLM_PROVIDERS:
click.echo(click.style(" \u2717 Invalid choice", fg="red"))
choice = click.prompt(" Select", default="1")

provider = _LLM_PROVIDERS[choice]
values["CALL_USE_LLM_PROVIDER"] = provider["value"] # type: ignore[assignment]
click.echo(click.style(f" \u2713 {provider['name']}", fg="green"))
click.echo()

for key_def in provider["keys"]: # type: ignore[union-attr]
_prompt_key(key_def, values) # type: ignore[arg-type]

# --- STT key ---
for key_def in _STT_KEYS:
_prompt_key(key_def, values)

# --- Optional keys ---
click.echo(click.style(f" {_sep * 3} Optional (press Enter to skip) {_sep * 5}", bold=True))
click.echo()

for key_def in _OPTIONAL_KEYS:
_prompt_key(key_def, values, required=False)

# --- Write .env ---
click.echo(click.style(f" {_sep * 3} Writing .env {_sep * 22}", bold=True))
click.echo()

lines = ["# Generated by call-use setup"]
for k, v in values.items():
safe_v = v.replace("\r", "").replace("\n", "")
lines.append(f"{k}={safe_v}")
env_path.write_text("\n".join(lines) + "\n")
env_path.chmod(0o600)
click.echo(click.style(f" \u2713 Created .env with {len(values)} variables", fg="green"))
click.echo()

# --- Run doctor ---
click.echo(click.style(f" {_sep * 3} Verification {_sep * 20}", bold=True))
click.echo()
click.echo(" Running call-use doctor...")
click.echo()

from dotenv import load_dotenv

load_dotenv(override=True)

for var, _desc in _doctor_env_vars().items():
if os.environ.get(var):
click.echo(click.style(f" \u2713 {var} set", fg="green"))
else:
click.echo(click.style(f" \u2717 {var} missing", fg="red"))

click.echo()
click.echo(' Setup complete! Try: call-use dial "+18001234567" -i "Ask about store hours"')
Loading
Loading