io.websockets

ServerConnection

class ServerConnection(Protocol)

send

def send(message: Union[Data, Iterable[Data]]) -> None

Send a message to the client.

Arguments:

• message Union[Data, Iterable[Data]] - The message to send.

recv

def recv(timeout: Optional[float] = None) -> Data

Receive a message from the client.

Arguments:

• timeout Optional[float], optional - The timeout for the receive operation. Defaults to None.

Returns:

• Data - The message received from the client.

close

def close() -> None

Close the connection.

WebSocketServer

class WebSocketServer(Protocol)

serve_forever

def serve_forever() -> None

Run the server forever.

shutdown

def shutdown() -> None

Shutdown the server.

__enter__

def __enter__() -> "WebSocketServer"

Enter the server context.

__exit__

def __exit__(exc_type: Any, exc_value: Any, traceback: Any) -> None

Exit the server context.

IOWebsockets

class IOWebsockets(IOStream)

A websocket input/output stream.

__init__

def __init__(websocket: ServerConnection) -> None

Initialize the websocket input/output stream.

Arguments:

• websocket ServerConnection - The websocket server.

Raises:

• ImportError - If the websockets module is not available.

run_server_in_thread

@staticmethod
@contextmanager
def run_server_in_thread(*,
                         host: str = "127.0.0.1",
                         port: int = 8765,
                         on_connect: Callable[["IOWebsockets"], None],
                         ssl_context: Optional[ssl.SSLContext] = None,
                         **kwargs: Any) -> Iterator[str]

Factory function to create a websocket input/output stream.

Arguments:

• host str, optional - The host to bind the server to. Defaults to “127.0.0.1”.
• port int, optional - The port to bind the server to. Defaults to 8765.
• on_connect Callable[[IOWebsockets], None] - The function to be executed on client connection. Typically creates agents and initiate chat.
• ssl_context Optional[ssl.SSLContext], optional - The SSL context to use for secure connections. Defaults to None.
• kwargs Any - Additional keyword arguments to pass to the websocket server.

Yields:

• str - The URI of the websocket server.

websocket

@property
def websocket() -> "ServerConnection"

The URI of the websocket server.

print

def print(*objects: Any,
          sep: str = " ",
          end: str = "\n",
          flush: bool = False) -> None

Print data to the output stream.

Arguments:

• objects any - The data to print.
• sep str, optional - The separator between objects. Defaults to ” ”.
• end str, optional - The end of the output. Defaults to ” ”.
• flush bool, optional - Whether to flush the output. Defaults to False.

input

def input(prompt: str = "", *, password: bool = False) -> str

Read a line from the input stream.

Arguments:

• prompt str, optional - The prompt to display. Defaults to "".
• password bool, optional - Whether to read a password. Defaults to False.

Returns:

• str - The line read from the input stream.