Media¶

The rtpengine namespace controls media anchoring and injection (announcements, DTMF, gating, subscriptions) via the RTPEngine / siphon-rtp NG control protocol. The qos namespace turns an SDP offer/answer pair into the media_components structure that diameter.rx_aar and sbi.create_session consume.

from siphon import rtpengine

@b2bua.on_invite
async def anchor(call):
    await rtpengine.play_media(call, file="/prompts/welcome.wav")

`rtpengine` namespace¶

Mock RTPEngine namespace — records media operations for assertions.

Example::

from siphon import rtpengine
# After running handler:
assert rtpengine.operations == [("offer", "srtp_to_rtp")]

Media-injection operations (play_media, stop_media, play_dtmf, silence_media, unsilence_media, block_media, unblock_media, echo) are also captured in operations as (name, detail) tuples so downstream apps can unit-test MMTEL announcement flows without a live rtpengine. Full parameter dicts are available on media_calls.

Every media verb's target accepts three forms (like the runtime's resolve_call_from_tag): a SIP object (Request/Reply/Call), a (call_id, from_tag) pair, or a bare call_id string — so an @rtpengine.on_dtmf handler can drive media from the call_id / from_tag it was handed. The resolved call_id / from_tag are recorded on each media_calls entry.

Valid profiles: "srtp_to_rtp", "ws_to_rtp", "wss_to_rtp", "rtp_passthrough".

operations `instance-attribute` ¶

operations: list[tuple[str, Optional[str]]] = []

List of (operation, profile_or_detail) tuples recorded.

media_calls `instance-attribute` ¶

media_calls: list[dict[str, Any]] = []

Full parameter dicts for each media-injection call.

ws_uris `instance-attribute` ¶

ws_uris: list[tuple[str, Optional[str]]] = []

(operation, resolved_ws_uri) per offer/answer/answer_local.

The URI is post-templating, so a test asserts the value the media engine would actually be handed. None means no WebSocket bridge was requested for that call.

media_overrides `instance-attribute` ¶

media_overrides: list[tuple[str, dict[str, Any]]] = []

(operation, overrides) per offer/answer/answer_local.

The per-call media knobs (beep_detection, ws_sample_rate, ws_vad_engine, ...) a script passed for that call, already validated. None values mean "leave the profile's value alone".

active_sessions `property` ¶

active_sessions: int

Number of active media sessions (mock: count of offer - delete).

instance_count `property` ¶

instance_count: int

Number of configured RTPEngine instances (mock: always 1).

offer `async` ¶

offer(
    request: Any,
    profile: Optional[str] = None,
    ws_uri: Optional[str] = None,
    beep_detection: Optional[bool] = None,
    beep_cadence_guard_ms: Optional[int] = None,
    ws_sample_rate: Optional[int] = None,
    ws_tee_sample_rate: Optional[int] = None,
    ws_vad_engine: Optional[str] = None,
    ws_vad_min_speech_ms: Optional[int] = None,
) -> bool

Send offer command to RTPEngine.

Extracts SDP from message body, sends to engine, replaces body with rewritten SDP.

Parameters:

Name	Type	Description	Default
`request`	`Any`	Request or Call object with SDP body.	required
`profile`	`Optional[str]`	RTP profile name. Defaults to `"rtp_passthrough"`.	`None`
`ws_uri`	`Optional[str]`	Bridge this leg's audio to an external WebSocket media server (`siphon-rtp` backend only), overriding the profile's own `ws_uri` for this call. Supports `{call_id}`, `{from_tag}`, `{from_user}` and `{to_user}` placeholders. The resolved URI is recorded on the media session, so a later `answer` reuses it automatically.	`None`

Returns:

Type	Description
`bool`	`True` on success.

Example::

@b2bua.on_invite
async def on_invite(call):
    sdp = await rtpengine.answer_local(
        call,
        profile="voice_ai",
        ws_uri=f"wss://ai.example.com/stream/{call.call_id}",
    )

answer `async` ¶

answer(
    reply: Any,
    profile: Optional[str] = None,
    call: Any = None,
    ws_uri: Optional[str] = None,
    beep_detection: Optional[bool] = None,
    beep_cadence_guard_ms: Optional[int] = None,
    ws_sample_rate: Optional[int] = None,
    ws_tee_sample_rate: Optional[int] = None,
    ws_vad_engine: Optional[str] = None,
    ws_vad_min_speech_ms: Optional[int] = None,
    sdp: Union[str, bytes, None] = None,
    to_tag: Optional[str] = None,
) -> Union[bool, str]

Send answer command to RTPEngine.

Two ways to hand it the answer:

A SIP reply (the usual proxy / B2BUA case): the SDP and To-tag come off reply, its body is replaced with the rewritten SDP, and the call resolves to True.
Raw SDP with sdp=: for a far side that is not a SIP agent and hands over its answer some other way (a media server behind its own API). No message body is read or written; the call resolves to the rewritten SDP as str for the script to send itself.

Profile precedence (matches the real implementation):

Explicit profile= argument (script override).
Profile recorded by the matching offer (looked up by A-leg Call-ID). Lets @b2bua.on_answer / @b2bua.on_early_media call rtpengine.answer(reply) with no profile= and still get the directional flags from the offer-side profile.
"rtp_passthrough" when no offer was ever recorded.

