Coverage for integrations / channels / extensions / voice_adapter.py: 35.1%

1""" 

2Voice Channel Adapter 

3 

4Implements voice/phone integration with Twilio and Vonage APIs. 

5Based on SantaClaw extension patterns for voice communication. 

6 

7Features: 

8- Twilio Voice API integration 

9- Vonage (Nexmo) Voice API integration 

10- Inbound/Outbound calls 

11- IVR (Interactive Voice Response) 

12- DTMF input handling 

13- Text-to-Speech (TTS) 

14- Speech-to-Text (STT) 

15- Call recording 

16- Call transfers 

17- Conference calls 

18- Webhook handling 

19- Real-time transcription 

20""" 

21 

22from __future__ import annotations 

23 

24import asyncio 

25import logging 

26import os 

27import json 

28import hmac 

29import hashlib 

30import base64 

31from typing import Optional, List, Dict, Any, Callable, Union 

32from datetime import datetime 

33from dataclasses import dataclass, field 

34from enum import Enum 

35from urllib.parse import urlencode 

36 

37try: 

38 import aiohttp 

39 HAS_HTTP = True 

40except ImportError: 

41 HAS_HTTP = False 

42 

43from ..base import ( 

44 ChannelAdapter, 

45 ChannelConfig, 

46 ChannelStatus, 

47 Message, 

48 MessageType, 

49 MediaAttachment, 

50 SendResult, 

51 ChannelConnectionError, 

52 ChannelSendError, 

53 ChannelRateLimitError, 

54) 

55 

56logger = logging.getLogger(__name__) 

57 

58 

59class VoiceProvider(Enum): 

60 """Voice service provider.""" 

61 TWILIO = "twilio" 

62 VONAGE = "vonage" 

63 

64 

65class CallStatus(Enum): 

66 """Call status.""" 

67 QUEUED = "queued" 

68 RINGING = "ringing" 

69 IN_PROGRESS = "in-progress" 

70 COMPLETED = "completed" 

71 BUSY = "busy" 

72 NO_ANSWER = "no-answer" 

73 FAILED = "failed" 

74 CANCELED = "canceled" 

75 

76 

77class DTMFKey(Enum): 

78 """DTMF key codes.""" 

79 KEY_0 = "0" 

80 KEY_1 = "1" 

81 KEY_2 = "2" 

82 KEY_3 = "3" 

83 KEY_4 = "4" 

84 KEY_5 = "5" 

85 KEY_6 = "6" 

86 KEY_7 = "7" 

87 KEY_8 = "8" 

88 KEY_9 = "9" 

89 KEY_STAR = "*" 

90 KEY_HASH = "#" 

91 

92 

93class TTSVoice(Enum): 

94 """Text-to-Speech voice options.""" 

95 # Twilio voices 

96 ALICE = "alice" 

97 POLLY_JOANNA = "Polly.Joanna" 

98 POLLY_MATTHEW = "Polly.Matthew" 

99 POLLY_AMY = "Polly.Amy" 

100 POLLY_BRIAN = "Polly.Brian" 

101 # Vonage voices 

102 VONAGE_EMMA = "emma" 

103 VONAGE_AMY = "amy" 

104 VONAGE_BRIAN = "brian" 

105 VONAGE_JOEY = "joey" 

106 

107 

108@dataclass 

109class VoiceConfig(ChannelConfig): 

110 """Voice-specific configuration.""" 

111 provider: VoiceProvider = VoiceProvider.TWILIO 

112 # Twilio 

113 twilio_account_sid: str = "" 

114 twilio_auth_token: str = "" 

115 twilio_phone_number: str = "" 

116 # Vonage 

117 vonage_api_key: str = "" 

118 vonage_api_secret: str = "" 

119 vonage_application_id: str = "" 

120 vonage_private_key: str = "" 

121 vonage_phone_number: str = "" 

122 # Common 

123 webhook_base_url: str = "" 

124 default_voice: str = "alice" 

125 default_language: str = "en-US" 

126 enable_recording: bool = False 

127 enable_transcription: bool = False 

128 max_call_duration: int = 3600 # seconds 

129 speech_timeout: int = 3 # seconds 

130 

131 

132@dataclass 

