┌─────────────────────────────────────────────────────────────────────────────┐
│                    GOTOCONNECT INTEGRATION CONTEXT                          │
├─────────────────────────────────────────────────────────────────────────────┤
│                                                                             │
│                              PSTN Network                                   │
│                                   │                                         │
│                                   │ Phone calls                             │
│                                   ▼                                         │
│  ┌─────────────────────────────────────────────────────────────────────┐   │
│  │                        GOTOCONNECT CLOUD                             │   │
│  │                                                                      │   │
│  │  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐               │   │
│  │  │     PBX      │  │   WebRTC     │  │    Event     │               │   │
│  │  │   (Jive)     │  │   Gateway    │  │    System    │               │   │
│  │  └──────────────┘  └──────────────┘  └──────────────┘               │   │
│  │         │                 │                 │                        │   │
│  │         │                 │                 │                        │   │
│  │  ┌──────▼─────────────────▼─────────────────▼──────┐                │   │
│  │  │                    API LAYER                     │                │   │
│  │  │                                                  │                │   │
│  │  │  • REST APIs (call control, users, lines)       │                │   │
│  │  │  • WebRTC signaling (SDP exchange)              │                │   │
│  │  │  • WebSocket (real-time events)                 │                │   │
│  │  │                                                  │                │   │
│  │  └──────────────────────┬───────────────────────────┘                │   │
│  │                         │                                            │   │
│  └─────────────────────────┼────────────────────────────────────────────┘   │
│                            │                                                │
│                            │ HTTPS / WSS                                    │
│                            │                                                │
│  ┌─────────────────────────▼────────────────────────────────────────────┐   │
│  │                    VOICE BY AICONNECTED                              │   │
│  │                                                                      │   │
│  │  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐               │   │
│  │  │    OAuth     │  │   WebRTC     │  │    Event     │               │   │
│  │  │   Manager    │  │   Bridge     │  │  Subscriber  │               │   │
│  │  └──────────────┘  └──────────────┘  └──────────────┘               │   │
│  │                                                                      │   │
│  └──────────────────────────────────────────────────────────────────────┘   │
│                                                                             │
└─────────────────────────────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────────────────────────────┐
│                      OAUTH 2.0 CLIENT CREDENTIALS FLOW                      │
├─────────────────────────────────────────────────────────────────────────────┤
│                                                                             │
│  ┌──────────────┐                              ┌──────────────┐             │
│  │   Voice by   │                              │     GoTo     │             │
│  │ aiConnected  │                              │   Auth API   │             │
│  └──────┬───────┘                              └──────┬───────┘             │
│         │                                             │                     │
│         │  1. POST /oauth/token                       │                     │
│         │     grant_type=password                     │                     │
│         │     username={service_user}                 │                     │
│         │     password={service_password}             │                     │
│         │     client_id={client_id}                   │                     │
│         │────────────────────────────────────────────▶│                     │
│         │                                             │                     │
│         │  2. 200 OK                                  │                     │
│         │     {                                       │                     │
│         │       "access_token": "eyJ...",             │                     │
│         │       "token_type": "Bearer",               │                     │
│         │       "expires_in": 3600,                   │                     │
│         │       "refresh_token": "abc...",            │                     │
│         │       "account_key": "123...",              │                     │
│         │       "organizer_key": "456..."             │                     │
│         │     }                                       │                     │
│         │◀────────────────────────────────────────────│                     │
│         │                                             │                     │
│         │  3. API Request                             │                     │
│         │     Authorization: Bearer eyJ...            │                     │
│         │────────────────────────────────────────────▶│                     │
│         │                                             │                     │
│                                                                             │
└─────────────────────────────────────────────────────────────────────────────┘

application:
  name: "Voice by aiConnected"
  description: "AI Voice Agent Platform"
  redirect_uris:
    - "https://api.aiconnected.io/oauth/callback"  # Not used for client credentials
  
  scopes:
    # Required scopes
    - "webrtc.v1.write"                    # WebRTC call management
    - "call-events.v1.notifications.manage" # Event subscriptions
    - "call-control.v1.calls.write"        # Call control operations
    - "users.v1.lines.read"                # Read line information
    - "users.v1.users.read"                # Read user information
    
  grant_types:
    - "password"           # For service user authentication
    - "refresh_token"      # For token refresh

service_user:
  username: "aiconnected-service@yourdomain.com"
  password: "${GOTO_SERVICE_PASSWORD}"  # Stored in secrets manager

# integrations/gotoconnect/auth.py

import httpx
from datetime import datetime, timedelta
from typing import Optional
import asyncio

class GoToAuthManager:
    """Manages OAuth tokens for GoToConnect API access."""
    
    AUTH_URL = "https://authentication.logmeininc.com/oauth/token"
    
    def __init__(
        self,
        client_id: str,
        username: str,
        password: str
    ):
        self.client_id = client_id
        self.username = username
        self.password = password
        
        self._access_token: Optional[str] = None
        self._refresh_token: Optional[str] = None
        self._expires_at: Optional[datetime] = None
        self._account_key: Optional[str] = None
        self._organizer_key: Optional[str] = None
        self._lock = asyncio.Lock()
    
    async def get_access_token(self) -> str:
        """Get a valid access token, refreshing if necessary."""
        async with self._lock:
            if self._is_token_valid():
                return self._access_token
            
            if self._refresh_token:
                try:
                    await self._refresh_access_token()
                    return self._access_token
                except Exception:
                    pass  # Fall through to full auth
            
            await self._authenticate()
            return self._access_token
    
    def _is_token_valid(self) -> bool:
        """Check if current token is valid with buffer."""
        if not self._access_token or not self._expires_at:
            return False
        # Refresh 5 minutes before expiry
        return datetime.utcnow() < (self._expires_at - timedelta(minutes=5))
    
    async def _authenticate(self) -> None:
        """Perform full authentication with username/password."""
        async with httpx.AsyncClient() as client:
            response = await client.post(
                self.AUTH_URL,
                data={
                    "grant_type": "password",
                    "username": self.username,
                    "password": self.password,
                    "client_id": self.client_id,
                },
                headers={
                    "Content-Type": "application/x-www-form-urlencoded",
                    "Accept": "application/json",
                }
            )
            response.raise_for_status()
            self._process_token_response(response.json())
    
    async def _refresh_access_token(self) -> None:
        """Refresh the access token using refresh token."""
        async with httpx.AsyncClient() as client:
            response = await client.post(
                self.AUTH_URL,
                data={
                    "grant_type": "refresh_token",
                    "refresh_token": self._refresh_token,
                    "client_id": self.client_id,
                },
                headers={
                    "Content-Type": "application/x-www-form-urlencoded",
                    "Accept": "application/json",
                }
            )
            response.raise_for_status()
            self._process_token_response(response.json())
    
    def _process_token_response(self, data: dict) -> None:
        """Process and store token response."""
        self._access_token = data["access_token"]
        self._refresh_token = data.get("refresh_token")
        self._account_key = data.get("account_key")
        self._organizer_key = data.get("organizer_key")
        
        expires_in = data.get("expires_in", 3600)
        self._expires_at = datetime.utcnow() + timedelta(seconds=expires_in)
    
    @property
    def account_key(self) -> Optional[str]:
        """Get the account key from authentication."""
        return self._account_key
    
    @property
    def organizer_key(self) -> Optional[str]:
        """Get the organizer key (user key) from authentication."""
        return self._organizer_key