Delayed offer (RFC 3264 §4): when the INVITE carried no SDP, the reply's SDP is the offer, not an answer. It goes to the engine as an offer from the replying party's side, and siphon completes it with the caller's answer from the ACK itself, so a script calls answer the same way for both. A raw sdp= is always an answer.

Either command is addressed by the call-id the engine knows the call by, which differs from the SIP Call-ID after a siphon-terminated transfer re-anchored the call.

Parameters:

Name	Type	Description	Default
`reply`	`Any`	Reply or Call object with SDP body. With `sdp=` it only names the offer being answered: a `Call`, `Request` or `Reply`, or a `(call_id, from_tag)` tuple. A bare `call_id` string names no from-tag and raises `TypeError`.	required
`profile`	`Optional[str]`	Optional explicit RTP profile name. When omitted, the profile recorded by the matching offer is used.	`None`
`call`	`Any`	Optional Call object — when provided, Call-ID and From-tag are taken from this object (matching the earlier `offer`), while To-tag and SDP body still come from `reply`.	`None`
`ws_uri`	`Optional[str]`	WebSocket bridge URI override for this call (`siphon-rtp` backend only). When omitted, the URI recorded by the matching `offer` is reused — the same precedence as `profile`.	`None`
`sdp`	`Union[str, bytes, None]`	The far side's answer SDP (`str` or `bytes`). Switches to raw mode. Blank raises `ValueError`. No source address is carried for the far side: siphon never heard from it, so a profile's `received_from` does not gate its media.	`None`
`to_tag`	`Optional[str]`	Tag naming the answering party to the engine. Raw mode only (`ValueError` without `sdp=`). When omitted: the To-tag on `reply` if it carries one, else the tag an earlier answer on this call recorded (so a re-answer reaches the same party), else a new one. The mock records what was passed, `None` when omitted.	`None`

Returns:

Type	Description
`Union[bool, str]`	`True`, or the rewritten SDP as `str` when `sdp=` was passed.
`Union[bool, str]`	The mock rewrites nothing and returns `sdp` as given.

Example::

@b2bua.on_invite
async def on_invite(call):
    await rtpengine.offer(call, profile="rtp_passthrough")
    far_sdp = await my_media_server.connect(call.body)  # not SIP
    sdp = await rtpengine.answer(call, sdp=far_sdp)
    call.answer(200, "OK", body=sdp, content_type="application/sdp")

answer_local `async` ¶

answer_local(
    call: Any,
    profile: Optional[str] = None,
    auto_reject: bool = True,
    ws_uri: Optional[str] = None,
    beep_detection: Optional[bool] = None,
    beep_cadence_guard_ms: Optional[int] = None,
    ws_sample_rate: Optional[int] = None,
    ws_tee_sample_rate: Optional[int] = None,
    ws_vad_engine: Optional[str] = None,
    ws_vad_min_speech_ms: Optional[int] = None,
) -> Optional[str]

Single-leg UAS answer — synthesise an RFC 3264 answer for the caller's own offer, with the media engine as the far side (IVR / echo / announcement server).

Unlike :meth:answer, this takes the offer (INVITE), not a peer's reply: there is no far leg, so the engine picks one encodable codec from the offer and returns the answer SDP for the script to put in its own 2xx.

Profile precedence matches :meth:answer (explicit profile= → profile recorded by a matching offer → "rtp_passthrough").

When the offer has no encodable codec (primed in tests via :meth:set_answer_local_no_codec), the engine cannot answer:

with auto_reject=True (default) and a Call target, a deferred 488 Not Acceptable Here is recorded on the call (call.reject(488, "Not Acceptable Here")) and the coroutine resolves to None;
with auto_reject=False (or a non-Call target) it raises ValueError instead.

Native siphon-rtp backend only.

Parameters:

Name	Type	Description	Default
`call`	`Any`	A `Call` (B2BUA) — or `Request` — carrying the INVITE offer whose Call-ID / From-tag drive the single-leg answer.	required
`profile`	`Optional[str]`	Optional explicit RTP profile name. When omitted, the profile recorded by a matching offer is used.	`None`
`auto_reject`	`bool`	When `True` (default) and `call` is a `Call`, a no-encodable-codec result records a deferred 488 and returns `None`. When `False` it raises `ValueError`.	`True`
`ws_uri`	`Optional[str]`	Bridge this leg's audio to an external WebSocket media server instead of a far SIP leg — the shape a voice-AI answer takes, since the WS server is the far side. Overrides the profile's own `ws_uri`; supports the same placeholders as :meth:`offer`.	`None`

Returns:

Type	Description
`Optional[str]`	The answer SDP as `str` on success, or `None` when the offer had
`Optional[str]`	no encodable codec and it was auto-rejected with a 488.

Example::

@b2bua.on_invite
async def on_invite(call):
    sdp = await rtpengine.answer_local(call, profile="ivr")
    if sdp is not None:
        call.answer(200, "OK", body=sdp, content_type="application/sdp")
        await rtpengine.play_media(call, file="/prompts/welcome.wav")

delete `async` ¶

delete(request: Any) -> bool

Send delete command to tear down media session.

Parameters:

Name	Type	Description	Default
`request`	`Any`	Request or Call object (uses Call-ID + From-tag).	required

Returns:

Type	Description
`bool`	`True` on success.

ping `async` ¶

ping() -> bool

Health check: ping RTPEngine instance(s).

Returns:

Type	Description
`bool`	`True` if healthy.

play_media `async` ¶

