Examples

Send TAK Data

Send a takPong CoT marker every 20 seconds over plain TCP. Save as send.py and run with python3 send.py.

#!/usr/bin/env python3

import asyncio
import xml.etree.ElementTree as ET

from configparser import ConfigParser

import pytak


class MySerializer(pytak.QueueWorker):
    """
    Defines how you process or generate your Cursor on Target Events.
    From there it adds the CoT Events to a queue for TX to a COT_URL.
    """

    async def handle_data(self, data):
        """Handle pre-CoT data, serialize to CoT Event, then puts on queue."""
        event = data
        await self.put_queue(event)

    async def run(self):
        """Run the loop for processing or generating pre-CoT data."""
        while True:
            data = tak_pong()
            await self.handle_data(data)
            await asyncio.sleep(20)


def tak_pong():
    """Generate a simple takPong CoT Event."""
    root = ET.Element("event")
    root.set("version", "2.0")
    root.set("type", "t-x-d-d")
    root.set("uid", "takPong")
    root.set("how", "m-g")
    root.set("time", pytak.cot_time())
    root.set("start", pytak.cot_time())
    root.set("stale", pytak.cot_time(3600))
    return ET.tostring(root)


async def main():
    """Main definition of your program, sets config params and
    adds your serializer to the asyncio task list.
    """
    config = ConfigParser()
    config["mycottool"] = {"COT_URL": "tcp://takserver.example.com:8087"}
    config = config["mycottool"]

    # Initializes worker queues and tasks.
    clitool = pytak.CLITool(config)
    await clitool.setup()

    # Add your serializer to the asyncio task list.
    clitool.add_tasks(set([MySerializer(clitool.tx_queue, config)]))

    # Start all tasks.
    await clitool.run()


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

---

Send & Receive TAK Data

Send a position marker every 5 seconds and receive all incoming CoT from the same TCP connection. Received events are logged to console.

Save as send_receive.py and run with python3 send_receive.py.

#!/usr/bin/env python3

import asyncio
import xml.etree.ElementTree as ET
import pytak

from configparser import ConfigParser


def gen_cot():
    """Generate CoT Event."""
    root = ET.Element("event")
    root.set("version", "2.0")
    root.set("type", "a-h-A-M-A")  # insert your type of marker
    root.set("uid", "name_your_marker")
    root.set("how", "m-g")
    root.set("time", pytak.cot_time())
    root.set("start", pytak.cot_time())
    root.set(
        "stale", pytak.cot_time(60)
    )  # time difference in seconds from 'start' when stale initiates

    pt_attr = {
        "lat": "40.781789",  # set your lat (this loc points to Central Park NY)
        "lon": "-73.968698",  # set your long (this loc points to Central Park NY)
        "hae": "0",
        "ce": "10",
        "le": "10",
    }

    ET.SubElement(root, "point", attrib=pt_attr)

    return ET.tostring(root)


class MySender(pytak.QueueWorker):
    """
    Defines how you process or generate your Cursor-On-Target Events.
    From there it adds the COT Events to a queue for TX to a COT_URL.
    """

    async def handle_data(self, data):
        """Handle pre-CoT data, serialize to CoT Event, then puts on queue."""
        event = data
        await self.put_queue(event)

    async def run(self):
        """Run the loop for processing or generating pre-CoT data."""
        while True:
            data = gen_cot()
            self._logger.info("Sending:\n%s\n", data.decode())
            await self.handle_data(data)
            await asyncio.sleep(5)


class MyReceiver(pytak.QueueWorker):
    """Defines how you will handle events from RX Queue."""

    async def handle_data(self, data):
        """Handle data from the receive queue."""
        self._logger.info("Received:\n%s\n", data.decode())

    async def run(self):
        """Read from the receive queue, put data onto handler."""
        while True:
            data = (
                await self.queue.get()
            )  # this is how we get the received CoT from rx_queue
            await self.handle_data(data)


async def main():
    """Main definition of your program, sets config params and
    adds your serializer to the asyncio task list.
    """
    config = ConfigParser()
    config["mycottool"] = {"COT_URL": "tcp://takserver.example.com:8087"}
    config = config["mycottool"]

    # Initializes worker queues and tasks.
    clitool = pytak.CLITool(config)
    await clitool.setup()

    # Add your serializer to the asyncio task list.
    clitool.add_tasks(
        set([MySender(clitool.tx_queue, config), MyReceiver(clitool.rx_queue, config)])
    )

    # Start all tasks.
    await clitool.run()


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