133class VoiceCall: 

134 """Active voice call information.""" 

135 call_sid: str 

136 from_number: str 

137 to_number: str 

138 status: CallStatus 

139 direction: str # inbound, outbound 

140 start_time: Optional[datetime] = None 

141 duration: int = 0 

142 recording_url: Optional[str] = None 

143 transcription: Optional[str] = None 

144 

145 

146@dataclass 

147class IVRMenu: 

148 """IVR menu configuration.""" 

149 name: str 

150 prompt: str 

151 options: Dict[str, Dict[str, Any]] = field(default_factory=dict) 

152 timeout_action: Optional[Dict[str, Any]] = None 

153 invalid_action: Optional[Dict[str, Any]] = None 

154 max_attempts: int = 3 

155 

156 

157class VoiceAdapter(ChannelAdapter): 

158 """ 

159 Voice/Phone adapter with Twilio and Vonage support. 

160 

161 Usage: 

162 # Twilio 

163 config = VoiceConfig( 

164 provider=VoiceProvider.TWILIO, 

165 twilio_account_sid="your-sid", 

166 twilio_auth_token="your-token", 

167 twilio_phone_number="+1234567890", 

168 webhook_base_url="https://your-server.com", 

169 ) 

170 

171 # Vonage 

172 config = VoiceConfig( 

173 provider=VoiceProvider.VONAGE, 

174 vonage_api_key="your-key", 

175 vonage_api_secret="your-secret", 

176 vonage_phone_number="+1234567890", 

177 ) 

178 

179 adapter = VoiceAdapter(config) 

180 adapter.on_message(my_handler) 

181 await adapter.start() 

182 """ 

183 

184 def __init__(self, config: VoiceConfig): 

185 if not HAS_HTTP: 

186 raise ImportError( 

187 "aiohttp not installed. " 

188 "Install with: pip install aiohttp" 

189 ) 

190 

191 super().__init__(config) 

192 self.voice_config: VoiceConfig = config 

193 self._session: Optional[aiohttp.ClientSession] = None 

194 self._active_calls: Dict[str, VoiceCall] = {} 

195 self._ivr_menus: Dict[str, IVRMenu] = {} 

196 self._dtmf_handlers: Dict[str, Callable] = {} 

197 self._speech_handlers: List[Callable] = [] 

198 self._call_start_handlers: List[Callable] = [] 

199 self._call_end_handlers: List[Callable] = [] 

200 

201 @property 

202 def name(self) -> str: 

203 return "voice" 

204 

205 async def connect(self) -> bool: 

206 """Initialize voice adapter.""" 

207 try: 

208 self._session = aiohttp.ClientSession() 

209 

210 # Validate provider credentials 

211 if self.voice_config.provider == VoiceProvider.TWILIO: 

212 if not self.voice_config.twilio_account_sid or not self.voice_config.twilio_auth_token: 

213 logger.error("Twilio credentials required") 

214 return False 

215 

216 # Verify credentials 

217 if not await self._verify_twilio(): 

218 return False 

219 

220 elif self.voice_config.provider == VoiceProvider.VONAGE: 

221 if not self.voice_config.vonage_api_key or not self.voice_config.vonage_api_secret: 

222 logger.error("Vonage credentials required") 

223 return False 

224 

225 # Verify credentials 

226 if not await self._verify_vonage(): 

227 return False 

228 

229 self.status = ChannelStatus.CONNECTED 

230 logger.info(f"Voice adapter connected ({self.voice_config.provider.value})") 

231 return True 

232 

233 except Exception as e: 

234 logger.error(f"Failed to connect voice adapter: {e}") 

235 self.status = ChannelStatus.ERROR 

236 return False 

237 

238 async def disconnect(self) -> None: 

239 """Disconnect voice adapter.""" 

240 if self._session: 

241 await self._session.close() 

242 self._session = None 

243 

244 self._active_calls.clear() 

245 self.status = ChannelStatus.DISCONNECTED 

246 

247 async def _verify_twilio(self) -> bool: 

248 """Verify Twilio credentials.""" 

249 if not self._session: 

250 return False 

251 

252 try: 

253 url = f"https://api.twilio.com/2010-04-01/Accounts/{self.voice_config.twilio_account_sid}.json" 