play_media(
    target: Any,
    file: Optional[str] = None,
    blob: Optional[bytes] = None,
    db_id: Optional[int] = None,
    tone: Optional[str] = None,
    url: Optional[str] = None,
    repeat: Optional[int] = None,
    start_ms: Optional[int] = None,
    duration_ms: Optional[int] = None,
    gain_decibels: Optional[int] = None,
    to_tag: Optional[str] = None,
    wait: bool = True,
) -> Optional[int]

Inject an audio prompt, replacing the party's live egress.

Exactly one of file/blob/db_id/tone/url must be supplied. Per rtpengine semantics, from-tag (derived from target) selects the monologue whose outgoing audio is replaced by the prompt — the peer of that monologue hears it. Pass to_tag to scope to a specific peer in MPTY scenarios.

Requires rtpengine built with --with-transcoding and launched with --audio-player=on-demand. AMR-NB/WB prompts need licensed codec plugins; G.711 and Opus prompts work without them.

Parameters:

Name	Type	Description	Default
`target`	`Any`	Request, Reply, or Call object.	required
`file`	`Optional[str]`	Absolute path to an audio file on the rtpengine host.	`None`
`blob`	`Optional[bytes]`	Raw audio bytes to play (e.g. TTS output).	`None`
`db_id`	`Optional[int]`	Reference to a prompt stored in rtpengine's prompt DB.	`None`
`tone`	`Optional[str]`	A synthesised call-progress tone, with no audio file to provision. Either a preset name (`"ringback_eu"`, `"busy_na"`, `"dial_uk"`, ...) or an explicit cadence spec in the engine's tone grammar (`"425/1000,0/4000inf"` is 425 Hz one second on, four seconds off, forever). The two are told apart by the `/`. Rendered at the leg's codec rate, so never resampled. Native siphon-rtp* backend only.	`None`
`url`	`Optional[str]`	An `http://` / `https://` WAV the engine fetches from its own network position. The fetch is bounded engine-side (connect, first-byte, deadline, size cap, redirect cap) and runs off the media path, so a URL that never answers ends the playback, never the leg. The accept carries no duration, since the length is unknown until the body arrives. Native siphon-rtp backend only.	`None`
`repeat`	`Optional[int]`	Number of times to repeat the prompt.	`None`
`start_ms`	`Optional[int]`	Offset into the file at which to start (ms).	`None`
`duration_ms`	`Optional[int]`	Cap on playback length (ms).	`None`
`gain_decibels`	`Optional[int]`	Playout gain in whole decibels relative to the source's own level, clamped engine-side to -60..=+12. Native siphon-rtp backend only.	`None`
`to_tag`	`Optional[str]`	Optional peer tag for MPTY scoping.	`None`
`wait`	`bool`	When `True` (default, native siphon-rtp backend), the real runtime blocks until the prompt finishes playing so a script can sequence a following action (e.g. `echo()`) after it. The coroutine parks while it waits. `wait=False` returns as soon as the engine accepts the prompt (fire-and-forget). In this mock the call always returns immediately (the completion event is a runtime behavior); `wait` is recorded for assertions.	`True`

Returns:

Type	Description
`Optional[int]`	Prompt duration in ms if rtpengine reports one (mock returns
`Optional[int]`	the value set via :meth:`set_play_media_duration`, else `None`).

Example::

@b2bua.on_invite
async def on_invite(call):
    await rtpengine.offer(call, profile="ivr")
    call.answer(200, "OK", body=call.body, content_type="application/sdp")
    await rtpengine.play_media(call, file="/prompts/welcome.wav")  # wait=True
    await rtpengine.echo(call)                                     # after prompt

play_overlay `async` ¶

play_overlay(
    target: Any,
    file: Optional[str] = None,
    blob: Optional[bytes] = None,
    db_id: Optional[int] = None,
    tone: Optional[str] = None,
    url: Optional[str] = None,
    repeat: Optional[int] = None,
    start_ms: Optional[int] = None,
    duration_ms: Optional[int] = None,
    gain_decibels: Optional[int] = None,
    to_tag: Optional[str] = None,
) -> Optional[int]

Start an overlay playback and return its play_id handle.

The additive twin of :meth:play_media: audio is mixed under the party's live egress instead of replacing it. Where play_media answers "how long did it play", this answers "which playback is it", because that is what an overlay is for -- a music bed you will duck with :meth:set_play_gain and stop individually with stop_media(target, play_id=...).

Up to four overlays run concurrently per direction, each with its own play_id and its own completion. Starting a fifth is rejected rather than displacing one, since a script that lost a playback it believes is running has no way to notice. An overlay never supersedes anything, including another overlay.

Returns on the engine's accept: an overlay is background audio, so there is no wait. Native siphon-rtp backend only.

Parameters:

Name	Type	Description	Default
`target`	`Any`	Request, Reply, or Call object.	required
`file`	`Optional[str]`	Absolute path to an audio file on the engine host.	`None`
`blob`	`Optional[bytes]`	Raw audio bytes to play.	`None`
`db_id`	`Optional[int]`	Reference to a prompt in the engine's prompt DB.	`None`
`tone`	`Optional[str]`	A preset name or cadence spec, as for :meth:`play_media`.	`None`
`url`	`Optional[str]`	An `http://` / `https://` WAV the engine fetches.	`None`
`repeat`	`Optional[int]`	Number of times to repeat.	`None`
`start_ms`	`Optional[int]`	Offset into the source at which to start (ms).	`None`
`duration_ms`	`Optional[int]`	Hard playout cap -- the only bound, short of a stop, on an endless (`*inf`) tone.	`None`
`gain_decibels`	`Optional[int]`	Playout gain relative to the source's own level.	`None`
`to_tag`	`Optional[str]`	Optional peer tag for MPTY scoping.	`None`