{
  "access_token": "eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9...",
  "token_type": "Bearer",
  "expires_in": 3600,
  "refresh_token": "abc123def456...",
  "scope": "webrtc.v1.write call-events.v1.notifications.manage ...",
  "account_key": "1234567890",
  "organizer_key": "9876543210",
  "principal": "aiconnected-service@yourdomain.com"
}

class GoToAuthError(Exception):
    """Base exception for GoTo authentication errors."""
    pass

class InvalidCredentialsError(GoToAuthError):
    """Raised when credentials are invalid."""
    pass

class TokenExpiredError(GoToAuthError):
    """Raised when token has expired and refresh failed."""
    pass

class InsufficientScopesError(GoToAuthError):
    """Raised when token lacks required scopes."""
    pass

# Error handling in auth manager
async def _authenticate(self) -> None:
    try:
        async with httpx.AsyncClient() as client:
            response = await client.post(self.AUTH_URL, ...)
            
            if response.status_code == 400:
                error_data = response.json()
                if error_data.get("error") == "invalid_grant":
                    raise InvalidCredentialsError(
                        "Invalid username or password"
                    )
                raise GoToAuthError(f"Auth failed: {error_data}")
            
            if response.status_code == 401:
                raise InvalidCredentialsError("Authentication failed")
            
            response.raise_for_status()
            self._process_token_response(response.json())
            
    except httpx.HTTPError as e:
        raise GoToAuthError(f"HTTP error during authentication: {e}")

┌─────────────────────────────────────────────────────────────────────────────┐
│                        WEBRTC CALL FLOW                                     │
├─────────────────────────────────────────────────────────────────────────────┤
│                                                                             │
│   PSTN Caller          GoToConnect           WebRTC Bridge        LiveKit   │
│        │                    │                      │                 │      │
│        │  1. Inbound call   │                      │                 │      │
│        │───────────────────▶│                      │                 │      │
│        │                    │                      │                 │      │
│        │                    │  2. Event: call.ringing               │      │
│        │                    │─────────────────────▶│                 │      │
│        │                    │                      │                 │      │
│        │                    │  3. POST /calls/{id}/answer            │      │
│        │                    │◀─────────────────────│                 │      │
│        │                    │                      │                 │      │
│        │                    │  4. SDP Offer        │                 │      │
│        │                    │─────────────────────▶│                 │      │
│        │                    │                      │                 │      │
│        │                    │                      │  5. Create room │      │
│        │                    │                      │────────────────▶│      │
│        │                    │                      │                 │      │
│        │                    │  6. SDP Answer       │                 │      │
│        │                    │◀─────────────────────│                 │      │
│        │                    │                      │                 │      │
│        │                    │  7. ICE Candidates   │                 │      │
│        │                    │◀────────────────────▶│                 │      │
│        │                    │                      │                 │      │
│        │  8. Audio stream   │  9. Audio stream     │  10. Audio      │      │
│        │◀──────────────────▶│◀────────────────────▶│◀───────────────▶│      │
│        │                    │                      │                 │      │
│                                                                             │
└─────────────────────────────────────────────────────────────────────────────┘

# integrations/gotoconnect/config.py

WEB_CALLS_BASE_URL = "https://api.goto.com/web-calls/v1"

# Endpoints
ENDPOINTS = {
    "calls": f"{WEB_CALLS_BASE_URL}/calls",
    "call": f"{WEB_CALLS_BASE_URL}/calls/{{call_id}}",
    "answer": f"{WEB_CALLS_BASE_URL}/calls/{{call_id}}/answer",
    "hangup": f"{WEB_CALLS_BASE_URL}/calls/{{call_id}}/hangup",
    "hold": f"{WEB_CALLS_BASE_URL}/calls/{{call_id}}/hold",
    "resume": f"{WEB_CALLS_BASE_URL}/calls/{{call_id}}/resume",
    "mute": f"{WEB_CALLS_BASE_URL}/calls/{{call_id}}/mute",
    "unmute": f"{WEB_CALLS_BASE_URL}/calls/{{call_id}}/unmute",
    "dtmf": f"{WEB_CALLS_BASE_URL}/calls/{{call_id}}/dtmf",
    "blind_transfer": f"{WEB_CALLS_BASE_URL}/calls/{{call_id}}/blind-transfer",
    "warm_transfer": f"{WEB_CALLS_BASE_URL}/calls/{{call_id}}/warm-transfer",
    "merge": f"{WEB_CALLS_BASE_URL}/calls/{{call_id}}/merge",
    "ice_candidates": f"{WEB_CALLS_BASE_URL}/calls/{{call_id}}/ice-candidates",
}

