mud_server.admin_tui.api.client

HTTP API client for PipeWorks MUD Server.

This module provides an async HTTP client for communicating with the MUD server’s REST API. It handles authentication, session management, and all administrative operations.

The client is designed to be used as an async context manager to ensure proper resource cleanup:

async with AdminAPIClient(config) as client:

await client.login(“admin”, “password”) health = await client.get_health()

Key Features:

• Async HTTP requests using httpx

• Automatic session management

• Custom exceptions for error handling

• Type hints for all public methods

Exceptions

APIError	Exception raised when an API request fails.
AuthenticationError	Exception raised for authentication-related failures.

Classes

SessionState	Tracks the current authentication session state.
AdminAPIClient	Async HTTP client for the MUD server admin API.

Module Contents

exception mud_server.admin_tui.api.client.APIError

Bases: Exception

Exception raised when an API request fails.

This exception is raised for any non-2xx response from the server, except for authentication-specific errors which raise AuthenticationError.

message

Human-readable error message.

status_code

HTTP status code from the response.

detail

Additional detail from the server response, if available.

Example

try:

await client.get_health()

except APIError as e:

print(f”API error {e.status_code}: {e.message}”)

Initialize self. See help(type(self)) for accurate signature.

message: str

status_code: int = 0

detail: str = ''

exception mud_server.admin_tui.api.client.AuthenticationError

Bases: APIError

Exception raised for authentication-related failures.

This includes:

• Invalid credentials (401)

• Permission denied (403)

• Session expired

• Not authenticated when required

Example

try:

await client.login(“admin”, “wrong_password”)

except AuthenticationError as e:

print(f”Login failed: {e.detail}”)

Initialize self. See help(type(self)) for accurate signature.

class mud_server.admin_tui.api.client.SessionState

Tracks the current authentication session state.

This dataclass maintains information about the current user session, including authentication status and user role. It is updated automatically by the AdminAPIClient during login/logout operations.

session_id

The session ID returned by the server after login. None if not authenticated.

username

The authenticated username. None if not authenticated.

role

The user’s role (e.g., “admin”, “superuser”, “player”). None if not authenticated.

Properties:

is_authenticated: True if currently logged in with a valid session. is_admin: True if the user has admin or superuser role. is_superuser: True if the user has superuser role.

Example

state = SessionState() print(state.is_authenticated) # False

state.session_id = “abc123” state.username = “admin” state.role = “admin” print(state.is_authenticated) # True print(state.is_admin) # True

session_id: str | None = None

username: str | None = None

role: str | None = None

property is_authenticated: bool

Check if we have an active session.

property is_admin: bool

Check if the user has admin privileges.

property is_superuser: bool

Check if the user has superuser privileges.

clear()

Clear all session state (logout).

class mud_server.admin_tui.api.client.AdminAPIClient

Async HTTP client for the MUD server admin API.

This client handles all communication with the MUD server’s REST API. It must be used as an async context manager to properly manage the underlying HTTP connection pool.

config

Configuration object with server URL and timeout settings.

session

Current authentication session state.

Example