254 auth = aiohttp.BasicAuth( 

255 self.voice_config.twilio_account_sid, 

256 self.voice_config.twilio_auth_token, 

257 ) 

258 

259 async with self._session.get(url, auth=auth) as resp: 

260 if resp.status == 200: 

261 data = await resp.json() 

262 logger.info(f"Twilio account: {data.get('friendly_name')}") 

263 return True 

264 else: 

265 logger.error(f"Twilio verification failed: {resp.status}") 

266 return False 

267 

268 except Exception as e: 

269 logger.error(f"Twilio verification error: {e}") 

270 return False 

271 

272 async def _verify_vonage(self) -> bool: 

273 """Verify Vonage credentials.""" 

274 if not self._session: 

275 return False 

276 

277 try: 

278 url = "https://api.nexmo.com/account/get-balance" 

279 params = { 

280 "api_key": self.voice_config.vonage_api_key, 

281 "api_secret": self.voice_config.vonage_api_secret, 

282 } 

283 

284 async with self._session.get(url, params=params) as resp: 

285 if resp.status == 200: 

286 data = await resp.json() 

287 logger.info(f"Vonage balance: {data.get('value')}") 

288 return True 

289 else: 

290 logger.error(f"Vonage verification failed: {resp.status}") 

291 return False 

292 

293 except Exception as e: 

294 logger.error(f"Vonage verification error: {e}") 

295 return False 

296 

297 def validate_twilio_signature( 

298 self, 

299 url: str, 

300 params: Dict[str, str], 

301 signature: str, 

302 ) -> bool: 

303 """Validate Twilio webhook signature.""" 

304 if not self.voice_config.twilio_auth_token: 

305 return False 

306 

307 try: 

308 # Build validation string 

309 s = url 

310 if params: 

311 s += "".join(f"{k}{v}" for k, v in sorted(params.items())) 

312 

313 expected = base64.b64encode( 

314 hmac.new( 

315 self.voice_config.twilio_auth_token.encode(), 

316 s.encode(), 

317 hashlib.sha1, 

318 ).digest() 

319 ).decode() 

320 

321 return hmac.compare_digest(signature, expected) 

322 

323 except Exception: 

324 return False 

325 

326 async def handle_webhook(self, body: Dict[str, Any], event_type: str = "") -> Dict[str, Any]: 

327 """ 

328 Handle incoming webhook from Twilio/Vonage. 

329 Returns TwiML/NCCO response. 

330 """ 

331 if self.voice_config.provider == VoiceProvider.TWILIO: 

332 return await self._handle_twilio_webhook(body, event_type) 

333 else: 

334 return await self._handle_vonage_webhook(body, event_type) 

335 

336 async def _handle_twilio_webhook( 

337 self, 

338 body: Dict[str, Any], 

339 event_type: str, 

340 ) -> Dict[str, Any]: 

341 """Handle Twilio webhook.""" 

342 call_sid = body.get("CallSid", "") 

343 call_status = body.get("CallStatus", "") 

344 

345 if event_type == "voice" or "CallSid" in body: 

346 # Incoming call or call status 

347 return await self._handle_twilio_call(body) 

348 

349 elif event_type == "gather" or "Digits" in body: 

350 # DTMF input 

351 return await self._handle_twilio_gather(body) 

352 

353 elif event_type == "speech" or "SpeechResult" in body: 

354 # Speech input 

355 return await self._handle_twilio_speech(body) 

356 

357 elif event_type == "recording": 

358 # Recording complete 

359 await self._handle_twilio_recording(body) 

360 

361 elif event_type == "transcription": 

362 # Transcription complete 

363 await self._handle_twilio_transcription(body) 

364 

365 return {"twiml": "<Response></Response>"} 

366 

367 async def _handle_twilio_call(self, body: Dict[str, Any]) -> Dict[str, Any]: 

368 """Handle Twilio incoming call.""" 

369 call_sid = body.get("CallSid", "") 

370 from_number = body.get("From", "") 

371 to_number = body.get("To", "") 

372 direction = body.get("Direction", "inbound") 

373 status = body.get("CallStatus", "ringing") 

374 

375 # Create/update call record 