Key differences from the send-only example:

• MySender uses clitool.tx_queue — events go to the network
• MyReceiver uses clitool.rx_queue — events come from the network
• Both workers are added to the task set so they run concurrently

---

TLS Send

Connect to a TAK Server over TLS using a client certificate and key. This is the standard connection method for production TAK Server deployments.

Generate a client cert on the TAK Server first:

sudo /opt/tak/certs/makeCert.sh client myclient

This produces myclient.pem and myclient.key. Copy them to your client machine, then:

#!/usr/bin/env python3

import asyncio

from configparser import ConfigParser

import pytak


class MySerializer(pytak.QueueWorker):
    """
    Defines how you process or generate your Cursor on Target Events.
    From there it adds the CoT Events to a queue for TX to a COT_URL.
    """

    async def handle_data(self, data):
        """Handle pre-CoT data, serialize to CoT Event, then puts on queue."""
        event = data
        await self.put_queue(event)

    async def run(self, number_of_iterations=-1):
        """Run the loop for processing or generating pre-CoT data."""
        while 1:
            data = pytak.gen_cot(lat=37.76, lon=-122.4975)
            await self.handle_data(data)
            await asyncio.sleep(20)


async def main():
    """Main definition of your program, sets config params and
    adds your serializer to the asyncio task list.
    """
    config = ConfigParser()
    # Generate certs with:
    # $ sudo /opt/tak/certs/makeCert.sh client pytak-test01
    config["tls_send"] = {
        "COT_URL": "tls://takserver.undef.net:8089",
        "PYTAK_TLS_CLIENT_CERT": "/Users/gba/src/SNS/pytak/examples/pytak-test01.pem",
        "PYTAK_TLS_CLIENT_KEY": "/Users/gba/src/SNS/pytak/examples/pytak-test01.key",
        "PYTAK_TLS_CLIENT_PASSWORD": "atakatak415",
        "PYTAK_TLS_DONT_VERIFY": True,
        "PYTAK_TLS_DONT_CHECK_HOSTNAME": True,
        # "DEBUG": True,
        "TAK_PROTO": 0,
    }
    config = config["tls_send"]

    # Initializes worker queues and tasks.
    clitool = pytak.CLITool(config)
    await clitool.setup()

    # Add your serializer to the asyncio task list.
    clitool.add_tasks(set([MySerializer(clitool.tx_queue, config)]))

    # Start all tasks.
    await clitool.run()


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

Certificate password

TAK Server's makeCert.sh creates password-protected keys. The default password is set in CoreConfig.xml (typically atakatak). Set it via PYTAK_TLS_CLIENT_PASSWORD.

PYTAK_TLS_DONT_VERIFY in production

PYTAK_TLS_DONT_VERIFY = True skips server cert validation. This is shown for development convenience; in production, provide PYTAK_TLS_CLIENT_CAFILE pointing to your TAK Server's CA chain instead.

---

TLS Send with Certificate Enrollment

Automatically enroll a client certificate from a TAK Server using your TAK account credentials. PyTAK handles the full CSR → PKCS#12 flow and caches the certificate locally.

Requires pip install pytak[with_aiohttp,with_crypto].

#!/usr/bin/env python3

import asyncio

from configparser import ConfigParser

import pytak


class MySerializer(pytak.QueueWorker):
    """
    Defines how you process or generate your Cursor on Target Events.
    From there it adds the CoT Events to a queue for TX to a COT_URL.
    """

    async def handle_data(self, data):
        """Handle pre-CoT data, serialize to CoT Event, then puts on queue."""
        event = data
        await self.put_queue(event)

    async def run(self, number_of_iterations=-1):
        """Run the loop for processing or generating pre-CoT data."""
        while 1:
            data = pytak.gen_cot(lat=37.76, lon=-122.4975)
            await self.handle_data(data)
            await asyncio.sleep(20)


