aiosipua 0.7.0

Asyncio SIP micro-library for Python

aiosipua

Asyncio SIP micro-library for Python. Companion to aiortp.

Built for voice AI backends that need SIP signaling without the bloat of a full SIP stack. Zero runtime dependencies, strict type hints, Python 3.11+.

Features

• SIP message parsing and serialization — RFC 3261 compliant, compact header expansion, multi-value header splitting, structured accessors; bodies are raw bytes with a .text view, so binary payloads survive intact
• Hardened parsing — header-injection guards, header/body size caps, required-header validation (400/drop), RFC 4475 torture-tested and property-tested with hypothesis
• SDP parsing, building, and negotiation — RFC 4566 / RFC 3264 (answers mirror every offered m-line), codec selection, DTMF, direction handling
• Video SDP negotiation — negotiate_video_sdp, negotiate_av_sdp for combined audio+video, build_video_sdp for outgoing video offers
• Transports — UDP (DatagramProtocol) and TCP (Content-Length framing), IPv4 and IPv6, received/rport handling (RFC 3581)
• UAS — INVITE/re-INVITE/BYE/CANCEL/OPTIONS/UPDATE/PRACK/REFER/NOTIFY dispatch, auto 100 Trying, dialog validation (tags + CSeq), IncomingCall high-level API
• UAC — outbound INVITE with send_invite, BYE, re-INVITE (hold/unhold), RFC-compliant CANCEL, UPDATE, REFER, INFO (DTMF)
• Reliability — 200 OK retransmitted until ACK over UDP (RFC 3261 §13.3.1.4), reliable provisionals with PRACK/100rel (RFC 3262), automatic transaction expiry
• Session timers — dead-call detection via UPDATE refreshes and expiry watchdogs (RFC 4028), negotiated on both sides
• REGISTER client — auto-refresh before expiry, 423 Min-Expires handling, expiry watchdog (RFC 3261 §10)
• Blind transfer — REFER with implicit-subscription NOTIFYs and transfer-progress callbacks (RFC 3515)
• SIP digest authentication — RFC 7616 (qop, MD5 and SHA-256), automatic 401/407 retry for INVITE and REGISTER
• Dialog management — RFC 3261 dialog state machine, Record-Route support, in-dialog request/response creation
• aiortp bridge — CallSession for audio RTP (jitter buffer, PLC, optional RFC 3389 comfort noise) and VideoCallSession for video RTP, bridging SDP negotiation to media with callbacks
• NAT traversal — advertised_ip for SDP and advertised_addr for Via/Contact: bind privately, signal publicly
• X-header support — pass application metadata (room ID, session ID, tenant) through SIP headers

Installation

pip install aiosipua

# With optional RTP support
pip install aiosipua[rtp]

Examples

Parse a SIP message

from aiosipua import SipMessage, parse_sdp

raw = (
    "INVITE sip:[e-mail] SIP/2.0\r\n"
    "Via: SIP/2.0/UDP 10.0.0.1:5060;branch=z9hG4bK776asdhds\r\n"
    "From: Alice <sip:[e-mail]>;tag=1928301774\r\n"
    "To: Bob <sip:[e-mail]>\r\n"
    "Call-ID: [e-mail]\r\n"
    "CSeq: 314159 INVITE\r\n"
    "Contact: <sip:alice@10.0.0.1:5060>\r\n"
    "Content-Type: application/sdp\r\n"
    "Content-Length: 162\r\n"
    "\r\n"
    "v=0\r\n"
    "o=- 2890844526 2890844526 IN IP4 10.0.0.1\r\n"
    "s=-\r\n"
    "c=IN IP4 10.0.0.1\r\n"
    "t=0 0\r\n"
    "m=audio 20000 RTP/AVP 0 8\r\n"
    "a=rtpmap:0 PCMU/8000\r\n"
    "a=rtpmap:8 PCMA/8000\r\n"
    "a=sendrecv\r\n"
)

msg = SipMessage.parse(raw)

# Structured header access
print(msg.from_addr.display_name)  # "Alice"
print(msg.from_addr.uri.user)      # "alice"
print(msg.to_addr.uri.host)        # "example.com"
print(msg.via[0].branch)           # "z9hG4bK776asdhds"
print(msg.cseq.method)             # "INVITE"
print(msg.call_id)                 # "[e-mail]"

# Parse the SDP body (message bodies are bytes; .text is the UTF-8 view)
sdp = parse_sdp(msg.text)
audio = sdp.audio
print(audio.port)                  # 20000
print(audio.codecs[0].encoding_name)  # "PCMU"
print(sdp.rtp_address)             # ("10.0.0.1", 20000)

SDP negotiation

from aiosipua import parse_sdp, negotiate_sdp, serialize_sdp

# Parse an incoming SDP offer
offer = parse_sdp(sdp_body)

# Negotiate: pick the best codec, build an answer
answer, chosen_pt = negotiate_sdp(
    offer=offer,
    local_ip="10.0.0.5",
    rtp_port=30000,
    supported_codecs=[0, 8],  # PCMU, PCMA
)