Returns:

Name	Type	Description
	`Optional[int]`	The `play_id` of the started overlay (mock returns the value set
`via`	`Optional[int]`	meth:`set_play_overlay_id`, default `1`).

Example::

bed = await rtpengine.play_overlay(call, file="/prompts/hold.wav")
await rtpengine.play_media(call, file="/prompts/agent.wav")
await rtpengine.set_play_gain(call, bed, -18)
await rtpengine.stop_media(call, play_id=bed)

stop_media `async` ¶

stop_media(
    target: Any, play_id: Optional[int] = None
) -> bool

Stop prompt playback on the selected monologue.

Parameters:

Name	Type	Description	Default
`target`	`Any`	Request, Reply, or Call object.	required
`play_id`	`Optional[int]`	Stop one specific playback (an individual overlay slot, from :meth:`play_overlay`). Omitting it stops everything playing on the leg. Native siphon-rtp backend only: the other backends have no handle on an individual playback and raise rather than widening this into "stop everything", which would kill playbacks the script meant to keep running.	`None`

Returns:

Type	Description
`bool`	`True` on success.

set_play_gain `async` ¶

set_play_gain(
    target: Any,
    play_id: int,
    gain_decibels: int,
    to_tag: Optional[str] = None,
) -> bool

Retune the playout gain of a playback that is already running.

How a script ducks a music bed under a prompt and lifts it again. A separate verb rather than a field on :meth:play_media because play_media is a start: reusing it would mean "start another playback", not "change this one". play_id is already the contract's handle on a running playback, so gain is addressed the same way.

The engine answers an error when no playback on the call holds that play_id, so a stale handle raises rather than silently doing nothing. Native siphon-rtp backend only.

Parameters:

Name	Type	Description	Default
`target`	`Any`	Request, Reply, or Call object.	required
`play_id`	`int`	The running playback to retune, from :meth:`play_overlay`.	required
`gain_decibels`	`int`	New gain in whole decibels, clamped engine-side to -60..=+12.	required
`to_tag`	`Optional[str]`	Optional peer tag for MPTY scoping.	`None`

Returns:

Type	Description
`bool`	`True` on success.

play_dtmf `async` ¶

play_dtmf(
    target: Any,
    code: str,
    duration_ms: Optional[int] = None,
    volume_dbm0: Optional[int] = None,
    pause_ms: Optional[int] = None,
    to_tag: Optional[str] = None,
) -> bool

Inject DTMF tone(s) into the call.

Parameters:

Name	Type	Description	Default
`target`	`Any`	Request, Reply, or Call object.	required
`code`	`str`	A single digit (`"0"`–`"9"`, `"*"`, `"#"`, `"A"`–`"D"`) or a string sequence of digits.	required
`duration_ms`	`Optional[int]`	Tone duration per digit.	`None`
`volume_dbm0`	`Optional[int]`	Tone volume in dBm0 (typically `-8`).	`None`
`pause_ms`	`Optional[int]`	Inter-tone gap when `code` is a sequence.	`None`
`to_tag`	`Optional[str]`	Optional peer tag for MPTY scoping.	`None`

Example::

await rtpengine.play_dtmf(call, "123#", duration_ms=100)

silence_media `async` ¶

silence_media(target: Any) -> bool

Replace outgoing audio on the selected monologue with silence.

Pair with :meth:unsilence_media to restore the original stream.

unsilence_media `async` ¶

unsilence_media(target: Any) -> bool

Stop replacing outgoing audio with silence (undo :meth:silence_media).

block_media `async` ¶

block_media(target: Any) -> bool

Drop outgoing packets on the selected monologue entirely.

Pair with :meth:unblock_media to resume.

unblock_media `async` ¶

unblock_media(target: Any) -> bool

Resume forwarding the selected monologue's packets.

echo `async` ¶

echo(target: Any, enabled: bool = True) -> bool

Toggle echo-test mode on a call — reflect the caller's ingress audio back to itself (single-leg IVR echo).

enabled=False stops echoing. Native siphon-rtp backend only; DTMF and media-timeout events still fire while echoing.

subscribe_request `async` ¶

subscribe_request(
    call_id: str,
    from_tag: str,
    to_tag: str,
    sdp: Optional[bytes] = None,
    profile: Optional[str] = None,
) -> bytes

Create a new subscription to an existing call's media (MPTY / MRF conference focus).

Parameters:

Name	Type	Description	Default
`call_id`	`str`	rtpengine call-id of the source session.	required
`from_tag`	`str`	source monologue tag whose outgoing audio is subscribed.	required
`to_tag`	`str`	subscriber tag to create.	required
`sdp`	`Optional[bytes]`	Optional inbound SDP for the subscriber.	`None`
`profile`	`Optional[str]`	RTP profile name (defaults to `"rtp_passthrough"`).	`None`

Returns:

Type	Description
`bytes`	The subscriber SDP as `bytes`.

subscribe_answer `async` ¶

