Proxy & B2BUA¶

The proxy and b2bua namespaces register the event handlers that make routing decisions, plus the helpers those handlers lean on (rate limiting, sanity checks, ENUM lookup) and the generic SUBSCRIBE-dialog state store.

from siphon import proxy, b2bua

@proxy.on_request
def route(request):
    request.relay()

@b2bua.on_invite
async def call(call):
    call.dial(call.ruri)

`proxy` namespace¶

Mock proxy namespace with decorator registration and utility stubs.

Decorators

@proxy.on_request / @proxy.on_request("INVITE")
@proxy.on_reply
@proxy.on_failure
@proxy.on_cancel
@proxy.on_register_reply

Example::

from siphon import proxy

@proxy.on_request("REGISTER")
def handle_register(request):
    request.reply(200, "OK")

sent_requests `property` ¶

sent_requests: list[dict]

List of requests sent via send_request() (for test assertions).

on_request ¶

on_request(
    fn_or_filter: Union[Callable, str, None] = None,
) -> Any

Register a handler for incoming SIP requests.

Can be used as

@proxy.on_request — handle all methods
@proxy.on_request() — same, explicit call
@proxy.on_request("REGISTER") — single method filter
@proxy.on_request("INVITE|SUBSCRIBE") — pipe-separated filter

on_reply `staticmethod` ¶

on_reply(fn: Callable) -> Callable

Register a handler for SIP replies.

Handler signature: (request, reply) -> None

on_failure `staticmethod` ¶

on_failure(fn: Callable) -> Callable

Register a handler for proxy failure (all branches failed).

Handler signature: (request, reply) -> None

on_cancel `staticmethod` ¶

on_cancel(fn: Callable) -> Callable

Register a handler for a CANCELled INVITE (RFC 3261 §9).

Handler signature: (request) -> None

Fires once, with the original INVITE, when a relayed INVITE is CANCELled before any final response — the one teardown that neither on_reply nor on_failure delivers (the proxy answers the CANCEL with 487 at the transaction layer and the session is gone). Use it to release per-call resources that no BYE will ever clear: Diameter Rx / N5 QoS sessions, rtpengine media anchors, charging maps.

Fire-and-forget — it does not gate or alter the 487 sent to the UAC.

Example::

@proxy.on_cancel
async def handle_cancel(request):
    await _release_qos(request.call_id)
    await rtpengine.delete(request)

on_register_reply `staticmethod` ¶

on_register_reply(fn: Callable) -> Callable

Register a handler for REGISTER replies.

Handler signature: (request, reply) -> None

send_request `async` ¶

send_request(
    method: str,
    ruri: str,
    headers: Optional[dict[str, str]] = None,
    body: Optional[Any] = None,
    next_hop: Optional[str] = None,
    wait_for_response: bool = False,
    timeout_ms: int = 2000,
) -> Any

Originate an outbound SIP request.

Always returns an awaitable — scripts must await it. Fire-and-forget by default; when wait_for_response=True, the awaitable resolves to a configured mock Reply (or None on timeout).

Parameters:

Name	Type	Description	Default
`method`	`str`	SIP method name (e.g. "NOTIFY", "OPTIONS", "MESSAGE").	required
`ruri`	`str`	Request-URI string (e.g. "sip:alice@10.0.0.1:5060").	required
`headers`	`Optional[dict[str, str]]`	Optional dict of header name → value to add. When a `Route` header is supplied without `next_hop`, the request is sent to the first `Route` entry's URI (its `;lr` loose-route target) per RFC 3261 §8.1.2 — the R-URI stays in the Request-Line. Use this to steer a request straight to a known next hop (e.g. a served IMPU's serving S-CSCF) instead of resolving the R-URI's home domain.	`None`
`body`	`Optional[Any]`	Optional body — `str` or `bytes`.	`None`
`next_hop`	`Optional[str]`	Optional next-hop URI override. Outranks a `Route` header for next-hop selection.	`None`
`wait_for_response`	`bool`	When `True`, return the configured mock reply.	`False`
`timeout_ms`	`int`	Response timeout (not meaningfully enforced in the mock).	`2000`