async def main():
    """Main definition of your program, sets config params and
    adds your serializer to the asyncio task list.
    """
    config = ConfigParser()
    # Generate certs with:
    # $ sudo /opt/tak/certs/makeCert.sh client pytak-test01
    config["tls_send"] = {
        "COT_URL": "tls://takserver.example.com:8089",
        "PYTAK_TLS_CERT_ENROLLMENT_USERNAME": "xxx",
        "PYTAK_TLS_CERT_ENROLLMENT_PASSWORD": "yyy",
        "PYTAK_TLS_DONT_VERIFY": True,
        "PYTAK_TLS_DONT_CHECK_HOSTNAME": True,
        # "DEBUG": True,
        "TAK_PROTO": 0,
    }
    config = config["tls_send"]

    # Initializes worker queues and tasks.
    clitool = pytak.CLITool(config)
    await clitool.setup()

    # Add your serializer to the asyncio task list.
    clitool.add_tasks(set([MySerializer(clitool.tx_queue, config)]))

    # Start all tasks.
    await clitool.run()


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

After the first run, the enrolled certificate is cached in ~/.pytak/certs/ and reused on subsequent runs (re-enrollment happens automatically when the cert is within 7 days of expiry).

---

TAK Enrollment URL

If your TAK admin provided a tak:// onboarding URL (e.g. via QR code), pass it directly as COT_URL. PyTAK handles enrollment and TLS setup automatically.

Command-line usage:

pytak "tak://com.atakmap.app/enroll?host=takserver.example.com&username=myuser&token=mytoken"

If you need to force a specific WebSocket/Marti port, include it in host=:

pytak "tak://com.atakmap.app/enroll?host=takserver.example.com:8443&username=myuser&token=mytoken"

import asyncio
from configparser import ConfigParser
import pytak


class MySender(pytak.QueueWorker):
    async def handle_data(self, data):
        await self.put_queue(data)

    async def run(self):
        import xml.etree.ElementTree as ET
        while True:
            root = ET.Element("event", version="2.0", type="t-x-d-d",
                              uid="myMarker", how="m-g",
                              time=pytak.cot_time(), start=pytak.cot_time(),
                              stale=pytak.cot_time(3600))
            await self.handle_data(ET.tostring(root))
            await asyncio.sleep(20)


async def main():
    config = ConfigParser()
    config["mytool"] = {
        # Paste your tak:// URL here
        "COT_URL": "tak://com.atakmap.app/enroll?host=takserver.example.com&username=myuser&token=mytoken",
    }
    config = config["mytool"]

    clitool = pytak.CLITool(config)
    await clitool.setup()
    clitool.add_tasks(set([MySender(clitool.tx_queue, config)]))
    await clitool.run()


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

---

Marti REST API

Send and receive CoT via the TAK Server's Marti HTTP API instead of a raw TCP/TLS socket. Useful when firewall rules block direct streaming ports.

Requires pip install pytak[with_aiohttp].

import asyncio
import xml.etree.ElementTree as ET
from configparser import ConfigParser
import pytak


class MySender(pytak.QueueWorker):
    async def handle_data(self, data):
        await self.put_queue(data)

    async def run(self):
        while True:
            root = ET.Element("event", version="2.0", type="t-x-d-d",
                              uid="martiMarker", how="m-g",
                              time=pytak.cot_time(), start=pytak.cot_time(),
                              stale=pytak.cot_time(3600))
            await self.handle_data(ET.tostring(root))
            await asyncio.sleep(10)


async def main():
    config = ConfigParser()
    config["mytool"] = {
        "COT_URL": "marti://takserver.example.com:8443",
        # Optional: provide client cert for mTLS
        # "PYTAK_TLS_CLIENT_CERT": "/etc/pytak/client.pem",
    }
    config = config["mytool"]

    clitool = pytak.CLITool(config)
    await clitool.setup()
    clitool.add_tasks(set([MySender(clitool.tx_queue, config)]))
    await clitool.run()


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

---

Send-only to a TAK Server

If your tool only publishes CoT and does not process inbound events (for example, a sensor gateway), use the +wo modifier so PyTAK does not enqueue received data:

COT_URL = tls+wo://takserver.example.com:8089

---

Debug: Print CoT to Console

Use log://stdout as the COT_URL to print CoT XML to your console without a network connection. Useful for verifying your CoT format before connecting to a real server.

COT_URL = log://stdout

Or in Python:

config["mytool"] = {"COT_URL": "log://stdout"}