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.