MCPcopy Index your code
hub / github.com/python-websockets/websockets / process_subprotocol

Method process_subprotocol

src/websockets/client.py:249–283  ·  view source on GitHub ↗

Handle the Sec-WebSocket-Protocol HTTP response header. If provided, check that it contains exactly one supported subprotocol. Args: headers: WebSocket handshake response headers. Returns: Subprotocol, if one was selected.

(self, headers: Headers)

Source from the content-addressed store, hash-verified

247 return accepted_extensions
248
249 def process_subprotocol(self, headers: Headers) -> Subprotocol | None:
250 """
251 Handle the Sec-WebSocket-Protocol HTTP response header.
252
253 If provided, check that it contains exactly one supported subprotocol.
254
255 Args:
256 headers: WebSocket handshake response headers.
257
258 Returns:
259 Subprotocol, if one was selected.
260
261 """
262 subprotocol: Subprotocol | None = None
263
264 subprotocols = headers.get_all("Sec-WebSocket-Protocol")
265
266 if subprotocols:
267 if self.available_subprotocols is None:
268 raise NegotiationError("no subprotocols supported")
269
270 parsed_subprotocols: Sequence[Subprotocol] = sum(
271 [parse_subprotocol(header_value) for header_value in subprotocols], []
272 )
273 if len(parsed_subprotocols) > 1:
274 raise InvalidHeader(
275 "Sec-WebSocket-Protocol",
276 f"multiple values: {', '.join(parsed_subprotocols)}",
277 )
278
279 subprotocol = parsed_subprotocols[0]
280 if subprotocol not in self.available_subprotocols:
281 raise NegotiationError(f"unsupported subprotocol: {subprotocol}")
282
283 return subprotocol
284
285 def send_request(self, request: Request) -> None:
286 """

Callers 1

process_responseMethod · 0.95

Calls 4

NegotiationErrorClass · 0.85
parse_subprotocolFunction · 0.85
InvalidHeaderClass · 0.85
get_allMethod · 0.80

Tested by

no test coverage detected