subscribe_answer(
    call_id: str,
    from_tag: str,
    to_tag: str,
    sdp: bytes,
    profile: Optional[str] = None,
) -> bytes

Complete the SDP negotiation for a subscription created via :meth:subscribe_request.

Returns:

Type	Description
`bytes`	The rewritten SDP as `bytes` (may be empty).

unsubscribe `async` ¶

unsubscribe(
    call_id: str, from_tag: str, to_tag: str
) -> bool

Tear down a subscription created via :meth:subscribe_request.

attach_ws_tee `async` ¶

attach_ws_tee(
    target: Any,
    ws_uri: str,
    direction: str = "both",
    channels: Optional[int] = None,
    sample_rate: Optional[int] = None,
) -> bool

Attach a WebSocket tee to a live call — stream a copy of its decoded audio to a WebSocket media server while the call keeps relaying.

The distinction from the ws_uri media-profile flag matters:

ws_uri is a takeover — the WebSocket server becomes leg A's far side and the A↔B relay is not wired. That is the voice-AI answer-the-call shape.
A tee is send-only and additive — the call relays (or transcodes) normally and streams a copy of its audio out. Any SIPREC subscription and recording on the same leg keep running untouched.

Use a tee for live transcription, agent-assist, sentiment or compliance monitoring on a call that is otherwise a normal two-party call.

A tee never affects the call: the engine drops frames rather than stalling the media path if the consumer cannot keep up, and a failure raises rather than tearing anything down — catch it and carry on.

Requires media.backend: siphon-rtp; the rtpengine and rtpproxy backends raise rather than silently doing nothing.

Parameters:

Name	Type	Description	Default
`target`	`Any`	Request, Reply, or Call object.	required
`ws_uri`	`str`	`ws://` or `wss://` URI the engine dials as a client.	required
`direction`	`str`	Which leg(s) to stream — `"both"` (default), `"caller"` (the offerer) or `"callee"` (the answerer).	`'both'`
`channels`	`Optional[int]`	Wire channel count — `2` interleaves caller/callee as stereo, `1` mixes them to mono. Only meaningful with `direction="both"`; a single-leg tee is always mono. `None` (default) leaves the engine's choice: 2 for both legs, 1 for one.	`None`
`sample_rate`	`Optional[int]`	L16 wire sample rate in Hz, independent of the legs' codec rates -- the engine resamples the teed copy into it. Must be a multiple of 1000 within 8000-48000; the engine fails the attach on anything else rather than clamping, so it is checked here first. `None` (default) leaves the engine's choice.	`None`

Returns:

Type	Description
`bool`	`True` on success.

Example::

@b2bua.on_answer
async def on_answer(call, reply):
    await rtpengine.answer(reply)
    try:
        await rtpengine.attach_ws_tee(call, f"wss://asr.internal/{call.call_id}")
    except RuntimeError as error:
        log.warn(f"transcription tee unavailable: {error}")

detach_ws_tee `async` ¶

detach_ws_tee(target: Any) -> bool

Detach a call's WebSocket tee, closing its stream.

Idempotent — detaching a call with no tee is not an error. A tee is also torn down automatically when the call ends, so an explicit detach is only needed to stop streaming mid-call.

Requires media.backend: siphon-rtp.

Parameters:

Name	Type	Description	Default
`target`	`Any`	Request, Reply, or Call object.	required

Returns:

Type	Description
`bool`	`True` on success.

Example::

await rtpengine.detach_ws_tee(call)

attach_ws_bridge `async` ¶

attach_ws_bridge(target: Any, ws_uri: str) -> bool

Attach a WebSocket takeover bridge to a live call, or re-point an existing one at a different server.

The opposite of :meth:attach_ws_tee in what it does to the call. A tee is additive — the call keeps relaying and a copy is streamed out. A bridge is a takeover: the WebSocket server becomes this leg's far side and A<->B is unwired for as long as the bridge lives.

Calling it on a call that already has a bridge re-points it rather than failing, and the media path never drops in between — which is what lets one party be moved from one media server to another without the other party hearing a gap.

Requires media.backend: siphon-rtp.

Parameters:

Name	Type	Description	Default
`target`	`Any`	Request, Reply, or Call object.	required
`ws_uri`	`str`	`ws://` or `wss://` URI the engine dials as a client.	required

Returns:

Type	Description
`bool`	`True` on success.

Example::

await rtpengine.attach_ws_bridge(call, "wss://ai.internal/session-1")
# ... later, hand the same caller to a different model session:
await rtpengine.attach_ws_bridge(call, "wss://ai.internal/session-2")

detach_ws_bridge `async` ¶

detach_ws_bridge(target: Any) -> bool

Detach a call's WebSocket takeover bridge, putting its media path back the way it was.

Not idempotent, unlike :meth:detach_ws_tee. The engine refuses a detach when there is no relay to return the call to — a bridge negotiated through ws_uri on the media profile is the call's media path, and a single-leg (answer_local) takeover has no second party that could ever be relayed to. Both raise rather than answering success, because the alternative is a live call with no audio path at all. Re-point those with :meth:attach_ws_bridge, or end the call.

Requires media.backend: siphon-rtp.

Parameters:

Name	Type	Description	Default
`target`	`Any`	Request, Reply, or Call object.	required

Returns:

Type	Description
`bool`	`True` on success.

Example::

await rtpengine.detach_ws_bridge(call)   # back to relaying A<->B

on_dtmf ¶