print(f"Chosen codec: payload type {chosen_pt}")
print(serialize_sdp(answer))

Video SDP negotiation

from aiosipua import parse_sdp, negotiate_av_sdp, serialize_sdp

# Negotiate both audio and video from a single offer
offer = parse_sdp(sdp_body)
answer, audio_pt, video_pt = negotiate_av_sdp(
    offer=offer,
    local_ip="10.0.0.5",
    audio_rtp_port=30000,
    video_rtp_port=30002,
    supported_video_codecs=["H264", "VP8"],
)

print(f"Audio PT: {audio_pt}, Video PT: {video_pt}")
print(serialize_sdp(answer))

Receive calls with the UAS

import asyncio
from aiosipua import IncomingCall, SipUAS
from aiosipua.rtp_bridge import CallSession
from aiosipua.transport import UdpSipTransport

async def handle_invite(call: IncomingCall):
    print(f"Incoming call: {call.caller} -> {call.callee}")
    print(f"X-headers: {call.x_headers}")

    if call.sdp_offer is None:
        call.reject(488, "Not Acceptable Here")
        return

    # Negotiate SDP and create RTP session
    session = CallSession(
        local_ip="10.0.0.5",
        rtp_port=30000,
        offer=call.sdp_offer,
    )

    # Accept the call with the SDP answer
    call.ringing()
    call.accept(session.sdp_answer)
    await session.start()

    # Wire up audio and DTMF callbacks
    session.on_audio = lambda pcm, ts: print(f"Audio: {len(pcm)} bytes")
    session.on_dtmf = lambda digit, dur: print(f"DTMF: {digit}")

def handle_bye(call: IncomingCall, request):
    print(f"Call ended: {call.call_id}")

async def main():
    transport = UdpSipTransport(local_addr=("0.0.0.0", 5060))
    uas = SipUAS(transport)
    uas.on_invite = lambda call: asyncio.get_running_loop().create_task(handle_invite(call))
    uas.on_bye = handle_bye

    await uas.start()
    print("Listening on port 5060...")
    await asyncio.Event().wait()

asyncio.run(main())

Video call session

from aiosipua import parse_sdp
from aiosipua.video_bridge import VideoCallSession

offer = parse_sdp(sdp_body)

session = VideoCallSession(
    local_ip="10.0.0.5",
    rtp_port=30002,
    offer=offer,
    supported_video_codecs=["H264"],
)
await session.start()

# Receive video frames
session.on_frame = lambda nal, ts, kf: process_video(nal, ts, kf)
session.on_keyframe_needed = lambda: encoder.force_keyframe()

# Send video frames
session.send_frame(nal_units, timestamp, keyframe=True)

# Request a keyframe from remote
session.request_keyframe()

await session.close()

NAT traversal with advertised_ip

When behind NAT, RTP sockets bind to a private IP but the SDP must advertise the public IP so the remote peer sends media to the right address:

from aiosipua.rtp_bridge import CallSession

# Behind NAT: bind RTP on 192.168.1.5, advertise 203.0.113.10 in SDP
session = CallSession(
    local_ip="192.168.1.5",           # RTP socket binds here
    rtp_port=30000,
    offer=call.sdp_offer,
    advertised_ip="203.0.113.10",     # SDP c=/o= lines use this
)

# SDP answer will contain:
#   c=IN IP4 203.0.113.10
#   o=... IN IP4 203.0.113.10
# But the RTP socket listens on 192.168.1.5:30000

# Works the same for build_sdp (outbound offers):
from aiosipua import build_sdp

sdp = build_sdp(
    local_ip="192.168.1.5",
    rtp_port=30000,
    payload_type=0,
    codec_name="PCMU",
    advertised_ip="203.0.113.10",
)

Backend-initiated actions with the UAC

from aiosipua import SipUAC
from aiosipua.transport import UdpSipTransport

transport = UdpSipTransport(local_addr=("0.0.0.0", 5060))
uac = SipUAC(transport)

# Hang up a call
uac.send_bye(dialog, remote_addr=("10.0.0.1", 5060))

# Put a call on hold with re-INVITE
from aiosipua import build_sdp
hold_sdp = build_sdp(
    local_ip="10.0.0.5",
    rtp_port=30000,
    payload_type=0,
    direction="sendonly",
)
uac.send_reinvite(dialog, sdp=hold_sdp, remote_addr=("10.0.0.1", 5060))

# Send DTMF via SIP INFO
uac.send_info(
    dialog,
    body="Signal=5\r\nDuration=250\r\n",
    content_type="application/dtmf-relay",
    remote_addr=("10.0.0.1", 5060),
)

Outbound calls with digest authentication

from aiosipua import SipUAC, SipDigestAuth, build_sdp
from aiosipua.transport import UdpSipTransport

transport = UdpSipTransport(local_addr=("0.0.0.0", 5060))
uac = SipUAC(transport)

sdp = build_sdp(local_ip="10.0.0.5", rtp_port=30000, payload_type=0, codec_name="PCMU")
auth = SipDigestAuth(username="alice", password="secret")

