Getting Started

Getting Started

Installation

From source

git clone https://github.com/sgofferj/python-tak-cot-streaming
cd python-tak-cot-streaming
poetry install

That is the intended way to use it at this stage — the package is not yet published to PyPI. Poetry creates and manages the virtualenv for you; run everything through poetry run.

Requirements

Dependency	Version	Use
Python	>= 3.11	Runtime
defusedxml	>= 0.7	Hardened XML parsing

Everything else is standard library. Yes, really — no pydantic, no lxml, no pytak.

Connecting to a server

Plaintext CoT (port 8087 by default)

from takstream import CotStream

stream = await CotStream.connect("tak.example.com", 8087)

TLS (port 8089 by default)

TLS is chosen by what you pass, not by flags: give cert and key and the client certificate context is built internally; give nothing and you get cleartext. I refuse to make you juggle ssl.SSLContext objects for the standard case.

from takstream import CotStream

stream = await CotStream.connect(
    "tak.example.com", 8089,
    cert="certs/client.pem",
    key="certs/client.key",
)

The client cert/key pair is used for mutual TLS authentication.

Warning

By default the server is not verified at all (CERT_NONE): the server authenticates your certificate, but nothing checks the identity of the host you talk to. This is a deliberate trade-off because TAK Servers typically run self-signed certificates - but it means an attacker who can intercept your connection could impersonate the server. Do not point this library at hosts you do not trust.

Verifying the server (optional)

Pass ca to pin a CA (or the server's self-signed certificate itself) and get full server verification:

stream = await CotStream.connect(
    "tak.example.com", 8089,
    cert="certs/client.pem",
    key="certs/client.key",
    ca="certs/ca.pem",  # verify the server
)

Ports

As usual with TAK: ports are configurable in CoreConfig.xml, never assume. Common defaults:

Port	Transport	Use
8089	TLS	Standard streaming port
8087	TCP	Plaintext streaming
8088	UDP	Multicast
8090	varies	Often data feeders

(8443 is the HTTPS API port — that one belongs to python-takserver-api, not here.)

A complete example

import asyncio

from takstream import CotStream


async def main() -> None:
    stream = await CotStream.connect(
        "tak.example.com", 8089,
        cert="certs/client.pem",
        key="certs/client.key",
        callsign="MYCALL", team="Cyan", role="Team Member",
    )
    try:
        async for event in stream:
            callsign = f" ({event.callsign})" if event.callsign else ""
            print(f"{event.cot_type}{callsign} lat={event.lat:.4f} lon={event.lon:.4f}")
    finally:
        await stream.close()


asyncio.run(main())

There is also a ready-made receiver script in the repository: examples/receive_demo.py. It reads its connection settings from an untracked .env file in the repository root (see .env.example) and prints incoming events until you hit Ctrl-C.