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"})

set_inbound_limit ¶

set_inbound_limit(
    group_name: str,
    max_concurrent_calls: Optional[int] = None,
    max_calls_per_second: Optional[int] = None,
    reject_code: int = 503,
    retry_after_secs: int = 1,
) -> None

Give a group the inbound limit siphon enforces for::

gateway:
  groups:
    - name: "carrier-a"
      inbound_limit:
        max_concurrent_calls: 300
        max_calls_per_second: 30
        reject_code: 486

Test-side, not a script API. The limit is configuration (siphon.yaml, or the inbound_* fields of a provisioned :class:siphon_sdk.gateways.GatewayRow); gateway.add_group() in a script cannot set one.

It caps the calls arriving from the group — from a source address :meth:contains_source calls a member — in B2BUA mode. With it set, :meth:siphon_sdk.testing.SipTestHarness.send_invite refuses a call from such a source once the group is at its ceiling, before @b2bua.on_invite runs, and before the instance-wide limit is consulted. A source two limited groups admit is counted against both. Calls to the group are not counted.

Example::

harness.gateway.add_group("carrier-a", [
    {"uri": "sip:gw1.carrier.example", "address": "203.0.113.10:5060"},
])
harness.gateway.set_inbound_limit("carrier-a", max_concurrent_calls=1)
harness.send_invite(source_ip="203.0.113.10")
refused = harness.send_invite(source_ip="203.0.113.10")
assert refused.was_refused
assert refused.refusal_gateway_group == "carrier-a"

Parameters:

Name	Type	Description	Default
`group_name`	`str`	A group added with :meth:`add_group`.	required
`max_concurrent_calls`	`Optional[int]`	Calls up at once. `None` or `0` is unlimited.	`None`
`max_calls_per_second`	`Optional[int]`	New calls per second, with a burst of one second's worth. `None` or `0` is unlimited.	`None`
`reject_code`	`int`	Status a refused call is answered with (400-699).	`503`
`retry_after_secs`	`int`	`Retry-After` on the refusal; `0` omits it.	`1`

Raises:

Type	Description
`KeyError`	No group of that name.
`ValueError`	`reject_code` is not a failure response.

clear_inbound_limit ¶

clear_inbound_limit(group_name: str) -> None

Remove a group's inbound limit (test helper).

inbound_calls_active ¶

inbound_calls_active(group_name: str) -> Optional[int]

Calls up from the group's sources — what siphon reports as siphon_gateway_inbound_calls_active{group}. None for a group with no inbound limit, which keeps no count.

inbound_limits_admitting ¶

inbound_limits_admitting(
    source_ip: str,
) -> "list[tuple[str, InboundLimit]]"

The limit of every limited group that admits source_ip, by group name. What the harness holds an inbound call to (test helper).

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.trunk.example.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`. `transport` is `"udp"` (default), `"tcp"` or `"tls"`, case-insensitive; anything else raises `ValueError` because siphon dials no outbound SCTP association and no WebSocket.	required
`algorithm`	`str`	Load-balancing algorithm: `"round_robin"`, `"weighted"` (default), `"hash"`.	`'weighted'`
`probe`	`bool`	Enable health probing (ignored in mock).	`False`

Raises:

Type	Description
`ValueError`	when a destination's `transport` is not udp/tcp/tls — mirroring the Rust binding, which used to fall through to udp and dial a TLS destination in the clear.

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"))

Gateway¶

gateway namespace¶

set_inbound_limit ¶

clear_inbound_limit ¶

inbound_calls_active ¶

inbound_limits_admitting ¶

select ¶

contains_source ¶

list ¶

status ¶

groups ¶

add_group ¶

remove_group ¶

mark_down ¶

mark_up ¶

clear ¶

Destination¶

`gateway` namespace¶

`Destination`¶