Session Initiation Protocol (SIP)

voip.sip.Dialog dataclass

Peer-to-peer SIP relationship between two user agents RFC 3261 §12.

Subclass Dialog to implement call handling. Set the subclass as dialog_class on the SIP session for inbound calls:

class MyDialog(Dialog):
    def call_received(self) -> None:
        self.ringing()
        self.answer(session_class=MyCall)

class MySession(SessionInitiationProtocol):
    dialog_class = MyDialog

For outbound calls:

dialog = Dialog(sip=my_sip_session)
await dialog.dial("sip:bob@biloxi.com", session_class=MyCall)

Parameters:

Name	Type	Description	Default
sip	SessionInitiationProtocol | None	The parent protocol the session belongs to.	None

Source code in voip/sip/dialog.py

@dataclasses.dataclass(kw_only=True, slots=True)
class Dialog:
    """Peer-to-peer SIP relationship between two user agents [RFC 3261 §12].

    Subclass `Dialog` to implement call handling.  Set the subclass as
    `dialog_class` on the SIP session for inbound calls:

    ```python
    class MyDialog(Dialog):
        def call_received(self) -> None:
            self.ringing()
            self.answer(session_class=MyCall)

    class MySession(SessionInitiationProtocol):
        dialog_class = MyDialog
    ```

    For outbound calls:

    ```python
    dialog = Dialog(sip=my_sip_session)
    await dialog.dial("sip:bob@biloxi.com", session_class=MyCall)
    ```


    [RFC 3261 §12]: https://datatracker.ietf.org/doc/html/rfc3261#section-12

    Args:
        sip: The parent protocol the session belongs to.
    """

    # https://datatracker.ietf.org/doc/html/rfc3261#section-17.1.1
    T1: typing.ClassVar[datetime.timedelta] = datetime.timedelta(milliseconds=500)

    # https://datatracker.ietf.org/doc/html/rfc3261#section-17.1.2
    BYE_ACK_TIMEOUT: typing.ClassVar[datetime.timedelta] = 64 * T1

    uac: SipURI = None
    call_id: str = dataclasses.field(
        default_factory=lambda: f"{uuid.uuid4()}@{socket.gethostname()}",
        compare=False,
    )
    local_tag: str = dataclasses.field(
        default_factory=lambda: str(uuid.uuid4()), compare=True
    )
    remote_tag: str | None = dataclasses.field(default=None, compare=True)
    remote_contact: SipURI | None = dataclasses.field(default=None, compare=True)
    route_set: list[SipURI] = dataclasses.field(default_factory=list)
    local_party: str | None = dataclasses.field(default=None, compare=False)
    remote_party: str | None = dataclasses.field(default=None, compare=False)
    outbound_cseq: int = dataclasses.field(default=1, compare=False)
    sip: transactions.SessionInitiationProtocol | None = dataclasses.field(
        default=None, compare=False, repr=False
    )
    invite_transaction: transactions.InviteTransaction | None = dataclasses.field(
        default=None, compare=False, repr=False
    )

    session: Session | None = dataclasses.field(default=None, init=False, compare=False)
    created: datetime.datetime = dataclasses.field(
        init=False, default_factory=datetime.datetime.now
    )

    @property
    def from_header(self) -> str:
        """The logical sender of a request."""
        return f"{self.uac.scheme}:{self.uac.user}@{self.uac.host};tag={self.local_tag}"

    @property
    def to_header(self) -> str:
        """The logical recipient of a request."""
        part = f"{self.uac.scheme}:{self.uac.user}@{self.uac.host}:{self.uac.port};transport={self.uac.parameters.get('transport', 'TLS')}"
        if self.remote_tag:
            part += f";tag={self.remote_tag}"
        return part

    @property
    def headers(self) -> dict[str, str]:
        """Return a dict of headers for this dialog."""
        return {
            "From": self.from_header,
            "To": self.to_header,
            "Call-ID": self.call_id,
        }

    def call_received(self) -> None:
        """
        Called when an INVITE is received from the remote party.

        Override in subclasses to [answer][voip.sip.Dialog.answer],
        [ring][voip.sip.Dialog.ringing],
        or [reject][voip.sip.Dialog.reject] the call.

        The base implementation rejects with a busy signal.
        """  # noqa: D401
        self.reject()

    def hangup_received(self) -> None:
        """
        Called when the remote party sends a BYE.

        Override in subclasses to perform teardown.
        """  # noqa: D401

    def ringing(self) -> None:
        """
        Send a ringing signal to the remote party.

        This is optional but recommended for good user experience.
        If not called, the caller will hear silence until the call is accepted or rejected.
        """
        if self.invite_transaction is not None:
            self.invite_transaction.ringing()

    def answer(
        self, *, session_class: type[Session], **session_kwargs: typing.Any
    ) -> None:
        """
        Accept the inbound call and start a multimedia session.

        Args:
            session_class: Session subclass to create for this call.
            **session_kwargs: Extra keyword arguments forwarded to `session_class`.
        """
        if self.invite_transaction is not None:
            self.invite_transaction.answer(
                session_class=session_class, **session_kwargs
            )

    def reject(self, status_code: types.SIPStatus = types.SIPStatus.BUSY_HERE) -> None:
        """
        Reject the inbound call with the given status code.

        Common status codes include:
            - [BUSY_HERE][voip.sip.types.SIPStatus.BUSY_HERE]: The remote party will hear a busy signal.
            - [DECLINE][voip.sip.types.SIPStatus.DECLINE]: The remote party will hear a decline signal.
            - [DOES_NOT_EXIST_ANYWHERE][voip.sip.types.SIPStatus.DOES_NOT_EXIST_ANYWHERE]: The remote party will hear a "The person you are trying to reach…" message.

        Args:
            status_code: SIP response status code (default: 486 Busy Here).
        """
        if self.invite_transaction is not None:
            self.invite_transaction.reject(status_code)

    async def bye(self) -> None:
        """End the call and terminate the dialog and multimedia session."""
        from voip.sip.transactions import ByeTransaction  # noqa: PLC0415

        try:
            await asyncio.wait_for(
                ByeTransaction.send(sip=self.sip, dialog=self),
                timeout=self.BYE_ACK_TIMEOUT.total_seconds(),
            )
        except TimeoutError:
            logger.warning(
                "BYE for dialog %s was not acknowledged within %r",
                self.call_id,
                self.BYE_ACK_TIMEOUT,
            )
        self.sip.drop_dialog(self)

    async def dial(
        self,
        target: SipURI,
        *,
        session_class: type[Session],
        **session_kwargs: typing.Any,
    ) -> None:
        """
        Initiate an outbound call to *target*.

        Offers SRTP (`RTP/SAVP` + SDES) and falls back to plain RTP if the far
        end rejects it (488/606/415) or answers with `RTP/AVP`.

        Args:
            target: SIP or tel URI of the remote party (e.g. `"sip:+15551234567@carrier.com"` or `"tel:+15551234567"`).
            session_class: Session subclass to create for this call.
            **session_kwargs: Extra keyword arguments forwarded to `session_class`.

        [RFC 3261 §13.1]: https://datatracker.ietf.org/doc/html/rfc3261#section-13.1
        """
        from voip.sip.transactions import InviteTransaction  # noqa: PLC0415

        await InviteTransaction.send(
            sip=self.sip,
            target=target,
            dialog=self,
            session_class=session_class,
            **session_kwargs,
        )

    @classmethod
    def from_request(cls, request: messages.Request, **kwargs) -> Dialog:
        """Create a dialog from a request, extracting relevant headers."""
        return cls(
            call_id=request.headers["Call-ID"],
            local_tag=request.remote_tag or str(uuid.uuid4()),
            remote_tag=request.local_tag,
            remote_contact=request.headers.get("Contact"),
            **kwargs,
        )

call_received()

Called when an INVITE is received from the remote party.

Override in subclasses to answer, ring, or reject the call.

The base implementation rejects with a busy signal.

Source code in voip/sip/dialog.py

def call_received(self) -> None:
    """
    Called when an INVITE is received from the remote party.

    Override in subclasses to [answer][voip.sip.Dialog.answer],
    [ring][voip.sip.Dialog.ringing],
    or [reject][voip.sip.Dialog.reject] the call.

    The base implementation rejects with a busy signal.
    """  # noqa: D401
    self.reject()

hangup_received()

Called when the remote party sends a BYE.

Override in subclasses to perform teardown.

Source code in voip/sip/dialog.py

def hangup_received(self) -> None:
    """
    Called when the remote party sends a BYE.

    Override in subclasses to perform teardown.
    """  # noqa: D401

ringing()

Send a ringing signal to the remote party.

This is optional but recommended for good user experience. If not called, the caller will hear silence until the call is accepted or rejected.

Source code in voip/sip/dialog.py

def ringing(self) -> None:
    """
    Send a ringing signal to the remote party.

    This is optional but recommended for good user experience.
    If not called, the caller will hear silence until the call is accepted or rejected.
    """
    if self.invite_transaction is not None:
        self.invite_transaction.ringing()

answer(*, session_class, **session_kwargs)

Accept the inbound call and start a multimedia session.

Parameters:

Name	Type	Description	Default
session_class	type[Session]	Session subclass to create for this call.	required
**session_kwargs	Any	Extra keyword arguments forwarded to session_class.	{}

Source code in voip/sip/dialog.py

def answer(
    self, *, session_class: type[Session], **session_kwargs: typing.Any
) -> None:
    """
    Accept the inbound call and start a multimedia session.

    Args:
        session_class: Session subclass to create for this call.
        **session_kwargs: Extra keyword arguments forwarded to `session_class`.
    """
    if self.invite_transaction is not None:
        self.invite_transaction.answer(
            session_class=session_class, **session_kwargs
        )

reject(status_code=types.SIPStatus.BUSY_HERE)

Reject the inbound call with the given status code.

Common status codes include

• BUSY_HERE: The remote party will hear a busy signal.
• DECLINE: The remote party will hear a decline signal.
• DOES_NOT_EXIST_ANYWHERE: The remote party will hear a "The person you are trying to reach…" message.

Parameters:

Name	Type	Description	Default
status_code	SIPStatus	SIP response status code (default: 486 Busy Here).	BUSY_HERE

Source code in voip/sip/dialog.py

def reject(self, status_code: types.SIPStatus = types.SIPStatus.BUSY_HERE) -> None:
    """
    Reject the inbound call with the given status code.

    Common status codes include:
        - [BUSY_HERE][voip.sip.types.SIPStatus.BUSY_HERE]: The remote party will hear a busy signal.
        - [DECLINE][voip.sip.types.SIPStatus.DECLINE]: The remote party will hear a decline signal.
        - [DOES_NOT_EXIST_ANYWHERE][voip.sip.types.SIPStatus.DOES_NOT_EXIST_ANYWHERE]: The remote party will hear a "The person you are trying to reach…" message.

    Args:
        status_code: SIP response status code (default: 486 Busy Here).
    """
    if self.invite_transaction is not None:
        self.invite_transaction.reject(status_code)

