CDPClient & CDPSession

CDPClient

Main entry point for cdpwave.

Manages a browser process (via launch()) or connects to an existing browser (via connect()). Provides session creation, event handling, and lifecycle management. Use async with for automatic cleanup.

Source code in cdpwave/client.py

class CDPClient:
    """Main entry point for cdpwave.

    Manages a browser process (via ``launch()``) or connects to an existing
    browser (via ``connect()``). Provides session creation, event handling,
    and lifecycle management. Use ``async with`` for automatic cleanup.
    """

    def __init__(
        self,
        connection: Connection,
        launcher: BrowserLauncher | None = None,
        discovery: TargetDiscovery | None = None,
        strict_events: bool = False,
        on_event_error: EventErrorCallback | None = None,
    ) -> None:
        self._connection = connection
        self._launcher = launcher
        self._discovery = discovery
        self._session_manager = SessionManager(connection)
        self._dispatcher = EventDispatcher(
            strict_events=strict_events,
            on_event_error=on_event_error,
        )
        self._strict_events = strict_events
        self._on_event_error = on_event_error
        self._session_dispatchers: dict[str, EventDispatcher] = {}
        self._sessions: dict[str, CDPSession] = {}
        self._managed_targets: set[str] = set()
        self._closed = False
        self._browser = BrowserDomain(self.send)

    async def _event_callback(
        self,
        event_name: str,
        params: dict[str, Any],
        session_id: str | None,
    ) -> None:
        """Internal callback for routing CDP events to the correct dispatcher."""
        if event_name == "Target.attachedToTarget":
            parent_session_id = params.get("sessionId")
            if parent_session_id is not None:
                parent = self._sessions.get(parent_session_id)
                if parent is not None:
                    parent._handle_attached_to_target(params)
                    return
            return

        if event_name == "Target.detachedFromTarget":
            detached_session_id = params.get("sessionId")
            if detached_session_id is not None:
                session = self._sessions.get(detached_session_id)
                if session is not None:
                    if session._auto_attach_enabled:
                        parent = self._sessions.get(session_id) if session_id else None
                        if parent is not None:
                            parent._handle_detached_from_target(params)
                            return
                    session._closed = True
                    session._dispatcher.clear()
                    self._session_dispatchers.pop(detached_session_id, None)
                    self._sessions.pop(detached_session_id, None)
                    logger.info(
                        "Session %s detached by browser",
                        detached_session_id,
                    )
                else:
                    logger.debug(
                        "Target.detachedFromTarget for unknown session %s",
                        detached_session_id,
                    )
            return

        if session_id is None:
            await self._dispatcher.dispatch(event_name, params)
        else:
            dispatcher = self._session_dispatchers.get(session_id)
            if dispatcher is not None:
                await dispatcher.dispatch(event_name, params)
            else:
                logger.debug(
                    "Event %s for unknown session %s",
                    event_name,
                    session_id,
                )

    def on(self, event_name: str, handler: EventHandler) -> Subscription:
        """Register an async handler for a browser-level CDP event.

        Args:
            event_name: CDP event name.
            handler: Async callable that receives the event params dict.

        Returns:
            A Subscription that can be used to unsubscribe.
        """
        return self._dispatcher.on(event_name, handler)

    def off(self, event_name: str, handler: EventHandler) -> None:
        """Remove a previously registered browser-level event handler.

        Args:
            event_name: CDP event name.
            handler: The handler to remove.
        """
        self._dispatcher.off(event_name, handler)

    async def send(
        self,
        method: str,
        params: dict[str, Any] | None = None,
    ) -> dict[str, Any]:
        """Send a raw CDP command to the browser target (no session).

        Use this for browser-level commands like ``SystemInfo.getInfo``
        that are only supported on the browser target, not page sessions.

        Args:
            method: CDP method name (e.g. ``"SystemInfo.getInfo"``).
            params: Optional command parameters.

        Returns:
            The CDP response result dict.
        """
        return await self._connection.send_command(method, params)

    @property
    def browser(self) -> BrowserDomain:
        """Browser domain wrapper for browser-level commands."""
        return self._browser

    @classmethod
    def launch(
        cls,
        headless: bool = True,
        browser_path: str | None = None,
        port: int = 0,
        user_data_dir: str | None = None,
        extra_args: list[str] | None = None,
        timeout: float = 10.0,
        max_retries: int = 0,
        backoff_base: float = 1.0,
        backoff_max: float = 30.0,
    ) -> _LaunchContext:
        """Launch a new browser and return a connected CDPClient.

        Supports both ``await CDPClient.launch()`` and
        ``async with CDPClient.launch() as client:`` patterns.

        Args:
            headless: Run browser in headless mode.
            browser_path: Optional path to browser executable.
            port: Optional debugging port (0 for auto-assigned).
            user_data_dir: Optional user data directory.
            extra_args: Optional extra command-line arguments.
            timeout: Maximum seconds to wait for browser startup.
            max_retries: Maximum WebSocket reconnection attempts (0 = no reconnect).
            backoff_base: Initial reconnection backoff delay in seconds.
            backoff_max: Maximum reconnection backoff delay in seconds.

        Returns:
            A _LaunchContext that resolves to a connected CDPClient.
        """

        async def _do_launch() -> CDPClient:
            launcher = BrowserLauncher(
                browser_path=browser_path,
                port=port,
                headless=headless,
                user_data_dir=user_data_dir,
                extra_args=extra_args,
            )
            info = await launcher.launch(timeout=timeout)
            discovery = TargetDiscovery(port=info.port)
            connection = Connection(
                info.web_socket_debugger_url,
                max_retries=max_retries,
                backoff_base=backoff_base,
                backoff_max=backoff_max,
            )
            await connection.connect()
            client = cls(connection, launcher=launcher, discovery=discovery)
            connection._event_callback = client._event_callback
            return client

        return _LaunchContext(_do_launch())

    @classmethod
    def connect(
        cls,
        host: str = "localhost",
        port: int = 9222,
        ws_url: str | None = None,
        max_retries: int = 0,
        backoff_base: float = 1.0,
        backoff_max: float = 30.0,
    ) -> _LaunchContext:
        """Connect to an existing browser's CDP endpoint.

        Supports both ``await CDPClient.connect()`` and
        ``async with CDPClient.connect() as client:`` patterns.

        If ``ws_url`` is provided, connects directly to that WebSocket
        URL without HTTP discovery. Otherwise, discovers the WebSocket
        URL via ``http://host:port/json/version``.

        Args:
            host: Host where the browser is running.
            port: Remote debugging port.
            ws_url: Optional direct WebSocket URL (skips discovery).
            max_retries: Maximum WebSocket reconnection attempts (0 = no reconnect).
            backoff_base: Initial reconnection backoff delay in seconds.
            backoff_max: Maximum reconnection backoff delay in seconds.

        Returns:
            A _LaunchContext that resolves to a connected CDPClient.
        """

        async def _do_connect() -> CDPClient:
            discovery = TargetDiscovery(host=host, port=port)
            if ws_url is not None:
                socket_url = ws_url
            else:
                version = await discovery.get_version()
                socket_url = version.web_socket_debugger_url
            connection = Connection(
                socket_url,
                max_retries=max_retries,
                backoff_base=backoff_base,
                backoff_max=backoff_max,
            )
            await connection.connect()
            client = cls(connection, launcher=None, discovery=discovery)
            connection._event_callback = client._event_callback
            return client

        return _LaunchContext(_do_connect())

    async def new_page(
        self,
        url: str = "about:blank",
        auto_attach: bool = False,
    ) -> CDPSession:
        """Create a new page target and return a CDPSession for it.

        Args:
            url: Initial URL for the new page.
            auto_attach: If True, auto-attach to child targets (iframes,
                workers) and expose them via ``session.sub_sessions``.

        Returns:
            A CDPSession connected to the new page.
        """
        target_id = await self._session_manager.create_target(url)
        try:
            session_id = await self._session_manager.attach_to_target(target_id)
        except Exception:
            with contextlib.suppress(Exception):
                await self._session_manager.close_target(target_id)
            raise
        self._managed_targets.add(target_id)
        session = CDPSession(
            connection=self._connection,
            session_id=session_id,
            target_id=target_id,
            client=self,
        )
        self._sessions[session_id] = session

        if auto_attach:
            await session._enable_auto_attach()

        return session

    async def get_pages(self) -> list[TargetInfo]:
        """List all open page targets in the browser."""
        if self._discovery is None:
            raise RuntimeError("Discovery is not available")
        targets = await self._discovery.list_targets()
        return [t for t in targets if t.type == "page"]

    async def connect_to_page(self, target_id: str) -> CDPSession:
        """Attach to an existing page target by ID.

        Args:
            target_id: The target ID to attach to.

        Returns:
            A CDPSession connected to the specified target.
        """
        session_id = await self._session_manager.attach_to_target(target_id)
        session = CDPSession(
            connection=self._connection,
            session_id=session_id,
            target_id=target_id,
            client=self,
        )
        self._sessions[session_id] = session
        return session

    async def new_context(self) -> BrowserContext:
        """Create a new isolated browser context.

        Returns:
            A BrowserContext that can be used to create pages with
            isolated cookies, storage, and permissions.
        """
        result = await self._connection.send_command(
            "Target.createBrowserContext",
            {"disposeOnDetach": False},
        )
        context_id = result.get("browserContextId")
        if context_id is None:
            raise KeyError(
                "Target.createBrowserContext response missing 'browserContextId'"
            )
        context = BrowserContext(self, str(context_id))
        logger.info("Browser context %s created", context_id)
        return context

    async def close(self) -> None:
        """Close all sessions, the WebSocket connection, and the browser process."""
        if self._closed:
            return
        self._closed = True

        for session in list(self._sessions.values()):
            with contextlib.suppress(Exception):
                session._closed = True
                session._dispatcher.clear()
            self._session_dispatchers.pop(session._session_id, None)
        self._sessions.clear()
        self._dispatcher.clear()

        with contextlib.suppress(Exception):
            await self._connection.close()

        if self._launcher is not None:
            with contextlib.suppress(Exception):
                await self._launcher.close()

        logger.info("CDPClient closed")

    @property
    def is_closed(self) -> bool:
        """Whether the client has been closed or the connection dropped."""
        return self._closed or self._connection.is_closed

    @property
    def is_connected(self) -> bool:
        """Whether the WebSocket connection is still open."""
        return not self._connection.is_closed

    @property
    def sessions(self) -> list[CDPSession]:
        """List of currently active sessions."""
        return list(self._sessions.values())

    async def __aenter__(self) -> CDPClient:
        """Enter async context manager."""
        return self

    async def __aexit__(
        self,
        exc_type: object,
        exc_val: object,
        exc_tb: object,
    ) -> None:
        """Exit async context manager and close the client."""
        await self.close()

