import os import telnyx import argparse import json from datetime import datetime # Import datetime # Retrieve API key from environment variable or direct definition for testing TELNYX_API_KEY = os.environ.get("TELNYX_API_KEY", "KEY019BE7A3A2D749FCA68681CFF8448A7F0_vTMM1n77CtQxLDT2ra3P1z") GOOGLE_API_KEY = os.environ.get("GOOGLE_API_KEY", "AIzaSyCT_rx0NusUJWoqtT7uxHAKEfHo129SJb8") if not TELNYX_API_KEY: raise ValueError("Telnyx API key not found. Please set TELNYX_API_KEY environment variable or provide it directly.") if not GOOGLE_API_KEY: raise ValueError("Google API key not found. Please set GOOGLE_API_KEY environment variable or provide it directly.") # Initialize the Telnyx client client = telnyx.Telnyx(api_key=TELNYX_API_KEY) # Custom JSON encoder for datetime objects class DateTimeEncoder(json.JSONEncoder): def default(self, obj): if isinstance(obj, datetime): return obj.isoformat() return json.JSONEncoder.default(self, obj) def create_ai_assistant( name: str, model: str, instructions: str, llm_api_key_ref: str, webhook_url: str = None, greeting: str = None, voice_id: str = None, ) -> str: """ Creates an AI Assistant in Telnyx and returns its UUID. """ print(f"Attempting to create AI Assistant: {name}...") try: assistant_data = { "name": name, "model": model, "instructions": instructions, "llm_api_key_ref": llm_api_key_ref, } if greeting: assistant_data["greeting"] = greeting if webhook_url: assistant_data["tools"] = [{ "type": "webhook", "webhook": { "url": webhook_url, "method": "POST", "name": "ghl_lead_capture", "description": "Captures lead information and sends it to GoHighLevel." } }] if voice_id: assistant_data["voice_settings"] = { "voice": voice_id } assistant = client.ai.assistants.create(**assistant_data) assistant_id = assistant.id print(f"Successfully created AI Assistant '{name}' with UUID: {assistant_id}") return assistant_id except telnyx.TelnyxError as e: print(f"Error creating AI Assistant: {e}") raise except Exception as e: print(f"An unexpected error occurred: {e}") raise def get_ai_assistant(assistant_id: str): """ Retrieves details of an AI Assistant by its UUID. """ print(f"Attempting to retrieve AI Assistant with UUID: {assistant_id}...") try: assistant = client.ai.assistants.retrieve(assistant_id) print(f"Successfully retrieved AI Assistant: {assistant.name}") # Use the custom encoder here return json.dumps(assistant.to_dict(), indent=2, cls=DateTimeEncoder) except telnyx.TelnyxError as e: print(f"Error retrieving AI Assistant: {e}") raise except Exception as e: print(f"An unexpected error occurred: {e}") raise def create_integration_secret(identifier: str, secret_type: str, token_value: str = None, username: str = None, password: str = None) -> str: """ Creates an integration secret in Telnyx and returns its identifier. """ print(f"Attempting to create integration secret: {identifier} of type {secret_type}...") try: secret_params = { "identifier": identifier, "type": secret_type, } if secret_type == "bearer" and token_value: secret_params["token"] = token_value elif secret_type == "basic" and username and password: secret_params["username"] = username secret_params["password"] = password else: raise ValueError(f"Invalid parameters for secret type '{secret_type}'.") client.integration_secrets.create(**secret_params) print(f"Successfully created integration secret '{identifier}'") return identifier except telnyx.TelnyxError as e: if e.code == 422 and "Duplicate resource" in str(e): print(f"Integration secret '{identifier}' already exists.") return identifier print(f"Error creating integration secret: {e}") raise except Exception as e: print(f"An unexpected error occurred: {e}") raise def get_integration_secret(identifier: str): """ Retrieves details of an integration secret by its identifier. """ print(f"Attempting to retrieve integration secret with identifier: {identifier}...") try: secret = client.integration_secrets.retrieve(identifier) print(f"Successfully retrieved integration secret: {secret.identifier}") return secret.to_dict() except telnyx.TelnyxError as e: print(f"Error retrieving integration secret: {e}") raise except Exception as e: print(f"An unexpected error occurred: {e}") raise def main(): parser = argparse.ArgumentParser(description="Manage Telnyx AI Assistants and Integration Secrets.") subparsers = parser.add_subparsers(dest="command", help="Available commands") # Create AI Assistant command create_assistant_parser = subparsers.add_parser("create-assistant", help="Create a new AI Assistant.") create_assistant_parser.add_argument("--name", required=True, help="Name of the AI Assistant.") create_assistant_parser.add_argument("--model", required=True, help="LLM model to use (e.g., google/gemini-2.5-flash).") create_assistant_parser.add_argument("--instructions", required=True, help="System instructions for the AI Assistant.") create_assistant_parser.add_argument("--llm-api-key-ref", required=True, help="Identifier of the LLM API Key Integration Secret.") create_assistant_parser.add_argument("--webhook-url", help="GHL webhook URL for lead capture.") create_assistant_parser.add_argument("--greeting", help="Initial greeting for the AI Assistant.") create_assistant_parser.add_argument("--voice-id", help="Voice ID (e.g., Eucalyptus).") # Get AI Assistant command get_assistant_parser = subparsers.add_parser("get-assistant", help="Retrieve details of an AI Assistant.") get_assistant_parser.add_argument("--assistant-id", required=True, help="UUID of the AI Assistant.") # Create Integration Secret command create_secret_parser = subparsers.add_parser("create-secret", help="Create a new Integration Secret.") create_secret_parser.add_argument("--identifier", required=True, help="Unique identifier for the secret.") create_secret_parser.add_argument("--type", required=True, choices=["bearer", "basic"], help="Type of the secret (bearer or basic).") create_secret_parser.add_argument("--token", help="The token value (required for bearer type).") create_secret_parser.add_argument("--username", help="The username (required for basic type).") create_secret_parser.add_argument("--password", help="The password (required for basic type).") # Get Integration Secret command get_secret_parser = subparsers.add_parser("get-secret", help="Retrieve details of an Integration Secret.") get_secret_parser.add_argument("--identifier", required=True, help="Identifier of the Integration Secret.") args = parser.parse_args() if args.command == "create-assistant": try: uuid = create_ai_assistant( name=args.name, model=args.model, instructions=args.instructions, llm_api_key_ref=args.llm_api_key_ref, webhook_url=args.webhook_url, greeting=args.greeting, voice_id=args.voice_id ) print(f"New AI Assistant created with UUID: {uuid}") except Exception as e: print(f"Failed to create AI Assistant: {e}") exit(1) elif args.command == "get-assistant": try: assistant_details_json_str = get_ai_assistant(args.assistant_id) print(assistant_details_json_str) # Print directly, as it's already JSON except Exception as e: print(f"Failed to retrieve AI Assistant: {e}") exit(1) elif args.command == "create-secret": try: identifier = create_integration_secret( identifier=args.identifier, secret_type=args.type, token_value=args.token, username=args.username, password=args.password ) print(f"New Integration Secret created with identifier: {identifier}") except Exception as e: print(f"Failed to create Integration Secret: {e}") exit(1) elif args.command == "get-secret": try: secret_details = get_integration_secret(args.identifier) print(json.dumps(secret_details, indent=2)) except Exception as e: print(f"Failed to retrieve Integration Secret: {e}") exit(1) else: parser.print_help() if __name__ == "__main__": main()