async def initiate_call(
    self,
    dial_string: str,
    caller_id: str,
    line_id: str
) -> dict:
    """
    Initiate an outbound call.
    
    Args:
        dial_string: Number to call (e.g., "tel:+15551234567")
        caller_id: Caller ID to display
        line_id: Line/extension to call from
    
    Returns:
        Call object with ID and initial SDP offer
    """
    token = await self.auth_manager.get_access_token()
    
    async with httpx.AsyncClient() as client:
        response = await client.post(
            ENDPOINTS["calls"],
            json={
                "dialString": dial_string,
                "callerId": caller_id,
                "lineId": line_id,
            },
            headers={
                "Authorization": f"Bearer {token}",
                "Content-Type": "application/json",
                "Accept": "application/json",
            }
        )
        response.raise_for_status()
        return response.json()

POST /web-calls/v1/calls
{
  "dialString": "tel:+15551234567",
  "callerId": "+15559876543",
  "lineId": "line_abc123"
}

{
  "callId": "call_xyz789",
  "state": "dialing",
  "direction": "outbound",
  "from": "+15559876543",
  "to": "+15551234567",
  "lineId": "line_abc123",
  "sdpOffer": "v=0\r\no=- 123456789 2 IN IP4 127.0.0.1\r\n...",
  "createdAt": "2026-01-16T10:30:00Z"
}

async def answer_call(
    self,
    call_id: str,
    sdp_answer: str
) -> dict:
    """
    Answer an inbound call with SDP answer.
    
    Args:
        call_id: The call to answer
        sdp_answer: Our SDP answer
    
    Returns:
        Updated call object
    """
    token = await self.auth_manager.get_access_token()
    
    endpoint = ENDPOINTS["answer"].format(call_id=call_id)
    
    async with httpx.AsyncClient() as client:
        response = await client.post(
            endpoint,
            json={
                "sdpAnswer": sdp_answer,
            },
            headers={
                "Authorization": f"Bearer {token}",
                "Content-Type": "application/json",
                "Accept": "application/json",
            }
        )
        response.raise_for_status()
        return response.json()

POST /web-calls/v1/calls/{call_id}/answer
{
  "sdpAnswer": "v=0\r\no=- 987654321 2 IN IP4 127.0.0.1\r\n..."
}

{
  "callId": "call_xyz789",
  "state": "connected",
  "direction": "inbound",
  "from": "+15551234567",
  "to": "+15559876543",
  "lineId": "line_abc123",
  "answeredAt": "2026-01-16T10:30:05Z"
}

v=0
o=- 1234567890 2 IN IP4 0.0.0.0
s=-
t=0 0
a=group:BUNDLE audio
a=msid-semantic: WMS
m=audio 9 UDP/TLS/RTP/SAVPF 111 103 104 9 0 8 106 105 13 110 112 113 126
c=IN IP4 0.0.0.0
a=rtcp:9 IN IP4 0.0.0.0
a=ice-ufrag:abcd
a=ice-pwd:efghijklmnopqrstuvwxyz
a=ice-options:trickle
a=fingerprint:sha-256 AA:BB:CC:DD:EE:FF:...
a=setup:actpass
a=mid:audio
a=extmap:1 urn:ietf:params:rtp-hdrext:ssrc-audio-level
a=sendrecv
a=rtcp-mux
a=rtpmap:111 opus/48000/2
a=rtcp-fb:111 transport-cc
a=fmtp:111 minptime=10;useinbandfec=1
a=rtpmap:103 ISAC/16000
a=rtpmap:104 ISAC/32000
a=rtpmap:9 G722/8000
a=rtpmap:0 PCMU/8000
a=rtpmap:8 PCMA/8000

from aiortc import RTCPeerConnection, RTCSessionDescription
from aiortc.contrib.media import MediaPlayer, MediaRecorder

async def create_sdp_answer(sdp_offer: str) -> tuple[RTCPeerConnection, str]:
    """
    Create an SDP answer for a GoToConnect offer.
    
    Args:
        sdp_offer: The SDP offer from GoToConnect
    
    Returns:
        Tuple of (peer_connection, sdp_answer)
    """
    # Create peer connection with our configuration
    pc = RTCPeerConnection(configuration={
        "iceServers": [
            {"urls": "stun:stun.l.google.com:19302"},
        ]
    })
    
    # Set the remote description (offer from GoToConnect)
    offer = RTCSessionDescription(sdp=sdp_offer, type="offer")
    await pc.setRemoteDescription(offer)
    
    # Create and set local description (our answer)
    answer = await pc.createAnswer()
    await pc.setLocalDescription(answer)
    
    return pc, pc.localDescription.sdp

CODEC_PREFERENCES = [
    {
        "name": "opus",
        "clockRate": 48000,
        "channels": 2,
        "priority": 1,
        "params": {
            "minptime": "10",
            "useinbandfec": "1",
        }
    },
    {
        "name": "PCMU",  # G.711 μ-law (fallback)
        "clockRate": 8000,
        "channels": 1,
        "priority": 2,
    },
    {
        "name": "PCMA",  # G.711 A-law (fallback)
        "clockRate": 8000,
        "channels": 1,
        "priority": 3,
    },
]

async def send_ice_candidate(
    self,
    call_id: str,
    candidate: dict
) -> None:
    """
    Send an ICE candidate to GoToConnect.
    
    Args:
        call_id: The call ID
        candidate: ICE candidate object
    """
    token = await self.auth_manager.get_access_token()
    endpoint = ENDPOINTS["ice_candidates"].format(call_id=call_id)
    
    async with httpx.AsyncClient() as client:
        response = await client.post(
            endpoint,
            json={
                "candidate": candidate["candidate"],
                "sdpMid": candidate["sdpMid"],
                "sdpMLineIndex": candidate["sdpMLineIndex"],
            },
            headers={
                "Authorization": f"Bearer {token}",
                "Content-Type": "application/json",
            }
        )
        response.raise_for_status()

POST /web-calls/v1/calls/{call_id}/ice-candidates
{
  "candidate": "candidate:1 1 UDP 2130706431 192.168.1.100 54321 typ host",
  "sdpMid": "audio",
  "sdpMLineIndex": 0
}