browser property

browser: BrowserDomain

Browser domain wrapper for browser-level commands.

is_closed property

is_closed: bool

Whether the client has been closed or the connection dropped.

is_connected property

is_connected: bool

Whether the WebSocket connection is still open.

sessions property

sessions: list[CDPSession]

List of currently active sessions.

on

on(event_name: str, handler: EventHandler) -> Subscription

Register an async handler for a browser-level CDP event.

Parameters:

Name	Type	Description	Default
event_name	str	CDP event name.	required
handler	EventHandler	Async callable that receives the event params dict.	required

Returns:

Type	Description
Subscription	A Subscription that can be used to unsubscribe.

Source code in cdpwave/client.py

def on(self, event_name: str, handler: EventHandler) -> Subscription:
    """Register an async handler for a browser-level CDP event.

    Args:
        event_name: CDP event name.
        handler: Async callable that receives the event params dict.

    Returns:
        A Subscription that can be used to unsubscribe.
    """
    return self._dispatcher.on(event_name, handler)

off

off(event_name: str, handler: EventHandler) -> None

Remove a previously registered browser-level event handler.

Parameters:

Name	Type	Description	Default
event_name	str	CDP event name.	required
handler	EventHandler	The handler to remove.	required

Source code in cdpwave/client.py

def off(self, event_name: str, handler: EventHandler) -> None:
    """Remove a previously registered browser-level event handler.

    Args:
        event_name: CDP event name.
        handler: The handler to remove.
    """
    self._dispatcher.off(event_name, handler)

send async

send(method: str, params: dict[str, Any] | None = None) -> dict[str, Any]

Send a raw CDP command to the browser target (no session).

Use this for browser-level commands like SystemInfo.getInfo that are only supported on the browser target, not page sessions.

Parameters:

Name	Type	Description	Default
method	str	CDP method name (e.g. "SystemInfo.getInfo").	required
params	dict[str, Any] | None	Optional command parameters.	None

Returns:

Type	Description
dict[str, Any]	The CDP response result dict.

Source code in cdpwave/client.py

async def send(
    self,
    method: str,
    params: dict[str, Any] | None = None,
) -> dict[str, Any]:
    """Send a raw CDP command to the browser target (no session).

    Use this for browser-level commands like ``SystemInfo.getInfo``
    that are only supported on the browser target, not page sessions.

    Args:
        method: CDP method name (e.g. ``"SystemInfo.getInfo"``).
        params: Optional command parameters.

    Returns:
        The CDP response result dict.
    """
    return await self._connection.send_command(method, params)

launch classmethod

launch(headless: bool = True, browser_path: str | None = None, port: int = 0, user_data_dir: str | None = None, extra_args: list[str] | None = None, timeout: float = 10.0, max_retries: int = 0, backoff_base: float = 1.0, backoff_max: float = 30.0) -> _LaunchContext

Launch a new browser and return a connected CDPClient.

Supports both await CDPClient.launch() and async with CDPClient.launch() as client: patterns.

Parameters:

Name	Type	Description	Default
headless	bool	Run browser in headless mode.	True
browser_path	str | None	Optional path to browser executable.	None
port	int	Optional debugging port (0 for auto-assigned).	0
user_data_dir	str | None	Optional user data directory.	None
extra_args	list[str] | None	Optional extra command-line arguments.	None
timeout	float	Maximum seconds to wait for browser startup.	10.0
max_retries	int	Maximum WebSocket reconnection attempts (0 = no reconnect).	0
backoff_base	float	Initial reconnection backoff delay in seconds.	1.0
backoff_max	float	Maximum reconnection backoff delay in seconds.	30.0

Returns:

Type	Description
_LaunchContext	A _LaunchContext that resolves to a connected CDPClient.

Source code in cdpwave/client.py