config = Config(server_url=”http://localhost:8000”, timeout=30.0)

async with AdminAPIClient(config) as client:

# Login await client.login(“admin”, “password123”)

# Check server health health = await client.get_health() print(f”Active players: {health[‘active_players’]}”)

# Get list of players (requires admin) players = await client.get_players()

# Logout await client.logout()

config: mud_server.admin_tui.config.Config

session: SessionState

property http_client: httpx.AsyncClient

Get the HTTP client, ensuring it’s been initialized.

Raises:

RuntimeError – If accessed outside of async context manager.

Returns:

The underlying httpx.AsyncClient.

Return type:

httpx.AsyncClient

async login(username, password)

Authenticate with the MUD server.

Sends credentials to the server and stores the session information on successful authentication.

Parameters:

• username (str) – The username to authenticate with.

• password (str) – The user’s password.

Returns:

The server response containing session_id, role, and message.

Return type:

dict

Raises:

• AuthenticationError – If credentials are invalid (401).

• APIError – If the server returns any other error.

Example

try:

result = await client.login(“admin”, “password123”) print(f”Logged in as {result[‘role’]}”)

except AuthenticationError:

print(“Invalid credentials”)

async logout()

End the current session.

Notifies the server of logout and clears local session state. The local state is always cleared, even if the server request fails.

Returns:

True if logout was performed, False if not authenticated.

Return type:

bool

Example

if await client.logout():

print(“Logged out successfully”)

else:

print(“Was not logged in”)

async get_health()

Get server health status.

This endpoint is public and does not require authentication.

Returns:

Server health information including:

• status: “ok” if healthy

• active_players: Number of currently connected players

Return type:

dict

Raises:

APIError – If the server returns an error.

Example

health = await client.get_health() if health[“status”] == “ok”:

print(f”{health[‘active_players’]} players online”)

async get_players()

Get list of all players in the database.

Requires admin privileges.

Returns:

List of player dictionaries with username, role, etc.

Return type:

list

Raises:

• AuthenticationError – If not authenticated or lacks permission.

• APIError – If the server returns an error.

Example

players = await client.get_players() for player in players:

print(f”{player[‘username’]} ({player[‘role’]})”)

async get_sessions()

Get list of active sessions.

Requires admin privileges.

Returns:

List of session dictionaries with username, created_at, etc.

Return type:

list

Raises:

• AuthenticationError – If not authenticated or lacks permission.

• APIError – If the server returns an error.

Example

sessions = await client.get_sessions() print(f”{len(sessions)} active sessions”)

async get_chat_messages(limit=100)

Get recent chat messages from the database.

Requires admin privileges.

Parameters:

limit (int) – Maximum number of messages to return (default 100).

Returns:

List of chat message dictionaries with username, message, etc.

Return type:

list

Raises:

• AuthenticationError – If not authenticated or lacks permission.

• APIError – If the server returns an error.

Example

messages = await client.get_chat_messages(limit=50) for msg in messages:

print(f”{msg[‘username’]}: {msg[‘message’]}”)

async get_tables()

Get list of database tables with schema metadata.

Requires admin privileges.

Returns:

List of table metadata dictionaries with name, columns, row_count.

Return type:

list

Raises:

• AuthenticationError – If not authenticated or lacks permission.

• APIError – If the server returns an error.

async get_table_rows(table_name, limit=100)

Get rows and columns for a specific database table.

Requires admin privileges.

Parameters:

• table_name (str) – Table name to query.

• limit (int) – Max number of rows to return.

Returns:

Contains table, columns, and rows.

Return type:

dict

Raises:

• AuthenticationError – If not authenticated or lacks permission.

• APIError – If the server returns an error.

async get_player_locations()

Get player locations with zone context for admin display.

Requires admin privileges.

Returns:

List of player location dictionaries.

Return type:

list

Raises:

• AuthenticationError – If not authenticated or lacks permission.

• APIError – If the server returns an error.

async get_connections()

Get active session connections with activity age.

Requires admin privileges.

Returns:

List of connection dictionaries.

Return type:

list

Raises:

• AuthenticationError – If not authenticated or lacks permission.

• APIError – If the server returns an error.

async kick_session(target_session_id, reason=None)

Force-disconnect a session (admin/superuser only).

Parameters:

• target_session_id (str) – Session id to disconnect.

• reason (str | None) – Optional reason for the action.

Returns:

Response payload with success/message.

Return type:

dict

async manage_user(target_username, action, *, new_role=None, new_password=None)

Perform a user management action via the admin API.

Parameters:

• target_username (str) – Username to manage.

• action (str) – Action string (change_role, ban, unban, delete, change_password).

• new_role (str | None) – Required when action is change_role.

• new_password (str | None) – Required when action is change_password.

Returns:

Response payload with success/message.

Return type:

dict

async create_user(username, password, password_confirm, role)

Create a new user account (admin/superuser only).

Parameters:

• username (str) – Username for the new account.

• password (str) – Initial password for the account.

• password_confirm (str) – Confirmation of the password.

• role (str) – Role to assign to the new account.

Returns:

Response payload with success/message.

Return type:

dict

Raises:

• AuthenticationError – If the current session lacks permissions.

• APIError – If validation fails or the server rejects the request.