dial(target, *, session_class, **session_kwargs) async

Initiate an outbound call to target.

Offers SRTP (RTP/SAVP + SDES) and falls back to plain RTP if the far end rejects it (488/606/415) or answers with RTP/AVP.

Parameters:

Name	Type	Description	Default
target	SipURI	SIP or tel URI of the remote party (e.g. "sip:+15551234567@carrier.com" or "tel:+15551234567").	required
session_class	type[Session]	Session subclass to create for this call.	required
**session_kwargs	Any	Extra keyword arguments forwarded to session_class.	{}

Source code in voip/sip/dialog.py

async def dial(
    self,
    target: SipURI,
    *,
    session_class: type[Session],
    **session_kwargs: typing.Any,
) -> None:
    """
    Initiate an outbound call to *target*.

    Offers SRTP (`RTP/SAVP` + SDES) and falls back to plain RTP if the far
    end rejects it (488/606/415) or answers with `RTP/AVP`.

    Args:
        target: SIP or tel URI of the remote party (e.g. `"sip:+15551234567@carrier.com"` or `"tel:+15551234567"`).
        session_class: Session subclass to create for this call.
        **session_kwargs: Extra keyword arguments forwarded to `session_class`.

    [RFC 3261 §13.1]: https://datatracker.ietf.org/doc/html/rfc3261#section-13.1
    """
    from voip.sip.transactions import InviteTransaction  # noqa: PLC0415

    await InviteTransaction.send(
        sip=self.sip,
        target=target,
        dialog=self,
        session_class=session_class,
        **session_kwargs,
    )

bye() async

End the call and terminate the dialog and multimedia session.

Source code in voip/sip/dialog.py

async def bye(self) -> None:
    """End the call and terminate the dialog and multimedia session."""
    from voip.sip.transactions import ByeTransaction  # noqa: PLC0415

    try:
        await asyncio.wait_for(
            ByeTransaction.send(sip=self.sip, dialog=self),
            timeout=self.BYE_ACK_TIMEOUT.total_seconds(),
        )
    except TimeoutError:
        logger.warning(
            "BYE for dialog %s was not acknowledged within %r",
            self.call_id,
            self.BYE_ACK_TIMEOUT,
        )
    self.sip.drop_dialog(self)

voip.sip.SessionInitiationProtocol dataclass

Bases: Protocol, DatagramProtocol

SIP User Agent Client (UAC) over TLS/TCP or UDP [RFC 3261].

Handles SIP message parsing, carrier registration, and transaction management. The transport is selected automatically from the AOR's transport parameter:

aor.transport	Underlying transport
UDP (default)	UDP datagram socket
TCP	plain TCP
TLS	TCP with TLS

Use run for a single outbound connection and serve for a persistent inbound server with automatic reconnection.

Example

import asyncio
from voip.sip import SessionInitiationProtocol, Dialog

async def main():
    protocol = await SessionInitiationProtocol.run(
        aor=SipURI.parse("sip:alice@carrier.example;transport=UDP"),
        dialog_class=Dialog,
    )
    # place outbound calls via protocol …

Note

The support is limited to UAC (client mode). This library currently does not implement server (UAS) functionality.

Parameters:

Name	Type	Description	Default
aor	SipURI	SIP Address of Record (AOR) to register with the carrier.	required
rtp	RealtimeTransportProtocol	Shared RTP mux for call media.	required
dialog_class	type[Dialog]	Dialog subclass used to create dialogs for incoming calls. Defaults to the base Dialog which rejects all calls with 486 Busy Here.	Dialog
keepalive_interval	timedelta	Keep-alive ping interval for TCP transports. Should be between 30 and 90 seconds (RFC 5626).	timedelta(seconds=30)

Source code in voip/sip/protocol.py