@classmethod
def launch(
    cls,
    headless: bool = True,
    browser_path: str | None = None,
    port: int = 0,
    user_data_dir: str | None = None,
    extra_args: list[str] | None = None,
    timeout: float = 10.0,
    max_retries: int = 0,
    backoff_base: float = 1.0,
    backoff_max: float = 30.0,
) -> _LaunchContext:
    """Launch a new browser and return a connected CDPClient.

    Supports both ``await CDPClient.launch()`` and
    ``async with CDPClient.launch() as client:`` patterns.

    Args:
        headless: Run browser in headless mode.
        browser_path: Optional path to browser executable.
        port: Optional debugging port (0 for auto-assigned).
        user_data_dir: Optional user data directory.
        extra_args: Optional extra command-line arguments.
        timeout: Maximum seconds to wait for browser startup.
        max_retries: Maximum WebSocket reconnection attempts (0 = no reconnect).
        backoff_base: Initial reconnection backoff delay in seconds.
        backoff_max: Maximum reconnection backoff delay in seconds.

    Returns:
        A _LaunchContext that resolves to a connected CDPClient.
    """

    async def _do_launch() -> CDPClient:
        launcher = BrowserLauncher(
            browser_path=browser_path,
            port=port,
            headless=headless,
            user_data_dir=user_data_dir,
            extra_args=extra_args,
        )
        info = await launcher.launch(timeout=timeout)
        discovery = TargetDiscovery(port=info.port)
        connection = Connection(
            info.web_socket_debugger_url,
            max_retries=max_retries,
            backoff_base=backoff_base,
            backoff_max=backoff_max,
        )
        await connection.connect()
        client = cls(connection, launcher=launcher, discovery=discovery)
        connection._event_callback = client._event_callback
        return client

    return _LaunchContext(_do_launch())

connect classmethod

connect(host: str = 'localhost', port: int = 9222, ws_url: str | None = None, max_retries: int = 0, backoff_base: float = 1.0, backoff_max: float = 30.0) -> _LaunchContext

Connect to an existing browser's CDP endpoint.

Supports both await CDPClient.connect() and async with CDPClient.connect() as client: patterns.

If ws_url is provided, connects directly to that WebSocket URL without HTTP discovery. Otherwise, discovers the WebSocket URL via http://host:port/json/version.

Parameters:

Name	Type	Description	Default
host	str	Host where the browser is running.	'localhost'
port	int	Remote debugging port.	9222
ws_url	str | None	Optional direct WebSocket URL (skips discovery).	None
max_retries	int	Maximum WebSocket reconnection attempts (0 = no reconnect).	0
backoff_base	float	Initial reconnection backoff delay in seconds.	1.0
backoff_max	float	Maximum reconnection backoff delay in seconds.	30.0

Returns:

Type	Description
_LaunchContext	A _LaunchContext that resolves to a connected CDPClient.

Source code in cdpwave/client.py

@classmethod
def connect(
    cls,
    host: str = "localhost",
    port: int = 9222,
    ws_url: str | None = None,
    max_retries: int = 0,
    backoff_base: float = 1.0,
    backoff_max: float = 30.0,
) -> _LaunchContext:
    """Connect to an existing browser's CDP endpoint.

    Supports both ``await CDPClient.connect()`` and
    ``async with CDPClient.connect() as client:`` patterns.

    If ``ws_url`` is provided, connects directly to that WebSocket
    URL without HTTP discovery. Otherwise, discovers the WebSocket
    URL via ``http://host:port/json/version``.

    Args:
        host: Host where the browser is running.
        port: Remote debugging port.
        ws_url: Optional direct WebSocket URL (skips discovery).
        max_retries: Maximum WebSocket reconnection attempts (0 = no reconnect).
        backoff_base: Initial reconnection backoff delay in seconds.
        backoff_max: Maximum reconnection backoff delay in seconds.

    Returns:
        A _LaunchContext that resolves to a connected CDPClient.
    """

    async def _do_connect() -> CDPClient:
        discovery = TargetDiscovery(host=host, port=port)
        if ws_url is not None:
            socket_url = ws_url
        else:
            version = await discovery.get_version()
            socket_url = version.web_socket_debugger_url
        connection = Connection(
            socket_url,
            max_retries=max_retries,
            backoff_base=backoff_base,
            backoff_max=backoff_max,
        )
        await connection.connect()
        client = cls(connection, launcher=None, discovery=discovery)
        connection._event_callback = client._event_callback
        return client

    return _LaunchContext(_do_connect())

new_page async

new_page(url: str = 'about:blank', auto_attach: bool = False) -> CDPSession

Create a new page target and return a CDPSession for it.

Parameters:

Name	Type	Description	Default
url	str	Initial URL for the new page.	'about:blank'
auto_attach	bool	If True, auto-attach to child targets (iframes, workers) and expose them via session.sub_sessions.	False

Returns:

Type	Description
CDPSession	A CDPSession connected to the new page.

Source code in cdpwave/client.py

async def new_page(
    self,
    url: str = "about:blank",
    auto_attach: bool = False,
) -> CDPSession:
    """Create a new page target and return a CDPSession for it.

    Args:
        url: Initial URL for the new page.
        auto_attach: If True, auto-attach to child targets (iframes,
            workers) and expose them via ``session.sub_sessions``.

    Returns:
        A CDPSession connected to the new page.
    """
    target_id = await self._session_manager.create_target(url)
    try:
        session_id = await self._session_manager.attach_to_target(target_id)
    except Exception:
        with contextlib.suppress(Exception):
            await self._session_manager.close_target(target_id)
        raise
    self._managed_targets.add(target_id)
    session = CDPSession(
        connection=self._connection,
        session_id=session_id,
        target_id=target_id,
        client=self,
    )
    self._sessions[session_id] = session

    if auto_attach:
        await session._enable_auto_attach()

    return session

get_pages async

get_pages() -> list[TargetInfo]

List all open page targets in the browser.

Source code in cdpwave/client.py

async def get_pages(self) -> list[TargetInfo]:
    """List all open page targets in the browser."""
    if self._discovery is None:
        raise RuntimeError("Discovery is not available")
    targets = await self._discovery.list_targets()
    return [t for t in targets if t.type == "page"]

connect_to_page async

connect_to_page(target_id: str) -> CDPSession

Attach to an existing page target by ID.

Parameters:

Name	Type	Description	Default
target_id	str	The target ID to attach to.	required

Returns:

Type	Description
CDPSession	A CDPSession connected to the specified target.

Source code in cdpwave/client.py

async def connect_to_page(self, target_id: str) -> CDPSession:
    """Attach to an existing page target by ID.

    Args:
        target_id: The target ID to attach to.

    Returns:
        A CDPSession connected to the specified target.
    """
    session_id = await self._session_manager.attach_to_target(target_id)
    session = CDPSession(
        connection=self._connection,
        session_id=session_id,
        target_id=target_id,
        client=self,
    )
    self._sessions[session_id] = session
    return session

new_context async

new_context() -> BrowserContext

Create a new isolated browser context.

Returns:

Type	Description
BrowserContext	A BrowserContext that can be used to create pages with
BrowserContext	isolated cookies, storage, and permissions.

Source code in cdpwave/client.py

async def new_context(self) -> BrowserContext:
    """Create a new isolated browser context.

    Returns:
        A BrowserContext that can be used to create pages with
        isolated cookies, storage, and permissions.
    """
    result = await self._connection.send_command(
        "Target.createBrowserContext",
        {"disposeOnDetach": False},
    )
    context_id = result.get("browserContextId")
    if context_id is None:
        raise KeyError(
            "Target.createBrowserContext response missing 'browserContextId'"
        )
    context = BrowserContext(self, str(context_id))
    logger.info("Browser context %s created", context_id)
    return context

close async

close() -> None

Close all sessions, the WebSocket connection, and the browser process.

Source code in cdpwave/client.py

async def close(self) -> None:
    """Close all sessions, the WebSocket connection, and the browser process."""
    if self._closed:
        return
    self._closed = True

    for session in list(self._sessions.values()):
        with contextlib.suppress(Exception):
            session._closed = True
            session._dispatcher.clear()
        self._session_dispatchers.pop(session._session_id, None)
    self._sessions.clear()
    self._dispatcher.clear()

    with contextlib.suppress(Exception):
        await self._connection.close()

    if self._launcher is not None:
        with contextlib.suppress(Exception):
            await self._launcher.close()

    logger.info("CDPClient closed")

