● Space Force UDL ● TAK Ecosystem Rev 1.0 — 2026

UDL ↔ TAK Integration Guide

A complete reference for connecting the U.S. Space Force Unified Data Library to the TAK ecosystem — bridging space domain awareness, sensor data, and space situational awareness into ATAK, WinTAK, WebTAK, and TAK Server via Cursor on Target (CoT) and the MARTI API.

⬡ Architecture ↔ Data Flows ⚙ Implementation

Integration patterns

Auth methods covered

10+

Code examples

JADC2

Architecture aligned

⬡

System Architecture

The UDL serves as the central space domain data layer — a cloud-hosted repository ingesting data from DoD, commercial, civil, and allied sensors. TAK provides the common operating picture at the tactical edge. This integration bridges space situational awareness, object tracking, and multi-domain sensor feeds into the TAK ecosystem in near-real time.

The bridge middleware authenticates against both the UDL REST API (using OAuth 2.0 bearer tokens) and TAK Server (using mTLS client certificates). It transforms UDL space objects and multi-domain tracks into Cursor on Target (CoT) XML and sends them into TAK Server's ingest port, where all TAK clients receive them as live map markers.

↔

Data Flow Patterns

Choose the pattern that matches your mission requirement. Patterns can be combined — for example, a bidirectional sync with event-driven alerts layered on top.

Push

UDL → TAK

UDL data is pulled on a schedule and pushed into TAK Server as CoT events. Satellite tracks, debris alerts, and space domain objects appear as live map markers.

Latency: 15–60s (poll) Protocol: TCP CoT :8089

Pull

TAK → UDL

TAK tracks (Blue Force positions, sensor reports) are read via MARTI REST and ingested into UDL as multi-domain ground truth to enrich space domain awareness.

Latency: On-demand (poll) Protocol: HTTPS MARTI :8443

Bidirectional

Full Duplex Sync

The middleware both subscribes to TAK Server (TCP :8087) and polls UDL. Changes in either system propagate to the other — supporting JADC2 kill chain requirements.

Latency: Real-time (subscribe) + 15–60s (UDL) Protocol: TCP + HTTPS

Event-Driven

UDL Alert → TAK

UDL conjunction warnings and object maneuver alerts trigger immediate CoT events in TAK — for example, a debris conjunction alert creates a t-b-a broadcast alert visible to all clients.

Latency: Near real-time (webhook) Protocol: UDL webhook → CoT

⚙

Implementation

Pattern 1: UDL → TAK (Scheduled Push)

This is the most common entry point. A Python bridge service polls the UDL REST API on a timer, converts space objects or tracks to CoT XML, and sends them to TAK Server over TLS.

Source

UDL REST API

OAuth 2.0 Bearer

→

Transform

UDL Object → CoT

Python / pyTAK

→

Destination

TAK Server :8089

TCP / TLS + mTLS cert

→

Clients

ATAK / WinTAK

Live map markers

udl_to_tak_bridge.py

import asyncio, aiohttp, pytak, datetime, xml.etree.ElementTree as ET

UDL_BASE    = "https://unifieddatalibrary.com/udl"
UDL_TOKEN   = "YOUR_UDL_BEARER_TOKEN"   # store in env var / Vault
TAK_HOST    = "your-tak-server.example.mil"
TAK_PORT    = 8089
POLL_SEC    = 30   # seconds between UDL polls

async def fetch_udl_tracks(session):
    """Fetch space objects / tracks from UDL."""
    headers = {"Authorization": f"Bearer {UDL_TOKEN}"}
    async with session.get(f"{UDL_BASE}/stateVector", headers=headers) as r:
        r.raise_for_status()
        return await r.json()

def udl_object_to_cot(obj) -> bytes:
    """Convert a UDL state vector entry to a CoT XML event."""
    uid      = f"UDL-{obj.get('satNo', obj.get('objectId','unknown'))}"
    lat      = float(obj.get("lat", 0.0))
    lon      = float(obj.get("lon", 0.0))
    alt_km   = float(obj.get("altitude", 400.0))
    hae      = alt_km * 1000           # metres above ellipsoid
    name     = obj.get("name", uid)
    now      = datetime.datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%S.000Z")
    stale    = (datetime.datetime.utcnow() +
                datetime.timedelta(seconds=90)).strftime("%Y-%m-%dT%H:%M:%S.000Z")

    evt = ET.Element("event", {
        "version": "2.0", "uid": uid,
        "type": "a-u-S",          # unknown space object
        "how": "m-g",
        "time": now, "start": now, "stale": stale
    })
    ET.SubElement(evt, "point", {
        "lat": str(lat), "lon": str(lon),
        "hae": str(hae), "ce": "9999999", "le": "9999999"
    })
    det = ET.SubElement(evt, "detail")
    ET.SubElement(det, "contact", {"callsign": name})
    ET.SubElement(det, "remarks").text = (
        f"UDL source | SatNo: {obj.get('satNo','?')} | "
        f"Alt: {alt_km:.0f} km | Inclination: {obj.get('inclination','?')}°"
    )
    return ET.tostring(evt, encoding="unicode").encode()