@dataclasses.dataclass(kw_only=True, slots=True)
class SessionInitiationProtocol(asyncio.Protocol, asyncio.DatagramProtocol):
    """
    SIP User Agent Client (UAC) over TLS/TCP or UDP [RFC 3261].

    Handles SIP message parsing, carrier registration, and transaction management.
    The transport is selected automatically from the AOR's `transport` parameter:

    | `aor.transport` | Underlying transport |
    |-----------------|----------------------|
    | `UDP` (default) | UDP datagram socket |
    | `TCP` | plain TCP |
    | `TLS` | TCP with TLS |

    Use [`run`][voip.sip.protocol.SessionInitiationProtocol.run] for a single
    outbound connection and [`serve`][voip.sip.protocol.SessionInitiationProtocol.serve]
    for a persistent inbound server with automatic reconnection.

    Example:
        ```python
        import asyncio
        from voip.sip import SessionInitiationProtocol, Dialog

        async def main():
            protocol = await SessionInitiationProtocol.run(
                aor=SipURI.parse("sip:alice@carrier.example;transport=UDP"),
                dialog_class=Dialog,
            )
            # place outbound calls via protocol …
        ```

    > [!Note]
    > The support is limited to UAC (client mode).
    > This library currently does not implement server (UAS) functionality.

    [RFC 3261]: https://datatracker.ietf.org/doc/html/rfc3261

    Args:
        aor: SIP Address of Record (AOR) to register with the carrier.
        rtp: Shared RTP mux for call media.
        dialog_class: [Dialog][voip.sip.Dialog] subclass used to
            create dialogs for incoming calls.  Defaults to the base
            [Dialog][voip.sip.Dialog] which rejects all calls with
            `486 Busy Here`.
        keepalive_interval: Keep-alive ping interval for TCP transports.
            Should be between 30 and 90 seconds (RFC 5626).

    """

    aor: types.SipURI
    rtp: RealtimeTransportProtocol
    dialog_class: type[Dialog] = dataclasses.field(default=Dialog)
    keepalive_interval: datetime.timedelta = datetime.timedelta(seconds=30)

    keepalive_task: asyncio.Task | None = dataclasses.field(init=False, default=None)
    public_address: NetworkAddress = None
    dialogs: dict[tuple[str, str], Dialog] = dataclasses.field(
        init=False, default_factory=dict
    )
    transactions: dict[str, Transaction] = dataclasses.field(
        init=False, default_factory=dict
    )
    disconnected_event: asyncio.Event = dataclasses.field(
        init=False, default_factory=asyncio.Event
    )
    registered_event: asyncio.Event = dataclasses.field(
        init=False, default_factory=asyncio.Event
    )
    transport: asyncio.BaseTransport | None = dataclasses.field(
        init=False, default=None
    )
    is_secure: bool = dataclasses.field(init=False, default=False)
    recv_buffer: bytearray = dataclasses.field(init=False, default_factory=bytearray)

    def __post_init__(self):
        if self.public_address is None and self.rtp.public_address is not None:
            self.public_address = self.rtp.public_address.result()

    @classmethod
    async def run(
        cls,
        aor: types.SipURI,
        dialog_class: type[Dialog],
        *,
        rtp: RealtimeTransportProtocol | None = None,
        no_verify_tls: bool = False,
        stun_server: NetworkAddress | None = None,
        **kwargs: typing.Any,
    ) -> SessionInitiationProtocol:
        """Connect to the SIP proxy and return once registered.

        Suspend until SIP registration is confirmed before returning the ready
        protocol. After this call returns, the caller may safely place outbound
        calls or start an MCP server.

        The transport protocol (TLS vs. plain TCP) and proxy address are read
        from *aor* directly — no extra arguments are needed.

        Args:
            aor: SIP Address of Record, e.g. `sip:alice@carrier.example`.
                The host, port, and `transport` parameter are used to connect
                to the SIP proxy.
            dialog_class: [`Dialog`][voip.sip.Dialog] subclass used for
                inbound calls.  Defaults to the base
                [`Dialog`][voip.sip.Dialog], which rejects all calls.
            rtp: Existing RTP endpoint to reuse.  When `None` (default) a
                new datagram endpoint is created from *aor* and *stun_server*.
                Pass an existing instance to share one endpoint across
                reconnections (see [`serve`][voip.sip.protocol.SessionInitiationProtocol.serve]).
            no_verify_tls: Disable TLS certificate verification. Insecure; for
                testing only. Defaults to `False`.
            stun_server: STUN server for RTP NAT traversal. Ignored when *rtp*
                is provided.
            **kwargs: Extra keyword arguments forwarded to the protocol constructor.

        Returns:
            The registered [`SessionInitiationProtocol`][voip.sip.protocol.SessionInitiationProtocol]
            instance, ready to place calls.
        """
        loop = asyncio.get_running_loop()
        if rtp is None:
            addr_info = socket.getaddrinfo(
                str(aor.maddr[0]), aor.maddr[1], type=socket.SOCK_DGRAM
            )
            rtp_bind_address = (
                "::" if addr_info[0][0] == socket.AF_INET6 else "0.0.0.0"  # noqa: S104
            )
            rtp = await RealtimeTransportProtocol.serve(rtp_bind_address, stun_server)
        if aor.transport == "UDP":
            _, protocol = await loop.create_datagram_endpoint(
                lambda: cls(aor=aor, rtp=rtp, dialog_class=dialog_class, **kwargs),
                remote_addr=(str(aor.maddr[0]), aor.maddr[1]),
            )
        else:
            ssl_context: ssl.SSLContext | None = None
            if aor.transport == "TLS":
                ssl_context = ssl.create_default_context()
                if no_verify_tls:
                    ssl_context.check_hostname = False
                    ssl_context.verify_mode = ssl.CERT_NONE
            _, protocol = await loop.create_connection(
                lambda: cls(aor=aor, rtp=rtp, dialog_class=dialog_class, **kwargs),
                host=str(aor.maddr[0]),
                port=aor.maddr[1],
                ssl=ssl_context,
            )
        await protocol.registered_event.wait()
        return protocol

    @classmethod
    async def serve(
        cls,
        aor: types.SipURI,
        dialog_class: type[Dialog],
        *,
        rtp: RealtimeTransportProtocol | None = None,
        no_verify_tls: bool = False,
        stun_server: NetworkAddress | None = None,
        **kwargs: typing.Any,
    ) -> None:
        """Register with a carrier and handle inbound calls, reconnecting on disconnect.

        Enter a persistent loop: connect to the SIP proxy, wait for the
        connection to drop, and reconnect with exponential back-off. Use this
        for long-running inbound-call servers.

        The transport protocol (TLS vs. plain TCP) and proxy address are read from
        *aor* directly.

        Args:
            aor: SIP Address of Record, e.g. `sip:alice@carrier.example`.
            dialog_class: [`Dialog`][voip.sip.Dialog] subclass used for
                inbound calls.
            rtp: Existing RTP endpoint to reuse.  When `None` (default) a
                new datagram endpoint is created from *aor* and *stun_server*.
                Pass an existing instance to share one endpoint across
                reconnections (see [`serve`][voip.sip.protocol.SessionInitiationProtocol.serve]).
            no_verify_tls: Disable TLS certificate verification. Insecure; for
                testing only. Defaults to `False`.
            stun_server: STUN server for RTP NAT traversal.
            **kwargs: Extra keyword arguments forwarded to the protocol constructor.
        """
        addr_info = socket.getaddrinfo(
            str(aor.maddr[0]), aor.maddr[1], type=socket.SOCK_DGRAM
        )
        if rtp is None:
            rtp_bind_address = (
                "::" if addr_info[0][0] == socket.AF_INET6 else "0.0.0.0"  # noqa: S104
            )
            rtp = await RealtimeTransportProtocol.serve(rtp_bind_address, stun_server)
        backoff_secs = 1
        while True:
            try:
                protocol = await cls.run(
                    aor, dialog_class, rtp=rtp, no_verify_tls=no_verify_tls, **kwargs
                )
                backoff_secs = 1
                await protocol.disconnected_event.wait()
                logger.info("SIP connection closed; reconnecting in %s s", backoff_secs)
            except (OSError, ssl.SSLError) as exc:
                logger.warning(
                    "SIP connection failed (%s); retrying in %s s", exc, backoff_secs
                )
            await asyncio.sleep(backoff_secs)
            backoff_secs = min(backoff_secs * 2, 60)

    def register_dialog(self, dialog: Dialog) -> None:
        """Register *dialog* for later lookup by its tag pair."""
        if dialog.remote_tag is None:
            logger.warning("Dialog without remote tag cannot be registered: %r", dialog)
        else:
            self.dialogs[dialog.local_tag, dialog.remote_tag] = dialog

    def drop_dialog(self, dialog: Dialog) -> None:
        """Remove *dialog* from the registry."""
        if dialog.remote_tag is None:
            logger.warning("Dialog without remote tag cannot be removed: %r", dialog)
        else:
            try:
                del self.dialogs[dialog.local_tag, dialog.remote_tag]
            except KeyError:
                logger.warning("Dialog not found for removal: %r", dialog)

    def register_transaction(self, tx: Transaction) -> None:
        """Register *tx* by its branch parameter."""
        self.transactions[tx.branch] = tx

    def drop_transaction(self, tx: Transaction) -> None:
        """Remove *tx* from the registry."""
        try:
            del self.transactions[tx.branch]
        except KeyError:
            logger.warning("Transaction not found for removal: %r", tx)

    def connection_made(self, transport: asyncio.BaseTransport) -> None:  # type: ignore[override]
        """Handle transport readiness by starting carrier registration.

        Keepalive pings (RFC 5626) are only started for TCP/TLS transports.
        """
        self.transport = transport
        self.is_secure = (
            not isinstance(transport, asyncio.DatagramTransport)
            and transport.get_extra_info("ssl_object") is not None
        )
        try:
            loop = asyncio.get_running_loop()
            tx = RegisterTransaction(sip=self, method=SIPMethod.REGISTER)
            self.register_transaction(tx)
            loop.create_task(self.handle_registration(tx))
            if not isinstance(transport, asyncio.DatagramTransport):
                self.keepalive_task = loop.create_task(self.send_keepalive())
        except RuntimeError:
            pass

    async def send_keepalive(self) -> None:
        while True:
            await asyncio.sleep(self.keepalive_interval.total_seconds())
            if self.transport is None or isinstance(
                self.transport, asyncio.DatagramTransport
            ):
                return
            logger.info("PING", extra={"addr": self.public_address})
            self.transport.write(PING)

    async def handle_registration(self, tx: RegisterTransaction) -> None:
        await tx
        self.on_registered()

    def data_received(self, data: bytes) -> None:
        self.recv_buffer.extend(data)
        for frame in self.extract_frames():
            self.dispatch_frame(frame)

    def datagram_received(self, data: bytes, addr: tuple) -> None:  # type: ignore[override]
        """Dispatch a complete UDP SIP datagram."""
        self.dispatch_frame(data)

    def error_received(self, exc: Exception) -> None:  # type: ignore[override]
        """Log a UDP transport error."""
        logger.warning("UDP error received", exc_info=exc)

    def extract_frames(self) -> typing.Generator[memoryview | bytes]:  # noqa: C901
        while self.recv_buffer:
            if self.recv_buffer[0:1] != b"\r":
                # SIP message: wait for the header-body separator.
                header_end = self.recv_buffer.find(b"\r\n\r\n")
                if header_end == -1:
                    break
                content_length = 0
                for line in self.recv_buffer[:header_end].split(b"\r\n")[1:]:
                    name, sep, value = line.partition(b":")
                    if sep and name.strip().lower() == b"content-length":
                        try:
                            content_length = int(value.strip())
                        except ValueError:
                            pass
                        break
                message_end = header_end + 4 + content_length
                if len(self.recv_buffer) < message_end:
                    break
                frame = memoryview(self.recv_buffer)[:message_end]
                yield frame
                frame.release()
                del self.recv_buffer[:message_end]
            elif len(self.recv_buffer) >= 4 and self.recv_buffer[:4] == PING:
                yield PING
                del self.recv_buffer[:4]
            elif len(self.recv_buffer) >= 3 and self.recv_buffer[2:3] == b"\r":
                # Third byte is CR – could be the start of PING; wait for the 4th.
                break
            elif self.recv_buffer[:2] == PONG:
                yield PONG
                del self.recv_buffer[:2]
            else:
                break

    def dispatch_frame(self, frame: memoryview | bytes) -> None:
        peer = NetworkAddress(*self.transport.get_extra_info("peername")[:2])
        if frame == PONG:
            logger.info("PONG", extra={"addr": peer})
        elif frame == PING:
            logger.info("PING", extra={"addr": peer})
            if self.transport and not isinstance(
                self.transport, asyncio.DatagramTransport
            ):
                logger.info("PONG", extra={"addr": self.public_address})
                self.transport.write(PONG)
        else:
            match Message.parse(bytes(frame)):
                case Request() as request:
                    logger.info(
                        "Request received: %r",
                        request,
                        extra={"addr": peer},
                    )
                    self.request_received(request)
                case Response() as response:
                    logger.info(
                        "Response received %r",
                        response,
                        extra={"addr": peer},
                    )
                    self.response_received(response)

    def send(self, message: Response | Request) -> None:
        """Serialize and send a SIP message over the active transport."""
        logger.debug("Sending %r", message)
        message.headers.setdefault("User-Agent", USER_AGENT)
        if self.transport is None:
            return
        if isinstance(self.transport, asyncio.DatagramTransport):
            self.transport.sendto(bytes(message))
        else:
            self.transport.write(bytes(message))

    def close(self) -> None:
        """Close the transport."""
        if self.transport is not None:
            self.transport.close()

    @property
    def allowed_methods(self) -> frozenset[SIPMethod]:
        """SIP methods supported by this UA."""
        return frozenset(
            {
                SIPMethod.INVITE,
                SIPMethod.ACK,
                SIPMethod.BYE,
                SIPMethod.CANCEL,
                SIPMethod.OPTIONS,
            }
        )

    @property
    def allow_header(self) -> str:
        """Comma-separated Allow header value in SIPMethod enum order."""
        return ",".join(m for m in SIPMethod if m in self.allowed_methods)

    def method_not_allowed(self, request: Request) -> None:
        """Respond with 405 Method Not Allowed.

        Override to customise the error response or add logging.

        Args:
            request: The unhandled SIP request.
        """
        logger.warning("SIP method %r is not supported", request.method)
        dialog_headers = {
            key: value
            for key, value in request.headers.items()
            if key in ("Via", "To", "From", "Call-ID", "CSeq")
        }
        self.send(
            Response(
                status_code=SIPStatus.METHOD_NOT_ALLOWED,
                phrase=SIPStatus.METHOD_NOT_ALLOWED.phrase,
                headers={**dialog_headers, "Allow": self.allow_header},
            ),
        )

    def request_received(self, request: Request) -> None:
        """Dispatch an incoming SIP request to the appropriate transaction."""
        match request.method:
            case SIPMethod.INVITE:
                asyncio.create_task(
                    InviteTransaction.receive(request=request, sip=self)
                )
            case SIPMethod.ACK:
                # For non-2xx ACKs the INVITE tx is still present; route by dialog.
                try:
                    tx = self.dialogs[
                        request.remote_tag, request.local_tag
                    ].invite_transaction
                except KeyError:
                    self.send(
                        Response.from_request(
                            request,
                            status_code=SIPStatus.GONE,
                            phrase=SIPStatus.GONE.phrase,
                        )
                    )
                else:
                    tx.ack_received(request)
            case SIPMethod.BYE:
                asyncio.create_task(ByeTransaction.receive(request=request, sip=self))
            case SIPMethod.CANCEL:
                try:
                    tx = self.transactions[request.branch]
                except KeyError:
                    self.send(
                        Response.from_request(
                            request,
                            status_code=SIPStatus.GONE,
                            phrase=SIPStatus.GONE.phrase,
                        )
                    )
                    return
                tx.cancel_received(request)
            case SIPMethod.OPTIONS:
                self.send(
                    Response.from_request(
                        request,
                        status_code=SIPStatus.OK,
                        phrase=SIPStatus.OK.phrase,
                        headers={"Allow": self.allow_header},
                    )
                )
            case _:
                self.method_not_allowed(request)

    def response_received(self, response: Response) -> None:
        """Dispatch an incoming SIP response to its originating transaction.

        Args:
            response: The parsed SIP response.
        """
        try:
            tx = self.transactions[response.branch]
        except KeyError:
            logger.warning(
                "Received response with unknown branch %r: %r",
                response.branch,
                response,
            )
        else:
            tx.response_received(response)

    def on_registered(self) -> None:
        """Handle successful carrier registration.

        Override in subclasses to initiate outbound calls or start other
        post-registration activity. The base implementation is a no-op.
        """
        self.registered_event.set()

    @property
    def contact(self) -> str:
        """Return a `Contact:` header value for this UA.

        The `ob` parameter ([RFC 5626 §5]) advertises outbound keep-alive
        support to the registrar for TCP/TLS transports.

        [RFC 5626 §5]: https://datatracker.ietf.org/doc/html/rfc5626#section-5
        """
        address = (
            f"{self.aor.user}@{self.public_address}"
            if self.aor.user
            else str(self.public_address)
        )
        if self.aor.scheme == "sips":
            return f"<sips:{address};ob>"
        if isinstance(self.transport, asyncio.DatagramTransport):
            return f"<sip:{address};transport=udp>"
        transport_param = "tls" if self.is_secure else "tcp"
        return f"<sip:{address};transport={transport_param};ob>"

    def connection_lost(self, exc: Exception | None) -> None:
        """Respond to a lost or closed transport connection."""
        if exc is not None:
            logger.exception("Connection lost", exc_info=exc)
        if self.keepalive_task is not None:
            self.keepalive_task.cancel()
            self.keepalive_task = None
        self.transport = None
        self.disconnected_event.set()