376 call = VoiceCall( 

377 call_sid=call_sid, 

378 from_number=from_number, 

379 to_number=to_number, 

380 status=CallStatus(status), 

381 direction=direction, 

382 start_time=datetime.now(), 

383 ) 

384 self._active_calls[call_sid] = call 

385 

386 # Notify handlers 

387 for handler in self._call_start_handlers: 

388 try: 

389 result = handler(call) 

390 if asyncio.iscoroutine(result): 

391 await result 

392 except Exception as e: 

393 logger.error(f"Call start handler error: {e}") 

394 

395 # Create message for incoming call 

396 message = Message( 

397 id=call_sid, 

398 channel=self.name, 

399 sender_id=from_number, 

400 sender_name=from_number, 

401 chat_id=f"call:{call_sid}", 

402 text="[Incoming Call]", 

403 timestamp=datetime.now(), 

404 is_group=False, 

405 raw={ 

406 "call_sid": call_sid, 

407 "direction": direction, 

408 "status": status, 

409 }, 

410 ) 

411 

412 await self._dispatch_message(message) 

413 

414 # Return default greeting TwiML 

415 twiml = self._build_default_greeting_twiml() 

416 return {"twiml": twiml} 

417 

418 async def _handle_twilio_gather(self, body: Dict[str, Any]) -> Dict[str, Any]: 

419 """Handle Twilio DTMF gather.""" 

420 call_sid = body.get("CallSid", "") 

421 digits = body.get("Digits", "") 

422 

423 # Create message for DTMF input 

424 message = Message( 

425 id=f"{call_sid}_{digits}", 

426 channel=self.name, 

427 sender_id=body.get("From", ""), 

428 chat_id=f"call:{call_sid}", 

429 text=f"[DTMF:{digits}]", 

430 timestamp=datetime.now(), 

431 raw={ 

432 "call_sid": call_sid, 

433 "digits": digits, 

434 "input_type": "dtmf", 

435 }, 

436 ) 

437 

438 await self._dispatch_message(message) 

439 

440 # Check for registered handler 

441 if digits in self._dtmf_handlers: 

442 handler = self._dtmf_handlers[digits] 

443 try: 

444 result = await handler(call_sid, digits) 

445 if isinstance(result, str): 

446 return {"twiml": result} 

447 except Exception as e: 

448 logger.error(f"DTMF handler error: {e}") 

449 

450 return {"twiml": "<Response></Response>"} 

451 

452 async def _handle_twilio_speech(self, body: Dict[str, Any]) -> Dict[str, Any]: 

453 """Handle Twilio speech recognition result.""" 

454 call_sid = body.get("CallSid", "") 

455 speech_result = body.get("SpeechResult", "") 

456 confidence = body.get("Confidence", 0) 

457 

458 # Create message for speech input 

459 message = Message( 

460 id=f"{call_sid}_speech", 

461 channel=self.name, 

462 sender_id=body.get("From", ""), 

463 chat_id=f"call:{call_sid}", 

464 text=speech_result, 

465 timestamp=datetime.now(), 

466 raw={ 

467 "call_sid": call_sid, 

468 "speech_result": speech_result, 

469 "confidence": confidence, 

470 "input_type": "speech", 

471 }, 

472 ) 

473 

474 await self._dispatch_message(message) 

475 

476 # Notify speech handlers 

477 for handler in self._speech_handlers: 

478 try: 

479 result = handler(call_sid, speech_result, confidence) 

480 if asyncio.iscoroutine(result): 

481 result = await result 

482 if isinstance(result, str): 

483 return {"twiml": result} 

484 except Exception as e: 

485 logger.error(f"Speech handler error: {e}") 

486 

487 return {"twiml": "<Response></Response>"} 

488 

489 async def _handle_twilio_recording(self, body: Dict[str, Any]) -> None: 

490 """Handle Twilio recording complete.""" 

491 call_sid = body.get("CallSid", "") 

492 recording_url = body.get("RecordingUrl", "") 

493 

494 if call_sid in self._active_calls: 

495 self._active_calls[call_sid].recording_url = recording_url 

496 

497 logger.info(f"Recording complete for {call_sid}: {recording_url}") 

498 

