# gql.transport.common.adapters.websockets

class gql.transport.common.adapters.websockets.WebSocketsAdapter(url: str, *, headers: Headers | Mapping[str, str] | Iterable[tuple[str, str]] | SupportsKeysAndGetItem | None = None, ssl: SSLContext | bool = False, connect_args: Dict[str, Any] | None = None)

  Bases: `AdapterConnection`

  AdapterConnection implementation using the websockets library.

  __init__(url: str, *, headers: Headers | Mapping[str, str] | Iterable[tuple[str, str]] | SupportsKeysAndGetItem | None = None, ssl: SSLContext | bool = False, connect_args: Dict[str, Any] | None = None) → None

    Initialize the transport with the given parameters.

    Parameters:

      - url – The GraphQL server URL. Example: ‘wss://server.com:PORT/graphql’.

      - headers – Dict of HTTP Headers.

      - ssl – ssl_context of the connection. Use ssl=False to disable encryption

      - connect_args – Other parameters forwarded to websockets.connect

  async connect() → None

    Connect to the WebSocket server.

  async send(message: str) → None

    Send message to the WebSocket server.

    Args:

      message: String message to send

    Raises:

      TransportConnectionFailed: If connection closed

  async receive() → str

    Receive message from the WebSocket server.

    Returns:

      String message received

    Raises:

      TransportConnectionFailed: If connection closed TransportProtocolError: If protocol error or binary data received

  async close() → None

    Close the WebSocket connection.

  property headers: Headers | Mapping[str, str] | Iterable[tuple[str, str]] | SupportsKeysAndGetItem | None

    Get the response headers from the WebSocket connection.

    Returns:

      Dictionary of response headers

  property response_headers: Dict[str, str]

    Get the response headers from the WebSocket connection.

    Returns:

      Dictionary of response headers

  url: str
  connect_args: Dict[str, Any]
  subprotocols: List[str] | None