run(aor, dialog_class, *, rtp=None, no_verify_tls=False, stun_server=None, **kwargs) async classmethod

Connect to the SIP proxy and return once registered.

Suspend until SIP registration is confirmed before returning the ready protocol. After this call returns, the caller may safely place outbound calls or start an MCP server.

The transport protocol (TLS vs. plain TCP) and proxy address are read from aor directly — no extra arguments are needed.

Parameters:

Name	Type	Description	Default
aor	SipURI	SIP Address of Record, e.g. sip:alice@carrier.example. The host, port, and transport parameter are used to connect to the SIP proxy.	required
dialog_class	type[Dialog]	Dialog subclass used for inbound calls. Defaults to the base Dialog, which rejects all calls.	required
rtp	RealtimeTransportProtocol | None	Existing RTP endpoint to reuse. When None (default) a new datagram endpoint is created from aor and stun_server. Pass an existing instance to share one endpoint across reconnections (see serve).	None
no_verify_tls	bool	Disable TLS certificate verification. Insecure; for testing only. Defaults to False.	False
stun_server	NetworkAddress | None	STUN server for RTP NAT traversal. Ignored when rtp is provided.	None
**kwargs	Any	Extra keyword arguments forwarded to the protocol constructor.	{}

Returns:

Type	Description
SessionInitiationProtocol	The registered SessionInitiationProtocol
SessionInitiationProtocol	instance, ready to place calls.

Source code in voip/sip/protocol.py

@classmethod
async def run(
    cls,
    aor: types.SipURI,
    dialog_class: type[Dialog],
    *,
    rtp: RealtimeTransportProtocol | None = None,
    no_verify_tls: bool = False,
    stun_server: NetworkAddress | None = None,
    **kwargs: typing.Any,
) -> SessionInitiationProtocol:
    """Connect to the SIP proxy and return once registered.

    Suspend until SIP registration is confirmed before returning the ready
    protocol. After this call returns, the caller may safely place outbound
    calls or start an MCP server.

    The transport protocol (TLS vs. plain TCP) and proxy address are read
    from *aor* directly — no extra arguments are needed.

    Args:
        aor: SIP Address of Record, e.g. `sip:alice@carrier.example`.
            The host, port, and `transport` parameter are used to connect
            to the SIP proxy.
        dialog_class: [`Dialog`][voip.sip.Dialog] subclass used for
            inbound calls.  Defaults to the base
            [`Dialog`][voip.sip.Dialog], which rejects all calls.
        rtp: Existing RTP endpoint to reuse.  When `None` (default) a
            new datagram endpoint is created from *aor* and *stun_server*.
            Pass an existing instance to share one endpoint across
            reconnections (see [`serve`][voip.sip.protocol.SessionInitiationProtocol.serve]).
        no_verify_tls: Disable TLS certificate verification. Insecure; for
            testing only. Defaults to `False`.
        stun_server: STUN server for RTP NAT traversal. Ignored when *rtp*
            is provided.
        **kwargs: Extra keyword arguments forwarded to the protocol constructor.

    Returns:
        The registered [`SessionInitiationProtocol`][voip.sip.protocol.SessionInitiationProtocol]
        instance, ready to place calls.
    """
    loop = asyncio.get_running_loop()
    if rtp is None:
        addr_info = socket.getaddrinfo(
            str(aor.maddr[0]), aor.maddr[1], type=socket.SOCK_DGRAM
        )
        rtp_bind_address = (
            "::" if addr_info[0][0] == socket.AF_INET6 else "0.0.0.0"  # noqa: S104
        )
        rtp = await RealtimeTransportProtocol.serve(rtp_bind_address, stun_server)
    if aor.transport == "UDP":
        _, protocol = await loop.create_datagram_endpoint(
            lambda: cls(aor=aor, rtp=rtp, dialog_class=dialog_class, **kwargs),
            remote_addr=(str(aor.maddr[0]), aor.maddr[1]),
        )
    else:
        ssl_context: ssl.SSLContext | None = None
        if aor.transport == "TLS":
            ssl_context = ssl.create_default_context()
            if no_verify_tls:
                ssl_context.check_hostname = False
                ssl_context.verify_mode = ssl.CERT_NONE
        _, protocol = await loop.create_connection(
            lambda: cls(aor=aor, rtp=rtp, dialog_class=dialog_class, **kwargs),
            host=str(aor.maddr[0]),
            port=aor.maddr[1],
            ssl=ssl_context,
        )
    await protocol.registered_event.wait()
    return protocol

serve(aor, dialog_class, *, rtp=None, no_verify_tls=False, stun_server=None, **kwargs) async classmethod

Register with a carrier and handle inbound calls, reconnecting on disconnect.

Enter a persistent loop: connect to the SIP proxy, wait for the connection to drop, and reconnect with exponential back-off. Use this for long-running inbound-call servers.

The transport protocol (TLS vs. plain TCP) and proxy address are read from aor directly.

Parameters:

Name	Type	Description	Default
aor	SipURI	SIP Address of Record, e.g. sip:alice@carrier.example.	required
dialog_class	type[Dialog]	Dialog subclass used for inbound calls.	required
rtp	RealtimeTransportProtocol | None	Existing RTP endpoint to reuse. When None (default) a new datagram endpoint is created from aor and stun_server. Pass an existing instance to share one endpoint across reconnections (see serve).	None
no_verify_tls	bool	Disable TLS certificate verification. Insecure; for testing only. Defaults to False.	False
stun_server	NetworkAddress | None	STUN server for RTP NAT traversal.	None
**kwargs	Any	Extra keyword arguments forwarded to the protocol constructor.	{}

Source code in voip/sip/protocol.py

@classmethod
async def serve(
    cls,
    aor: types.SipURI,
    dialog_class: type[Dialog],
    *,
    rtp: RealtimeTransportProtocol | None = None,
    no_verify_tls: bool = False,
    stun_server: NetworkAddress | None = None,
    **kwargs: typing.Any,
) -> None:
    """Register with a carrier and handle inbound calls, reconnecting on disconnect.

    Enter a persistent loop: connect to the SIP proxy, wait for the
    connection to drop, and reconnect with exponential back-off. Use this
    for long-running inbound-call servers.

    The transport protocol (TLS vs. plain TCP) and proxy address are read from
    *aor* directly.

    Args:
        aor: SIP Address of Record, e.g. `sip:alice@carrier.example`.
        dialog_class: [`Dialog`][voip.sip.Dialog] subclass used for
            inbound calls.
        rtp: Existing RTP endpoint to reuse.  When `None` (default) a
            new datagram endpoint is created from *aor* and *stun_server*.
            Pass an existing instance to share one endpoint across
            reconnections (see [`serve`][voip.sip.protocol.SessionInitiationProtocol.serve]).
        no_verify_tls: Disable TLS certificate verification. Insecure; for
            testing only. Defaults to `False`.
        stun_server: STUN server for RTP NAT traversal.
        **kwargs: Extra keyword arguments forwarded to the protocol constructor.
    """
    addr_info = socket.getaddrinfo(
        str(aor.maddr[0]), aor.maddr[1], type=socket.SOCK_DGRAM
    )
    if rtp is None:
        rtp_bind_address = (
            "::" if addr_info[0][0] == socket.AF_INET6 else "0.0.0.0"  # noqa: S104
        )
        rtp = await RealtimeTransportProtocol.serve(rtp_bind_address, stun_server)
    backoff_secs = 1
    while True:
        try:
            protocol = await cls.run(
                aor, dialog_class, rtp=rtp, no_verify_tls=no_verify_tls, **kwargs
            )
            backoff_secs = 1
            await protocol.disconnected_event.wait()
            logger.info("SIP connection closed; reconnecting in %s s", backoff_secs)
        except (OSError, ssl.SSLError) as exc:
            logger.warning(
                "SIP connection failed (%s); retrying in %s s", exc, backoff_secs
            )
        await asyncio.sleep(backoff_secs)
        backoff_secs = min(backoff_secs * 2, 60)

Types

voip.sip.SipURI

Bases: str

A parsed SIP or SIPS URI per RFC 3261 §19.1.

Format: sip:user:password@host:port;uri-parameters?headers

Behaves as a plain str holding the canonical URI, so instances can be stored in header dicts unchanged. The parse classmethod decodes a raw SIP URI string into structured fields. IPv6 addresses in the host part must be enclosed in square brackets per RFC 2732 (e.g. sip:alice@[::1]:5060); the stored host is the bare address without brackets.

Examples:

>>> SipURI.parse("sip:alice@example.com")
'sip:alice@example.com:5060'
>>> SipURI.parse("sips:+15551234567@carrier.com:5061")
'sips:%2B15551234567@carrier.com:5061'
>>> SipURI.parse("sip:alice@[::1]:5060")
'sip:alice@[::1]:5060'

Source code in voip/sip/types.py

