"""Place a short call to yourself through the public Cleo API (standard library only).""" from __future__ import annotations import argparse import getpass import json import os import re import sys import time import warnings from http.client import HTTPException from urllib.error import HTTPError, URLError from urllib.parse import urlsplit from urllib.request import HTTPRedirectHandler, Request, build_opener from uuid import UUID, uuid4 DEFAULT_BASE_URL = "https://api.cleolabs.com" USER_AGENT = "CleoLiveCallTest/1.0" TERMINAL_STATES = {"completed", "failed", "cancelled"} CALL_STATES = TERMINAL_STATES | {"queued", "dialing", "ringing", "in_progress"} POLL_SECONDS = 3 WAIT_SECONDS = 300 REQUEST_TIMEOUT = 60 class CallError(RuntimeError): """An API or transport failure suitable for displaying in the terminal.""" class RequestRejected(CallError): """The server explicitly refused this request before call submission.""" class NoRedirects(HTTPRedirectHandler): def redirect_request(self, req, fp, code, msg, headers, newurl): # Never forward the customer's bearer key to a redirected destination. return None def base_url(value: str) -> str: value = value.strip().rstrip("/") try: parsed = urlsplit(value) port = parsed.port except ValueError as exc: raise argparse.ArgumentTypeError("Invalid API base URL.") from exc if ( not parsed.hostname or parsed.username is not None or parsed.password is not None or parsed.query or parsed.fragment or parsed.path or port == 0 or any(char.isspace() for char in value) or parsed.scheme not in {"http", "https"} ): raise argparse.ArgumentTypeError("Use an API origin, e.g. https://api.cleolabs.com.") if parsed.scheme == "http" and parsed.hostname not in {"localhost", "127.0.0.1", "::1"}: raise argparse.ArgumentTypeError("Use HTTPS, or HTTP on localhost for development.") return value def phone_number(value: str) -> str: value = re.sub(r"[\s().-]", "", value) if not re.fullmatch(r"\+[1-9][0-9]{7,14}", value): raise argparse.ArgumentTypeError("Include the country code, e.g. +15555550123.") return value def idempotency_key(value: str) -> str: value = value.strip() if not re.fullmatch(r"[A-Za-z0-9._-]{1,200}", value): raise argparse.ArgumentTypeError("Use 1-200 letters, digits, dots, underscores or hyphens.") return value def sample_task(phone: str) -> dict: return { "to": phone, "objective": ( "Run a short, friendly test call with the account owner, who is pretending to be " "the receptionist at fictional Maple Bike Repair. Ask for Saturday opening hours " "and the price of a basic bicycle tune-up. Repeat both answers to confirm them, " "thank the person, and end the call." ), "context": { "test_call": True, "recipient": "The account owner is calling their own phone for this test.", "business": "Maple Bike Repair is fictional; this is a role-play exercise.", }, "constraints": [ "Introduce yourself as Cleo, an AI assistant, and explain this is a test call.", "Ask one question at a time and let the person finish speaking.", "Use only answers heard during the call; return null for information not obtained.", "Do not book anything, make purchases, or request personal or payment information.", "If the person wants to stop, thank them and end the call immediately.", "If voicemail answers, leave no message and end the call.", "Aim to finish within one minute.", ], "success_criteria": [ "Obtain Saturday opening hours and the basic tune-up price.", "Read both answers back and allow the person to correct them before ending.", ], "result_schema": { "type": "object", "properties": { "saturday_hours": {"type": ["string", "null"]}, "basic_tune_up_price": {"type": ["string", "null"]}, }, "required": ["saturday_hours", "basic_tune_up_price"], "additionalProperties": False, }, "limits": {"max_duration_seconds": 120}, } class CallClient: def __init__(self, origin: str, api_key: str): self.origin = origin self.api_key = api_key self.opener = build_opener(NoRedirects()) def request(self, method: str, path: str, payload=None, *, key: str | None = None) -> dict: headers = { "Authorization": f"Bearer {self.api_key}", "Accept": "application/json", # Identify this API client explicitly; the edge rejects Python-urllib's default. "User-Agent": USER_AGENT, } if key: headers["Idempotency-Key"] = key data = None if payload is not None: data = json.dumps(payload).encode("utf-8") headers["Content-Type"] = "application/json" request = Request(self.origin + path, data=data, headers=headers, method=method) try: with self.opener.open(request, timeout=REQUEST_TIMEOUT) as response: result = json.load(response) except HTTPError as exc: message = f"HTTP {exc.code}" problem = {} try: with exc: body = json.load(exc) if isinstance(body, dict): problem = body code = problem.get("code") or problem.get("error_name") or "request_failed" message += f": {code} - {problem.get('detail', '')}" except (ValueError, OSError, HTTPException): pass if exc.code == 403 and problem.get("cloudflare_error") is True: message += "\nCloudflare blocked this request before it reached the call API." if problem.get("error_code") == 1010: message += ( " Review Browser Integrity Check for the API hostname if this persists." ) if problem.get("ray_id"): message += f"\nCloudflare Ray ID: {problem['ray_id']}" # These are pre-submission rejections. Keep conflicts, request timeouts, # redirects and server/transport failures uncertain: a call may exist. if exc.code in {400, 401, 403, 404, 405, 413, 415, 422, 429}: raise RequestRejected(message.replace(self.api_key, "[redacted]")) from None raise CallError(message.replace(self.api_key, "[redacted]")) from None except (URLError, OSError, HTTPException) as exc: raise CallError( f"Cannot reach the API: {exc}".replace(self.api_key, "[redacted]") ) from None except ValueError: raise CallError("The API returned invalid JSON. Check the API base URL.") from None try: UUID(result["id"]) if result["state"] not in CALL_STATES: raise ValueError except (KeyError, TypeError, ValueError, AttributeError): raise CallError("The API returned an invalid call resource.") from None return result def read_api_key() -> str: key = os.environ.get("CLEO_API_KEY") or os.environ.get("NIGHTDIAL_API_KEY") if not key: # getpass can otherwise fall back to echoing a key in a redirected terminal. with warnings.catch_warnings(): warnings.simplefilter("error", getpass.GetPassWarning) try: key = getpass.getpass("Cleo API key (hidden): ") except getpass.GetPassWarning: raise CallError("Use an interactive terminal or set CLEO_API_KEY.") from None key = key.strip() if not key or any(ord(char) < 33 or ord(char) > 126 for char in key): raise CallError("Enter the API key from Cleo Settings, without the 'Bearer ' prefix.") return key def show_result(call: dict) -> int: print(f"\nFinal state: {call['state']}") print(f"Summary: {call.get('summary') or '(not available)'}") print("Result:") print(json.dumps(call.get("result"), indent=2, ensure_ascii=True)) valid = call.get("result_schema_valid") print(f"Result schema valid: {valid if valid is not None else 'not available'}") return 0 if call["state"] == "completed" else 1 def watch_call(client: CallClient, call: dict) -> int: deadline = time.monotonic() + WAIT_SECONDS previous = None while True: if call["state"] != previous: print(f"State: {call['state']}", flush=True) previous = call["state"] if call["state"] in TERMINAL_STATES: return show_result(call) if time.monotonic() >= deadline: raise CallError("Stopped waiting after five minutes. The call may still be active.") time.sleep(POLL_SECONDS) # GET reads stored state; /sync refreshes it from the telephony gateway. call = client.request("POST", f"/v1/calls/{call['id']}/sync") def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument( "--base-url", type=base_url, default=DEFAULT_BASE_URL, help="API origin (default: %(default)s); local: http://127.0.0.1:8010", ) parser.add_argument("--phone", type=phone_number, help="your phone number with country code") parser.add_argument( "--dry-run", action="store_true", help="preview JSON without a key or network" ) parser.add_argument( "--yes", action="store_true", help="place the live call without confirmation" ) parser.add_argument( "--idempotency-key", type=idempotency_key, help="reuse the printed key and same phone to recover an uncertain submission", ) parser.add_argument( "--call-id", type=UUID, help="resume watching an existing call without dialing" ) parser.add_argument("--cancel", action="store_true", help="cancel the call given by --call-id") args = parser.parse_args(argv) if args.cancel and not args.call_id: parser.error("--cancel requires --call-id") if args.call_id and (args.phone or args.dry_run or args.idempotency_key or args.yes): parser.error("--call-id cannot be combined with options for creating a call") call_id = str(args.call_id) if args.call_id else None client = None submitted = False rejected = False try: print(f"API: {args.base_url}") if not call_id: phone = args.phone or phone_number(input("Your phone number (with country code): ")) payload = sample_task(phone) print( f"\nSample task: call {phone}; you play fictional Maple Bike Repair's receptionist." ) print("Cleo will ask for Saturday opening hours and a basic tune-up price.") print('You can reply: "Saturday, 10 AM to 4 PM. A basic tune-up is 45 dollars."') print("You can change an answer to test whether Cleo catches the correction.") print("Maximum call duration: 120 seconds.") if args.dry_run: print(json.dumps(payload, indent=2)) return 0 client = CallClient(args.base_url, read_api_key()) if call_id: action = "cancel" if args.cancel else "sync" call = client.request("POST", f"/v1/calls/{call_id}/{action}") else: if ( not args.yes and input("\nPlace a real call to your phone now? [y/N]: ").lower().strip() != "y" ): print("No call placed.") return 0 key = args.idempotency_key or f"self-test-{uuid4()}" print(f"\nIdempotency key: {key}", flush=True) print("Submitting a live call. Pick up your phone when it rings.", flush=True) submitted = True call = client.request("POST", "/v1/calls", payload, key=key) call_id = call["id"] print(f"Call ID: {call_id}", flush=True) print("Press Ctrl+C to request cancellation while waiting.") return watch_call(client, call) except KeyboardInterrupt: if call_id and client: print("\nRequesting cancellation...") try: call = client.request("POST", f"/v1/calls/{call_id}/cancel") if call["state"] in TERMINAL_STATES: show_result(call) else: print("Cancellation is pending. Check the call again with --call-id.") except (CallError, KeyboardInterrupt) as exc: print(f"Cancellation was not confirmed: {exc}", file=sys.stderr) else: print("\nStopped.") return 130 except RequestRejected as exc: rejected = True print(f"Error: {exc}", file=sys.stderr) return 1 except (CallError, EOFError, argparse.ArgumentTypeError) as exc: print(f"Error: {exc or 'Input ended.'}", file=sys.stderr) return 1 finally: if call_id: print( f"\nCheck again: python scripts/test_live_call.py --base-url {args.base_url} --call-id {call_id}" ) print("Add --cancel to that command to request cancellation.") elif submitted: if rejected: print("\nThis request was rejected before a call could start.") else: print("\nThe submission outcome is unknown; a call may already exist.") print(f"Rerun with --idempotency-key {key} and the SAME phone number and API base URL.") print("Keep the sample task unchanged so the API can safely return the original call.") if __name__ == "__main__": raise SystemExit(main())