set_response_for ¶

set_response_for(
    method: str, ruri: str, reply: Any
) -> None

Test helper: configure the mock reply returned by send_request(wait_for_response=True) for a given (method, ruri).

Parameters:

Name	Type	Description	Default
`method`	`str`	SIP method (e.g. "OPTIONS").	required
`ruri`	`str`	Request-URI the script will pass.	required
`reply`	`Any`	Any object (often a `MockReply`) — returned to the script.	required

`proxy` utilities¶

Reached as proxy.rate_limit, proxy.sanity_check, proxy.enum_lookup, and proxy.memory_used_pct.

Mock proxy._utils namespace.

Provides rate limiting, sanity checking, ENUM lookup, and memory stats. In the mock, these return configurable defaults.

rate_limit ¶

rate_limit(
    request: Any, window_secs: float, max_requests: int
) -> bool

Check if a request is within the rate limit.

Parameters:

Name	Type	Description	Default
`request`	`Any`	The SIP request object.	required
`window_secs`	`float`	Sliding window duration in seconds.	required
`max_requests`	`int`	Maximum requests allowed in the window.	required

Returns:

Type	Description
`bool`	`True` if allowed, `False` if rate-limited.

In the mock, returns the value of _rate_limit_allow (default True).

sanity_check ¶

sanity_check(request: Any) -> bool

Validate request per RFC 3261 (mandatory headers, Max-Forwards, etc.).

Returns:

Type	Description
`bool`	`True` if valid, `False` otherwise.

In the mock, returns _sanity_check_pass (default True).

enum_lookup `async` ¶

enum_lookup(
    number: str,
    suffix: str = "e164.arpa.",
    service: str = "E2U+sip",
) -> Optional[str]

DNS NAPTR lookup for phone number to SIP URI.

Parameters:

Name	Type	Description	Default
`number`	`str`	E.164 number (e.g. `"+14155552671"`).	required
`suffix`	`str`	DNS suffix (default `"e164.arpa."`).	`'e164.arpa.'`
`service`	`str`	Service type (default `"E2U+sip"`).	`'E2U+sip'`

Returns:

Type	Description
`Optional[str]`	SIP URI string or `None`.

In the mock, looks up _enum_results dict.

memory_used_pct ¶

memory_used_pct() -> int

Process RSS memory usage as percentage (0–100).

In the mock, returns _memory_pct (default 25).

`b2bua` namespace¶

Mock B2BUA namespace with decorator registration.

Decorators

@b2bua.on_invite — new call
@b2bua.on_early_media — provisional response with SDP (183/180)
@b2bua.on_answer — call answered
@b2bua.on_failure — all B-legs failed
@b2bua.on_bye — call ended
@b2bua.on_refer — call transfer (RFC 3515)
@b2bua.on_cancel — unanswered call cancelled (RFC 3261 §9)

Imperative

b2bua.terminate(call_id) — end a call by SIP Call-ID from any context (records onto terminates for test assertions)
b2bua.refer(call_id, target) — transfer a call by SIP Call-ID from any context (records onto refers for test assertions)

clear ¶

clear() -> None

Reset recorded imperative calls (called by reset()).

terminate ¶

terminate(
    call_id: str, reason: str = "Normal Clearing"
) -> bool

Imperatively end a B2BUA call by its SIP Call-ID.

Unlike call.terminate() (deferred until its handler returns), this acts immediately and is keyed by SIP Call-ID, so it works from an out-of-band event callback (@rtpengine.on_dtmf, @rtpengine.on_media_timeout), a timer, or a normal handler.

Parameters:

Name	Type	Description	Default
`call_id`	`str`	the SIP Call-ID of the call to end.	required
`reason`	`str`	free-text hangup reason (RFC 3326 `Reason` on the BYE).	`'Normal Clearing'`

Returns:

Name	Type	Description
`bool`	`bool`	True if a matching call was found and torn down, False if the
	`bool`	Call-ID is unknown / already gone. Never raises.