async def handle_ice_candidate_event(
    self,
    event: dict,
    peer_connection: RTCPeerConnection
) -> None:
    """
    Handle an ICE candidate event from GoToConnect.
    
    Args:
        event: The ICE candidate event
        peer_connection: Our RTCPeerConnection
    """
    candidate_data = event.get("candidate")
    
    if candidate_data:
        from aiortc import RTCIceCandidate
        
        candidate = RTCIceCandidate(
            candidate=candidate_data["candidate"],
            sdpMid=candidate_data["sdpMid"],
            sdpMLineIndex=candidate_data["sdpMLineIndex"],
        )
        
        await peer_connection.addIceCandidate(candidate)

from aiortc import MediaStreamTrack

class GoToAudioReceiver:
    """Receives audio from GoToConnect and forwards to LiveKit."""
    
    def __init__(
        self,
        peer_connection: RTCPeerConnection,
        livekit_room: Room
    ):
        self.pc = peer_connection
        self.livekit_room = livekit_room
        
        # Set up track handler
        self.pc.on("track", self._on_track)
    
    async def _on_track(self, track: MediaStreamTrack) -> None:
        """Handle incoming audio track from GoToConnect."""
        if track.kind != "audio":
            return
        
        logger.info(f"Received audio track from GoToConnect: {track.id}")
        
        # Create a track forwarder to LiveKit
        forwarder = AudioTrackForwarder(track, self.livekit_room)
        await forwarder.start()
    

class AudioTrackForwarder:
    """Forwards audio frames from GoToConnect to LiveKit."""
    
    def __init__(
        self,
        source_track: MediaStreamTrack,
        livekit_room: Room
    ):
        self.source_track = source_track
        self.livekit_room = livekit_room
        self._running = False
    
    async def start(self) -> None:
        """Start forwarding audio frames."""
        self._running = True
        
        # Create LiveKit audio source
        audio_source = rtc.AudioSource(
            sample_rate=48000,
            num_channels=1
        )
        
        # Publish track to LiveKit
        track = rtc.LocalAudioTrack.create_audio_track(
            "caller-audio",
            audio_source
        )
        
        await self.livekit_room.local_participant.publish_track(track)
        
        # Forward frames
        while self._running:
            try:
                frame = await self.source_track.recv()
                
                # Convert frame format if needed
                audio_frame = self._convert_frame(frame)
                
                # Push to LiveKit
                await audio_source.capture_frame(audio_frame)
                
            except MediaStreamError:
                logger.info("Audio track ended")
                break
    
    def _convert_frame(self, frame) -> rtc.AudioFrame:
        """Convert aiortc frame to LiveKit frame."""
        # Frame conversion logic
        return rtc.AudioFrame(
            data=frame.to_ndarray().tobytes(),
            sample_rate=frame.sample_rate,
            num_channels=frame.layout.channels,
            samples_per_channel=frame.samples,
        )
    
    async def stop(self) -> None:
        """Stop forwarding."""
        self._running = False

class GoToAudioSender:
    """Sends audio from LiveKit to GoToConnect."""
    
    def __init__(
        self,
        peer_connection: RTCPeerConnection,
        livekit_room: Room
    ):
        self.pc = peer_connection
        self.livekit_room = livekit_room
        self._audio_track = None
    
    async def start(self) -> None:
        """Start sending audio to GoToConnect."""
        # Create an audio track for GoToConnect
        self._audio_track = AudioStreamTrack()
        
        # Add track to peer connection
        self.pc.addTrack(self._audio_track)
        
        # Subscribe to AI agent's audio in LiveKit
        @self.livekit_room.on("track_subscribed")
        async def on_track_subscribed(
            track: rtc.Track,
            publication: rtc.TrackPublication,
            participant: rtc.RemoteParticipant
        ):
            if track.kind == rtc.TrackKind.KIND_AUDIO:
                # Forward AI audio to GoToConnect
                async for frame in track:
                    await self._audio_track.send_frame(frame)


class AudioStreamTrack(MediaStreamTrack):
    """Custom audio track for sending to GoToConnect."""
    
    kind = "audio"
    
    def __init__(self):
        super().__init__()
        self._queue = asyncio.Queue()
    
    async def send_frame(self, frame: rtc.AudioFrame) -> None:
        """Queue a frame for sending."""
        await self._queue.put(frame)
    
    async def recv(self):
        """Receive next frame to send."""
        frame = await self._queue.get()
        return self._convert_to_aiortc_frame(frame)
    
    def _convert_to_aiortc_frame(self, frame: rtc.AudioFrame):
        """Convert LiveKit frame to aiortc frame."""
        from av import AudioFrame as AVAudioFrame
        
        av_frame = AVAudioFrame(
            format="s16",
            layout="mono",
            samples=frame.samples_per_channel
        )
        av_frame.sample_rate = frame.sample_rate
        av_frame.planes[0].update(frame.data)
        
        return av_frame

async def hold_call(self, call_id: str) -> dict:
    """
    Place a call on hold.
    
    Args:
        call_id: The call to hold
    
    Returns:
        Updated call object
    """
    token = await self.auth_manager.get_access_token()
    endpoint = ENDPOINTS["hold"].format(call_id=call_id)
    
    async with httpx.AsyncClient() as client:
        response = await client.put(
            endpoint,
            headers={
                "Authorization": f"Bearer {token}",
                "Accept": "application/json",
            }
        )
        response.raise_for_status()
        return response.json()

PUT /web-calls/v1/calls/{call_id}/hold

{
  "callId": "call_xyz789",
  "state": "held",
  "holdStartedAt": "2026-01-16T10:35:00Z"
}

async def resume_call(self, call_id: str) -> dict:
    """
    Resume a held call.
    
    Args:
        call_id: The call to resume
    
    Returns:
        Updated call object
    """
    token = await self.auth_manager.get_access_token()
    endpoint = ENDPOINTS["resume"].format(call_id=call_id)
    
    async with httpx.AsyncClient() as client:
        response = await client.put(
            endpoint,
            headers={
                "Authorization": f"Bearer {token}",
                "Accept": "application/json",
            }
        )
        response.raise_for_status()
        return response.json()