__aenter__ async

__aenter__() -> CDPClient

Enter async context manager.

Source code in cdpwave/client.py

async def __aenter__(self) -> CDPClient:
    """Enter async context manager."""
    return self

__aexit__ async

__aexit__(exc_type: object, exc_val: object, exc_tb: object) -> None

Exit async context manager and close the client.

Source code in cdpwave/client.py

async def __aexit__(
    self,
    exc_type: object,
    exc_val: object,
    exc_tb: object,
) -> None:
    """Exit async context manager and close the client."""
    await self.close()

CDPSession

Represents a single CDP target session.

Provides access to CDP domain wrappers (page, runtime, network, etc.) and event handling for a single browser target. Use async with for automatic cleanup.

Source code in cdpwave/client.py

class CDPSession:
    """Represents a single CDP target session.

    Provides access to CDP domain wrappers (page, runtime, network, etc.)
    and event handling for a single browser target. Use ``async with`` for
    automatic cleanup.
    """

    def __init__(
        self,
        connection: Connection,
        session_id: str,
        target_id: str,
        client: CDPClient | None = None,
    ) -> None:
        self._connection = connection
        self._session_id = session_id
        self._target_id = target_id
        self._closed = False
        self._dispatcher = EventDispatcher()

        if client is not None:
            client._session_dispatchers[session_id] = self._dispatcher
            self._dispatcher = EventDispatcher(
                strict_events=client._strict_events,
                on_event_error=client._on_event_error,
            )
            client._session_dispatchers[session_id] = self._dispatcher
        self._client = client

        self._sub_sessions: dict[str, CDPSession] = {}
        self._auto_attach_enabled = False

        async def _send(
            method: str,
            params: dict[str, Any] | None = None,
        ) -> dict[str, Any]:
            return await connection.send_command(
                method,
                params,
                session_id=session_id,
            )

        self._sender: CommandSender = _send
        self._page = PageDomain(self._sender)
        self._runtime = RuntimeDomain(self._sender)
        self._target = TargetDomain(self._sender)
        self._network = NetworkDomain(self._sender)
        self._dom = DOMDomain(self._sender)
        self._log = LogDomain(self._sender)
        self._console = ConsoleDomain(self._sender)
        self._input = InputDomain(self._sender)
        self._emulation = EmulationDomain(self._sender)
        self._fetch = FetchDomain(self._sender)
        self._performance = PerformanceDomain(self._sender)
        self._profiler = ProfilerDomain(self._sender)
        self._debugger = DebuggerDomain(self._sender)
        self._overlay = OverlayDomain(self._sender)
        self._security = SecurityDomain(self._sender)
        self._audits = AuditsDomain(self._sender)
        self._accessibility = AccessibilityDomain(self._sender)
        self._storage = StorageDomain(self._sender)
        self._dom_storage = DOMStorageDomain(self._sender)
        self._tracing = TracingDomain(self._sender)
        self._animation = AnimationDomain(self._sender)
        self._service_worker = ServiceWorkerDomain(self._sender)
        self._system_info = SystemInfoDomain(self._sender)
        self._web_authn = WebAuthnDomain(self._sender)
        self._io = IODomain(self._sender)
        self._memory = MemoryDomain(self._sender)
        self._schema = SchemaDomain(self._sender)
        self._device_orientation = DeviceOrientationDomain(self._sender)
        self._sensor = SensorDomain(self._sender)
        self._headless_experimental = HeadlessExperimentalDomain(self._sender)
        self._tethering = TetheringDomain(self._sender)
        self._background_service = BackgroundServiceDomain(self._sender)
        self._cast = CastDomain(self._sender)
        self._preload = PreloadDomain(self._sender)
        self._indexed_db = IndexedDBDomain(self._sender)
        self._media = MediaDomain(self._sender)
        self._device_access = DeviceAccessDomain(self._sender)
        self._extensions = ExtensionsDomain(self._sender)
        self._pwa = PWADomain(self._sender)
        self._worker = WorkerDomain(self._sender)
        self._inspector = InspectorDomain(self._sender)
        self._cache_storage = CacheStorageDomain(self._sender)
        self._css = CSSDomain(self._sender)
        self._dom_debugger = DOMDebuggerDomain(self._sender)
        self._dom_snapshot = DOMSnapshotDomain(self._sender)
        self._event_breakpoints = EventBreakpointsDomain(self._sender)
        self._heap_profiler = HeapProfilerDomain(self._sender)
        self._layer_tree = LayerTreeDomain(self._sender)
        self._performance_timeline = PerformanceTimelineDomain(self._sender)
        self._autofill = AutofillDomain(self._sender)
        self._web_audio = WebAudioDomain(self._sender)
        self._ads = AdsDomain(self._sender)
        self._bluetooth_emulation = BluetoothEmulationDomain(self._sender)
        self._crash_report_context = CrashReportContextDomain(self._sender)
        self._digital_credentials = DigitalCredentialsDomain(self._sender)
        self._fed_cm = FedCmDomain(self._sender)
        self._file_system = FileSystemDomain(self._sender)
        self._smart_card_emulation = SmartCardEmulationDomain(self._sender)
        self._web_mcp = WebMCPDomain(self._sender)

    @property
    def page(self) -> PageDomain:
        """Page domain wrapper for navigation, screenshots, and PDF."""
        return self._page

    @property
    def runtime(self) -> RuntimeDomain:
        """Runtime domain wrapper for JS evaluation and remote objects."""
        return self._runtime

    @property
    def target(self) -> TargetDomain:
        """Target domain wrapper for session management."""
        return self._target

    @property
    def network(self) -> NetworkDomain:
        """Network domain wrapper for monitoring, cookies, and cache."""
        return self._network

    @property
    def dom(self) -> DOMDomain:
        """DOM domain wrapper for document inspection and manipulation."""
        return self._dom

    @property
    def log(self) -> LogDomain:
        """Log domain wrapper for browser log entries."""
        return self._log

    @property
    def console(self) -> ConsoleDomain:
        """Console domain wrapper (deprecated, use Runtime events)."""
        return self._console

    @property
    def input(self) -> InputDomain:
        """Input domain wrapper for synthetic input events (mouse, keyboard, touch)."""
        return self._input

    @property
    def emulation(self) -> EmulationDomain:
        """Emulation domain wrapper for device metrics, sensors, and throttling."""
        return self._emulation

    @property
    def fetch(self) -> FetchDomain:
        """Fetch domain wrapper for request interception and modification."""
        return self._fetch

    @property
    def performance(self) -> PerformanceDomain:
        """Performance domain wrapper for runtime metrics and timeline."""
        return self._performance

    @property
    def profiler(self) -> ProfilerDomain:
        """Profiler domain wrapper for CPU profiling and code coverage."""
        return self._profiler

    @property
    def debugger(self) -> DebuggerDomain:
        """Debugger domain wrapper for breakpoints, stepping, and script inspection."""
        return self._debugger

    @property
    def overlay(self) -> OverlayDomain:
        """Overlay domain wrapper for visual highlighting and inspect mode."""
        return self._overlay

    @property
    def security(self) -> SecurityDomain:
        """Security domain wrapper for certificate error handling."""
        return self._security

    @property
    def audits(self) -> AuditsDomain:
        """Audits domain wrapper for Lighthouse-style audits and contrast checks."""
        return self._audits

    @property
    def accessibility(self) -> AccessibilityDomain:
        """Accessibility domain wrapper for AX tree inspection."""
        return self._accessibility

    @property
    def storage(self) -> StorageDomain:
        """Storage domain wrapper for cookies, IndexedDB, and cache storage."""
        return self._storage

    @property
    def dom_storage(self) -> DOMStorageDomain:
        """DOMStorage domain wrapper for localStorage and sessionStorage."""
        return self._dom_storage

    @property
    def tracing(self) -> TracingDomain:
        """Tracing domain wrapper for performance tracing and timeline recording."""
        return self._tracing

    @property
    def animation(self) -> AnimationDomain:
        """Animation domain wrapper for CSS/Web animation inspection and control."""
        return self._animation

    @property
    def service_worker(self) -> ServiceWorkerDomain:
        """ServiceWorker domain wrapper for service worker inspection and control."""
        return self._service_worker

    @property
    def system_info(self) -> SystemInfoDomain:
        """SystemInfo domain wrapper for system and GPU information."""
        return self._system_info

    @property
    def web_authn(self) -> WebAuthnDomain:
        """WebAuthn domain wrapper for virtual authenticator management."""
        return self._web_authn

    @property
    def io(self) -> IODomain:
        """IO domain wrapper for reading stream handles."""
        return self._io

    @property
    def memory(self) -> MemoryDomain:
        """Memory domain wrapper for DOM counters, sampling, and GC control."""
        return self._memory

    @property
    def schema(self) -> SchemaDomain:
        """Schema domain wrapper for CDP domain discovery."""
        return self._schema

    @property
    def device_orientation(self) -> DeviceOrientationDomain:
        """DeviceOrientation domain wrapper for sensor simulation."""
        return self._device_orientation

    @property
    def sensor(self) -> SensorDomain:
        """Sensor domain wrapper for device sensor simulation."""
        return self._sensor

    @property
    def headless_experimental(self) -> HeadlessExperimentalDomain:
        """HeadlessExperimental domain wrapper for headless window bounds."""
        return self._headless_experimental

    @property
    def tethering(self) -> TetheringDomain:
        """Tethering domain wrapper for port binding."""
        return self._tethering

    @property
    def background_service(self) -> BackgroundServiceDomain:
        """BackgroundService domain wrapper for background service event observation."""
        return self._background_service

    @property
    def cast(self) -> CastDomain:
        """Cast domain wrapper for sink discovery and tab mirroring."""
        return self._cast

    @property
    def preload(self) -> PreloadDomain:
        """Preload domain wrapper for speculative loading control."""
        return self._preload

    @property
    def indexed_db(self) -> IndexedDBDomain:
        """IndexedDB domain wrapper for IndexedDB inspection and manipulation."""
        return self._indexed_db

    @property
    def media(self) -> MediaDomain:
        """Media domain wrapper for media player inspection."""
        return self._media

    @property
    def device_access(self) -> DeviceAccessDomain:
        """DeviceAccess domain wrapper for Bluetooth/USB device prompts."""
        return self._device_access

    @property
    def extensions(self) -> ExtensionsDomain:
        """Extensions domain wrapper for loading and managing extensions."""
        return self._extensions

    @property
    def pwa(self) -> PWADomain:
        """PWA domain wrapper for installing and managing PWAs."""
        return self._pwa

    @property
    def worker(self) -> WorkerDomain:
        """Worker domain wrapper for dedicated worker lifecycle events."""
        return self._worker

    @property
    def inspector(self) -> InspectorDomain:
        """Inspector domain wrapper for inspector lifecycle events."""
        return self._inspector

    @property
    def cache_storage(self) -> CacheStorageDomain:
        """CacheStorage domain wrapper for Cache API inspection."""
        return self._cache_storage

    @property
    def css(self) -> CSSDomain:
        """CSS domain wrapper for CSS styles and stylesheets."""
        return self._css

    @property
    def dom_debugger(self) -> DOMDebuggerDomain:
        """DOMDebugger domain wrapper for DOM and event breakpoints."""
        return self._dom_debugger

    @property
    def dom_snapshot(self) -> DOMSnapshotDomain:
        """DOMSnapshot domain wrapper for efficient flattened DOM capture."""
        return self._dom_snapshot

    @property
    def event_breakpoints(self) -> EventBreakpointsDomain:
        """EventBreakpoints domain wrapper for instrumentation breakpoints."""
        return self._event_breakpoints

    @property
    def heap_profiler(self) -> HeapProfilerDomain:
        """HeapProfiler domain wrapper for heap snapshots and allocation profiling."""
        return self._heap_profiler

    @property
    def layer_tree(self) -> LayerTreeDomain:
        """LayerTree domain wrapper for compositing layer inspection."""
        return self._layer_tree

    @property
    def performance_timeline(self) -> PerformanceTimelineDomain:
        """PerformanceTimeline domain wrapper for timeline events."""
        return self._performance_timeline

    @property
    def autofill(self) -> AutofillDomain:
        """Autofill domain wrapper for form autofill testing."""
        return self._autofill

    @property
    def web_audio(self) -> WebAudioDomain:
        """WebAudio domain wrapper for Web Audio API inspection and testing."""
        return self._web_audio

    @property
    def ads(self) -> AdsDomain:
        """Ads domain wrapper for ad metrics inspection."""
        return self._ads

    @property
    def bluetooth_emulation(self) -> BluetoothEmulationDomain:
        """BluetoothEmulation domain wrapper for Bluetooth testing."""
        return self._bluetooth_emulation

    @property
    def crash_report_context(self) -> CrashReportContextDomain:
        """CrashReportContext domain wrapper for crash report entries."""
        return self._crash_report_context

    @property
    def digital_credentials(self) -> DigitalCredentialsDomain:
        """DigitalCredentials domain wrapper for digital wallet behavior."""
        return self._digital_credentials

    @property
    def fed_cm(self) -> FedCmDomain:
        """FedCm domain wrapper for Federated Credential Management."""
        return self._fed_cm

    @property
    def file_system(self) -> FileSystemDomain:
        """FileSystem domain wrapper for File System Access API."""
        return self._file_system

    @property
    def smart_card_emulation(self) -> SmartCardEmulationDomain:
        """SmartCardEmulation domain wrapper for smart card testing."""
        return self._smart_card_emulation

    @property
    def web_mcp(self) -> WebMCPDomain:
        """WebMCP domain wrapper for Web MCP tool invocation."""
        return self._web_mcp

    @property
    def session_id(self) -> str:
        """The CDP session ID for this target."""
        return self._session_id

    @property
    def target_id(self) -> str:
        """The CDP target ID for this session."""
        return self._target_id

    @property
    def is_closed(self) -> bool:
        """Whether this session has been closed."""
        return self._closed

    @property
    def sub_sessions(self) -> list[CDPSession]:
        """Sub-sessions for auto-attached iframes and workers."""
        return list(self._sub_sessions.values())

    async def _enable_auto_attach(self) -> None:
        """Enable auto-attach to child targets (iframes, workers).

        Uses ``Target.setAutoAttach`` with ``flatten=True`` so sub-sessions
        share the same WebSocket. Incoming ``Target.attachedToTarget`` events
        create sub-sessions automatically.
        """
        self._auto_attach_enabled = True
        await self._target.set_auto_attach(
            auto_attach=True,
            flatten=True,
            wait_for_debugger_on_start=False,
        )

    def _handle_attached_to_target(self, params: dict[str, Any]) -> None:
        """Handle Target.attachedToTarget event for sub-session creation.

        Args:
            params: Event params with targetInfo and sessionId.
        """
        sub_session_id = params.get("sessionId")
        target_info = params.get("targetInfo", {})
        sub_target_id = target_info.get("targetId", "")
        if sub_session_id is None:
            return
        sub_session = CDPSession(
            connection=self._connection,
            session_id=sub_session_id,
            target_id=sub_target_id,
            client=self._client,
        )
        self._sub_sessions[sub_session_id] = sub_session
        if self._client is not None:
            self._client._sessions[sub_session_id] = sub_session
        logger.info(
            "Sub-session %s attached (target=%s)",
            sub_session_id,
            sub_target_id,
        )

    def _handle_detached_from_target(self, params: dict[str, Any]) -> None:
        """Handle Target.detachedFromTarget event for sub-session cleanup.

        Args:
            params: Event params with sessionId.
        """
        sub_session_id = params.get("sessionId")
        if sub_session_id is None:
            return
        sub_session = self._sub_sessions.pop(sub_session_id, None)
        if sub_session is not None:
            sub_session._closed = True
            sub_session._dispatcher.clear()
            if self._client is not None:
                self._client._session_dispatchers.pop(sub_session_id, None)
                self._client._sessions.pop(sub_session_id, None)
            logger.info("Sub-session %s detached", sub_session_id)

    async def send(
        self,
        method: str,
        params: dict[str, Any] | None = None,
    ) -> dict[str, Any]:
        """Send a raw CDP command as an escape hatch.

        Args:
            method: CDP method name (e.g. ``"Page.navigate"``).
            params: Optional command parameters.

        Returns:
            The CDP response result dict.

        Raises:
            SessionClosedError: If the session is closed.
        """
        if self._closed:
            raise SessionClosedError(
                f"Session {self._session_id} is closed"
            )
        return await self._sender(method, params)

    async def close(self) -> None:
        """Close the session and detach from the target.

        If the session's target was created by this client (via
        ``new_page()``), the target is also closed. If the session
        was attached to an existing target (via ``connect_to_page()``),
        only the session is detached. Sub-sessions from auto-attach are
        also cleaned up.
        """
        if self._closed:
            return
        self._closed = True
        self._dispatcher.clear()

        for sub_id, sub in list(self._sub_sessions.items()):
            sub._closed = True
            sub._dispatcher.clear()
            if self._client is not None:
                self._client._session_dispatchers.pop(sub_id, None)
                self._client._sessions.pop(sub_id, None)
        self._sub_sessions.clear()

        if self._client is not None:
            self._client._session_dispatchers.pop(self._session_id, None)
        with contextlib.suppress(Exception):
            await self._connection.send_command(
                "Target.detachFromTarget",
                {"sessionId": self._session_id},
            )
        if self._client is not None and self._target_id in self._client._managed_targets:
            with contextlib.suppress(Exception):
                await self._connection.send_command(
                    "Target.closeTarget",
                    {"targetId": self._target_id},
                )
            self._client._managed_targets.discard(self._target_id)
        logger.info("Session %s closed", self._session_id)

    def on(self, event_name: str, handler: EventHandler) -> Subscription:
        """Register an async handler for a CDP event.

        Args:
            event_name: CDP event name (e.g. ``"Page.loadEventFired"``).
            handler: Async callable that receives the event params dict.

        Returns:
            A Subscription that can be used to unsubscribe.
        """
        return self._dispatcher.on(event_name, handler)

    def off(self, event_name: str, handler: EventHandler) -> None:
        """Remove a previously registered event handler.

        Args:
            event_name: CDP event name.
            handler: The handler to remove.
        """
        self._dispatcher.off(event_name, handler)

    async def wait_for_event(
        self,
        event_name: str,
        timeout: float = 30.0,
    ) -> dict[str, Any]:
        """Wait for a single CDP event and return its params.

        Registers a one-shot handler that resolves when the event fires.
        The handler is automatically removed after the event is received
        or on timeout.

        Args:
            event_name: CDP event name (e.g. ``"Page.loadEventFired"``).
            timeout: Maximum seconds to wait.

        Returns:
            The event params dict.

        Raises:
            TimeoutError: If the event does not fire within ``timeout``.
        """
        event = asyncio.Event()
        captured: list[dict[str, Any]] = []

        async def _handler(params: dict[str, Any]) -> None:
            captured.append(params)
            event.set()

        sub = self.on(event_name, _handler)
        try:
            await asyncio.wait_for(event.wait(), timeout=timeout)
            return captured[0]
        finally:
            sub.unsubscribe()

    async def wait_for_navigation(
        self,
        url: str | None = None,
        timeout: float = 30.0,
    ) -> dict[str, Any]:
        """Wait for a navigation event.

        Args:
            url: Optional URL to wait for (substring match).
            timeout: Maximum seconds to wait.

        Returns:
            The ``Page.frameNavigated`` event params.
        """
        from cdpwave.waiters import wait_for_navigation as _wait

        return await _wait(self, url=url, timeout=timeout)

    async def wait_for_load_state(
        self,
        state: str = "load",
        timeout: float = 30.0,
    ) -> dict[str, Any]:
        """Wait for a specific page lifecycle event.

        Args:
            state: Lifecycle state (``"DOMContentLoaded"``, ``"load"``,
                ``"networkIdle"``, etc.).
            timeout: Maximum seconds to wait.

        Returns:
            The ``Page.lifecycleEvent`` event params.
        """
        from cdpwave.waiters import wait_for_load_state as _wait

        return await _wait(self, state=state, timeout=timeout)

    async def wait_for_selector(
        self,
        selector: str,
        root_node_id: int = 1,
        timeout: float = 30.0,
        poll_interval: float = 0.1,
    ) -> int:
        """Wait for a CSS selector to appear in the DOM.

        Args:
            selector: CSS selector to wait for.
            root_node_id: Root node ID to query from (default: document).
            timeout: Maximum seconds to wait.
            poll_interval: Seconds between polls.

        Returns:
            The DOM node ID of the matched element.
        """
        from cdpwave.waiters import wait_for_selector as _wait

        return await _wait(
            self,
            selector,
            root_node_id=root_node_id,
            timeout=timeout,
            poll_interval=poll_interval,
        )

    async def wait_for_network_idle(
        self,
        idle_time: float = 0.5,
        timeout: float = 30.0,
    ) -> None:
        """Wait until network activity settles.

        Args:
            idle_time: Seconds of no new requests before resolving.
            timeout: Maximum seconds to wait overall.
        """
        from cdpwave.waiters import wait_for_network_idle as _wait

        await _wait(self, idle_time=idle_time, timeout=timeout)

    async def __aenter__(self) -> CDPSession:
        """Enter async context manager."""
        return self

    async def __aexit__(
        self,
        exc_type: object,
        exc_val: object,
        exc_tb: object,
    ) -> None:
        """Exit async context manager and close the session."""
        await self.close()

