Sitelet https://github.com/sgofferj/python-tak-cot-streaming/wiki
Skip to content
Stefan Gofferje edited this page Aug 26, 2026 · 1 revision

python-tak-cot-streaming

Streaming-friendly TAK Cursor-on-Target library for Python. Work with typed Python objects — never with XML trees, sockets or protocol details.

©2026 Stefan Gofferje | License: GPL-3.0-or-later

Description

Async Python 3 library for TAK CoT streaming. You send and receive plain Python objects; the library takes care of the wire format.

  • CotEvent — one CoT message: envelope, position, detail block
  • CotStream — async bidirectional event stream over TCP/TLS with initial identification, position updates, parent-link attribution, graceful disconnect (t-x-d-d) and spec-conformant ping/pong keepalive plus a periodic SA refresh
  • Detail helpers — typed setters and factories for markers, SA reports, GeoChat messages and emergencies

The design goal is simple: if you need to look at raw XML, that's a bug in this library. Everything known gets a typed interface, everything unknown is preserved verbatim so nothing is lost on a round trip.

Note

This is an early proof of concept. The object model and the streaming client work; reconnect handling and TAK Protocol (CoT over protobuf) are on the roadmap.

Need the TAK Server REST API instead (missions, data feeds, user management, certificate enrollment)? That is the sister project python-takserver-api. It can also obtain client certificates through enrollment for you, ready to use with CotStream.connect(cert=..., key=...).

Quick start

import asyncio

from takstream import CotEvent, CotStream


async def main() -> None:
    stream = await CotStream.connect(
        "tak.example.com",
        8089,
        cert="client.pem",
        key="client.key",
        callsign="MYCALL", team="Cyan", role="Team Member",
    )

    # Send a marker; time/start/stale are filled automatically
    marker = CotEvent.marker(48.208, 16.373, callsign="PIT", remarks="landmark")
    await stream.send(marker)

    # Receive events as an async iterator
    async for event in stream:
        print(event.uid, event.cot_type, event.lat, event.lon)


asyncio.run(main())

Wiki contents

Clone this wiki locally