​
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.
consolebase_logger