page property

page: PageDomain

Page domain wrapper for navigation, screenshots, and PDF.

runtime property

runtime: RuntimeDomain

Runtime domain wrapper for JS evaluation and remote objects.

target property

target: TargetDomain

Target domain wrapper for session management.

network property

network: NetworkDomain

Network domain wrapper for monitoring, cookies, and cache.

dom property

dom: DOMDomain

DOM domain wrapper for document inspection and manipulation.

log property

log: LogDomain

Log domain wrapper for browser log entries.

console property

console: ConsoleDomain

Console domain wrapper (deprecated, use Runtime events).

input property

input: InputDomain

Input domain wrapper for synthetic input events (mouse, keyboard, touch).

emulation property

emulation: EmulationDomain

Emulation domain wrapper for device metrics, sensors, and throttling.

fetch property

fetch: FetchDomain

Fetch domain wrapper for request interception and modification.

performance property

performance: PerformanceDomain

Performance domain wrapper for runtime metrics and timeline.

profiler property

profiler: ProfilerDomain

Profiler domain wrapper for CPU profiling and code coverage.

debugger property

debugger: DebuggerDomain

Debugger domain wrapper for breakpoints, stepping, and script inspection.

overlay property

overlay: OverlayDomain

Overlay domain wrapper for visual highlighting and inspect mode.