In the mock, records {"call_id", "reason"} on terminates and returns True. Inspect via siphon.get_b2bua().terminates.

Usage::

@rtpengine.on_dtmf
def on_ivr_dtmf(call_id, from_tag, digit, duration_ms, volume):
    if digit == "#":
        b2bua.terminate(call_id)

refer ¶

refer(
    call_id: str,
    target: str,
    replaces: Optional[dict] = None,
) -> bool

Imperatively transfer a B2BUA call by its SIP Call-ID.

The imperative twin of :meth:Call.refer. Unlike call.refer() (a deferred call action, honoured after its handler returns), this acts immediately and is keyed by SIP Call-ID, so it works from an out-of-band event callback (@rtpengine.on_dtmf, a timer) where no call object is in scope and deferred actions are no-ops — the same reason :meth:terminate exists alongside call.terminate().

Parameters:

Name	Type	Description	Default
`call_id`	`str`	the SIP Call-ID of the call to transfer.	required
`target`	`str`	the Refer-To URI (transfer destination).	required
`replaces`	`Optional[dict]`	optional attended-transfer dict (RFC 3891) with `call_id` / `from_tag` / `to_tag` (and an optional `early_only`); `None` for a blind transfer.	`None`

Returns:

Name	Type	Description
`bool`	`bool`	True if a matching call was found and the REFER was
	`bool`	originated, False if the Call-ID is unknown / already gone.
	`bool`	Never raises for a missing call.

Raises:

Type	Description
`ValueError`	if `replaces` is given but missing any of `call_id` / `from_tag` / `to_tag`.

In the mock, records {"call_id", "target", "replaces"} on refers and returns True. Inspect via siphon.get_b2bua().refers.

Usage::

@rtpengine.on_dtmf
def on_ivr_dtmf(call_id, from_tag, digit, duration_ms, volume):
    if digit == "*":
        b2bua.refer(call_id, "sip:+15550142@example.com")

on_invite `staticmethod` ¶

on_invite(fn: Callable) -> Callable

Register handler for new INVITE (new call).

Handler signature: (call) -> None

on_early_media `staticmethod` ¶

on_early_media(fn: Callable) -> Callable

Register handler for provisional response with SDP (183/180).

Called when the B-leg sends a provisional response containing SDP (early media). Use this to process the SDP through RTPEngine so early media is anchored correctly.

Handler signature: (call, reply) -> None

Example::

@b2bua.on_early_media
async def early_media(call, reply):
    await rtpengine.answer(reply)

on_answer `staticmethod` ¶

on_answer(fn: Callable) -> Callable

Register handler for call answered (200 OK on B-leg).

Handler signature: (call, reply) -> None

on_failure `staticmethod` ¶

on_failure(fn: Callable) -> Callable

Register handler for B-leg failure.

Handler signature: (call, code, reason) -> None

on_bye `staticmethod` ¶

on_bye(fn: Callable) -> Callable

Register handler for BYE (call ended).

Handler signature: (call, initiator) -> None

initiator is a :class:ByeInitiator with a .side property ("a" or "b").

on_refer `staticmethod` ¶

on_refer(fn: Callable) -> Callable

Register handler for REFER (call transfer, RFC 3515).

Handler signature is single-arg (call) -> None. A REFER is a SIP request, not a response, so there is no reply object — do NOT write (call, reply) and do NOT call rtpengine.answer() here. Read the transfer target off :attr:Call.refer_to (and :attr:Call.refer_replaces for an attended transfer), then decide with :meth:Call.accept_refer or :meth:Call.reject_refer.

Example::

@b2bua.on_refer
def handle_refer(call):
    log.info(f"Transfer requested to {call.refer_to}")
    call.accept_refer()

on_cancel `staticmethod` ¶

on_cancel(fn: Callable) -> Callable

Register handler for a CANCELled call (RFC 3261 §9).

Handler signature: (call) -> None