class UDLtoTAKWorker(pytak.Worker):
    async def run(self):
        async with aiohttp.ClientSession() as session:
            while True:
                objects = await fetch_udl_tracks(session)
                for obj in objects:
                    cot = udl_object_to_cot(obj)
                    await self.queue.put(cot)
                await asyncio.sleep(POLL_SEC)

async def main():
    clitool = pytak.CLITool(pytak.QueueWorker)
    config  = {"COT_URL": f"tcps://{TAK_HOST}:{TAK_PORT}",
               "PYTAK_TLS_CLIENT_CERT": "client.pem",
               "PYTAK_TLS_CLIENT_KEY":  "client.key",
               "PYTAK_TLS_CA_CERT":     "tak-ca.pem"}
    await clitool.setup(config)
    clitool.add_tasks({UDLtoTAKWorker(clitool.tx_queue, config)})
    await clitool.run()

if __name__ == "__main__":
    asyncio.run(main())

Note

CoT type for space objects: Use a-u-S (unknown space) or a-f-S (friendly space) for orbital objects. For debris/hazards use a-h-S. The hae field should be set to the object's altitude above the ellipsoid in metres (convert from km × 1000). TAK clients will display these at ground level by default unless 3D rendering is enabled.

Pattern 2: TAK → UDL (MARTI Pull)

Read current TAK track state via the MARTI REST API and forward it to UDL as ground-truth multi-domain data. This enriches the space picture with friendly force positions.

Source

MARTI REST API

HTTPS :8443 + mTLS

→

Transform

CoT → UDL Schema

JSON mapping

→

Destination

UDL REST API

POST /udl/groundTracks

tak_to_udl_bridge.py

import asyncio, aiohttp, ssl, xml.etree.ElementTree as ET

MARTI_URL   = "https://your-tak-server.example.mil:8443"
UDL_BASE    = "https://unifieddatalibrary.com/udl"
UDL_TOKEN   = "YOUR_UDL_BEARER_TOKEN"
CERT_FILE   = "client.pem"
KEY_FILE    = "client.key"
CA_FILE     = "tak-ca.pem"
POLL_SEC    = 30

def build_ssl():
    ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
    ctx.load_verify_locations(CA_FILE)
    ctx.load_cert_chain(CERT_FILE, KEY_FILE)
    return ctx

async def fetch_tak_tracks(session):
    url = f"{MARTI_URL}/Marti/api/cot/sa"
    async with session.get(url, ssl=build_ssl()) as r:
        r.raise_for_status()
        return await r.text()

def parse_cot_to_udl(xml_text: str) -> list:
    records = []
    root = ET.fromstring(xml_text)
    for evt in root.findall(".//event"):
        pt  = evt.find("point")
        det = evt.find("detail")
        if pt is None:
            continue
        cs = ""
        if det is not None:
            c = det.find("contact")
            if c is not None:
                cs = c.get("callsign", "")
        records.append({
            "source": "TAK",
            "uid":    evt.get("uid"),
            "type":   evt.get("type"),
            "callsign": cs,
            "lat":    float(pt.get("lat", 0)),
            "lon":    float(pt.get("lon", 0)),
            "hae":    float(pt.get("hae", 0)),
            "timestamp": evt.get("time")
        })
    return records

async def push_to_udl(session, records):
    headers = {"Authorization": f"Bearer {UDL_TOKEN}",
               "Content-Type": "application/json"}
    async with session.post(
        f"{UDL_BASE}/groundTracks", headers=headers, json=records
    ) as r:
        r.raise_for_status()

async def main():
    async with aiohttp.ClientSession() as session:
        while True:
            xml_text = await fetch_tak_tracks(session)
            records  = parse_cot_to_udl(xml_text)
            if records:
                await push_to_udl(session, records)
                print(f"Pushed {len(records)} TAK tracks to UDL")
            await asyncio.sleep(POLL_SEC)