on_dtmf(
    func_or_none: Any = None,
    *,
    call_id: Optional[str] = None,
    from_tag: Optional[str] = None
) -> Any

Register a handler for inbound DTMF events from rtpengine.

Usage::

@rtpengine.on_dtmf
def handle_any(call_id, from_tag, digit, duration_ms, volume):
    ...

@rtpengine.on_dtmf(call_id="abc", from_tag="ftag1")
def handle_specific(call_id, from_tag, digit, duration_ms, volume):
    ...

fire_dtmf ¶

fire_dtmf(
    call_id: str,
    from_tag: str,
    digit: str,
    duration_ms: int = 0,
    volume: int = 0,
) -> int

Test helper: fire a DTMF event. Returns the number of handlers that matched (and were invoked).

on_media_timeout ¶

on_media_timeout(
    func_or_none: Any = None,
    *,
    call_id: Optional[str] = None,
    from_tag: Optional[str] = None
) -> Any

Register a handler for media-timeout events from the media engine.

The engine reaps a call whose media went dead and pushes a media-timeout event; the handler releases the per-call state no BYE will now clear (Rx/N5 QoS, charging, dialog).

Usage::

@rtpengine.on_media_timeout
def handle_any(call_id, from_tag):
    ...

@rtpengine.on_media_timeout(call_id="abc", from_tag="ftag1")
def handle_specific(call_id, from_tag):
    ...

fire_media_timeout ¶

fire_media_timeout(call_id: str, from_tag: str) -> int

Test helper: fire a media-timeout event. Returns the number of handlers that matched (and were invoked).

on_text ¶

on_text(
    func_or_none: Any = None,
    *,
    call_id: Optional[str] = None,
    from_tag: Optional[str] = None
) -> Any

Register a handler for RFC 4103 real-time text (T.140) increments.

Fires once per increment the engine's text processor recovers on the call's m=text stream, carrying the UTF-8 text that packet newly delivered. Only non-empty increments are reported — a duplicate, a reordered packet or an idle keepalive produces no event — so the handler firing always means new characters arrived. A � in the text is a gap RED redundancy could not repair (RFC 4103 §5.3), left in place so a consumer sees where loss occurred rather than silently reading a shorter message.

Requires the call's media profile to set text_events. Delivered by the native siphon-rtp backend only.

Usage::

@rtpengine.on_text
def transcript(call_id, from_tag, to_tag, text, direction):
    log.info(f"[{call_id}] {direction}: {text}")

@rtpengine.on_text(call_id="abc", from_tag="ftag1")
def transcript_specific(call_id, from_tag, to_tag, text, direction):
    ...

fire_text ¶

fire_text(
    call_id: str,
    from_tag: str,
    text: str,
    to_tag: Optional[str] = None,
    direction: Optional[str] = "a_to_b",
) -> int

Test helper: fire a real-time text event. Returns the number of handlers that matched (and were invoked).

on_beep ¶

on_beep(
    func_or_none: Any = None,
    *,
    call_id: Optional[str] = None,
    from_tag: Optional[str] = None
) -> Any

Register a handler for record-tone ("voicemail beep") events.

Fires when the engine hears the short single tone an answering machine plays before it starts recording, on a leg whose media profile set beep_detection. This is the media half of answering-machine detection: a script can abort an attended transfer here instead of bridging the caller into a voicemail box.

Arm it per leg -- the profile used toward the callee is what watches the party that might be a machine. It fires once per leg per call (the engine drops the detector after the first tone, so a handler never has to de-duplicate, and there is no mid-call re-arm).

offset_ms is how much decoded audio was seen on the leg before the tone started -- the offset of the tone itself, not of this event. The event trails it by roughly the profile's beep_cadence_guard_ms (4500 ms by default), which is the detector's cadence guard and its detection latency.

Delivered by the native siphon-rtp backend only.

Usage::

@rtpengine.on_beep
def machine(call_id, from_tag, to_tag, frequency_hz, duration_ms, offset_ms):
    log.info(f"{call_id}: answering machine ({frequency_hz:.0f} Hz)")
    b2bua.terminate(call_id, "Answering machine detected")

@rtpengine.on_beep(call_id="abc", from_tag="ftag1")
def machine_specific(call_id, from_tag, to_tag, frequency_hz, duration_ms, offset_ms):
    ...

fire_beep ¶

fire_beep(
    call_id: str,
    from_tag: str,
    to_tag: Optional[str] = None,
    frequency_hz: float = 1000.0,
    duration_ms: int = 420,
    offset_ms: int = 7300,
) -> int

Test helper: fire a record-tone event. Returns the number of handlers that matched (and were invoked).

The defaults describe a typical voicemail beep: a ~1 kHz tone a few hundred milliseconds long, several seconds into the leg's audio.

on_ws_tee_started ¶

on_ws_tee_started(
    func_or_none: Any = None,
    *,
    call_id: Optional[str] = None,
    from_tag: Optional[str] = None
) -> Any

Register a handler for WebSocket tee started events.

Fires once the engine has dialled the tee's WebSocket server, sent its start envelope, and begun streaming. The handler receives the negotiated wire shape, so it can decode the binary frames without guessing — stream_id is the correlator between this control event and the media stream on the socket.

Delivered by the native siphon-rtp backend only.

Usage::