PUT /web-calls/v1/calls/{call_id}/resume

async def mute_call(self, call_id: str) -> dict:
    """Mute our audio (caller won't hear us)."""
    token = await self.auth_manager.get_access_token()
    endpoint = ENDPOINTS["mute"].format(call_id=call_id)
    
    async with httpx.AsyncClient() as client:
        response = await client.put(
            endpoint,
            headers={"Authorization": f"Bearer {token}"}
        )
        response.raise_for_status()
        return response.json()

async def unmute_call(self, call_id: str) -> dict:
    """Unmute our audio."""
    token = await self.auth_manager.get_access_token()
    endpoint = ENDPOINTS["unmute"].format(call_id=call_id)
    
    async with httpx.AsyncClient() as client:
        response = await client.put(
            endpoint,
            headers={"Authorization": f"Bearer {token}"}
        )
        response.raise_for_status()
        return response.json()

async def send_dtmf(
    self,
    call_id: str,
    digits: str,
    duration_ms: int = 100,
    gap_ms: int = 50
) -> dict:
    """
    Send DTMF tones.
    
    Args:
        call_id: The call ID
        digits: Digits to send (0-9, *, #)
        duration_ms: Duration of each tone
        gap_ms: Gap between tones
    
    Returns:
        Confirmation response
    """
    token = await self.auth_manager.get_access_token()
    endpoint = ENDPOINTS["dtmf"].format(call_id=call_id)
    
    async with httpx.AsyncClient() as client:
        response = await client.post(
            endpoint,
            json={
                "digits": digits,
                "durationMs": duration_ms,
                "gapMs": gap_ms,
            },
            headers={
                "Authorization": f"Bearer {token}",
                "Content-Type": "application/json",
            }
        )
        response.raise_for_status()
        return response.json()

POST /web-calls/v1/calls/{call_id}/dtmf
{
  "digits": "1234#",
  "durationMs": 100,
  "gapMs": 50
}

async def hangup_call(
    self,
    call_id: str,
    reason: str = "normal"
) -> dict:
    """
    End a call.
    
    Args:
        call_id: The call to end
        reason: Reason for ending (normal, busy, rejected)
    
    Returns:
        Final call object
    """
    token = await self.auth_manager.get_access_token()
    endpoint = ENDPOINTS["hangup"].format(call_id=call_id)
    
    async with httpx.AsyncClient() as client:
        response = await client.post(
            endpoint,
            json={"reason": reason},
            headers={
                "Authorization": f"Bearer {token}",
                "Content-Type": "application/json",
            }
        )
        response.raise_for_status()
        return response.json()

async def blind_transfer(
    self,
    call_id: str,
    dial_string: str
) -> dict:
    """
    Perform a blind (cold) transfer.
    
    The caller is immediately connected to the transfer target.
    Our connection is terminated.
    
    Args:
        call_id: The call to transfer
        dial_string: Target (e.g., "ext:1001" or "tel:+15551234567")
    
    Returns:
        Transfer result
    """
    token = await self.auth_manager.get_access_token()
    endpoint = ENDPOINTS["blind_transfer"].format(call_id=call_id)
    
    async with httpx.AsyncClient() as client:
        response = await client.post(
            endpoint,
            json={
                "dialString": dial_string,
            },
            headers={
                "Authorization": f"Bearer {token}",
                "Content-Type": "application/json",
            }
        )
        response.raise_for_status()
        return response.json()

POST /web-calls/v1/calls/{call_id}/blind-transfer
{
  "dialString": "ext:1001"
}

{
  "callId": "call_xyz789",
  "state": "transferred",
  "transferType": "blind",
  "transferTarget": "ext:1001",
  "transferredAt": "2026-01-16T10:40:00Z"
}

**Dial String Formats:** | Format | Example | Description | |--------|---------|-------------| | `ext:{extension}` | `ext:1001` | Internal extension | | `tel:{number}` | `tel:+15551234567` | External phone number | | `sip:{uri}` | `sip:user@domain.com` | SIP URI | | `voicemail:{extension}` | `voicemail:1001` | Extension's voicemail |

#### 5.2.2 Warm Transfer (Attended Transfer) {#5.2.2-warm-transfer-(attended-transfer)}

┌─────────────────────────────────────────────────────────────────────────────┐
│                        WARM TRANSFER FLOW                                   │
├─────────────────────────────────────────────────────────────────────────────┤
│                                                                             │
│   Caller              AI Agent            Human Agent          GoToConnect  │
│     │                    │                     │                    │       │
│     │  1. Talking        │                     │                    │       │
│     │◀──────────────────▶│                     │                    │       │
│     │                    │                     │                    │       │
│     │                    │  2. Hold caller     │                    │       │
│     │  (hold music)      │────────────────────────────────────────▶│       │
│     │◀─ ─ ─ ─ ─ ─ ─ ─ ─ ─│                     │                    │       │
│     │                    │                     │                    │       │
│     │                    │  3. Call agent      │                    │       │
│     │                    │────────────────────────────────────────▶│       │
│     │                    │                     │◀───────────────────│       │
│     │                    │                     │                    │       │
│     │                    │  4. Brief agent     │                    │       │
│     │                    │◀───────────────────▶│                    │       │
│     │                    │  "Customer John,    │                    │       │
│     │                    │   billing issue"    │                    │       │
│     │                    │                     │                    │       │
│     │                    │  5. Complete transfer                    │       │
│     │                    │────────────────────────────────────────▶│       │
│     │                    │                     │                    │       │
│     │  6. Connected to agent                   │                    │       │
│     │◀────────────────────────────────────────▶│                    │       │
│     │                    │                     │                    │       │
│     │                    │  7. AI disconnected │                    │       │
│     │                    │◀────────────────────────────────────────│       │
│     │                    │                     │                    │       │
│                                                                             │
└─────────────────────────────────────────────────────────────────────────────┘