class SipURI(str):
    """A parsed SIP or SIPS URI per [RFC 3261 §19.1].

    Format: `sip:user:password@host:port;uri-parameters?headers`

    Behaves as a plain `str` holding the canonical URI, so instances can be
    stored in header dicts unchanged.  The `parse` classmethod decodes a raw
    SIP URI string into structured fields.  IPv6 addresses in the host part
    must be enclosed in square brackets per [RFC 2732]
    (e.g. `sip:alice@[::1]:5060`); the stored `host` is the bare address
    without brackets.

    [RFC 3261 §19.1]: https://datatracker.ietf.org/doc/html/rfc3261#section-19.1
    [RFC 2732]: https://datatracker.ietf.org/doc/html/rfc2732

    Examples:
        >>> SipURI.parse("sip:alice@example.com")
        'sip:alice@example.com:5060'
        >>> SipURI.parse("sips:+15551234567@carrier.com:5061")
        'sips:%2B15551234567@carrier.com:5061'
        >>> SipURI.parse("sip:alice@[::1]:5060")
        'sip:alice@[::1]:5060'

    """

    __slots__ = ("scheme", "host", "user", "password", "port", "parameters", "headers")

    SIP_URL_PATTERN: typing.ClassVar[re.Pattern[str]] = re.compile(
        r"^(?P<scheme>sips?):"
        r"((?P<user>[^@;:]+)(?P<password>:[^@;]*)?@)?"
        r"(?P<host>(\[[0-9a-fA-F:]+]|[^;?:@\[\]]+))"
        r"(?P<port>:[0-9]+)?"
        r"(?P<parameters>;[^?]+)?"
        r"(?P<headers>\?[^?]+)?$",
        re.IGNORECASE,
    )

    def __new__(
        cls,
        scheme: str,
        host: str | ipaddress.IPv6Address | ipaddress.IPv4Address,
        user: str | None = None,
        password: str | None = None,
        port: int | None = None,
        parameters: dict[str, str] | None = None,
        headers: dict[str, str] | None = None,
    ) -> SipURI:
        try:
            host = ipaddress.ip_address(host)
        except ValueError:
            pass
        port = port if port is not None else (5061 if scheme == "sips" else 5060)
        parameters = parameters or {}
        headers = headers or {}
        parts = [f"{scheme}:"]
        if user:
            parts.append(urllib.parse.quote(user))
            if password:
                parts.append(f":{urllib.parse.quote(password)}")
            parts.append("@")
        parts.append(
            f"[{host}]" if isinstance(host, ipaddress.IPv6Address) else str(host)
        )
        parts.append(f":{port}")
        for name, val in parameters.items():
            parts.append(
                f";{urllib.parse.quote(name)}={urllib.parse.quote(val)}"
                if val is not None
                else f";{urllib.parse.quote(name)}"
            )
        if headers:
            parts.append("?")
            parts.append(
                "&".join(
                    f"{urllib.parse.quote(name)}={urllib.parse.quote(val)}"
                    for name, val in headers.items()
                )
            )
        instance = super().__new__(cls, "".join(parts))
        instance.scheme = scheme
        instance.host = host
        instance.user = user
        instance.password = password
        instance.port = port
        instance.parameters = parameters
        instance.headers = headers
        return instance

    @classmethod
    def parse(cls, value: str) -> SipURI:
        """
        Parse a SIP or SIPS URI string into a `SipUri` instance.

        Returns:
            Parsed `SipUri` instance.

        Raises:
            ValueError: When the URI is malformed (missing scheme, invalid
                characters, unclosed IPv6 bracket, empty host, or invalid port).
        """
        if match := cls.SIP_URL_PATTERN.fullmatch(value):
            host = match.group("host")
            if host.startswith("[") and host.endswith("]"):
                host = host[1:-1]
            host = urllib.parse.unquote(host)

            return cls(
                scheme=match.group("scheme").lower(),
                user=urllib.parse.unquote(match.group("user"))
                if match.group("user")
                else None,
                host=host,
                password=urllib.parse.unquote(match.group("password")[1:])
                if match.group("password")
                else None,
                port=int(match.group("port")[1:]) if match.group("port") else None,
                parameters=dict(cls.parse_parameters(match.group("parameters")))
                if match.group("parameters")
                else {},
                headers=dict(cls.parse_headers(match.group("headers")[1:]))
                if match.group("headers")
                else {},
            )
        raise ValueError(f"Invalid SIP URI: {value!r}")

    @classmethod
    def parse_parameters(cls, params: str) -> Iterator[tuple[str, str | None]]:
        for part in params[1:].split(";"):
            if "=" in part:
                name, val = part.split("=", 1)
                yield urllib.parse.unquote(name), urllib.parse.unquote(val)
            elif part:
                yield urllib.parse.unquote(part), None

    @classmethod
    def parse_headers(cls, headers: str) -> Iterator[tuple[str, str]]:
        for part in headers.split("&"):
            if "=" in part:
                name, val = part.split("=", 1)
                yield urllib.parse.unquote(name), urllib.parse.unquote(val)
            elif part:
                yield urllib.parse.unquote(part), ""

    @property
    def maddr(self) -> NetworkAddress:
        try:
            return NetworkAddress.parse(self.parameters["maddr"])
        except KeyError:
            return NetworkAddress(self.host, self.port)

    @property
    def ttl(self) -> int | None:
        try:
            return int(self.parameters["ttl"])
        except KeyError:
            return None

    @property
    def transport(self) -> str:
        return (
            self.parameters.get("transport", "UDP").upper()
            if self.scheme == "sip"
            else "TLS"
        )

parse(value) classmethod

Parse a SIP or SIPS URI string into a SipUri instance.

Returns:

Type	Description
SipURI	Parsed SipUri instance.

Raises:

Type	Description
ValueError	When the URI is malformed (missing scheme, invalid characters, unclosed IPv6 bracket, empty host, or invalid port).

Source code in voip/sip/types.py

@classmethod
def parse(cls, value: str) -> SipURI:
    """
    Parse a SIP or SIPS URI string into a `SipUri` instance.

    Returns:
        Parsed `SipUri` instance.

    Raises:
        ValueError: When the URI is malformed (missing scheme, invalid
            characters, unclosed IPv6 bracket, empty host, or invalid port).
    """
    if match := cls.SIP_URL_PATTERN.fullmatch(value):
        host = match.group("host")
        if host.startswith("[") and host.endswith("]"):
            host = host[1:-1]
        host = urllib.parse.unquote(host)

        return cls(
            scheme=match.group("scheme").lower(),
            user=urllib.parse.unquote(match.group("user"))
            if match.group("user")
            else None,
            host=host,
            password=urllib.parse.unquote(match.group("password")[1:])
            if match.group("password")
            else None,
            port=int(match.group("port")[1:]) if match.group("port") else None,
            parameters=dict(cls.parse_parameters(match.group("parameters")))
            if match.group("parameters")
            else {},
            headers=dict(cls.parse_headers(match.group("headers")[1:]))
            if match.group("headers")
            else {},
        )
    raise ValueError(f"Invalid SIP URI: {value!r}")

voip.sip.CallerID

Bases: str

SIP From/To header value with structured access and privacy-safe repr.

Behaves as a plain str so it is wire-format compatible and can be stored in header dicts unchanged. repr() returns a short anonymized form that shows only the last four characters of the user part and the carrier domain — useful for log messages.

Examples:

>>> str(CallerID('"08001234567" <sip:08001234567@telefonica.de>;tag=abc'))
'"08001234567" <sip:08001234567@telefonica.de>;tag=abc'
>>> repr(CallerID('"08001234567" <sip:08001234567@telefonica.de>;tag=abc'))
'***4567@telefonica.de'
>>> repr(CallerID('sip:alice@example.com'))
'*lice@example.com'

Source code in voip/sip/types.py

class CallerID(str):
    """SIP From/To header value with structured access and privacy-safe repr.

    Behaves as a plain `str` so it is wire-format compatible and can be
    stored in header dicts unchanged.  `repr()` returns a short anonymized
    form that shows only the last four characters of the user part and the
    carrier domain — useful for log messages.

    Examples:
        >>> str(CallerID('"08001234567" <sip:08001234567@telefonica.de>;tag=abc'))
        '"08001234567" <sip:08001234567@telefonica.de>;tag=abc'
        >>> repr(CallerID('"08001234567" <sip:08001234567@telefonica.de>;tag=abc'))
        '***4567@telefonica.de'
        >>> repr(CallerID('sip:alice@example.com'))
        '*lice@example.com'
    """

    @property
    def display_name(self) -> str | None:
        """Display name from the From/To header, if present."""
        if m := re.match(r'^"([^"]+)"\s*<|^([^<"]+?)\s*<', self):
            return (m.group(1) or m.group(2) or "").strip() or None
        return None

    @property
    def uri(self) -> SipURI | None:
        """Parsed SIP or tel URI embedded in the header value, if present."""
        if m := re.search(r"<?((?:sips?|tel):[^>\s]+)>?", self):
            return SipURI.parse(m.group(1))

    @property
    def user(self) -> str | None:
        """SIP user part (phone number or username)."""
        if m := re.search(r"sips?:([^@>;\s]+)@", self):
            return m.group(1)

    @property
    def host(self) -> str | None:
        """Carrier domain extracted from the SIP URI."""
        if m := re.search(r"sips?:[^@>;\s]+@([^>;)\s,]+)", self):
            return m.group(1)

    @property
    def tag(self) -> str | None:
        """Dialog tag parameter value, if present."""
        m = re.search(r";tag=([^\s;]+)", self)
        return m.group(1) if m else None

    def __repr__(self) -> str:
        user = self.display_name or self.user or ""
        host = self.host or ""
        masked = ("*" * max(0, len(user) - 4)) + user[-4:] if user else "****"
        return f"{masked}@{host}" if host else masked

display_name property

Display name from the From/To header, if present.

host property

Carrier domain extracted from the SIP URI.

tag property

Dialog tag parameter value, if present.

uri property

Parsed SIP or tel URI embedded in the header value, if present.

user property

SIP user part (phone number or username).

voip.sip.SIPStatus

Bases: IntEnum

SIP Status Codes based on RFC 3261.

Source code in voip/sip/types.py