call = uac.send_invite(
    from_uri="sip:[e-mail]",
    to_uri="sip:[e-mail]",
    remote_addr=("proxy.example.com", 5060),
    sdp_offer=sdp,
    auth=auth,  # auto-retries on 401/407
)

Register with a registrar

from aiosipua import Registration, SipDigestAuth, SipUAC
from aiosipua.transport import UdpSipTransport

transport = UdpSipTransport(local_addr=("0.0.0.0", 5060))
uac = SipUAC(transport)

reg = Registration(
    uac,
    "sip:[e-mail]",
    ("registrar.example.com", 5060),
    auth=SipDigestAuth("alice", "secret"),
    expires=300,
)
reg.on_registered = lambda r: print(f"registered for {r.granted_expires}s")
reg.register()   # refreshes itself until reg.unregister()

Blind transfer and session timers

# Ask the remote party to call an agent (RFC 3515); progress arrives as NOTIFYs
uac.send_refer(call.dialog, "sip:[e-mail]", remote_addr)
uas.on_transfer_progress = lambda call_id, status, reason: print(status, reason)

# Dead-call detection (RFC 4028): UAS side negotiates timers on incoming calls,
# UAC side requests them on outgoing ones
uas = SipUAS(transport, session_expires=1800)
call = uac.send_invite(
    "sip:[e-mail]", "sip:[e-mail]", remote_addr,
    sdp_offer=sdp, session_expires=1800,
)

Build a SIP message from scratch

from aiosipua import SipRequest, SipResponse, generate_branch, generate_call_id, generate_tag

# Build a SIP request
request = SipRequest(method="OPTIONS", uri="sip:[e-mail]")
request.headers.set_single("Via", f"SIP/2.0/UDP 10.0.0.1:5060;branch={generate_branch()}")
request.headers.set_single("From", f"<sip:[e-mail]>;tag={generate_tag()}")
request.headers.set_single("To", "<sip:[e-mail]>")
request.headers.set_single("Call-ID", generate_call_id("example.com"))
request.headers.set_single("CSeq", "1 OPTIONS")

# Serialize to bytes for the wire
raw_bytes = bytes(request)

Modify and re-serialize

from aiosipua import SipMessage

msg = SipMessage.parse(raw_sip_text)

# Add a Via header
msg.headers.append("Via", "SIP/2.0/UDP proxy.example.com:5060;branch=z9hG4bKnew")

# Change the Contact
msg.headers.set_single("Contact", "<sip:newhost@10.0.0.99:5060>")

# Add custom X-headers
msg.headers.set_single("X-Room-ID", "room-42")
msg.headers.set_single("X-Session-ID", "sess-abc123")

# Re-serialize (Content-Length auto-updated)
print(msg.serialize())

TCP transport

import asyncio
from aiosipua.transport import TcpSipTransport

async def main():
    transport = TcpSipTransport(local_addr=("0.0.0.0", 5060))

    # As a server
    transport.on_message = lambda msg, addr: print(f"Received from {addr}")
    await transport.start()

    # Or connect as a client
    await transport.connect(("proxy.example.com", 5060))
    transport.send(request, ("proxy.example.com", 5060))

asyncio.run(main())

Architecture

┌─────────────┐     ┌──────────────┐     ┌────────────┐     ┌──────────────┐
│  SipUAS     │────▶│  Dialog      │────▶│  SipUAC    │────▶│ Registration │
│  (incoming) │     │  (state mgr) │     │  (outgoing)│     │ (REGISTER)   │
└──────┬──────┘     └──────────────┘     └─────┬──────┘     └──────────────┘
       │                                       │
       │    ┌───────────────┐   ┌──────────┐   │
       ├───▶│ SessionTimer  │   │  REFER / │◀──┤
       │    │  (RFC 4028)   │   │  NOTIFY  │   │
       │    └───────────────┘   └──────────┘   │
       ▼                                       ▼
┌──────────────┐    ┌──────────────┐    ┌──────────────┐
│  Transaction │    │  SDP/Codec   │    │  CallSession │
│  Layer       │    │  Negotiation │    │  (audio RTP) │
└──────┬───────┘    └──────┬───────┘    └──────┬───────┘
       │                   │                   │
       │            ┌──────┴───────┐    ┌──────┴────────┐
       │            │ Video SDP    │    │ VideoCall     │
       │            │ Negotiation  │    │ Session       │
       │            └──────────────┘    │ (video RTP)   │
       │                                └──────┬────────┘
       ▼                                       ▼
┌──────────────┐                        ┌──────────────┐
│  Transport   │                        │  aiortp      │
│  (UDP / TCP) │                        │  (optional)  │
└──────────────┘                        └──────────────┘

More examples

See the examples/ directory:

• echo_server.py — Receives audio via RTP and echoes it back
• dtmf_ivr.py — Collects DTMF digits and hangs up on #
• lossy_caller.py — Dials an agent with controlled RTP packet loss
• roomkit_prototype.py — Voice AI backend integration with X-header metadata

License

MIT. See LICENSE for details.