499 async def _handle_twilio_transcription(self, body: Dict[str, Any]) -> None: 

500 """Handle Twilio transcription complete.""" 

501 call_sid = body.get("CallSid", "") 

502 transcription = body.get("TranscriptionText", "") 

503 

504 if call_sid in self._active_calls: 

505 self._active_calls[call_sid].transcription = transcription 

506 

507 logger.info(f"Transcription for {call_sid}: {transcription}") 

508 

509 async def _handle_vonage_webhook( 

510 self, 

511 body: Dict[str, Any], 

512 event_type: str, 

513 ) -> Dict[str, Any]: 

514 """Handle Vonage webhook.""" 

515 conversation_uuid = body.get("conversation_uuid", "") 

516 

517 if body.get("status") == "started": 

518 return await self._handle_vonage_call_started(body) 

519 elif body.get("dtmf"): 

520 return await self._handle_vonage_dtmf(body) 

521 elif body.get("speech"): 

522 return await self._handle_vonage_speech(body) 

523 

524 return {"ncco": []} 

525 

526 async def _handle_vonage_call_started(self, body: Dict[str, Any]) -> Dict[str, Any]: 

527 """Handle Vonage call started.""" 

528 uuid = body.get("uuid", "") 

529 from_number = body.get("from", "") 

530 to_number = body.get("to", "") 

531 

532 call = VoiceCall( 

533 call_sid=uuid, 

534 from_number=from_number, 

535 to_number=to_number, 

536 status=CallStatus.IN_PROGRESS, 

537 direction=body.get("direction", "inbound"), 

538 start_time=datetime.now(), 

539 ) 

540 self._active_calls[uuid] = call 

541 

542 # Create message 

543 message = Message( 

544 id=uuid, 

545 channel=self.name, 

546 sender_id=from_number, 

547 chat_id=f"call:{uuid}", 

548 text="[Incoming Call]", 

549 timestamp=datetime.now(), 

550 raw={"uuid": uuid}, 

551 ) 

552 

553 await self._dispatch_message(message) 

554 

555 # Return default NCCO 

556 ncco = self._build_default_greeting_ncco() 

557 return {"ncco": ncco} 

558 

559 async def _handle_vonage_dtmf(self, body: Dict[str, Any]) -> Dict[str, Any]: 

560 """Handle Vonage DTMF input.""" 

561 uuid = body.get("uuid", "") 

562 dtmf = body.get("dtmf", {}) 

563 digits = dtmf.get("digits", "") 

564 

565 message = Message( 

566 id=f"{uuid}_{digits}", 

567 channel=self.name, 

568 sender_id=body.get("from", ""), 

569 chat_id=f"call:{uuid}", 

570 text=f"[DTMF:{digits}]", 

571 timestamp=datetime.now(), 

572 raw={"uuid": uuid, "digits": digits}, 

573 ) 

574 

575 await self._dispatch_message(message) 

576 return {"ncco": []} 

577 

578 async def _handle_vonage_speech(self, body: Dict[str, Any]) -> Dict[str, Any]: 

579 """Handle Vonage speech input.""" 

580 uuid = body.get("uuid", "") 

581 speech = body.get("speech", {}) 

582 results = speech.get("results", []) 

583 

584 if results: 

585 text = results[0].get("text", "") 

586 confidence = results[0].get("confidence", 0) 

587 

588 message = Message( 

589 id=f"{uuid}_speech", 

590 channel=self.name, 

591 sender_id=body.get("from", ""), 

592 chat_id=f"call:{uuid}", 

593 text=text, 

594 timestamp=datetime.now(), 

595 raw={"uuid": uuid, "confidence": confidence}, 

596 ) 

597 

598 await self._dispatch_message(message) 

599 

600 return {"ncco": []} 

601 

602 def _build_default_greeting_twiml(self) -> str: 

603 """Build default TwiML greeting.""" 

604 voice = self.voice_config.default_voice 

605 language = self.voice_config.default_language 

606 

607 return f"""<?xml version="1.0" encoding="UTF-8"?> 

608<Response> 

609 <Say voice="{voice}" language="{language}"> 

610 Hello, thank you for calling. How can I help you today? 

611 </Say> 

612 <Gather input="speech dtmf" timeout="{self.voice_config.speech_timeout}" action="/voice/gather"> 

613 <Say voice="{voice}">Please speak or press a key.</Say> 

614 </Gather> 

615</Response>""" 