security property

security: SecurityDomain

Security domain wrapper for certificate error handling.

audits property

audits: AuditsDomain

Audits domain wrapper for Lighthouse-style audits and contrast checks.

accessibility property

accessibility: AccessibilityDomain

Accessibility domain wrapper for AX tree inspection.

storage property

storage: StorageDomain

Storage domain wrapper for cookies, IndexedDB, and cache storage.

dom_storage property

dom_storage: DOMStorageDomain

DOMStorage domain wrapper for localStorage and sessionStorage.

tracing property

tracing: TracingDomain

Tracing domain wrapper for performance tracing and timeline recording.

animation property

animation: AnimationDomain

Animation domain wrapper for CSS/Web animation inspection and control.

service_worker property

service_worker: ServiceWorkerDomain

ServiceWorker domain wrapper for service worker inspection and control.

system_info property

system_info: SystemInfoDomain

SystemInfo domain wrapper for system and GPU information.

web_authn property

web_authn: WebAuthnDomain

WebAuthn domain wrapper for virtual authenticator management.

io property

io: IODomain

IO domain wrapper for reading stream handles.

memory property

memory: MemoryDomain

Memory domain wrapper for DOM counters, sampling, and GC control.

schema property

schema: SchemaDomain

Schema domain wrapper for CDP domain discovery.

