Gateway

The gateway namespace manages named groups of SIP destinations with health probing and pluggable load-balancing (round-robin, weighted, consistent hash).

from siphon import gateway

gw = gateway.select("carriers", key=request.call_id)
if gw:
    request.relay(gw.uri)

gateway namespace

Mock gateway namespace — manages named groups of SIP destinations.

Pre-populate groups for testing::

from siphon import gateway
gateway.add_group("carriers", [
    {"uri": "sip:gw1.carrier.com:5060", "address": "10.0.0.1:5060", "weight": 3},
    {"uri": "sip:gw2.carrier.com:5060", "address": "10.0.0.2:5060"},
], algorithm="weighted")

Then in your script::

gw = gateway.select("carriers")
gw = gateway.select("sbc-pool", key=request.call_id)
gw = gateway.select("carriers", attrs={"region": "us-east"})

select

select(
    group_name: str,
    /,
    key: Optional[str] = None,
    attrs: Optional[dict[str, str]] = None,
) -> Optional[MockDestination]

Select a destination from a named group.

Parameters:

Name	Type	Description	Default
group_name	str	Name of the gateway group (e.g. "carriers").	required
key	Optional[str]	Optional hash key for sticky sessions (e.g. call_id). Used by the "hash" algorithm.	None
attrs	Optional[dict[str, str]]	Optional dict of attribute filters. Only destinations matching all key-value pairs are considered.	None

Returns:

Name	Type	Description
A	Optional[MockDestination]	class:MockDestination object, or None if no healthy
	Optional[MockDestination]	destination matches.

Example::

gw = gateway.select("carriers")
gw = gateway.select("sbc-pool", key=request.call_id)
gw = gateway.select("carriers", attrs={"region": "us-east"})

contains_source

contains_source(group_name: str, source_ip: str) -> bool

True when source_ip is a member IP of the named group.

Mirrors the Rust DispatcherManager::source_in_group — the backing check for request.from_gateway / call.from_gateway. Matches on IP only (destination port ignored) against every destination's address. Returns False (never raises) for an unknown group or an unparseable source_ip, so callers stay infallible.

In the mock, membership is the set of destination-address IP literals you registered via :meth:add_group (no DNS is performed — give destinations literal IP addresses to model resolved gateways).

Example::

gateway.add_group("teams", [
    {"uri": "sip:sip.pstnhub.microsoft.com", "address": "203.0.113.10:5061"},
])
gateway.contains_source("teams", "203.0.113.10")  # True

list

list(group_name: str) -> list[MockDestination]

List all destinations in a group.

Returns:

Type	Description
list[MockDestination]	List of :class:MockDestination objects (healthy and unhealthy).

status

status(group_name: str) -> list[tuple[str, bool]]

Get status of all destinations in a group.

Returns:

Type	Description
list[tuple[str, bool]]	List of (uri, is_healthy) tuples.

groups

groups() -> list[str]

List all group names.

add_group

add_group(
    name: str,
    destinations: list[dict[str, Any]],
    /,
    algorithm: str = "weighted",
    probe: bool = False,
) -> None

Dynamically add a new gateway group.

Parameters:

Name	Type	Description	Default
name	str	Group name.	required
destinations	list[dict[str, Any]]	List of dicts with keys: uri (required), address, weight, priority, transport, attrs.	required
algorithm	str	Load-balancing algorithm: "round_robin", "weighted" (default), "hash".	'weighted'
probe	bool	Enable health probing (ignored in mock).	False

Example::

gateway.add_group("overflow", [
    {"uri": "sip:gw3.carrier.com", "address": "10.0.0.3:5060", "weight": 2},
    {"uri": "sip:gw4.carrier.com", "address": "10.0.0.4:5060"},
], algorithm="weighted")

remove_group

remove_group(name: str) -> bool

Remove a group by name.

Returns:

Type	Description
bool	True if the group existed and was removed.

mark_down

mark_down(group_name: str, uri: str) -> bool

Manually mark a destination as down.

Returns:

Type	Description
bool	True if the destination was found.

mark_up

mark_up(group_name: str, uri: str) -> bool

Manually mark a destination as up.

Returns:

Type	Description
bool	True if the destination was found.

clear

clear() -> None

Remove all groups (test helper).

Destination

A single destination returned by gateway.select() / gateway.list().

A destination returned by gateway.select() or gateway.list().

Attributes:

Name	Type	Description
uri		SIP URI to route to (e.g. "sip:gw1.carrier.com:5060").
address		Socket address string (e.g. "10.0.0.1:5060").
healthy		Whether the destination is healthy.
weight		Weight for load balancing.
priority		Priority tier (lower = higher priority).
attrs	dict[str, str]	User-defined attributes dict.

Example::

gw = gateway.select("carriers")
if gw:
    request.relay(gw.uri)
    print(gw.attrs.get("region"))