class SIPStatus(enum.IntEnum):
    """SIP Status Codes based on [RFC 3261].

    [RFC 3261]: https://datatracker.ietf.org/doc/html/rfc3261#section-21
    """

    def __new__(cls, value: int, phrase: str) -> SIPStatus:
        obj = int.__new__(cls, value)
        obj._value_ = value
        obj.phrase = phrase
        return obj

    TRYING = 100, "Trying"
    """The request is being processed. No final response is available yet."""

    RINGING = 180, "Ringing"
    """The called party is being alerted of the call."""

    CALL_IS_BEING_FORWARDED = 181, "Call Is Being Forwarded"
    """The called party is being alerted of the call, but the call is not yet established."""

    QUEUED = 182, "Queued"
    """The called party is being alerted of the call, but the call is not yet established."""

    SESSION_PROGRESS = 183, "Session Progress"
    """The called party is being alerted of the call, but the call is not yet established."""

    OK = 200, "OK"
    """The request has succeeded."""

    MULTIPLE_CHOICES = 300, "Multiple Choices"
    """The requested resource has multiple representations, each with its own specific location."""

    MOVED_PERMANENTLY = 301, "Moved Permanently"
    """The requested resource has been assigned a new permanent URI and any future references to this resource ought to use one of the returned URIs."""

    MOVED_TEMPORARILY = 302, "Moved Temporarily"
    """The requested resource is temporarily unavailable and the server is asking the client to try again later."""

    USE_PROXY = 305, "Use Proxy"
    """The requested resource is available only through a proxy, the address for which is provided in the response."""

    ALTERNATIVE_SERVICE = 380, "Alternative Service"
    """The server has fulfilled a request for the service indicated by the URI."""

    BAD_REQUEST = 400, "Bad Request"
    """The request has bad syntax or cannot be fulfilled due to bad syntax."""

    UNAUTHORIZED = 401, "Unauthorized"
    """The request requires user authentication."""

    PAYMENT_REQUIRED = 402, "Payment Required"
    """Further action is required."""

    FORBIDDEN = 403, "Forbidden"
    """The server understood the request but refuses to fulfill it."""

    NOT_FOUND = 404, "Not Found"
    """The requested resource could not be found."""

    METHOD_NOT_ALLOWED = 405, "Method Not Allowed"
    """The method specified in the Request-URI is not allowed for the resource identified by the request URI."""

    NOT_ACCEPTABLE = 406, "Not Acceptable"
    """The server cannot produce a response matching the Accept headers."""

    PROXY_AUTHENTICATION_REQUIRED = 407, "Proxy Authentication Required"
    """The client must authenticate itself with the proxy."""

    REQUEST_TIMEOUT = 408, "Request Timeout"
    """The server timed out waiting for the request."""

    GONE = 410, "Gone"
    """The requested resource is no longer available at the server and no longer exists."""

    REQUEST_ENTITY_TOO_LARGE = 413, "Request Entity Too Large"
    """The server will not accept the request, because the entity of the request is too large."""

    REQUEST_URI_TOO_LONG = 414, "Request-URI Too Long"
    """The server will not accept the request, because the Request-URI is too long."""

    UNSUPPORTED_MEDIA_TYPE = 415, "Unsupported Media Type"
    """The server will not accept the request, because the media type of the request is unsupported."""

    UNSUPPORTED_URI_SCHEME = 416, "Unsupported URI Scheme"
    """The server will not accept the request, because the URI scheme of the request is unsupported."""

    BAD_EXTENSION = 420, "Bad Extension"
    """Server does not recognize any of the parameters needed to understand the request."""

    EXTENSION_REQUIRED = 421, "Extension Required"
    """Server requires the client to identify itself (usually, using the Contact header field) before proceeding."""

    INTERVAL_TOO_BRIEF = 423, "Interval Too Brief"
    """Server is unwilling to process the request because header fields are too large."""

    TEMPORARILY_UNAVAILABLE = 480, "Temporarily Unavailable"
    """Server is currently unable to process the request due to temporary overloading or maintenance."""

    CALL_TRANSACTION_DOES_NOT_EXIST = 481, "Call/Transaction Does Not Exist"
    """Server has received a final response for the transaction it is still attempting to complete."""

    LOOP_DETECTED = 482, "Loop Detected"
    """Server has detected an infinite loop while processing the request."""

    TOO_MANY_HOPS = 483, "Too Many Hops"
    """Server has exceeded the maximum number of hops allowed in the request URI."""

    ADDRESS_INCOMPLETE = 484, "Address Incomplete"
    """Server has received a final response for the transaction it is still attempting to complete, but has an invalid value for one or more header fields."""

    AMBIGUOUS = 485, "Ambiguous"
    """Server cannot decide on a response because multiple responses are possible."""

    BUSY_HERE = 486, "Busy Here"
    """Server is busy."""

    REQUEST_TERMINATED = 487, "Request Terminated"
    """Server has received a final response for the transaction it is still attempting to complete, but has received a termination request for it from the client."""

    NOT_ACCEPTABLE_HERE = 488, "Not Acceptable Here"
    """Server is unable to produce a response acceptable to the client according to proactive negotiation, and is unwilling to supply a default reason phrase."""

    REQUEST_PENDING = 491, "Request Pending"
    """Server has received a final response for the transaction it is still attempting to complete, but has not yet delivered it to the client."""

    UNDECIPHERABLE = 493, "Undecipherable"
    """Server was unable to decrypt a message after performing necessary decryption(s)."""

    SERVER_INTERNAL_ERROR = 500, "Server Internal Error"
    """The server encountered an unexpected condition which prevented it from fulfilling the request."""

    NOT_IMPLEMENTED = 501, "Not Implemented"
    """The server does not support the functionality required to fulfill the request."""

    BAD_GATEWAY = 502, "Bad Gateway"
    """The server, while acting as a gateway or proxy, received an invalid response from the upstream server it accessed in attempting to fulfill the request."""

    SERVICE_UNAVAILABLE = 503, "Service Unavailable"
    """Server is currently unable to process the request due to temporary overloading or maintenance."""

    SERVER_TIME_OUT = 504, "Server Time-out"
    """The server, while acting as a gateway or proxy, did not receive a timely response from the upstream server specified by the URI (e.g., HTTP, FTP, LDAP) or some other auxiliary server (e.g., DNS) it needed to access in attempting to complete the request."""

    VERSION_NOT_SUPPORTED = 505, "Version Not Supported"
    """Server does not support, or refuses to support, the protocol version used in the request."""

    MESSAGE_TOO_LARGE = 513, "Message Too Large"
    """The server is unwilling to process the request because its header fields are too large."""

    BUSY_EVERYWHERE = 600, "Busy Everywhere"
    """The server is not able to process the request because it is busy. For example, this error might be given if a server is overloaded with requests and is unable to process one of the requests."""

    DECLINE = 603, "Decline"
    """The call has been declined."""

    DOES_NOT_EXIST_ANYWHERE = 604, "Does Not Exist Anywhere"
    """Server has received a final response for the transaction it is still attempting to complete, but has received a termination request for it from a server it does not control."""

    NOT_ACCEPTABLE_ANYWHERE = 606, "Not Acceptable"
    """The server is not able to produce a response which is acceptable to the client, according to the proactive negotiation header fields received in the request, and the server is unwilling to supply a default reason phrase."""

ADDRESS_INCOMPLETE = (484, 'Address Incomplete') class-attribute instance-attribute

Server has received a final response for the transaction it is still attempting to complete, but has an invalid value for one or more header fields.

ALTERNATIVE_SERVICE = (380, 'Alternative Service') class-attribute instance-attribute

The server has fulfilled a request for the service indicated by the URI.

AMBIGUOUS = (485, 'Ambiguous') class-attribute instance-attribute

Server cannot decide on a response because multiple responses are possible.

BAD_EXTENSION = (420, 'Bad Extension') class-attribute instance-attribute

Server does not recognize any of the parameters needed to understand the request.

BAD_GATEWAY = (502, 'Bad Gateway') class-attribute instance-attribute

The server, while acting as a gateway or proxy, received an invalid response from the upstream server it accessed in attempting to fulfill the request.

BAD_REQUEST = (400, 'Bad Request') class-attribute instance-attribute

The request has bad syntax or cannot be fulfilled due to bad syntax.

BUSY_EVERYWHERE = (600, 'Busy Everywhere') class-attribute instance-attribute

The server is not able to process the request because it is busy. For example, this error might be given if a server is overloaded with requests and is unable to process one of the requests.

BUSY_HERE = (486, 'Busy Here') class-attribute instance-attribute

Server is busy.

CALL_IS_BEING_FORWARDED = (181, 'Call Is Being Forwarded') class-attribute instance-attribute

The called party is being alerted of the call, but the call is not yet established.

CALL_TRANSACTION_DOES_NOT_EXIST = (481, 'Call/Transaction Does Not Exist') class-attribute instance-attribute

Server has received a final response for the transaction it is still attempting to complete.

DECLINE = (603, 'Decline') class-attribute instance-attribute

The call has been declined.

DOES_NOT_EXIST_ANYWHERE = (604, 'Does Not Exist Anywhere') class-attribute instance-attribute

Server has received a final response for the transaction it is still attempting to complete, but has received a termination request for it from a server it does not control.

EXTENSION_REQUIRED = (421, 'Extension Required') class-attribute instance-attribute

Server requires the client to identify itself (usually, using the Contact header field) before proceeding.

FORBIDDEN = (403, 'Forbidden') class-attribute instance-attribute

The server understood the request but refuses to fulfill it.

GONE = (410, 'Gone') class-attribute instance-attribute

The requested resource is no longer available at the server and no longer exists.

INTERVAL_TOO_BRIEF = (423, 'Interval Too Brief') class-attribute instance-attribute

Server is unwilling to process the request because header fields are too large.

LOOP_DETECTED = (482, 'Loop Detected') class-attribute instance-attribute

Server has detected an infinite loop while processing the request.

MESSAGE_TOO_LARGE = (513, 'Message Too Large') class-attribute instance-attribute

The server is unwilling to process the request because its header fields are too large.

METHOD_NOT_ALLOWED = (405, 'Method Not Allowed') class-attribute instance-attribute

The method specified in the Request-URI is not allowed for the resource identified by the request URI.

MOVED_PERMANENTLY = (301, 'Moved Permanently') class-attribute instance-attribute

The requested resource has been assigned a new permanent URI and any future references to this resource ought to use one of the returned URIs.

MOVED_TEMPORARILY = (302, 'Moved Temporarily') class-attribute instance-attribute

The requested resource is temporarily unavailable and the server is asking the client to try again later.

MULTIPLE_CHOICES = (300, 'Multiple Choices') class-attribute instance-attribute

The requested resource has multiple representations, each with its own specific location.

NOT_ACCEPTABLE = (406, 'Not Acceptable') class-attribute instance-attribute

The server cannot produce a response matching the Accept headers.

NOT_ACCEPTABLE_ANYWHERE = (606, 'Not Acceptable') class-attribute instance-attribute

The server is not able to produce a response which is acceptable to the client, according to the proactive negotiation header fields received in the request, and the server is unwilling to supply a default reason phrase.

NOT_ACCEPTABLE_HERE = (488, 'Not Acceptable Here') class-attribute instance-attribute

Server is unable to produce a response acceptable to the client according to proactive negotiation, and is unwilling to supply a default reason phrase.

NOT_FOUND = (404, 'Not Found') class-attribute instance-attribute

The requested resource could not be found.

NOT_IMPLEMENTED = (501, 'Not Implemented') class-attribute instance-attribute

The server does not support the functionality required to fulfill the request.

OK = (200, 'OK') class-attribute instance-attribute

The request has succeeded.

PAYMENT_REQUIRED = (402, 'Payment Required') class-attribute instance-attribute

Further action is required.

PROXY_AUTHENTICATION_REQUIRED = (407, 'Proxy Authentication Required') class-attribute instance-attribute

The client must authenticate itself with the proxy.

QUEUED = (182, 'Queued') class-attribute instance-attribute

The called party is being alerted of the call, but the call is not yet established.

REQUEST_ENTITY_TOO_LARGE = (413, 'Request Entity Too Large') class-attribute instance-attribute

The server will not accept the request, because the entity of the request is too large.

REQUEST_PENDING = (491, 'Request Pending') class-attribute instance-attribute

Server has received a final response for the transaction it is still attempting to complete, but has not yet delivered it to the client.

REQUEST_TERMINATED = (487, 'Request Terminated') class-attribute instance-attribute

Server has received a final response for the transaction it is still attempting to complete, but has received a termination request for it from the client.

REQUEST_TIMEOUT = (408, 'Request Timeout') class-attribute instance-attribute

The server timed out waiting for the request.

REQUEST_URI_TOO_LONG = (414, 'Request-URI Too Long') class-attribute instance-attribute

The server will not accept the request, because the Request-URI is too long.

RINGING = (180, 'Ringing') class-attribute instance-attribute

The called party is being alerted of the call.

SERVER_INTERNAL_ERROR = (500, 'Server Internal Error') class-attribute instance-attribute

The server encountered an unexpected condition which prevented it from fulfilling the request.

SERVER_TIME_OUT = (504, 'Server Time-out') class-attribute instance-attribute

The server, while acting as a gateway or proxy, did not receive a timely response from the upstream server specified by the URI (e.g., HTTP, FTP, LDAP) or some other auxiliary server (e.g., DNS) it needed to access in attempting to complete the request.

