add sip endpoints

video/src/vonage_video/errors.py

@@ -47,3 +47,7 @@ class InvalidOutputOptionsError(VideoError):
 
 class InvalidBroadcastStateError(VideoError):
     """The broadcast state was invalid for the specified operation."""
+
+
+class RoutedSessionRequiredError(VideoError):
+    """The operation requires a session with `media_mode=routed`."""

video/src/vonage_video/models/enums.py

@@ -2,23 +2,32 @@
 
 
 class TokenRole(str, Enum):
+    """The role assigned to the token."""
+
     SUBSCRIBER = 'subscriber'
     PUBLISHER = 'publisher'
     PUBLISHER_ONLY = 'publisheronly'
     MODERATOR = 'moderator'
 
 
 class ArchiveMode(str, Enum):
+    """Whether the session is archived automatically ("always") or not ("manual")."""
+
     MANUAL = 'manual'
     ALWAYS = 'always'
 
 
 class MediaMode(str, Enum):
+    """Whether the session uses the Vonage Video media router ("routed")
+    or peers connect directly (relayed)."""
+
     ROUTED = 'routed'
     RELAYED = 'relayed'
 
 
 class P2pPreference(str, Enum):
+    """The preference for peer-to-peer connections."""
+
     DISABLED = 'disabled'
     ALWAYS = 'always'
 
@@ -40,6 +49,8 @@ class LanguageCode(str, Enum):
 
 
 class AudioSampleRate(int, Enum):
+    """Audio sample rate, in Hertz."""
+
     KHZ_8 = 8000
     KHZ_16 = 16000
 

video/src/vonage_video/models/sip.py

@@ -0,0 +1,78 @@
+from typing import Optional
+from pydantic import BaseModel, Field
+
+
+class SipAuth(BaseModel):
+    """
+    Model representing the authentication details for the SIP INVITE request
+    for HTTP digest authentication, if it is required by your SIP platform.
+
+    Attributes:
+        username (str): The username for HTTP digest authentication.
+        password (str): The password for HTTP digest authentication.
+    """
+
+    username: str
+    password: str
+
+
+class SipOptions(BaseModel):
+    """
+    Model representing the SIP options for the call.
+
+    Attributes:
+        uri (str): The SIP URI to be used as the destination of the SIP call.
+        from_ (Optional[str]): The number or string sent to the final SIP number
+        as the caller. It must be a string in the form of `from@example.com`, where
+        `from` can be a string or a number.
+        headers (Optional[dict]): Custom headers to be added to the SIP INVITE request.
+        auth (Optional[SipAuth]): Authentication details for the SIP INVITE request.
+        secure (Optional[bool]): Indicates whether the media must be transmitted encrypted.
+            Default is false.
+        video (Optional[bool]): Indicates whether the SIP call will include video.
+            Default is false.
+        observe_force_mute (Optional[bool]): Indicates whether the SIP endpoint observes
+            force mute moderation.
+    """
+
+    uri: str
+    from_: Optional[str] = Field(None, serialization_alias='from')
+    headers: Optional[dict] = None
+    auth: Optional[SipAuth] = None
+    secure: Optional[bool] = None
+    video: Optional[bool] = None
+    observe_force_mute: Optional[bool] = Field(
+        None, serialization_alias='observeForceMute'
+    )
+
+
+class InitiateSipRequest(BaseModel):
+    """
+    Model representing the SIP options for joining a Vonage Video session.
+
+    Attributes:
+        session_id (str): The Vonage Video session ID for the SIP call to join.
+        token (str): The Vonage Video token to be used for the participant being called.
+        sip (Sip): The SIP options for the call.
+    """
+
+    session_id: str = Field(..., serialization_alias='sessionId')
+    token: str
+    sip: SipOptions
+
+
+class SipCall(BaseModel):
+    """
+    Model representing the details of a SIP call.
+
+    Attributes:
+        id (str): A unique ID for the SIP call.
+        connection_id (str): The Vonage Video connection ID for the SIP call's connection
+            in the Vonage Video session.
+        stream_id (str): The Vonage Video stream ID for the SIP call's stream in the
+            Vonage Video session.
+    """
+
+    id: str
+    connection_id: str = Field(..., serialization_alias='connectionId')
+    stream_id: str = Field(..., serialization_alias='streamId')

video/src/vonage_video/video.py

@@ -3,9 +3,11 @@
 from pydantic import validate_call
 from vonage_http_client.errors import HttpRequestError
 from vonage_http_client.http_client import HttpClient