async def warm_transfer(
    self,
    call_id: str,
    dial_string: str,
    announcement: str = None
) -> dict:
    """
    Perform a warm (attended) transfer.
    
    This is a multi-step process:
    1. Place caller on hold
    2. Call the transfer target
    3. Brief the target (optional)
    4. Complete the transfer
    
    Args:
        call_id: The original call ID
        dial_string: Target to transfer to
        announcement: Message to speak to target before transfer
    
    Returns:
        Transfer result
    """
    # Step 1: Hold the original caller
    await self.hold_call(call_id)
    
    # Step 2: Initiate call to transfer target
    consult_call = await self.initiate_call(
        dial_string=dial_string,
        caller_id=self._get_caller_id(call_id),
        line_id=self._get_line_id(call_id),
    )
    
    consult_call_id = consult_call["callId"]
    
    # Step 3: Wait for target to answer
    # (This would be handled via events in practice)
    
    # Step 4: Complete the transfer
    token = await self.auth_manager.get_access_token()
    endpoint = ENDPOINTS["warm_transfer"].format(call_id=call_id)
    
    async with httpx.AsyncClient() as client:
        response = await client.post(
            endpoint,
            json={
                "referId": consult_call_id,
            },
            headers={
                "Authorization": f"Bearer {token}",
                "Content-Type": "application/json",
            }
        )
        response.raise_for_status()
        return response.json()

POST /web-calls/v1/calls/{call_id}/warm-transfer
{
  "referId": "call_consult_abc"
}

async def merge_calls(
    self,
    call_id: str,
    other_call_id: str
) -> dict:
    """
    Merge two calls into a conference.
    
    Both parties will be connected together with us.
    
    Args:
        call_id: First call ID
        other_call_id: Second call ID to merge
    
    Returns:
        Merged call result
    """
    token = await self.auth_manager.get_access_token()
    endpoint = ENDPOINTS["merge"].format(call_id=call_id)
    
    async with httpx.AsyncClient() as client:
        response = await client.post(
            endpoint,
            json={
                "referId": other_call_id,
            },
            headers={
                "Authorization": f"Bearer {token}",
                "Content-Type": "application/json",
            }
        )
        response.raise_for_status()
        return response.json()

POST /web-calls/v1/calls/{call_id}/merge
{
  "referId": "call_supervisor_xyz"
}

┌─────────────────────────────────────────────────────────────────────────────┐
│                     CALL CONTROL STATE MACHINE                              │
├─────────────────────────────────────────────────────────────────────────────┤
│                                                                             │
│                                                                             │
│                    ┌─────────────────────────────────────────┐             │
│                    │                                         │             │
│                    │              ┌─────────┐                │             │
│                    │              │ INITIAL │                │             │
│                    │              └────┬────┘                │             │
│                    │                   │                     │             │
│                    │    ┌──────────────┼──────────────┐      │             │
│                    │    │              │              │      │             │
│                    │    ▼              ▼              ▼      │             │
│                    │ ┌──────┐    ┌─────────┐    ┌────────┐   │             │
│                    │ │DIALING    │ RINGING │    │OFFERING│   │             │
│                    │ │(outbound) │(inbound)│    │(inbound│   │             │
│                    │ └────┬─┘    └────┬────┘    │ WebRTC)│   │             │
│                    │      │           │         └───┬────┘   │             │
│                    │      │           │             │        │             │
│                    │      └───────────┼─────────────┘        │             │
│                    │                  │                      │             │
│                    │                  │ answer               │             │
│                    │                  ▼                      │             │
│                    │            ┌───────────┐                │             │
│       resume       │     ┌─────▶│ CONNECTED │◀─────┐        │             │
│          │         │     │      └─────┬─────┘      │        │             │
│          │         │     │            │            │        │             │
│          │         │     │    ┌───────┴───────┐    │        │             │
│          │         │     │    │               │    │        │             │
│          │         │     │    ▼               ▼    │        │             │
│          │         │   ┌──────────┐     ┌──────────┐│        │             │
│          └─────────┼──▶│   HELD   │     │  MUTED   ││        │             │
│                    │   └──────────┘     └──────────┘│        │             │
│                    │                                │        │             │
│                    │                                │ unmute │             │
│                    │                                │        │             │
│                    │                                         │             │
│                    │    From any state:                      │             │
│                    │                                         │             │
│                    │         hangup      transfer            │             │
│                    │            │            │                │             │
│                    │            ▼            ▼                │             │
│                    │      ┌─────────┐  ┌────────────┐        │             │
│                    │      │  ENDED  │  │TRANSFERRED │        │             │
│                    │      └─────────┘  └────────────┘        │             │
│                    │                                         │             │
│                    └─────────────────────────────────────────┘             │
│                                                                             │
└─────────────────────────────────────────────────────────────────────────────┘

# integrations/gotoconnect/call_control.py

from dataclasses import dataclass
from enum import Enum
from typing import Optional
import httpx

class CallState(Enum):
    INITIAL = "initial"
    DIALING = "dialing"
    RINGING = "ringing"
    OFFERING = "offering"
    CONNECTED = "connected"
    HELD = "held"
    MUTED = "muted"
    TRANSFERRED = "transferred"
    ENDED = "ended"

@dataclass
class Call:
    """Represents a GoToConnect call."""
    call_id: str
    state: CallState
    direction: str
    from_number: str
    to_number: str
    line_id: str
    created_at: str
    answered_at: Optional[str] = None
    ended_at: Optional[str] = None
    sdp_offer: Optional[str] = None
    sdp_answer: Optional[str] = None