Fires once, with the Call object, when an unanswered call (Calling/Ringing) is CANCELled — the teardown that on_failure (B-leg error) and on_bye (answered call) never cover. A 2xx that wins the CANCEL/answer glare is ACK+BYE'd by the framework and never delivers on_answer, so this hook only ever sees a genuinely abandoned call. Use it to release per-call resources that no BYE will clear: rtpengine media anchors, QoS sessions.

Example::

@b2bua.on_cancel
async def handle_cancel(call):
    await rtpengine.delete(call)

`proxy.subscribe_state`¶

Generic SUBSCRIBE-dialog state (RFC 6665) for any event package, with optional Redis-backed persistence.

Mock of the Rust proxy.subscribe_state namespace.

Used from scripts under test as proxy.subscribe_state.create(request). Records NOTIFY and terminate invocations on notifies / terminates lists for test assertions.

send ¶

send(
    ruri: str,
    event: str,
    expires: int,
    accept: Optional[str] = None,
    target_uri: Optional[str] = None,
    headers: Optional[dict] = None,
    timeout_ms: int = 2000,
) -> MockSubscribeHandle

Mock outbound SUBSCRIBE — records the call and synthesises a dialog.

Tests can assert on the recorded self.sends list to verify a script originated a SUBSCRIBE with the expected parameters.

find ¶

find(
    call_id: str, local_tag: str, remote_tag: str
) -> Optional[MockSubscribeHandle]

Mock dialog lookup by tags. Returns the first live dialog matching all three identity fields, or None.

`SubscribeHandle`¶

A single subscription dialog returned by proxy.subscribe_state.create(...).

Mock of the Rust SubscribeHandle.

In the mock, NOTIFY / terminate calls are recorded on the parent MockSubscribeState for test assertions. No real SIP message is produced.

event_version `property` ¶

event_version: int

Current event-package body version (read-only).

Mirrors the Rust SubscribeHandle.event_version — used for RFC 3680 reginfo / RFC 4235 dialog-info / RFC 4575 conference bodies that require a monotonic version= attribute.

next_event_version ¶

next_event_version() -> int

Atomically increment and return the next event-package body version.

Call before building a NOTIFY body whose monotonicity matters::

version = handle.next_event_version()
body = registrar.reginfo_xml(aor, state="full", version=version)
handle.notify(body=body, content_type="application/reginfo+xml")

refresh ¶

refresh(
    expires: Optional[int] = None, timeout_ms: int = 2000
) -> bool

Mock refresh — records the call and updates the dialog's expiry.

Tests can assert on the parent's refreshes list. Raises if the dialog wasn't created via send() (consistent with the Rust contract that refresh is only valid on outbound dialogs).

Proxy & B2BUA¶

proxy namespace¶

sent_requests property ¶

on_request ¶

on_reply staticmethod ¶

on_failure staticmethod ¶

on_cancel staticmethod ¶

on_register_reply staticmethod ¶

send_request async ¶

set_response_for ¶

proxy utilities¶

rate_limit ¶

sanity_check ¶

enum_lookup async ¶

memory_used_pct ¶

b2bua namespace¶

clear ¶

terminate ¶

refer ¶

on_invite staticmethod ¶

on_early_media staticmethod ¶

on_answer staticmethod ¶

on_failure staticmethod ¶

on_bye staticmethod ¶

on_refer staticmethod ¶

on_cancel staticmethod ¶

proxy.subscribe_state¶

send ¶

find ¶

SubscribeHandle¶

event_version property ¶

next_event_version ¶

refresh ¶

`proxy` namespace¶

sent_requests `property` ¶

on_reply `staticmethod` ¶

on_failure `staticmethod` ¶

on_cancel `staticmethod` ¶

on_register_reply `staticmethod` ¶

send_request `async` ¶

`proxy` utilities¶

enum_lookup `async` ¶

`b2bua` namespace¶

on_invite `staticmethod` ¶

on_early_media `staticmethod` ¶

on_answer `staticmethod` ¶

on_failure `staticmethod` ¶

on_bye `staticmethod` ¶

on_refer `staticmethod` ¶

on_cancel `staticmethod` ¶

`proxy.subscribe_state`¶

`SubscribeHandle`¶

event_version `property` ¶