+from vonage_utils.types import Dtmf
 from vonage_video.errors import (
     InvalidArchiveStateError,
     InvalidBroadcastStateError,
+    RoutedSessionRequiredError,
     VideoError,
 )
 from vonage_video.models.archive import (
@@ -29,6 +31,7 @@
 )
 from vonage_video.models.session import SessionOptions, VideoSession
 from vonage_video.models.signal import SignalData
+from vonage_video.models.sip import SipCall, InitiateSipRequest
 from vonage_video.models.stream import StreamInfo, StreamLayoutOptions
 from vonage_video.models.token import TokenOptions
 
@@ -649,6 +652,51 @@ def remove_stream_from_broadcast(self, broadcast_id: str, stream_id: str) -> Non
             params={'removeStream': stream_id},
         )
 
+    @validate_call
+    def initiate_sip_call(self, sip_request_params: InitiateSipRequest) -> SipCall:
+        """Initiates a SIP call using the Vonage Video API.
+
+        Args:
+            sip_request_params (SipParams): Model containing the session ID and a valid token,
+                as well as options for the SIP call.
+
+        Returns:
+            SipCall: The SIP call object.
+        """
+        try:
+            response = self._http_client.post(
+                self._http_client.video_host,
+                f'/v2/project/{self._http_client.auth.application_id}/dial',
+                sip_request_params.model_dump(exclude_none=True, by_alias=True),
+            )
+        except HttpRequestError as e:
+            conflict_error_message = 'SIP calling can only be used in a session with'
+            ' `media_mode=routed`.'
+            self._check_conflict_error(
+                e, RoutedSessionRequiredError, conflict_error_message
+            )
+
+        return SipCall(**response)
+
+    @validate_call
+    def play_dtmf(self, session_id: str, digits: Dtmf, connection_id: str = None) -> None:
+        """Plays DTMF tones into one or all SIP connections in a session using the Vonage
+        Video API.
+
+        Args:
+            session_id (str): The session ID.
+            digits (Dtmf): The DTMF digits to play.
+            connection_id (str, Optional): The connection ID to send the DTMF tones to.
+                If not provided, the DTMF tones will be played on all connections in
+                the session.
+        """
+        if connection_id is not None:
+            url = f'/v2/project/{self._http_client.auth.application_id}/session/{session_id}/connection/{connection_id}/play-dtmf'
+        else:
+            url = f'/v2/project/{self._http_client.auth.application_id}/session/{session_id}/play-dtmf'
+
+        self._http_client.post(self._http_client.video_host, url, {'digits': digits})
+
     @validate_call
     def _list_video_objects(
         self,

video/tests/test_sip.py

@@ -0,0 +1,76 @@
+from os.path import abspath
+
+import responses
+from vonage_http_client import HttpClient
+
+from vonage_video.models.sip import SipAuth, SipOptions, InitiateSipRequest
+from vonage_video.models.token import TokenOptions
+from vonage_video.video import Video
+
+from testutils import build_response, get_mock_jwt_auth
+
+path = abspath(__file__)
+
+
+video = Video(HttpClient(get_mock_jwt_auth()))
+
+
+def test_sip_params_model():
+    sip_request_params = InitiateSipRequest(
+        session_id='test_session_id',
+        token='test_token',
+        sip=SipOptions(
+            uri='sip:user@sip.partner.com;transport=tls',
+            from_='example@example.com',
+            headers={'header_key': 'header_value'},
+            auth=SipAuth(username='username', password='password'),
+            secure=True,
+            video=True,
+            observe_force_mute=True,
+        ),
+    )
+
+    assert sip_request_params.model_dump(by_alias=True) == {
+        'sessionId': 'test_session_id',
+        'token': 'test_token',
+        'sip': {
+            'uri': 'sip:user@sip.partner.com;transport=tls',
+            'from': 'example@example.com',
+            'headers': {'header_key': 'header_value'},
+            'auth': {'username': 'username', 'password': 'password'},
+            'secure': True,
+            'video': True,
+            'observeForceMute': True,
+        },
+    }
+
+
+@responses.activate
+def test_initiate_sip_call():
+    build_response(
+        path,
+        'POST',
+        'https://video.api.vonage.com/v2/project/test_application_id/dial',
+        'sip.json',
+        200,
+    )
+
+    sip_request_params = InitiateSipRequest(
+        session_id='test_session_id',
+        token='test_token',
+        sip=SipOptions(
+            uri='sip:user@sip.partner.com;transport=tls',
+            from_='example@example.com',
+            headers={'header_key': 'header_value'},
+            auth=SipAuth(username='username', password='password'),
+            secure=True,
+            video=True,
+            observe_force_mute=True,
+        ),
+    )
+
+    sip_call = video.initiate_sip_call(sip_request_params)
+
+    assert sip_call.id == ''
+    assert sip_call.connection_id == ''
+    assert sip_call.stream_id == ''