if __name__ == "__main__":
    asyncio.run(main())

Pattern 3: Bidirectional Full Duplex

Run both workers concurrently within the same process. The TAK TCP subscriber (port 8087) receives events the moment they are published; the UDL poller runs on a timer. A shared asyncio queue decouples ingest from dispatch.

bidirectional_bridge.py (skeleton)

import asyncio
from udl_to_tak_bridge import UDLtoTAKWorker, fetch_udl_tracks, udl_object_to_cot
from tak_to_udl_bridge  import fetch_tak_tracks, parse_cot_to_udl, push_to_udl
import aiohttp, pytak

TAK_HOST = "your-tak-server.example.mil"

async def udl_push_loop(tx_queue, session):
    """Poll UDL and push space objects into TAK."""
    while True:
        objects = await fetch_udl_tracks(session)
        for obj in objects:
            await tx_queue.put(udl_object_to_cot(obj))
        await asyncio.sleep(30)

async def tak_pull_loop(session):
    """Pull TAK tracks and push to UDL."""
    while True:
        xml_text = await fetch_tak_tracks(session)
        records  = parse_cot_to_udl(xml_text)
        if records:
            await push_to_udl(session, records)
        await asyncio.sleep(30)

async def main():
    config = {"COT_URL": f"tcps://{TAK_HOST}:8089",
              "PYTAK_TLS_CLIENT_CERT": "client.pem",
              "PYTAK_TLS_CLIENT_KEY":  "client.key",
              "PYTAK_TLS_CA_CERT":     "tak-ca.pem"}
    clitool = pytak.CLITool(pytak.QueueWorker)
    await clitool.setup(config)
    async with aiohttp.ClientSession() as session:
        await asyncio.gather(
            clitool.run(),
            udl_push_loop(clitool.tx_queue, session),
            tak_pull_loop(session),
        )

if __name__ == "__main__":
    asyncio.run(main())

Pattern 4: Event-Driven — UDL Alerts → TAK

UDL conjunction warnings and maneuver alerts can be forwarded to TAK as broadcast alert CoT events (t-b-a), making them visible to all connected clients immediately.

udl_alert_webhook.py — Flask receiver

from flask import Flask, request, jsonify
import socket, datetime, xml.etree.ElementTree as ET, threading

app = Flask(__name__)
TAK_HOST, TAK_PORT = "your-tak-server.example.mil", 8089

def send_cot(xml_bytes: bytes):
    with socket.create_connection((TAK_HOST, TAK_PORT), timeout=10) as s:
        s.sendall(xml_bytes)

def build_alert_cot(alert: dict) -> bytes:
    now   = datetime.datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%S.000Z")
    stale = (datetime.datetime.utcnow() +
             datetime.timedelta(minutes=5)).strftime("%Y-%m-%dT%H:%M:%S.000Z")
    uid   = f"UDL-ALERT-{alert.get('alertId','unknown')}"
    evt   = ET.Element("event", {
        "version": "2.0", "uid": uid,
        "type": "t-b-a",          # broadcast alert
        "how": "m-g",
        "time": now, "start": now, "stale": stale
    })
    # Alerts may not have a precise ground lat/lon; use 0,0 with high CE
    ET.SubElement(evt, "point", {"lat":"0","lon":"0","hae":"0","ce":"9999999","le":"9999999"})
    det = ET.SubElement(evt, "detail")
    ET.SubElement(det, "contact", {"callsign": "UDL ALERT"})
    ET.SubElement(det, "remarks").text = (
        f"[UDL] {alert.get('alertType','Alert')}: "
        f"{alert.get('message','See UDL for details')} | "
        f"Object: {alert.get('objectId','?')}"
    )
    return ET.tostring(evt, encoding="unicode").encode()

@app.route("/udl-webhook", methods=["POST"])
def udl_webhook():
    alert = request.get_json(force=True)
    cot   = build_alert_cot(alert)
    threading.Thread(target=send_cot, args=(cot,), daemon=True).start()
    return jsonify({"status": "dispatched"}), 200

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5050)

Warning

Use port 8089 (TLS) in production. The plain TCP port 8087 is for LAN development only. All production bridges must use mTLS with client certificates issued by the TAK Server CA. See the Security Checklist below.

〈/〉

CoT Message Format for UDL Objects

Every UDL-sourced object entering TAK must be expressed as a Cursor on Target (CoT) XML event. Use the type codes and field conventions below for space domain objects.

