Metadata-Version: 2.4
Name: aiosipua
Version: 0.7.0
Summary: Asyncio SIP micro-library for Python
Project-URL: Homepage, https://github.com/anganyAI/aiosipua
Project-URL: Repository, https://github.com/anganyAI/aiosipua
Author-email: Sylvain Boily <sylvainboilydroid@gmail.com>
License-Expression: MIT
License-File: LICENSE
Classifier: Development Status :: 3 - Alpha
Classifier: Framework :: AsyncIO
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Communications :: Telephony
Classifier: Typing :: Typed
Requires-Python: >=3.11
Provides-Extra: rtp
Requires-Dist: aiortp>=0.7.0; extra == 'rtp'
Description-Content-Type: text/markdown

# aiosipua

[![CI](https://github.com/anganyAI/aiosipua/actions/workflows/ci.yml/badge.svg)](https://github.com/anganyAI/aiosipua/actions/workflows/ci.yml)
[![PyPI](https://img.shields.io/pypi/v/aiosipua)](https://pypi.org/project/aiosipua/)

Asyncio SIP micro-library for Python. Companion to [aiortp](https://github.com/anganyAI/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

```bash
pip install aiosipua

# With optional RTP support
pip install aiosipua[rtp]
```

## Examples

### Parse a SIP message

```python
from aiosipua import SipMessage, parse_sdp

raw = (
    "INVITE sip:bob@example.com SIP/2.0\r\n"
    "Via: SIP/2.0/UDP 10.0.0.1:5060;branch=z9hG4bK776asdhds\r\n"
    "From: Alice <sip:alice@example.com>;tag=1928301774\r\n"
    "To: Bob <sip:bob@example.com>\r\n"
    "Call-ID: a84b4c76e66710@example.com\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)                 # "a84b4c76e66710@example.com"

# 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

```python
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

```python
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

```python
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

```python
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:

```python
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

```python
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

```python
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:alice@example.com",
    to_uri="sip:bob@example.com",
    remote_addr=("proxy.example.com", 5060),
    sdp_offer=sdp,
    auth=auth,  # auto-retries on 401/407
)
```

### Register with a registrar

```python
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:alice@example.com",
    ("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

```python
# Ask the remote party to call an agent (RFC 3515); progress arrives as NOTIFYs
uac.send_refer(call.dialog, "sip:agent@example.com", 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:me@example.com", "sip:them@example.com", remote_addr,
    sdp_offer=sdp, session_expires=1800,
)
```

### Build a SIP message from scratch

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

# Build a SIP request
request = SipRequest(method="OPTIONS", uri="sip:bob@example.com")
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:alice@example.com>;tag={generate_tag()}")
request.headers.set_single("To", "<sip:bob@example.com>")
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

```python
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

```python
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/`](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](LICENSE) for details.
