import asyncio import socket import logging logging.basicConfig(level=logging.DEBUG) class Prot(asyncio.DatagramProtocol): def __init__(self): super().__init__() def error_received(self, exc): logging.exception('error_received', exc_info=exc) def datagram_received(self, data, addr): print(f'{addr=} {data=}') async def amain(): loop = asyncio.get_event_loop() trsp, _ = await loop.create_datagram_endpoint( Prot, local_addr=('::', 11111), family=socket.AF_INET6, proto=socket.IPPROTO_UDP) while True: sender_sock = socket.socket(socket.AF_INET6, socket.SOCK_DGRAM, socket.IPPROTO_UDP) sender_sock.sendto(b'test', ('::1', 11111, 0, 0)) await asyncio.sleep(1) def main(): asyncio.run(amain(), debug=True) if __name__ == '__main__': main()