Object Type	CoT Type Code	hae Value	Callsign Convention
Friendly satellite (US/ally)	`a-f-S`	Altitude × 1000 m	NORAD name or SatNo
Unknown space object	`a-u-S`	Altitude × 1000 m	UDL-{objectId}
Hostile / threatening object	`a-h-S`	Altitude × 1000 m	THREAT-{satNo}
Debris / conjunction hazard	`a-u-S-D`	Altitude × 1000 m	DEBRIS-{piece}
Ground track (Blue Force)	`a-f-G-U-C`	MSL altitude (m)	TAK callsign
Conjunction warning alert	`t-b-a`	0 (no ground point)	UDL ALERT
UDL sensor marker	`b-m-p-s-m`	0 (ground sensor)	Sensor name

Minimal CoT event — space object

<?xml version="1.0" encoding="UTF-8"?>
<event version="2.0"
       uid="UDL-25544"
       type="a-f-S"
       how="m-g"
       time="2026-05-27T14:00:00.000Z"
       start="2026-05-27T14:00:00.000Z"
       stale="2026-05-27T14:01:30.000Z">
  <point lat="38.8977" lon="-77.0365"
         hae="408000" ce="9999999" le="9999999"/>
  <detail>
    <contact callsign="ISS (ZARYA)"/>
    <remarks>UDL source | SatNo: 25544 | Alt: 408 km | Inc: 51.6°</remarks>
  </detail>
</event>

🔑

Authentication & Certificate Setup

UDL Authentication (OAuth 2.0)

UDL uses OAuth 2.0 bearer tokens. Request a token from the UDL Auth service, then include it in every API call. Tokens typically expire after 1 hour; implement automatic refresh.

udl_auth.py

import aiohttp, os, time

UDL_AUTH_URL = "https://unifieddatalibrary.com/auth/realms/udl/protocol/openid-connect/token"

class UDLAuth:
    def __init__(self):
        self._token   = None
        self._expires = 0

    async def get_token(self, session: aiohttp.ClientSession) -> str:
        if time.time() < self._expires - 60:
            return self._token   # still valid
        data = {
            "grant_type":    "client_credentials",
            "client_id":     os.environ["UDL_CLIENT_ID"],
            "client_secret": os.environ["UDL_CLIENT_SECRET"],
        }
        async with session.post(UDL_AUTH_URL, data=data) as r:
            r.raise_for_status()
            j = await r.json()
            self._token   = j["access_token"]
            self._expires = time.time() + j["expires_in"]
        return self._token

TAK Server mTLS Certificate Setup

In TAK Server Admin Console: Certificate Enrollment → Create Client Certificate for the bridge service. Name it udl-bridge.
Download the generated .p12 file, then extract PEM files:
openssl pkcs12 -in udl-bridge.p12 -out client.pem -nodes
Extract the key separately if needed:
openssl pkcs12 -in udl-bridge.p12 -nocerts -nodes -out client.key
Export the TAK CA certificate from the admin console and save as tak-ca.pem.
Store all three files in a secrets vault (HashiCorp Vault, AWS Secrets Manager) — never commit to git.
Revoke the certificate immediately if the bridge service is decommissioned.

API

UDL API Quick Reference

Endpoint	Method	Description	Key Params
`/udl/stateVector`	GET	Current state vectors for all tracked objects	`epoch`, `satNo`, `objectType`
`/udl/conjunction`	GET	Active conjunction warnings	`minPc` (min collision prob.)
`/udl/launchEvent`	GET	Recent launch events	`msgEpoch`
`/udl/maneuver`	GET	Detected object maneuvers	`satNo`, `source`
`/udl/groundTracks`	POST	Ingest multi-domain ground tracks (TAK → UDL)	JSON body array
`/udl/sensor`	GET	Space sensor status and positions	`sensorType`

Info

Access control: UDL data access is tiered by classification and data-provider authorization. Ensure your application's client_id has been granted access to the data types it queries. Contact the Space Systems Command Cross-Mission Data Branch for onboarding and steering committee approval.

API

MARTI API Quick Reference

Endpoint	Method	Description	Auth
`/Marti/api/cot/sa`	GET	All current situational awareness tracks (CoT XML)	mTLS
`/Marti/api/cot/sa/group/{group}`	GET	SA tracks filtered to a specific TAK group	mTLS
`/Marti/api/missions`	GET	List active TAK missions	mTLS
`/Marti/api/missions/{name}/contents`	GET	Contents (KML, overlays, points) of a mission	mTLS
`/Marti/sync/upload`	POST	Upload a data package (KML, GeoPackage, etc.)	mTLS
`/Marti/api/cot`	POST	Inject a CoT event via HTTPS (alternative to TCP)	mTLS