class GoToCallControlClient:
    """
    Client for GoToConnect Call Control operations.
    
    Provides methods for all call control operations including
    hold, transfer, merge, and DTMF.
    """
    
    BASE_URL = "https://api.goto.com/web-calls/v1"
    
    def __init__(self, auth_manager: GoToAuthManager):
        self.auth_manager = auth_manager
    
    async def _request(
        self,
        method: str,
        endpoint: str,
        json: dict = None
    ) -> dict:
        """Make an authenticated request to GoToConnect."""
        token = await self.auth_manager.get_access_token()
        
        async with httpx.AsyncClient() as client:
            response = await client.request(
                method=method,
                url=f"{self.BASE_URL}{endpoint}",
                json=json,
                headers={
                    "Authorization": f"Bearer {token}",
                    "Content-Type": "application/json",
                    "Accept": "application/json",
                },
                timeout=30.0,
            )
            
            if response.status_code >= 400:
                await self._handle_error(response)
            
            if response.status_code == 204:
                return {}
            
            return response.json()
    
    async def _handle_error(self, response: httpx.Response) -> None:
        """Handle API error responses."""
        try:
            error_data = response.json()
        except Exception:
            error_data = {"message": response.text}
        
        status = response.status_code
        
        if status == 401:
            raise GoToAuthError("Authentication failed")
        elif status == 403:
            raise GoToPermissionError(
                f"Permission denied: {error_data.get('message')}"
            )
        elif status == 404:
            raise GoToNotFoundError(
                f"Resource not found: {error_data.get('message')}"
            )
        elif status == 409:
            raise GoToConflictError(
                f"Conflict: {error_data.get('message')}"
            )
        elif status == 429:
            raise GoToRateLimitError(
                f"Rate limit exceeded: {error_data.get('message')}"
            )
        else:
            raise GoToAPIError(
                f"API error {status}: {error_data.get('message')}"
            )
    
    # Call initiation
    async def initiate_call(
        self,
        dial_string: str,
        caller_id: str,
        line_id: str
    ) -> Call:
        """Initiate an outbound call."""
        data = await self._request(
            "POST",
            "/calls",
            json={
                "dialString": dial_string,
                "callerId": caller_id,
                "lineId": line_id,
            }
        )
        return self._parse_call(data)
    
    async def answer_call(
        self,
        call_id: str,
        sdp_answer: str
    ) -> Call:
        """Answer an inbound call."""
        data = await self._request(
            "POST",
            f"/calls/{call_id}/answer",
            json={"sdpAnswer": sdp_answer}
        )
        return self._parse_call(data)
    
    # Call control
    async def hangup(self, call_id: str) -> Call:
        """End a call."""
        data = await self._request("POST", f"/calls/{call_id}/hangup")
        return self._parse_call(data)
    
    async def hold(self, call_id: str) -> Call:
        """Place call on hold."""
        data = await self._request("PUT", f"/calls/{call_id}/hold")
        return self._parse_call(data)
    
    async def resume(self, call_id: str) -> Call:
        """Resume held call."""
        data = await self._request("PUT", f"/calls/{call_id}/resume")
        return self._parse_call(data)
    
    async def mute(self, call_id: str) -> Call:
        """Mute our audio."""
        data = await self._request("PUT", f"/calls/{call_id}/mute")
        return self._parse_call(data)
    
    async def unmute(self, call_id: str) -> Call:
        """Unmute our audio."""
        data = await self._request("PUT", f"/calls/{call_id}/unmute")
        return self._parse_call(data)
    
    async def send_dtmf(
        self,
        call_id: str,
        digits: str,
        duration_ms: int = 100,
        gap_ms: int = 50
    ) -> dict:
        """Send DTMF tones."""
        return await self._request(
            "POST",
            f"/calls/{call_id}/dtmf",
            json={
                "digits": digits,
                "durationMs": duration_ms,
                "gapMs": gap_ms,
            }
        )
    
    # Transfers
    async def blind_transfer(
        self,
        call_id: str,
        dial_string: str
    ) -> Call:
        """Perform blind transfer."""
        data = await self._request(
            "POST",
            f"/calls/{call_id}/blind-transfer",
            json={"dialString": dial_string}
        )
        return self._parse_call(data)
    
    async def warm_transfer(
        self,
        call_id: str,
        refer_id: str
    ) -> Call:
        """Complete warm transfer."""
        data = await self._request(
            "POST",
            f"/calls/{call_id}/warm-transfer",
            json={"referId": refer_id}
        )
        return self._parse_call(data)
    
    async def merge(
        self,
        call_id: str,
        refer_id: str
    ) -> Call:
        """Merge calls into conference."""
        data = await self._request(
            "POST",
            f"/calls/{call_id}/merge",
            json={"referId": refer_id}
        )
        return self._parse_call(data)
    
    # ICE candidates
    async def send_ice_candidate(
        self,
        call_id: str,
        candidate: str,
        sdp_mid: str,
        sdp_m_line_index: int
    ) -> None:
        """Send ICE candidate."""
        await self._request(
            "POST",
            f"/calls/{call_id}/ice-candidates",
            json={
                "candidate": candidate,
                "sdpMid": sdp_mid,
                "sdpMLineIndex": sdp_m_line_index,
            }
        )
    
    # Helpers
    def _parse_call(self, data: dict) -> Call:
        """Parse API response into Call object."""
        return Call(
            call_id=data["callId"],
            state=CallState(data["state"]),
            direction=data.get("direction", "unknown"),
            from_number=data.get("from", ""),
            to_number=data.get("to", ""),
            line_id=data.get("lineId", ""),
            created_at=data.get("createdAt", ""),
            answered_at=data.get("answeredAt"),
            ended_at=data.get("endedAt"),
            sdp_offer=data.get("sdpOffer"),
            sdp_answer=data.get("sdpAnswer"),
        )

┌─────────────────────────────────────────────────────────────────────────────┐
│                         EVENT SUBSCRIPTION FLOW                             │
├─────────────────────────────────────────────────────────────────────────────┤
│                                                                             │
│  ┌──────────────┐                              ┌──────────────┐             │
│  │   Voice by   │                              │  GoToConnect │             │
│  │ aiConnected  │                              │              │             │
│  └──────┬───────┘                              └──────┬───────┘             │
│         │                                             │                     │
│         │  1. POST /notifications/channels            │                     │
│         │     Create channel for events               │                     │
│         │────────────────────────────────────────────▶│                     │
│         │                                             │                     │
│         │  2. Response: channel_id, websocket_url     │                     │
│         │◀────────────────────────────────────────────│                     │
│         │                                             │                     │
│         │  3. POST /notifications/subscriptions       │                     │
│         │     Subscribe to call events                │                     │
│         │────────────────────────────────────────────▶│                     │
│         │                                             │                     │
│         │  4. Connect WebSocket                       │                     │
│         │═══════════════════════════════════════════▶│                     │
│         │                                             │                     │
│         │  5. Events flow via WebSocket               │                     │
│         │◀══════════════════════════════════════════▶│                     │
│         │     { "event": "call.ringing", ... }        │                     │
│         │     { "event": "call.connected", ... }      │                     │
│         │     { "event": "call.ended", ... }          │                     │
│         │                                             │                     │
│                                                                             │
└─────────────────────────────────────────────────────────────────────────────┘

