Skip to content

Subscription API

Jump to bottom
sgofferj edited this page Aug 25, 2026 · 1 revision

Subscription API

Accessor: server.subscriptions

Reference: https://docs.tak.gov/api/takserver#tag/subscription-api

The server's view of its connected clients: subscription records (callsign, groups, client software, traffic counters), static outbound subscriptions, the per-client incognito flag and per-client CoT filters.

Warning

Every mutation here targets a connected client's state. Toggling incognito or installing a filter on someone else's client changes what the server delivers to them - only operate on UIDs you control.

Two quirks verified live on 5.7-RELEASE-43-HEAD (2026-08-25):

  1. toggle_incognito() answers HTTP 200 but does not change the flag.
  2. A client only appears in the registry after it has sent a proper CoT contact announcement (<contact> + <takv> in the event detail). Bare placenta events are not enough, and a fresh announcement resets any mutation you performed.

The group-subscription toggles (PUT /Marti/api/groups/active*) belong to this tag in the OpenAPI spec but live in the Groups API wrapper.

get_all_subscriptions()

async def get_all_subscriptions() -> tuple[int, Any]

Returns all current client subscriptions. Each record carries clientUid, callsign, username, groups, incognito, client metadata (takClient, takVersion, protocol, port) and traffic counters.

Example:

status, subs = await srv.subscriptions.get_all_subscriptions()
for s in subs:
    print(s["clientUid"], s["callsign"], s["takClient"])

get_subscription()

async def get_subscription(uid: str) -> tuple[int, Any]

Single subscription record by UID. Transient registry entries with an empty UID can exist - filter those out before use.

Static subscriptions

async def add_static_subscription(sub: dict) -> tuple[int, Any]
async def delete_subscription(uid: str) -> tuple[int, Any]

A static subscription makes the server treat a remote endpoint as a permanent subscriber. The body is a tmpStaticSub: uid, to, subaddr, subport, protocol, optional filterGroups (comma-separated string) and xpath. Note the server will attempt outbound connections to the given target.

toggle_incognito()

async def toggle_incognito(uid: str) -> tuple[int, Any]

Flips the incognito flag (an incognito client's markers are not broadcast). On the reference server this currently has no effect despite answering 200 - see the warning above.

set_filter() / delete_filter() 

async def set_filter(client_uid: str, filter_xml: str) -> tuple[int, Any]
async def delete_filter(client_uid: str) -> tuple[int, Any]

Installs or removes a per-client CoT filter. The server requires a non-null <geospatialFilter> element inside the document; anything else answers 400:

filter_xml = (
    '<filter xmlns="http://bbn.com/marti/xml/config">'
    '<geospatialFilter filterTAKClients="true">'
    '<boundingBox minLongitude="0.0" minLatitude="0.0" '
    'maxLongitude="10.0" maxLatitude="10.0"/>'
    "</geospatialFilter></filter>"
)
await srv.subscriptions.set_filter("SOME-UID", filter_xml)

A too-restrictive filter silences the client entirely - test on your own UIDs first.

bulk_groups_updated()

async def bulk_groups_updated(usernames: list[str]) -> tuple[int, Any]

Bulk trigger of the group-change notification: pushes all listed users' clients to re-fetch their channel list (the push counterpart to GroupApi.wait_for_group_update()).

Clone this wiki locally