SERVICE_UNAVAILABLE = (503, 'Service Unavailable') class-attribute instance-attribute

Server is currently unable to process the request due to temporary overloading or maintenance.

SESSION_PROGRESS = (183, 'Session Progress') class-attribute instance-attribute

The called party is being alerted of the call, but the call is not yet established.

TEMPORARILY_UNAVAILABLE = (480, 'Temporarily Unavailable') class-attribute instance-attribute

Server is currently unable to process the request due to temporary overloading or maintenance.

TOO_MANY_HOPS = (483, 'Too Many Hops') class-attribute instance-attribute

Server has exceeded the maximum number of hops allowed in the request URI.

TRYING = (100, 'Trying') class-attribute instance-attribute

The request is being processed. No final response is available yet.

UNAUTHORIZED = (401, 'Unauthorized') class-attribute instance-attribute

The request requires user authentication.

UNDECIPHERABLE = (493, 'Undecipherable') class-attribute instance-attribute

Server was unable to decrypt a message after performing necessary decryption(s).

UNSUPPORTED_MEDIA_TYPE = (415, 'Unsupported Media Type') class-attribute instance-attribute

The server will not accept the request, because the media type of the request is unsupported.

UNSUPPORTED_URI_SCHEME = (416, 'Unsupported URI Scheme') class-attribute instance-attribute

The server will not accept the request, because the URI scheme of the request is unsupported.

USE_PROXY = (305, 'Use Proxy') class-attribute instance-attribute

The requested resource is available only through a proxy, the address for which is provided in the response.

VERSION_NOT_SUPPORTED = (505, 'Version Not Supported') class-attribute instance-attribute

Server does not support, or refuses to support, the protocol version used in the request.

voip.sdp.types

SDP field types as defined by RFC 4566.

Attribute dataclass

Bases: ByteSerializableObject

Attribute field (a=) as defined by RFC 4566 §5.13.

Source code in voip/sdp/types.py

@dataclasses.dataclass(slots=True, frozen=True)
class Attribute(ByteSerializableObject):
    """Attribute field (a=) as defined by RFC 4566 §5.13."""

    letter: ClassVar[str] = "a"
    session_attr: ClassVar[str] = "attributes"
    is_list: ClassVar[bool] = True
    media_attr: ClassVar[str | None] = "attributes"

    name: str
    value: str | None = None

    def __bytes__(self) -> bytes:
        match self.value:
            case None:
                return self.name.encode()
            case _:
                return f"{self.name}:{self.value}".encode()

    @classmethod
    def parse(cls, data: bytes | str) -> Attribute:
        value = data.decode() if isinstance(data, bytes) else data
        name, _, attr_value = value.partition(":")
        return cls(name=name, value=attr_value or None)

Bandwidth dataclass

Bases: ByteSerializableObject

Bandwidth field (b=) as defined by RFC 4566 §5.8.

Source code in voip/sdp/types.py

@dataclasses.dataclass(slots=True, frozen=True)
class Bandwidth(ByteSerializableObject):
    """Bandwidth field (b=) as defined by RFC 4566 §5.8."""

    letter: ClassVar[str] = "b"
    session_attr: ClassVar[str] = "bandwidths"
    is_list: ClassVar[bool] = True
    media_attr: ClassVar[str | None] = "bandwidths"

    bwtype: str
    bandwidth: int

    def __bytes__(self) -> bytes:
        return f"{self.bwtype}:{self.bandwidth}".encode()

    @classmethod
    def parse(cls, data: bytes | str) -> Bandwidth:
        value = data.decode() if isinstance(data, bytes) else data
        bwtype, _, bandwidth = value.partition(":")
        return cls(bwtype=bwtype, bandwidth=int(bandwidth))

ConnectionData dataclass

Bases: ByteSerializableObject

Connection data field (c=) as defined by RFC 4566 §5.7.

Source code in voip/sdp/types.py

@dataclasses.dataclass(slots=True, frozen=True)
class ConnectionData(ByteSerializableObject):
    """Connection data field (c=) as defined by RFC 4566 §5.7."""

    letter: ClassVar[str] = "c"
    session_attr: ClassVar[str] = "connection"
    is_list: ClassVar[bool] = False
    media_attr: ClassVar[str | None] = "connection"

    nettype: str
    addrtype: str
    connection_address: str

    def __bytes__(self) -> bytes:
        return f"{self.nettype} {self.addrtype} {self.connection_address}".encode()

    @classmethod
    def parse(cls, data: bytes | str) -> ConnectionData:
        value = data.decode() if isinstance(data, bytes) else data
        nettype, addrtype, connection_address = value.split(" ", 2)
        return cls(
            nettype=nettype,
            addrtype=addrtype,
            connection_address=connection_address,
        )

Field

Bases: Protocol

SDP field descriptor protocol.

Source code in voip/sdp/types.py

@runtime_checkable
class Field(Protocol):
    """SDP field descriptor protocol."""

    letter: str
    session_attr: str
    is_list: bool
    media_attr: str | None

    @staticmethod
    def parse(value: str) -> object:
        """Parse a raw SDP line value."""

parse(value) staticmethod

Parse a raw SDP line value.

Source code in voip/sdp/types.py

@staticmethod
def parse(value: str) -> object:
    """Parse a raw SDP line value."""

IntField dataclass

Descriptor for SDP fields parsing and serializing as integers.

Source code in voip/sdp/types.py

@dataclasses.dataclass(slots=True, frozen=True)
class IntField:
    """Descriptor for SDP fields parsing and serializing as integers."""

    letter: str
    session_attr: str
    is_list: bool = False
    media_attr: str | None = None

    @staticmethod
    def parse(value: str) -> int:
        return int(value)

MediaDescription dataclass

Bases: ByteSerializableObject

Media description section (m=) as defined by RFC 4566 §5.14.

Source code in voip/sdp/types.py

@dataclasses.dataclass(slots=True)
class MediaDescription(ByteSerializableObject):
    """Media description section (m=) as defined by RFC 4566 §5.14."""

    letter: ClassVar[str] = "m"
    session_attr: ClassVar[str] = "media"
    is_list: ClassVar[bool] = True
    media_attr: ClassVar[str | None] = None

    media: str
    port: int
    proto: str
    fmt: list[RTPPayloadFormat]
    title: str | None = None
    connection: ConnectionData | None = None
    bandwidths: list[Bandwidth] = dataclasses.field(default_factory=list)
    attributes: list[Attribute] = dataclasses.field(default_factory=list)

    def get_format(self, pt: int | str) -> RTPPayloadFormat | None:
        """Return the `RTPPayloadFormat` for payload type *pt*, or `None`."""
        target = int(pt)
        return next((f for f in self.fmt if f.payload_type == target), None)

    def apply_attribute(self, attr: Attribute) -> bool:
        """Apply a media-level `a=` attribute, returning `True` if consumed.

        `a=rtpmap` and `a=fmtp` update the matching `RTPPayloadFormat` entry;
        other attributes go to `attributes`.
        """
        if attr.name == "rtpmap" and attr.value is not None:
            rtpfmt = RTPPayloadFormat.parse(attr.value)
            for i, f in enumerate(self.fmt):
                if f.payload_type == rtpfmt.payload_type:
                    if f.fmtp is not None and rtpfmt.fmtp is None:
                        rtpfmt.fmtp = f.fmtp
                    self.fmt[i] = rtpfmt
                    break
            return True
        if attr.name == "fmtp" and attr.value is not None:
            pt_str, _, params = attr.value.partition(" ")
            try:
                pt = int(pt_str)
            except ValueError:
                return False
            for f in self.fmt:
                if f.payload_type == pt:
                    f.fmtp = params
                    return True
        return False

    def lines(self) -> Generator[str]:
        """Yield each SDP line in canonical field order."""
        yield f"m={self.media} {self.port} {self.proto} {' '.join(str(f.payload_type) for f in self.fmt)}"
        match self.title:
            case str() as title:
                yield f"i={title}"
        match self.connection:
            case ConnectionData() as connection:
                yield f"c={connection}"
        yield from (f"b={b}" for b in self.bandwidths)
        for fmt in self.fmt:
            match fmt.encoding_name, fmt.sample_rate:
                case (str(), int()):
                    yield f"a=rtpmap:{fmt}"
            match fmt.fmtp:
                case str() as fmtp:
                    yield f"a=fmtp:{fmt.payload_type} {fmtp}"
        yield from (f"a={a}" for a in self.attributes)

    def __bytes__(self) -> bytes:
        return "\r\n".join(self.lines()).encode()

    @classmethod
    def parse(cls, data: bytes | str) -> MediaDescription:
        value = data.decode() if isinstance(data, bytes) else data
        lines = value.splitlines()
        first = lines[0].rstrip("\r").removeprefix("m=")
        media_type, port_str, proto, *fmts = first.split()
        fmt = [RTPPayloadFormat.from_pt(int(pt)) for pt in fmts]
        obj = cls(media=media_type, port=int(port_str), proto=proto, fmt=fmt)
        for line in lines[1:]:
            line = line.rstrip("\r")
            if not line or "=" not in line:
                continue
            letter, _, attr_value = line.partition("=")
            match letter:
                case "i":
                    obj.title = attr_value
                case "c":
                    obj.connection = ConnectionData.parse(attr_value)
                case "b":
                    obj.bandwidths.append(Bandwidth.parse(attr_value))
                case "a":
                    attr = Attribute.parse(attr_value)
                    if not obj.apply_attribute(attr):
                        obj.attributes.append(attr)
        return obj

apply_attribute(attr)

Apply a media-level a= attribute, returning True if consumed.

a=rtpmap and a=fmtp update the matching RTPPayloadFormat entry; other attributes go to attributes.

Source code in voip/sdp/types.py

def apply_attribute(self, attr: Attribute) -> bool:
    """Apply a media-level `a=` attribute, returning `True` if consumed.

    `a=rtpmap` and `a=fmtp` update the matching `RTPPayloadFormat` entry;
    other attributes go to `attributes`.
    """
    if attr.name == "rtpmap" and attr.value is not None:
        rtpfmt = RTPPayloadFormat.parse(attr.value)
        for i, f in enumerate(self.fmt):
            if f.payload_type == rtpfmt.payload_type:
                if f.fmtp is not None and rtpfmt.fmtp is None:
                    rtpfmt.fmtp = f.fmtp
                self.fmt[i] = rtpfmt
                break
        return True
    if attr.name == "fmtp" and attr.value is not None:
        pt_str, _, params = attr.value.partition(" ")
        try:
            pt = int(pt_str)
        except ValueError:
            return False
        for f in self.fmt:
            if f.payload_type == pt:
                f.fmtp = params
                return True
    return False

get_format(pt)

Return the RTPPayloadFormat for payload type pt, or None.

Source code in voip/sdp/types.py