# integrations/gotoconnect/events.py

class GoToEventManager:
    """Manages GoToConnect event subscriptions."""
    
    EVENTS_BASE_URL = "https://api.goto.com/call-events/v1"
    
    def __init__(self, auth_manager: GoToAuthManager):
        self.auth_manager = auth_manager
        self._channel_id: Optional[str] = None
        self._websocket_url: Optional[str] = None
        self._subscriptions: dict[str, str] = {}
    
    async def create_channel(self) -> tuple[str, str]:
        """
        Create a notification channel.
        
        Returns:
            Tuple of (channel_id, websocket_url)
        """
        token = await self.auth_manager.get_access_token()
        
        async with httpx.AsyncClient() as client:
            response = await client.post(
                f"{self.EVENTS_BASE_URL}/notifications/channels",
                headers={
                    "Authorization": f"Bearer {token}",
                    "Content-Type": "application/json",
                }
            )
            response.raise_for_status()
            data = response.json()
            
            self._channel_id = data["channelId"]
            self._websocket_url = data["websocketUrl"]
            
            return self._channel_id, self._websocket_url

POST /call-events/v1/notifications/channels

{
  "channelId": "ch_abc123def456",
  "websocketUrl": "wss://realtime.goto.com/v1/notifications?channelId=ch_abc123def456",
  "expiresAt": "2026-01-16T11:30:00Z"
}

async def subscribe_to_calls(
    self,
    line_ids: list[str] = None,
    event_types: list[str] = None
) -> str:
    """
    Subscribe to call events.
    
    Args:
        line_ids: Specific lines to subscribe to (None = all)
        event_types: Specific events (None = all)
    
    Returns:
        Subscription ID
    """
    if not self._channel_id:
        await self.create_channel()
    
    token = await self.auth_manager.get_access_token()
    
    # Default to all call events
    if event_types is None:
        event_types = [
            "call.ringing",
            "call.connected",
            "call.ended",
            "call.held",
            "call.resumed",
            "call.transferred",
            "call.dtmf",
            "call.recording",
            "call.ice_candidate",
        ]
    
    subscription_request = {
        "channelId": self._channel_id,
        "eventTypes": event_types,
    }
    
    if line_ids:
        subscription_request["lineIds"] = line_ids
    
    async with httpx.AsyncClient() as client:
        response = await client.post(
            f"{self.EVENTS_BASE_URL}/notifications/subscriptions",
            json=subscription_request,
            headers={
                "Authorization": f"Bearer {token}",
                "Content-Type": "application/json",
            }
        )
        response.raise_for_status()
        data = response.json()
        
        subscription_id = data["subscriptionId"]
        self._subscriptions[subscription_id] = data
        
        return subscription_id

POST /call-events/v1/notifications/subscriptions
{
  "channelId": "ch_abc123def456",
  "eventTypes": [
    "call.ringing",
    "call.connected",
    "call.ended",
    "call.held",
    "call.resumed",
    "call.transferred"
  ],
  "lineIds": ["line_001", "line_002"]
}

{
  "subscriptionId": "sub_xyz789",
  "channelId": "ch_abc123def456",
  "eventTypes": ["call.ringing", "call.connected", ...],
  "lineIds": ["line_001", "line_002"],
  "createdAt": "2026-01-16T10:30:00Z"
}

import websockets
import json
from typing import Callable, Awaitable

EventHandler = Callable[[dict], Awaitable[None]]

class GoToEventListener:
    """
    WebSocket listener for GoToConnect events.
    """
    
    def __init__(
        self,
        websocket_url: str,
        auth_manager: GoToAuthManager
    ):
        self.websocket_url = websocket_url
        self.auth_manager = auth_manager
        self._handlers: dict[str, list[EventHandler]] = {}
        self._running = False
        self._ws = None
    
    def on(self, event_type: str, handler: EventHandler) -> None:
        """Register an event handler."""
        if event_type not in self._handlers:
            self._handlers[event_type] = []
        self._handlers[event_type].append(handler)
    
    async def start(self) -> None:
        """Start listening for events."""
        self._running = True
        
        token = await self.auth_manager.get_access_token()
        
        # Add auth token to WebSocket URL
        ws_url = f"{self.websocket_url}&token={token}"
        
        while self._running:
            try:
                async with websockets.connect(ws_url) as ws:
                    self._ws = ws
                    await self._listen(ws)
            except websockets.ConnectionClosed:
                if self._running:
                    logger.warning("WebSocket disconnected, reconnecting...")
                    await asyncio.sleep(1)
            except Exception as e:
                logger.error(f"WebSocket error: {e}")
                if self._running:
                    await asyncio.sleep(5)
    
    async def _listen(self, ws) -> None:
        """Listen for events on the WebSocket."""
        async for message in ws:
            try:
                event = json.loads(message)
                await self._dispatch(event)
            except json.JSONDecodeError:
                logger.warning(f"Invalid JSON: {message}")
            except Exception as e:
                logger.error(f"Error handling event: {e}")
    
    async def _dispatch(self, event: dict) -> None:
        """Dispatch event to registered handlers."""
        event_type = event.get("type") or event.get("event")
        
        if not event_type:
            logger.warning(f"Event without type: {event}")
            return
        
        handlers = self._handlers.get(event_type, [])
        handlers += self._handlers.get("*", [])  # Wildcard handlers
        
        for handler in handlers:
            try:
                await handler(event)
            except Exception as e:
                logger.error(f"Handler error for {event_type}: {e}")
    
    async def stop(self) -> None:
        """Stop listening."""
        self._running = False
        if self._ws:
            await self._ws.close()