device_orientation property

device_orientation: DeviceOrientationDomain

DeviceOrientation domain wrapper for sensor simulation.

sensor property

sensor: SensorDomain

Sensor domain wrapper for device sensor simulation.

headless_experimental property

headless_experimental: HeadlessExperimentalDomain

HeadlessExperimental domain wrapper for headless window bounds.

tethering property

tethering: TetheringDomain

Tethering domain wrapper for port binding.

background_service property

background_service: BackgroundServiceDomain

BackgroundService domain wrapper for background service event observation.

cast property

cast: CastDomain

Cast domain wrapper for sink discovery and tab mirroring.

preload property

preload: PreloadDomain

Preload domain wrapper for speculative loading control.

indexed_db property

indexed_db: IndexedDBDomain

IndexedDB domain wrapper for IndexedDB inspection and manipulation.

media property

media: MediaDomain

Media domain wrapper for media player inspection.

device_access property

device_access: DeviceAccessDomain

DeviceAccess domain wrapper for Bluetooth/USB device prompts.

extensions property

extensions: ExtensionsDomain

Extensions domain wrapper for loading and managing extensions.

pwa property

pwa: PWADomain

PWA domain wrapper for installing and managing PWAs.

worker property

worker: WorkerDomain

Worker domain wrapper for dedicated worker lifecycle events.

inspector property

inspector: InspectorDomain

Inspector domain wrapper for inspector lifecycle events.

cache_storage property

cache_storage: CacheStorageDomain

CacheStorage domain wrapper for Cache API inspection.

css property

css: CSSDomain

CSS domain wrapper for CSS styles and stylesheets.

dom_debugger property

dom_debugger: DOMDebuggerDomain

DOMDebugger domain wrapper for DOM and event breakpoints.

dom_snapshot property

dom_snapshot: DOMSnapshotDomain

DOMSnapshot domain wrapper for efficient flattened DOM capture.

event_breakpoints property

event_breakpoints: EventBreakpointsDomain

EventBreakpoints domain wrapper for instrumentation breakpoints.

heap_profiler property

heap_profiler: HeapProfilerDomain

HeapProfiler domain wrapper for heap snapshots and allocation profiling.

layer_tree property

layer_tree: LayerTreeDomain

LayerTree domain wrapper for compositing layer inspection.

performance_timeline property

performance_timeline: PerformanceTimelineDomain

PerformanceTimeline domain wrapper for timeline events.

autofill property

autofill: AutofillDomain

Autofill domain wrapper for form autofill testing.

web_audio property

web_audio: WebAudioDomain

WebAudio domain wrapper for Web Audio API inspection and testing.

ads property

ads: AdsDomain

Ads domain wrapper for ad metrics inspection.

bluetooth_emulation property

bluetooth_emulation: BluetoothEmulationDomain

BluetoothEmulation domain wrapper for Bluetooth testing.

crash_report_context property

crash_report_context: CrashReportContextDomain

CrashReportContext domain wrapper for crash report entries.

digital_credentials property

digital_credentials: DigitalCredentialsDomain

DigitalCredentials domain wrapper for digital wallet behavior.

fed_cm property

fed_cm: FedCmDomain

FedCm domain wrapper for Federated Credential Management.

file_system property

file_system: FileSystemDomain

FileSystem domain wrapper for File System Access API.

smart_card_emulation property

smart_card_emulation: SmartCardEmulationDomain

SmartCardEmulation domain wrapper for smart card testing.

web_mcp property

web_mcp: WebMCPDomain

WebMCP domain wrapper for Web MCP tool invocation.

session_id property

session_id: str

The CDP session ID for this target.

target_id property

target_id: str

The CDP target ID for this session.

is_closed property

is_closed: bool

Whether this session has been closed.

sub_sessions property

sub_sessions: list[CDPSession]

Sub-sessions for auto-attached iframes and workers.

send async

send(method: str, params: dict[str, Any] | None = None) -> dict[str, Any]

Send a raw CDP command as an escape hatch.

Parameters:

Name	Type	Description	Default
method	str	CDP method name (e.g. "Page.navigate").	required
params	dict[str, Any] | None	Optional command parameters.	None

Returns:

Type	Description
dict[str, Any]	The CDP response result dict.

Raises:

Type	Description
SessionClosedError	If the session is closed.

Source code in cdpwave/client.py

async def send(
    self,
    method: str,
    params: dict[str, Any] | None = None,
) -> dict[str, Any]:
    """Send a raw CDP command as an escape hatch.

    Args:
        method: CDP method name (e.g. ``"Page.navigate"``).
        params: Optional command parameters.

    Returns:
        The CDP response result dict.

    Raises:
        SessionClosedError: If the session is closed.
    """
    if self._closed:
        raise SessionClosedError(
            f"Session {self._session_id} is closed"
        )
    return await self._sender(method, params)

close async

close() -> None

Close the session and detach from the target.

If the session's target was created by this client (via new_page()), the target is also closed. If the session was attached to an existing target (via connect_to_page()), only the session is detached. Sub-sessions from auto-attach are also cleaned up.

Source code in cdpwave/client.py

async def close(self) -> None:
    """Close the session and detach from the target.

    If the session's target was created by this client (via
    ``new_page()``), the target is also closed. If the session
    was attached to an existing target (via ``connect_to_page()``),
    only the session is detached. Sub-sessions from auto-attach are
    also cleaned up.
    """
    if self._closed:
        return
    self._closed = True
    self._dispatcher.clear()

    for sub_id, sub in list(self._sub_sessions.items()):
        sub._closed = True
        sub._dispatcher.clear()
        if self._client is not None:
            self._client._session_dispatchers.pop(sub_id, None)
            self._client._sessions.pop(sub_id, None)
    self._sub_sessions.clear()

    if self._client is not None:
        self._client._session_dispatchers.pop(self._session_id, None)
    with contextlib.suppress(Exception):
        await self._connection.send_command(
            "Target.detachFromTarget",
            {"sessionId": self._session_id},
        )
    if self._client is not None and self._target_id in self._client._managed_targets:
        with contextlib.suppress(Exception):
            await self._connection.send_command(
                "Target.closeTarget",
                {"targetId": self._target_id},
            )
        self._client._managed_targets.discard(self._target_id)
    logger.info("Session %s closed", self._session_id)

