#!/usr/bin/env python3 """ Register Genesis Voice Bridge MCP Server with Telnyx ===================================================== Creates the MCP server in Telnyx and attaches it to the AI assistant. Usage: # Set your deployed server URL first export GENESIS_MCP_PUBLIC_URL="https://your-server.com/sse" python register_telnyx.py Steps: 1. Creates an integration secret in Telnyx for the MCP auth token 2. Registers the MCP server with Telnyx 3. Attaches the MCP server to the Claude AI assistant 4. Verifies the configuration """ import os import sys import json import requests # --------------------------------------------------------------------------- # Configuration # --------------------------------------------------------------------------- TELNYX_API_KEY = os.environ.get( "TELNYX_API_KEY", "KEY019BE7A3A2D749FCA8681CFF8448A7F0_vTMM1n77CtQxLDT2ra3P1z", ) TELNYX_API_BASE = "https://api.telnyx.com/v2" # The assistant to attach the MCP server to ASSISTANT_ID = os.environ.get( "TELNYX_ASSISTANT_ID", "assistant-cffc79bc-fd3b-4f96-a8e1-31a360100eb5", ) # MCP server configuration MCP_SERVER_NAME = "genesis-memory-bridge" MCP_AUTH_TOKEN = os.environ.get( "GENESIS_MCP_AUTH_TOKEN", "genesis-voice-bridge-2026-production-key", ) # The public URL where the MCP server is deployed MCP_PUBLIC_URL = os.environ.get("GENESIS_MCP_PUBLIC_URL", "") HEADERS = { "Authorization": f"Bearer {TELNYX_API_KEY}", "Content-Type": "application/json", } def step(n: int, title: str): """Print step header.""" print(f"\n[Step {n}] {title}") print("-" * 50) def create_integration_secret() -> str: """Create an integration secret for the MCP auth token.""" step(1, "Create Integration Secret for MCP Auth Token") # Check if secret already exists resp = requests.get(f"{TELNYX_API_BASE}/integration_secrets", headers=HEADERS) if resp.status_code == 200: for secret in resp.json().get("data", []): if secret["identifier"] == "genesis_mcp_key": print(f" Secret 'genesis_mcp_key' already exists (id: {secret['id']})") return "genesis_mcp_key" # Create new secret resp = requests.post( f"{TELNYX_API_BASE}/integration_secrets", headers=HEADERS, json={ "identifier": "genesis_mcp_key", "type": "bearer", "token": MCP_AUTH_TOKEN, }, ) if resp.status_code in (200, 201): data = resp.json() secret_id = data.get("id", data.get("data", {}).get("id", "unknown")) print(f" Created integration secret: genesis_mcp_key (id: {secret_id})") return "genesis_mcp_key" else: print(f" ERROR: Failed to create secret: {resp.status_code} {resp.text}") return "" def register_mcp_server(secret_ref: str) -> str: """Register the MCP server with Telnyx.""" step(2, "Register MCP Server with Telnyx") if not MCP_PUBLIC_URL: print(" ERROR: GENESIS_MCP_PUBLIC_URL not set!") print(" Set it to your deployed server URL, e.g.:") print(" export GENESIS_MCP_PUBLIC_URL='https://your-server.elestio.app/sse'") return "" # Check if server already exists resp = requests.get(f"{TELNYX_API_BASE}/ai/mcp_servers", headers=HEADERS) if resp.status_code == 200: for server in resp.json().get("data", []): if server["name"] == MCP_SERVER_NAME: print(f" MCP server '{MCP_SERVER_NAME}' already exists (id: {server['id']})") # Update the URL if needed if server.get("url") != MCP_PUBLIC_URL: print(f" Updating URL to: {MCP_PUBLIC_URL}") update_resp = requests.patch( f"{TELNYX_API_BASE}/ai/mcp_servers/{server['id']}", headers=HEADERS, json={"url": MCP_PUBLIC_URL}, ) if update_resp.status_code == 200: print(" URL updated successfully.") return server["id"] # Create new MCP server payload = { "name": MCP_SERVER_NAME, "url": MCP_PUBLIC_URL, "type": "sse", # Telnyx supports sse and http } if secret_ref: payload["api_key_ref"] = secret_ref resp = requests.post( f"{TELNYX_API_BASE}/ai/mcp_servers", headers=HEADERS, json=payload, ) if resp.status_code in (200, 201): data = resp.json() server_id = data.get("id", "unknown") print(f" Registered MCP server: {MCP_SERVER_NAME}") print(f" Server ID: {server_id}") print(f" URL: {MCP_PUBLIC_URL}") print(f" Type: sse") print(f" Auth: {secret_ref or 'none'}") return server_id else: print(f" ERROR: Failed to register: {resp.status_code} {resp.text}") return "" def attach_to_assistant(mcp_server_id: str): """Attach the MCP server to the AI assistant. Note: Telnyx uses POST (not PATCH) for assistant updates, and mcp_servers must be an array of objects [{"id": "..."}] not strings. """ step(3, "Attach MCP Server to AI Assistant") # Get current assistant config resp = requests.get( f"{TELNYX_API_BASE}/ai/assistants/{ASSISTANT_ID}", headers=HEADERS, ) if resp.status_code != 200: print(f" ERROR: Failed to get assistant: {resp.status_code}") return assistant = resp.json() current_mcp = assistant.get("mcp_servers", []) print(f" Current MCP servers: {current_mcp}") # Check if already attached (mcp_servers is a list of objects with 'id' key) existing_ids = [ s["id"] if isinstance(s, dict) else s for s in current_mcp ] if mcp_server_id in existing_ids: print(f" MCP server already attached to assistant.") return # Build updated list as objects (Telnyx requires this format) updated_mcp = list(current_mcp) # keep existing updated_mcp.append({"id": mcp_server_id}) # Update assistant via POST (Telnyx uses POST, not PATCH, for updates) resp = requests.post( f"{TELNYX_API_BASE}/ai/assistants/{ASSISTANT_ID}", headers=HEADERS, json={ "name": assistant.get("name", "AI Assistant"), "mcp_servers": updated_mcp, }, ) if resp.status_code == 200: print(f" Attached MCP server to assistant: {ASSISTANT_ID}") print(f" MCP servers now: {resp.json().get('mcp_servers', [])}") else: print(f" ERROR: Failed to attach: {resp.status_code} {resp.text}") def verify(): """Verify the full configuration.""" step(4, "Verify Configuration") # Check assistant resp = requests.get( f"{TELNYX_API_BASE}/ai/assistants/{ASSISTANT_ID}", headers=HEADERS, ) if resp.status_code == 200: assistant = resp.json() mcp_servers = assistant.get("mcp_servers", []) print(f" Assistant: {assistant['name']}") print(f" MCP Servers: {mcp_servers}") print(f" Model: {assistant['model']}") if mcp_servers: print("\n MCP Server Details:") for entry in mcp_servers: sid = entry["id"] if isinstance(entry, dict) else entry srv_resp = requests.get( f"{TELNYX_API_BASE}/ai/mcp_servers/{sid}", headers=HEADERS, ) if srv_resp.status_code == 200: srv = srv_resp.json() print(f" - {srv['name']}: {srv['url']} (type: {srv['type']})") print("\n VERIFICATION COMPLETE") else: print(f" ERROR: {resp.status_code}") def main(): print("=" * 60) print("Genesis Voice Bridge - Telnyx Registration") print("=" * 60) # Step 1: Create integration secret secret_ref = create_integration_secret() # Step 2: Register MCP server server_id = register_mcp_server(secret_ref) if not server_id: print("\nABORTED: Could not register MCP server.") return 1 # Step 3: Attach to assistant attach_to_assistant(server_id) # Step 4: Verify verify() print("\n" + "=" * 60) print("DONE. The voice assistant now has access to Genesis memory.") print("=" * 60) return 0 if __name__ == "__main__": sys.exit(main())