swarmauri_transport_tls_unicast 0.11.0.dev1

TLS / mTLS unicast transport for Swarmauri.

srv_ctx.verify_mode = ssl.CERT_REQUIRED

srv_ctx.load_verify_locations("ca.pem")

cli_ctx = ssl.create_default_context()

cli_ctx.load_cert_chain("cli.pem", "cli.key")

async def main(): server = TlsUnicastTransport(srv_ctx)

async def run_server():
    async with server.server(host="0.0.0.0", port=8443):
        data = await server.recv()
        await server.send("peer", b"tls:" + data)

async def run_client():
    client = TlsUnicastTransport(cli_ctx, sni="localhost")
    async with client.client(host="127.0.0.1", port=8443):
        await client.send("server", b"hello")
        response = await client.recv()
        print(response.decode())

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

asyncio.run(main())

Configure the SSL contexts with your own certificates (and CA trust) to enable
TLS or full mTLS verification.