on

on(event_name: str, handler: EventHandler) -> Subscription

Register an async handler for a CDP event.

Parameters:

Name	Type	Description	Default
event_name	str	CDP event name (e.g. "Page.loadEventFired").	required
handler	EventHandler	Async callable that receives the event params dict.	required

Returns:

Type	Description
Subscription	A Subscription that can be used to unsubscribe.

Source code in cdpwave/client.py

def on(self, event_name: str, handler: EventHandler) -> Subscription:
    """Register an async handler for a CDP event.

    Args:
        event_name: CDP event name (e.g. ``"Page.loadEventFired"``).
        handler: Async callable that receives the event params dict.

    Returns:
        A Subscription that can be used to unsubscribe.
    """
    return self._dispatcher.on(event_name, handler)

off

off(event_name: str, handler: EventHandler) -> None

Remove a previously registered event handler.

Parameters:

Name	Type	Description	Default
event_name	str	CDP event name.	required
handler	EventHandler	The handler to remove.	required

Source code in cdpwave/client.py

def off(self, event_name: str, handler: EventHandler) -> None:
    """Remove a previously registered event handler.

    Args:
        event_name: CDP event name.
        handler: The handler to remove.
    """
    self._dispatcher.off(event_name, handler)

wait_for_event async

wait_for_event(event_name: str, timeout: float = 30.0) -> dict[str, Any]

Wait for a single CDP event and return its params.

Registers a one-shot handler that resolves when the event fires. The handler is automatically removed after the event is received or on timeout.

Parameters:

Name	Type	Description	Default
event_name	str	CDP event name (e.g. "Page.loadEventFired").	required
timeout	float	Maximum seconds to wait.	30.0

Returns:

Type	Description
dict[str, Any]	The event params dict.

Raises:

Type	Description
TimeoutError	If the event does not fire within timeout.

Source code in cdpwave/client.py

async def wait_for_event(
    self,
    event_name: str,
    timeout: float = 30.0,
) -> dict[str, Any]:
    """Wait for a single CDP event and return its params.

    Registers a one-shot handler that resolves when the event fires.
    The handler is automatically removed after the event is received
    or on timeout.

    Args:
        event_name: CDP event name (e.g. ``"Page.loadEventFired"``).
        timeout: Maximum seconds to wait.

    Returns:
        The event params dict.

    Raises:
        TimeoutError: If the event does not fire within ``timeout``.
    """
    event = asyncio.Event()
    captured: list[dict[str, Any]] = []

    async def _handler(params: dict[str, Any]) -> None:
        captured.append(params)
        event.set()

    sub = self.on(event_name, _handler)
    try:
        await asyncio.wait_for(event.wait(), timeout=timeout)
        return captured[0]
    finally:
        sub.unsubscribe()

wait_for_navigation async

wait_for_navigation(url: str | None = None, timeout: float = 30.0) -> dict[str, Any]

Wait for a navigation event.

Parameters:

Name	Type	Description	Default
url	str | None	Optional URL to wait for (substring match).	None
timeout	float	Maximum seconds to wait.	30.0

Returns:

Type	Description
dict[str, Any]	The Page.frameNavigated event params.

Source code in cdpwave/client.py

async def wait_for_navigation(
    self,
    url: str | None = None,
    timeout: float = 30.0,
) -> dict[str, Any]:
    """Wait for a navigation event.

    Args:
        url: Optional URL to wait for (substring match).
        timeout: Maximum seconds to wait.

    Returns:
        The ``Page.frameNavigated`` event params.
    """
    from cdpwave.waiters import wait_for_navigation as _wait

    return await _wait(self, url=url, timeout=timeout)

wait_for_load_state async

wait_for_load_state(state: str = 'load', timeout: float = 30.0) -> dict[str, Any]

Wait for a specific page lifecycle event.

Parameters:

Name	Type	Description	Default
state	str	Lifecycle state ("DOMContentLoaded", "load", "networkIdle", etc.).	'load'
timeout	float	Maximum seconds to wait.	30.0

Returns:

Type	Description
dict[str, Any]	The Page.lifecycleEvent event params.

Source code in cdpwave/client.py

async def wait_for_load_state(
    self,
    state: str = "load",
    timeout: float = 30.0,
) -> dict[str, Any]:
    """Wait for a specific page lifecycle event.

    Args:
        state: Lifecycle state (``"DOMContentLoaded"``, ``"load"``,
            ``"networkIdle"``, etc.).
        timeout: Maximum seconds to wait.

    Returns:
        The ``Page.lifecycleEvent`` event params.
    """
    from cdpwave.waiters import wait_for_load_state as _wait

    return await _wait(self, state=state, timeout=timeout)

wait_for_selector async

wait_for_selector(selector: str, root_node_id: int = 1, timeout: float = 30.0, poll_interval: float = 0.1) -> int

Wait for a CSS selector to appear in the DOM.

Parameters:

Name	Type	Description	Default
selector	str	CSS selector to wait for.	required
root_node_id	int	Root node ID to query from (default: document).	1
timeout	float	Maximum seconds to wait.	30.0
poll_interval	float	Seconds between polls.	0.1

Returns:

Type	Description
int	The DOM node ID of the matched element.

Source code in cdpwave/client.py

async def wait_for_selector(
    self,
    selector: str,
    root_node_id: int = 1,
    timeout: float = 30.0,
    poll_interval: float = 0.1,
) -> int:
    """Wait for a CSS selector to appear in the DOM.

    Args:
        selector: CSS selector to wait for.
        root_node_id: Root node ID to query from (default: document).
        timeout: Maximum seconds to wait.
        poll_interval: Seconds between polls.

    Returns:
        The DOM node ID of the matched element.
    """
    from cdpwave.waiters import wait_for_selector as _wait

    return await _wait(
        self,
        selector,
        root_node_id=root_node_id,
        timeout=timeout,
        poll_interval=poll_interval,
    )

wait_for_network_idle async

wait_for_network_idle(idle_time: float = 0.5, timeout: float = 30.0) -> None

Wait until network activity settles.

Parameters:

Name	Type	Description	Default
idle_time	float	Seconds of no new requests before resolving.	0.5
timeout	float	Maximum seconds to wait overall.	30.0

Source code in cdpwave/client.py

async def wait_for_network_idle(
    self,
    idle_time: float = 0.5,
    timeout: float = 30.0,
) -> None:
    """Wait until network activity settles.

    Args:
        idle_time: Seconds of no new requests before resolving.
        timeout: Maximum seconds to wait overall.
    """
    from cdpwave.waiters import wait_for_network_idle as _wait

    await _wait(self, idle_time=idle_time, timeout=timeout)

__aenter__ async

__aenter__() -> CDPSession

Enter async context manager.

Source code in cdpwave/client.py

async def __aenter__(self) -> CDPSession:
    """Enter async context manager."""
    return self

__aexit__ async

__aexit__(exc_type: object, exc_val: object, exc_tb: object) -> None

Exit async context manager and close the session.

Source code in cdpwave/client.py

async def __aexit__(
    self,
    exc_type: object,
    exc_val: object,
    exc_tb: object,
) -> None:
    """Exit async context manager and close the session."""
    await self.close()