def get_format(self, pt: int | str) -> RTPPayloadFormat | None:
    """Return the `RTPPayloadFormat` for payload type *pt*, or `None`."""
    target = int(pt)
    return next((f for f in self.fmt if f.payload_type == target), None)

lines()

Yield each SDP line in canonical field order.

Source code in voip/sdp/types.py

def lines(self) -> Generator[str]:
    """Yield each SDP line in canonical field order."""
    yield f"m={self.media} {self.port} {self.proto} {' '.join(str(f.payload_type) for f in self.fmt)}"
    match self.title:
        case str() as title:
            yield f"i={title}"
    match self.connection:
        case ConnectionData() as connection:
            yield f"c={connection}"
    yield from (f"b={b}" for b in self.bandwidths)
    for fmt in self.fmt:
        match fmt.encoding_name, fmt.sample_rate:
            case (str(), int()):
                yield f"a=rtpmap:{fmt}"
        match fmt.fmtp:
            case str() as fmtp:
                yield f"a=fmtp:{fmt.payload_type} {fmtp}"
    yield from (f"a={a}" for a in self.attributes)

Origin dataclass

Bases: ByteSerializableObject

Origin field (o=) as defined by RFC 4566 §5.2.

Source code in voip/sdp/types.py

@dataclasses.dataclass(slots=True, frozen=True)
class Origin(ByteSerializableObject):
    """Origin field (o=) as defined by RFC 4566 §5.2."""

    letter: ClassVar[str] = "o"
    session_attr: ClassVar[str] = "origin"
    is_list: ClassVar[bool] = False
    media_attr: ClassVar[str | None] = None

    username: str
    sess_id: str
    sess_version: str
    nettype: str
    addrtype: str
    unicast_address: str

    def __bytes__(self) -> bytes:
        return (
            f"{self.username} {self.sess_id} {self.sess_version}"
            f" {self.nettype} {self.addrtype} {self.unicast_address}"
        ).encode()

    @classmethod
    def parse(cls, data: bytes | str) -> Origin:
        value = data.decode() if isinstance(data, bytes) else data
        username, sess_id, sess_version, nettype, addrtype, unicast_address = (
            value.split(" ", 5)
        )
        return cls(
            username=username,
            sess_id=sess_id,
            sess_version=sess_version,
            nettype=nettype,
            addrtype=addrtype,
            unicast_address=unicast_address,
        )

PayloadTypeSpec

Bases: NamedTuple

Typed specification for a static RTP payload type (RFC 3551 §6).

Source code in voip/sdp/types.py

class PayloadTypeSpec(NamedTuple):
    """Typed specification for a static RTP payload type (RFC 3551 §6)."""

    pt: int
    sample_rate: int
    encoding_name: str
    channels: int = 1
    frame_size: int = 0

RTPPayloadFormat dataclass

Bases: ByteSerializableObject

RTP payload format descriptor (RFC 3551 §6 / RFC 4566 §6).

Static payload types fall back to the StaticPayloadType table. Dynamic payload types (PT ≥ 96) require an explicit a=rtpmap.

Serialises to the a=rtpmap value when codec fields are present.

Source code in voip/sdp/types.py

@dataclasses.dataclass(slots=True)
class RTPPayloadFormat(ByteSerializableObject):
    """RTP payload format descriptor (RFC 3551 §6 / RFC 4566 §6).

    Static payload types fall back to the `StaticPayloadType` table.
    Dynamic payload types (PT ≥ 96) require an explicit `a=rtpmap`.

    Serialises to the `a=rtpmap` value when codec fields are present.
    """

    payload_type: int
    fmtp: str | None = None
    encoding_name: str | None = None
    channels: int | None = None
    sample_rate: int | None = None

    def __post_init__(self):
        try:
            default = StaticPayloadType.from_pt(self.payload_type)
        except ValueError:
            self.channels = self.channels or 1
        else:
            self.sample_rate = self.sample_rate or default.sample_rate
            self.encoding_name = self.encoding_name or default.encoding_name
            self.channels = self.channels or default.channels

    def __bytes__(self) -> bytes:
        base = f"{self.payload_type} {self.encoding_name}/{self.sample_rate}"
        match self.channels:
            case 1:
                return base.encode()
            case _:
                return f"{base}/{self.channels}".encode()

    @classmethod
    def parse(cls, data: bytes | str) -> RTPPayloadFormat:
        value = data.decode() if isinstance(data, bytes) else data
        fmt, _, rest = value.partition(" ")
        parts = rest.split("/")
        if len(parts) < 2:
            raise ValueError(f"Invalid rtpmap value: {value!r}")
        return cls(
            payload_type=int(fmt),
            encoding_name=parts[0],
            sample_rate=int(parts[1]),
            channels=int(parts[2]) if len(parts) > 2 else 1,
        )

    @classmethod
    def from_pt(cls, pt: int) -> RTPPayloadFormat:
        """Create an `RTPPayloadFormat` from a payload type number."""
        return cls(payload_type=pt)

    @property
    def frame_size(self) -> int:
        """Samples per standard 20 ms RTP frame.

        For static payload types the value comes from `StaticPayloadType`.
        For dynamic payload types (e.g. Opus, PT ≥ 96) it is derived from
        `sample_rate` assuming a 20 ms packetisation interval.
        """
        try:
            spec = StaticPayloadType.from_pt(self.payload_type)
            if spec.frame_size:
                return spec.frame_size
        except ValueError:
            pass
        return (self.sample_rate or 8000) * 20 // 1000

frame_size property

Samples per standard 20 ms RTP frame.

For static payload types the value comes from StaticPayloadType. For dynamic payload types (e.g. Opus, PT ≥ 96) it is derived from sample_rate assuming a 20 ms packetisation interval.

from_pt(pt) classmethod

Create an RTPPayloadFormat from a payload type number.

Source code in voip/sdp/types.py

@classmethod
def from_pt(cls, pt: int) -> RTPPayloadFormat:
    """Create an `RTPPayloadFormat` from a payload type number."""
    return cls(payload_type=pt)

StaticPayloadType

Bases: PayloadTypeSpec, Enum

Static RTP payload types as defined by RFC 3551 §6.

Each member's value is a PayloadTypeSpec carrying the payload type number, sample rate, canonical encoding name, channel count, and frame size. Use from_pt to look up a member by its PT number.

Source code in voip/sdp/types.py

class StaticPayloadType(PayloadTypeSpec, enum.Enum):
    """Static RTP payload types as defined by RFC 3551 §6.

    Each member's `value` is a `PayloadTypeSpec` carrying the
    payload type number, sample rate, canonical encoding name, channel count,
    and frame size.  Use `from_pt` to look up a member by its PT number.
    """

    #: G.711 µ-law
    PCMU = PayloadTypeSpec(0, 8000, "PCMU", frame_size=160)
    GSM = PayloadTypeSpec(3, 8000, "GSM", frame_size=160)
    G723 = PayloadTypeSpec(4, 8000, "G723", frame_size=160)
    #: DVI4 at 8 kHz
    DVI4_8K = PayloadTypeSpec(5, 8000, "DVI4", frame_size=160)
    #: DVI4 at 16 kHz
    DVI4_16K = PayloadTypeSpec(6, 16000, "DVI4", frame_size=320)
    LPC = PayloadTypeSpec(7, 8000, "LPC", frame_size=160)
    #: G.711 A-law
    PCMA = PayloadTypeSpec(8, 8000, "PCMA", frame_size=160)
    #: RTP clock rate is 8000 per RFC 3551 even though wideband
    G722 = PayloadTypeSpec(9, 8000, "G722", frame_size=160)
    L16_STEREO = PayloadTypeSpec(10, 44100, "L16", 2, frame_size=882)
    L16_MONO = PayloadTypeSpec(11, 44100, "L16", frame_size=882)
    QCELP = PayloadTypeSpec(12, 8000, "QCELP", frame_size=160)
    CN = PayloadTypeSpec(13, 8000, "CN", frame_size=160)
    MPA = PayloadTypeSpec(14, 90000, "MPA")
    G728 = PayloadTypeSpec(15, 8000, "G728", frame_size=160)
    #: DVI4 at 11.025 kHz
    DVI4_11K = PayloadTypeSpec(16, 11025, "DVI4", frame_size=220)
    #: DVI4 at 22.05 kHz
    DVI4_22K = PayloadTypeSpec(17, 22050, "DVI4", frame_size=441)
    G729 = PayloadTypeSpec(18, 8000, "G729", frame_size=160)
    CELB = PayloadTypeSpec(25, 90000, "CelB")
    JPEG = PayloadTypeSpec(26, 90000, "JPEG")
    NV = PayloadTypeSpec(28, 90000, "nv")
    H261 = PayloadTypeSpec(31, 90000, "H261")
    #: MPEG-1 and MPEG-2 video
    MPV = PayloadTypeSpec(32, 90000, "MPV")
    #: MPEG-2 transport stream
    MP2T = PayloadTypeSpec(33, 90000, "MP2T")
    H263 = PayloadTypeSpec(34, 90000, "H263")

    @classmethod
    def from_pt(cls, pt: int) -> StaticPayloadType:
        """Look up a static payload type by its PT number."""
        for member in cls:
            if member.value.pt == pt:
                return member
        raise ValueError(f"No static payload type with PT {pt}")

from_pt(pt) classmethod

Look up a static payload type by its PT number.

Source code in voip/sdp/types.py

@classmethod
def from_pt(cls, pt: int) -> StaticPayloadType:
    """Look up a static payload type by its PT number."""
    for member in cls:
        if member.value.pt == pt:
            return member
    raise ValueError(f"No static payload type with PT {pt}")

StrField dataclass

Descriptor for SDP fields parsing and serializing as plain strings.

Source code in voip/sdp/types.py

@dataclasses.dataclass(slots=True, frozen=True)
class StrField:
    """Descriptor for SDP fields parsing and serializing as plain strings."""

    letter: str
    session_attr: str
    is_list: bool = False
    media_attr: str | None = None

    @staticmethod
    def parse(value: str) -> str:
        return value

Timing dataclass

Bases: ByteSerializableObject

Timing field (t=) as defined by RFC 4566 §5.9.

Source code in voip/sdp/types.py

@dataclasses.dataclass(slots=True, frozen=True)
class Timing(ByteSerializableObject):
    """Timing field (t=) as defined by RFC 4566 §5.9."""

    letter: ClassVar[str] = "t"
    session_attr: ClassVar[str] = "timings"
    is_list: ClassVar[bool] = True
    media_attr: ClassVar[str | None] = None

    start_time: int
    stop_time: int

    def __bytes__(self) -> bytes:
        return f"{self.start_time} {self.stop_time}".encode()

    @classmethod
    def parse(cls, data: bytes | str) -> Timing:
        value = data.decode() if isinstance(data, bytes) else data
        start_time, stop_time = value.split(" ", 1)
        return cls(start_time=int(start_time), stop_time=int(stop_time))