616 

617 def _build_default_greeting_ncco(self) -> List[Dict[str, Any]]: 

618 """Build default Vonage NCCO greeting.""" 

619 return [ 

620 { 

621 "action": "talk", 

622 "text": "Hello, thank you for calling. How can I help you today?", 

623 "voiceName": self.voice_config.default_voice, 

624 }, 

625 { 

626 "action": "input", 

627 "type": ["speech", "dtmf"], 

628 "speech": { 

629 "language": self.voice_config.default_language, 

630 }, 

631 }, 

632 ] 

633 

634 async def send_message( 

635 self, 

636 chat_id: str, 

637 text: str, 

638 reply_to: Optional[str] = None, 

639 media: Optional[List[MediaAttachment]] = None, 

640 buttons: Optional[List[Dict]] = None, 

641 ) -> SendResult: 

642 """Send TTS message to active call.""" 

643 if not chat_id.startswith("call:"): 

644 return SendResult(success=False, error="Invalid chat_id for voice") 

645 

646 call_sid = chat_id.replace("call:", "") 

647 

648 if call_sid not in self._active_calls: 

649 return SendResult(success=False, error="Call not found") 

650 

651 # In practice, TTS is handled via TwiML/NCCO response 

652 # This would be used for out-of-band updates 

653 logger.info(f"TTS message for {call_sid}: {text}") 

654 return SendResult(success=True) 

655 

656 async def edit_message( 

657 self, 

658 chat_id: str, 

659 message_id: str, 

660 text: str, 

661 buttons: Optional[List[Dict]] = None, 

662 ) -> SendResult: 

663 """Not applicable for voice.""" 

664 return SendResult(success=False, error="Not supported for voice") 

665 

666 async def delete_message(self, chat_id: str, message_id: str) -> bool: 

667 """Not applicable for voice.""" 

668 return False 

669 

670 async def send_typing(self, chat_id: str) -> None: 

671 """Not applicable for voice.""" 

672 pass 

673 

674 async def get_chat_info(self, chat_id: str) -> Optional[Dict[str, Any]]: 

675 """Get call information.""" 

676 if not chat_id.startswith("call:"): 

677 return None 

678 

679 call_sid = chat_id.replace("call:", "") 

680 

681 if call_sid in self._active_calls: 

682 call = self._active_calls[call_sid] 

683 return { 

684 "call_sid": call.call_sid, 

685 "from": call.from_number, 

686 "to": call.to_number, 

687 "status": call.status.value, 

688 "direction": call.direction, 

689 "duration": call.duration, 

690 } 

691 

692 return None 

693 

694 # Voice-specific methods 

695 

696 async def make_call( 

697 self, 

698 to_number: str, 

699 twiml_url: Optional[str] = None, 

700 twiml: Optional[str] = None, 

701 ) -> Optional[str]: 

702 """Make an outbound call (Twilio).""" 

703 if self.voice_config.provider != VoiceProvider.TWILIO: 

704 logger.error("make_call only supported for Twilio") 

705 return None 

706 

707 if not self._session: 

708 return None 

709 

710 try: 

711 url = f"https://api.twilio.com/2010-04-01/Accounts/{self.voice_config.twilio_account_sid}/Calls.json" 

712 auth = aiohttp.BasicAuth( 

713 self.voice_config.twilio_account_sid, 

714 self.voice_config.twilio_auth_token, 

715 ) 

716 

717 data = { 

718 "From": self.voice_config.twilio_phone_number, 

719 "To": to_number, 

720 } 

721 

722 if twiml_url: 

723 data["Url"] = twiml_url 

724 elif twiml: 

725 data["Twiml"] = twiml 

726 else: 

727 data["Twiml"] = self._build_default_greeting_twiml() 

728 

729 async with self._session.post(url, auth=auth, data=data) as resp: 

730 if resp.status == 201: 

731 result = await resp.json() 

732 call_sid = result.get("sid") 

733 

