Способ или библиотека, для написания в асинхронной программе логики нахождения открытых портов(python)

import socket

#функция проверки порта
def is_port_open(host, port, timeout=10):
    try:
        with socket.create_connection((host, port), timeout=timeout):
            return True
    except (socket.timeout, ConnectionRefusedError, OSError):
        return False

# Пример вызова
if is_port_open("8.8.8.8.", 53):
    print("Порт доступен")
else:
    print("Порт недоступен")

Асинхронная реализация

import asyncio

async def check_port(host: str, port: int, sem: asyncio.Semaphore) -> tuple[int, bool]:
    try:
        async with sem:
            #print(f"Scanning port {port}")
            _, writer = await asyncio.wait_for(
                asyncio.open_connection(host, port),
                timeout=1.0
            )
            writer.close()
            await writer.wait_closed()
            #print(f"Port {port} OPENED")
            return (port, True)
    except OSError as e :
        #print(f"Port {port} CLOSED ({e})")
        return (port, False)

async def main():
    host = "8.8.8.8"
    ports = range(20, 1024)
    max_concurrency = 200
    
    sem = asyncio.Semaphore(max_concurrency)
    tasks = [check_port(host, port, sem) for port in ports]
    results = await asyncio.gather(*tasks)
    
    for port, is_open in results:
        print(f"Port {port}: {'OPENED' if is_open else 'CLOSED'}")

asyncio.run(main())

https://godbolt.org/z/q4WGM6rKd

Короче, есть ли какая-то библиотека, которая будет работать с asyncio, или другой способ?

Как бы, конечно, для эффективного сканирования TCP портов надо на RAW уровне работать, или ниже. Но уровень TCP - это ж базовая функциональность asyncio. К примеру, так:

import asyncio

async def probe_one(host=None, port=None):
    loop = asyncio.get_running_loop()
    try:
        transport, _ = await loop.create_connection(asyncio.BaseProtocol,
                                                    host, port)
        transport.close()
        print(f"{host}:{port} - хорь")
        return (host, port, True)
    except (OSError, asyncio.CancelledError) as exc:
        print(f"{host}:{port} - НЕУД {exc=}")
        return (host, port, False)

async def probe_list(host_port_list):
    tasks = []
    try:
        # <= 3.10: await asyncio.gather(*tasks)
        async with asyncio.TaskGroup() as tg:
            tasks = [tg.create_task(probe_one(hp[0], hp[1])) 
                     for hp in host_port_list]
    except asyncio.CancelledError as exc:
        print(f"probe_list: {exc=}")
    return [t.result() for t in tasks]

async def main():
    return await asyncio.wait_for(
                    probe_list([('1.1.1.1', 1234),
                                ('127.0.0.1', 8890),
                                ('127.0.0.1', 8891),
                                ('127.0.0.1', 8892),
                                ('127.0.0.1', 8893),]),
                    timeout=0.1)  
    # Для очень малых timeout может возникать TimeoutError в самом wait_for()

await main()

Если запустить после старта nc -l 127.0.0.1 8892 в фоне, то постепенно выдаст что-то в духе:

127.0.0.1:8890 - НЕУД exc=ConnectionRefusedError(61, "Connect call failed ('127.0.0.1', 8890)")
127.0.0.1:8891 - НЕУД exc=ConnectionRefusedError(61, "Connect call failed ('127.0.0.1', 8891)")
127.0.0.1:8893 - НЕУД exc=ConnectionRefusedError(61, "Connect call failed ('127.0.0.1', 8893)")
127.0.0.1:8892 - хорь
1.1.1.1:1234 - НЕУД exc=CancelledError()
probe_list: exc=CancelledError()

[('1.1.1.1', 1234, False),
 ('127.0.0.1', 8890, False),
 ('127.0.0.1', 8891, False),
 ('127.0.0.1', 8892, True),
 ('127.0.0.1', 8893, False)]

Если есть спешка, то количество используемых сокетов, так же как и таймауты стоит контролировать на верхнем уровне.