Function watch_event

python/samples/manual_tool_resume.py:19–29

(session, data_type: type[T], predicate=None)

def watch_event(session, data_type: type[T], predicate=None) -> asyncio.Future:
    loop = asyncio.get_running_loop()
    future = loop.create_future()

    def on_event(event):
        if isinstance(event.data, data_type) and (predicate is None or predicate(event.data)):
            unsubscribe()
            future.set_result(event)

    unsubscribe = session.on(on_event)
    return future


async def wait_for_event(future: asyncio.Future) -> SessionEvent: