Swarmauri UDS Unicast Transport

The Swarmauri UDS Unicast Transport offers reliable, ordered, local-only communication over Unix domain sockets. It embraces the unified transport lifecycle so the server and client contexts are created directly from the transport instance (.server(...) / .client(...)).

Installation

Using uv

uv add --directory pkgs/standards/swarmauri_transport_uds_unicast swarmauri_transport_uds_unicast

Using pip

pip install swarmauri_transport_uds_unicast

Usage

The example below demonstrates setting up a server and client that echo messages over a Unix domain socket path.

import asyncio
from swarmauri_transport_uds_unicast import UdsUnicastTransport

SOCKET_PATH = "/tmp/swm-uds.sock"

async def main() -> None:
    server = UdsUnicastTransport(SOCKET_PATH)

    async def run_server():
        async with server.server():
            data = await server.recv()
            await server.send("peer", b"echo:" + data)

    async def run_client():
        client = UdsUnicastTransport(SOCKET_PATH)
        async with client.client():
            await client.send("server", b"ping")
            response = await client.recv()
            print(response.decode())

    await asyncio.gather(run_server(), run_client())

asyncio.run(main())

This transport is ideal for same-host communication where low latency and security isolation are required.