@rtpengine.on_ws_tee_started
def tee_up(call_id, from_tag, stream_id, ws_uri, direction, channels, sample_rate):
    log.info(f"tee {stream_id}: {channels}ch @ {sample_rate}Hz -> {ws_uri}")

@rtpengine.on_ws_tee_started(call_id="abc", from_tag="ftag1")
def tee_up_specific(call_id, from_tag, stream_id, ws_uri, direction, channels, sample_rate):
    ...

fire_ws_tee_started ¶

fire_ws_tee_started(
    call_id: str,
    from_tag: str,
    stream_id: str,
    ws_uri: str,
    direction: str = "both",
    channels: int = 2,
    sample_rate: int = 8000,
) -> int

Test helper: fire a ws-tee-started event. Returns the number of handlers that matched (and were invoked).

on_ws_tee_ended ¶

on_ws_tee_ended(
    func_or_none: Any = None,
    *,
    call_id: Optional[str] = None,
    from_tag: Optional[str] = None
) -> Any

Register a handler for WebSocket tee ended events.

Fires exactly once per started tee, including when the server ends it. That is the point of the hook: any reason other than "detached" means the audio stream died while the call is still up, which is otherwise invisible — the call carries on and nothing reaches the consumer. Re-attach, fail over, or alert from here.

reason is one of "detached" (the script or the call teardown asked for it — the only orderly end), "server_closed", "server_stopped", "call_ended" or "transport_error".

frames_dropped non-zero means the consumer could not keep up; the call itself was never affected.

Delivered by the native siphon-rtp backend only.

Usage::

@rtpengine.on_ws_tee_ended
async def tee_down(call_id, from_tag, stream_id, reason, frames_sent, frames_dropped):
    if reason != "detached":
        log.warn(f"tee {stream_id} died: {reason}")

fire_ws_tee_ended ¶

fire_ws_tee_ended(
    call_id: str,
    from_tag: str,
    stream_id: str,
    reason: str = "detached",
    frames_sent: Optional[int] = None,
    frames_dropped: Optional[int] = None,
) -> int

Test helper: fire a ws-tee-ended event. Returns the number of handlers that matched (and were invoked).

on_play_finished ¶

on_play_finished(
    func_or_none: Any = None,
    *,
    call_id: Optional[str] = None,
    from_tag: Optional[str] = None
) -> Any

Register a handler for playback finished events.

Fires when a playback the engine accepted ends — for every play, including a fire-and-forget one. A blocking play_media(wait=True) already learns the outcome as its return value; this is for everything else, and it is what an announcement-then-act flow hangs on instead of a timer that a stop or a decode error would make wrong.

reason is one of completed, stopped, superseded or error. Only completed means the prompt was heard in full. play_id correlates with the accept and with PlayStarted.

Delivered by the native siphon-rtp backend only.

Usage::

@rtpengine.on_play_finished
async def prompt_done(call_id, from_tag, play_id, reason, played_ms):
    if reason == "completed":
        await rtpengine.play_dtmf(call_id, "1")

fire_play_finished ¶

fire_play_finished(
    call_id: str,
    from_tag: str,
    play_id: int,
    reason: str = "completed",
    played_ms: Optional[int] = None,
) -> int

Test helper: fire a play-finished event. Returns the number of handlers that matched (and were invoked).

on_ws_bridge_started ¶

on_ws_bridge_started(
    func_or_none: Any = None,
    *,
    call_id: Optional[str] = None,
    from_tag: Optional[str] = None
) -> Any

Register a handler for WebSocket takeover bridge started events.

Fires once the engine has dialled the bridge's WebSocket server and the leg's far side is that server — A<->B is unwired for the bridge's lifetime. stream_id is the correlator between this control event and the media stream on the socket.

A re-point (attach_ws_bridge on a call that already had one) ends the old bridge and starts a new one, so it delivers an ended with reason detached followed by a fresh started carrying the new stream_id.

Delivered by the native siphon-rtp backend only.

Usage::

@rtpengine.on_ws_bridge_started
def bridge_up(call_id, from_tag, stream_id, ws_uri, sample_rate):
    log.info(f"bridge {stream_id} @ {sample_rate}Hz -> {ws_uri}")

fire_ws_bridge_started ¶

fire_ws_bridge_started(
    call_id: str,
    from_tag: str,
    stream_id: str,
    ws_uri: str,
    sample_rate: int = 8000,
) -> int

Test helper: fire a ws-bridge-started event. Returns the number of handlers that matched (and were invoked).

on_ws_bridge_ended ¶

on_ws_bridge_ended(
    func_or_none: Any = None,
    *,
    call_id: Optional[str] = None,
    from_tag: Optional[str] = None
) -> Any

Register a handler for WebSocket takeover bridge ended events.

Fires exactly once per started bridge, including when the server ends it. reason is one of detached, server_closed, server_stopped, call_ended or transport_error.

Only detached is orderly. Every other reason leaves a live call with no media far side — both parties are up and hearing nothing — so unlike the tee's equivalent this handler usually has to act: re-point with attach_ws_bridge, fall back with detach_ws_bridge, or tear the call down. siphon logs an unexpected end at WARN even when no handler is registered.

Delivered by the native siphon-rtp backend only.

Usage::

@rtpengine.on_ws_bridge_ended
async def bridge_down(call_id, from_tag, stream_id, reason):
    if reason != "detached":
        log.warn(f"{call_id}: bridge died ({reason})")

fire_ws_bridge_ended ¶