⌁

Ports & Protocol Reference

:8089 TCP (TLS)

CoT inbound to TAK Server over TLS. Primary production port for the UDL→TAK bridge. Requires mTLS client certificate.

Inbound to TAK

:8087 TCP

CoT cleartext. Development and LAN use only. Disable in any production or classified deployment.

Dev only

:8443 HTTPS

MARTI REST API. Used by TAK→UDL pull pattern. Also used for admin console — restrict access to bridge service IPs only.

Bidirectional

:9000 TCP (TLS)

TAK Server federation port. Used if a secondary TAK Server federates with the primary to relay UDL data to partner agencies.

Federation

UDL :443 HTTPS

UDL REST API over HTTPS. All UDL queries use OAuth 2.0 bearer tokens on this port.

Outbound to UDL

:5050 HTTP

Local Flask/FastAPI webhook receiver for UDL alert events (Pattern 4). Do not expose externally; sit behind a reverse proxy with TLS.

Webhook receiver

📦

Libraries & Tools

Tool / Library	Language	Purpose	Get It
pyTAK	Python	High-level async CoT producer / consumer — the recommended TAK integration library	`pip install pytak`
aiohttp	Python	Async HTTP client for UDL REST API calls and OAuth token refresh	`pip install aiohttp`
Flask / FastAPI	Python	Webhook receiver for UDL alert events (Pattern 4)	`pip install fastapi`
ATAK SDK	Java / Android	Build an ATAK plugin to display UDL data natively inside ATAK	tak.gov/products
FreeTAKServer	Python	Open-source TAK Server with REST API — easier for dev/test than full TAK Server	github.com/FreeTAKTeam
shapely	Python	Geofence detection — trigger CoT events when space objects pass over AOIs	`pip install shapely`
sgp4	Python	Propagate TLE orbital elements to lat/lon/alt for CoT position generation	`pip install sgp4`
HashiCorp Vault	Any	Secure storage for UDL bearer tokens and TAK mTLS certificates	vaultproject.io

Tip

Orbital propagation: UDL state vectors include position at epoch. For real-time CoT, propagate using the sgp4 library with the UDL-provided TLE data. Refresh propagation every 30–60 seconds to maintain accuracy — low-Earth objects move ~7 km/s and a stale position will be visibly wrong on the TAK map within minutes.

🛡

Security Checklist

Zero Trust Required UDL Auth: OAuth 2.0 TAK Auth: mTLS

Item	Requirement
TAK Server port	Use :8089 TLS only in production. Disable :8087 cleartext.
UDL credentials	Store `UDL_CLIENT_ID` and `UDL_CLIENT_SECRET` in Vault or environment injection — never hardcoded or committed to git.
TAK certificates	Issue a dedicated client certificate named `udl-bridge` from TAK Server CA. Never share certificates between services.
Certificate rotation	Revoke and reissue TAK client certificates every 90 days or immediately upon suspected compromise.
CoT input validation	Validate all UDL fields before constructing CoT XML. Sanitize `uid`, `callsign`, and `remarks` against injection. See pyTAK's built-in validators.
Network segmentation	The bridge host should have outbound-only access to UDL (:443) and inbound+outbound to TAK Server (:8089, :8443). Block all other egress.
Webhook receiver	Place Flask/FastAPI behind Nginx with TLS. Validate a shared secret header on every UDL webhook call before processing.
Classification handling	Verify UDL data classification before injecting into a TAK Server network. Do not mix classification levels without a data guard or cross-domain solution.

🔗

Key Resources

Resource	URL
UDL Portal (SSC)	ssc.spaceforce.mil
TAK Product Center	tak.gov
ATAK SDK & Documentation	tak.gov/products
ATAK Civilian Source (GitHub)	deptofdefense/AndroidTacticalAssaultKit-CIV
pyTAK Library	github.com/snstac/pytak
FreeTAKServer	github.com/FreeTAKTeam/FreeTakServer
CoT Schema Reference (MITRE)	mitre.org — CoT Schema PDF
sgp4 Orbital Propagator	pypi.org/project/sgp4
AFT AK COE TAK Integration Guide	aftakcoe.org
JADC2 Architecture Reference	defense.gov