734 call = VoiceCall( 

735 call_sid=call_sid, 

736 from_number=self.voice_config.twilio_phone_number, 

737 to_number=to_number, 

738 status=CallStatus.QUEUED, 

739 direction="outbound", 

740 ) 

741 self._active_calls[call_sid] = call 

742 

743 return call_sid 

744 else: 

745 error = await resp.text() 

746 logger.error(f"Failed to make call: {error}") 

747 return None 

748 

749 except Exception as e: 

750 logger.error(f"Make call error: {e}") 

751 return None 

752 

753 async def hangup_call(self, call_sid: str) -> bool: 

754 """Hang up an active call.""" 

755 if self.voice_config.provider == VoiceProvider.TWILIO: 

756 return await self._hangup_twilio(call_sid) 

757 else: 

758 return await self._hangup_vonage(call_sid) 

759 

760 async def _hangup_twilio(self, call_sid: str) -> bool: 

761 """Hang up Twilio call.""" 

762 if not self._session: 

763 return False 

764 

765 try: 

766 url = f"https://api.twilio.com/2010-04-01/Accounts/{self.voice_config.twilio_account_sid}/Calls/{call_sid}.json" 

767 auth = aiohttp.BasicAuth( 

768 self.voice_config.twilio_account_sid, 

769 self.voice_config.twilio_auth_token, 

770 ) 

771 

772 data = {"Status": "completed"} 

773 

774 async with self._session.post(url, auth=auth, data=data) as resp: 

775 if resp.status == 200: 

776 if call_sid in self._active_calls: 

777 self._active_calls[call_sid].status = CallStatus.COMPLETED 

778 return True 

779 

780 except Exception as e: 

781 logger.error(f"Hangup error: {e}") 

782 

783 return False 

784 

785 async def _hangup_vonage(self, uuid: str) -> bool: 

786 """Hang up Vonage call.""" 

787 if not self._session: 

788 return False 

789 

790 try: 

791 url = f"https://api.nexmo.com/v1/calls/{uuid}" 

792 headers = { 

793 "Authorization": f"Bearer {self._get_vonage_jwt()}", 

794 "Content-Type": "application/json", 

795 } 

796 

797 data = {"action": "hangup"} 

798 

799 async with self._session.put(url, headers=headers, json=data) as resp: 

800 return resp.status == 204 

801 

802 except Exception as e: 

803 logger.error(f"Vonage hangup error: {e}") 

804 

805 return False 

806 

807 def _get_vonage_jwt(self) -> str: 

808 """Generate Vonage JWT token.""" 

809 # Simplified - in production use proper JWT library 

810 # This is a placeholder 

811 return "vonage_jwt_token" 

812 

813 def register_dtmf_handler( 

814 self, 

815 key: str, 

816 handler: Callable[[str, str], Any], 

817 ) -> None: 

818 """Register handler for DTMF key press.""" 

819 self._dtmf_handlers[key] = handler 

820 

821 def on_speech(self, handler: Callable[[str, str, float], Any]) -> None: 

822 """Register speech recognition handler.""" 

823 self._speech_handlers.append(handler) 

824 

825 def on_call_start(self, handler: Callable[[VoiceCall], Any]) -> None: 

826 """Register call start handler.""" 

827 self._call_start_handlers.append(handler) 

828 

829 def on_call_end(self, handler: Callable[[VoiceCall], Any]) -> None: 

830 """Register call end handler.""" 

831 self._call_end_handlers.append(handler) 

832 

833 def register_ivr_menu(self, menu: IVRMenu) -> None: 

834 """Register an IVR menu.""" 

835 self._ivr_menus[menu.name] = menu 

836 

837 def build_twiml_say( 

838 self, 

839 text: str, 

840 voice: Optional[str] = None, 

841 language: Optional[str] = None, 

842 ) -> str: 

843 """Build TwiML Say element.""" 

844 voice = voice or self.voice_config.default_voice 

845 language = language or self.voice_config.default_language 

846 return f'<Say voice="{voice}" language="{language}">{text}</Say>' 

847 

848 def build_twiml_gather( 

849 self, 

850 prompt: str, 

851 input_type: str = "dtmf speech", 

852 action_url: str = "/voice/gather", 

853 timeout: Optional[int] = None, 

854 num_digits: Optional[int] = None,