fire_ws_bridge_ended(
    call_id: str,
    from_tag: str,
    stream_id: str,
    reason: str = "detached",
) -> int

Test helper: fire a ws-bridge-ended event. Returns the number of handlers that matched (and were invoked).

set_subscribe_request_sdp ¶

set_subscribe_request_sdp(sdp: bytes) -> None

Configure the SDP returned by :meth:subscribe_request (test helper).

set_subscribe_answer_sdp ¶

set_subscribe_answer_sdp(sdp: bytes) -> None

Configure the SDP returned by :meth:subscribe_answer (test helper).

set_play_media_duration ¶

set_play_media_duration(duration_ms: Optional[int]) -> None

Configure the duration returned by :meth:play_media (test helper).

set_play_overlay_id ¶

set_play_overlay_id(play_id: Optional[int]) -> None

Configure the play_id returned by :meth:play_overlay (test helper).

Set it to None to model an engine that accepted the overlay without assigning a handle, which is what a script's "can I duck this later" branch has to cope with.

set_answer_local_sdp ¶

set_answer_local_sdp(sdp: str) -> None

Configure the answer SDP returned by :meth:answer_local (test helper).

set_answer_local_no_codec ¶

set_answer_local_no_codec(no_codec: bool = True) -> None

Prime :meth:answer_local to model a no-encodable-codec offer (test helper) — the next answer_local records a deferred 488 (auto-reject) or raises ValueError (auto_reject=False).

clear ¶

clear() -> None

Clear recorded operations and registered event handlers (test helper).

`qos` namespace¶

Mock qos namespace — turns SDP offer/answer pairs into the media_components structure consumed by diameter.rx_aar and sbi.create_session.

The mock parses SDP just enough to produce a usable media_components list with RTP + RTCP sub-components for each m= section. Disabled streams (port 0) are skipped and a=rtcp-mux collapses RTCP into the RTP sub-component.

Example::

from siphon import qos, diameter

components = qos.media_flows_from_sdp(
    offer=request.body, answer=reply.body, direction="orig",
)
await diameter.rx_aar(framed_ip=request.source_ip, media_components=components)

media_flows_from_sdp ¶

media_flows_from_sdp(
    *, offer: Any, answer: Any, direction: str = "orig"
) -> list[dict]

Translate an SDP offer/answer pair into a media_components list.

Parameters:

Name	Type	Description	Default
`offer`	`Any`	the original (offer) SDP — `str`, `bytes`, or a Request/Reply/Call mock with a `body` attribute.	required
`answer`	`Any`	the answer SDP (typically post `rtpengine.answer()`).	required
`direction`	`str`	`"orig"` (UE is offerer — UE addr from `offer`, remote from `answer`) or `"term"` (UE is answerer — addresses flipped).	`'orig'`

Returns:

Type	Description
`list[dict]`	list[dict]: one entry per non-disabled `m=` section.

Media¶

rtpengine namespace¶

operations instance-attribute ¶

media_calls instance-attribute ¶

ws_uris instance-attribute ¶

media_overrides instance-attribute ¶

active_sessions property ¶

instance_count property ¶

offer async ¶

answer async ¶

answer_local async ¶

delete async ¶

ping async ¶

play_media async ¶

play_overlay async ¶

stop_media async ¶

set_play_gain async ¶

play_dtmf async ¶

silence_media async ¶

unsilence_media async ¶

block_media async ¶

unblock_media async ¶

echo async ¶

subscribe_request async ¶

subscribe_answer async ¶

unsubscribe async ¶

attach_ws_tee async ¶

detach_ws_tee async ¶

attach_ws_bridge async ¶

detach_ws_bridge async ¶

on_dtmf ¶

fire_dtmf ¶

on_media_timeout ¶

fire_media_timeout ¶

on_text ¶

fire_text ¶

on_beep ¶

fire_beep ¶

on_ws_tee_started ¶

fire_ws_tee_started ¶

on_ws_tee_ended ¶

fire_ws_tee_ended ¶

on_play_finished ¶

fire_play_finished ¶

on_ws_bridge_started ¶

fire_ws_bridge_started ¶

on_ws_bridge_ended ¶

fire_ws_bridge_ended ¶

set_subscribe_request_sdp ¶

set_subscribe_answer_sdp ¶

set_play_media_duration ¶

set_play_overlay_id ¶

set_answer_local_sdp ¶

set_answer_local_no_codec ¶

clear ¶

qos namespace¶

media_flows_from_sdp ¶

`rtpengine` namespace¶

operations `instance-attribute` ¶

media_calls `instance-attribute` ¶

ws_uris `instance-attribute` ¶

media_overrides `instance-attribute` ¶

active_sessions `property` ¶

instance_count `property` ¶

offer `async` ¶

answer `async` ¶

answer_local `async` ¶

delete `async` ¶

ping `async` ¶

play_media `async` ¶

play_overlay `async` ¶

stop_media `async` ¶

set_play_gain `async` ¶

play_dtmf `async` ¶

silence_media `async` ¶

unsilence_media `async` ¶

block_media `async` ¶

unblock_media `async` ¶

echo `async` ¶

subscribe_request `async` ¶

subscribe_answer `async` ¶

unsubscribe `async` ¶

attach_ws_tee `async` ¶

detach_ws_tee `async` ¶

attach_ws_bridge `async` ¶

detach_ws_bridge `async` ¶

`qos` namespace¶