appmesh package

Submodules

appmesh.app module

Application definition

class appmesh.app.App(data=None)

Bases: object

An application in App Mesh, include all the process attributes, resource limitations, behaviors, and permissions.

Initialize an App instance with optional configuration data.

class Behavior(data=None)

Bases: object

Application error handling behavior, including exit and control behaviors.

class Action(value)

Bases: Enum

Actions for application exit behaviors.

KEEPALIVE = 'keepalive'

REMOVE = 'remove'

RESTART = 'restart'

STANDBY = 'standby'

control

standby), higher priority than default exit behavior

Type:

Exit code specific behavior (e.g, –control 0

Type:

restart –control 1

exit

‘restart’, ‘standby’, ‘keepalive’, ‘remove’.

Type:

Default exit behavior, options

set_control_behavior(control_code, action)

Define behavior for specific exit codes.

Return type:

None

set_exit_behavior(action)

Set default behavior for application exit.

Return type:

None

class DailyLimitation(data=None)

Bases: object

Application availability within a daily time range.

daily_end

00+08).

Type:

End time for application availability (e.g., 09

Type:

00

daily_start

00+08).

Type:

Start time for application availability (e.g., 09

Type:

00

set_daily_range(start, end)

Set the valid daily start and end times.

Return type:

None

class Permission(value)

Bases: Enum

Application permission levels.

DENY = '1'

READ = '2'

WRITE = '3'

class ResourceLimitation(data=None)

Bases: object

Application resource limits, such as CPU and memory usage.

cpu_shares

CPU shares, relative weight of CPU usage.

memory_mb

Physical memory limit in MB.

memory_virt_mb

Virtual memory limit in MB.

command

full command line with arguments

container_id

docker container id

cpu

cpu usage

cron

Whether the interval is specified as a cron expression

description

app description string

docker_image

Docker image for containerized execution

end_time

05’)

Type:

end date time for app (ISO8601 time format, e.g., ‘2020-10-11T10

Type:

22

env

environment variables (e.g., -e env1=value1 -e env2=value2, APP_DOCKER_OPTS is used to input docker run parameters)

fd

file descriptor usage

property group_permission: Permission | None

Typed view of the group-user digit (ones place) of the permission int.

health

0 for healthy, 1 for unhealthy

Type:

health status

health_check_cmd

port/health’, return 0 is health)

Type:

health check script command (e.g., sh -x ‘curl host

property is_enabled: bool | None

True when enabled (1), False when disabled (0), None when unset.

Type:

Typed view of status

last_error

last error message

last_exit_time

last exit time

last_start_time

last start time

memory

memory usage

metadata

metadata string/JSON (input for app, pass to process stdin)

name

app name (unique)

next_start_time

next start time

property others_permission: Permission | None

Typed view of the other-users digit (tens place) of the permission int.

owner

owner name of app mesh user who created the app

permission

2, write:3 (see set_permission(), group_permission/others_permission).

Type:

app user permission, two decimal digits [others][group], each digit deny

Type:

1, read

pid

process id used to attach to the running process

pstree

process tree

register_time

app register time

retention

extra timeout seconds for stopping current process, support ISO 8601 durations (e.g., ‘P1Y2M3DT4H5M6S’ ‘P5W’).

return_code

last process exit code

sec_env

security environment variables, encrypt in server side with app owner’s cipher

session_login

Whether to run the app in session login mode (inheriting the user’s full login environment)

set_env(key, value, secure=False)

Set an environment variable, marking it secure if specified.

Return type:

None

set_permission(group_user, others_user)

Define application permissions based on user roles.

Return type:

None

set_valid_time(start, end)

Define the valid time window for the application.

Return type:

None

shell

Whether run command in shell mode (enables shell syntax such as pipes and compound commands)

start_interval_seconds

start interval seconds for short running app, support integer seconds, ISO 8601 durations and cron expression (e.g., 30, ‘P1Y2M3DT4H5M6S’, ‘P5W’, ‘* */5 * * * *’)

start_time

05’)

Type:

start date time for app (ISO8601 time format, e.g., ‘2020-10-11T09

Type:

22

starts

number of times started

status

1 for enabled, 0 for disabled

Type:

app status

stdout_cache_num

maximum number of stdout log files to retain

stdout_cache_size

number of stdout log files currently retained

subscription_id

subscription id returned by the daemon when add_app is called atomically with subscribe_events on a TCP/WSS client; empty for HTTP or when no subscribe_events was supplied

task_id

current task id

task_status

task status

to_dict()

Convert the application data into a JSON-compatible dictionary, removing empty items. Attributes are serialized under their wire names (user -> pid_user).

Return type:

Dict[str, Any]

user

process OS user name (wire field pid_user); distinct from owner, the App Mesh user who created the app

version

app version

working_dir

working directory

appmesh.app_output module

Application output information.

class appmesh.app_output.AppOutput(status_code, output, output_position, exit_code)

Bases: object

Output information returned by the get_app_output() API.

Includes the application’s stdout, current read position, HTTP status code, and process exit code.

exit_code: int | None

Exit code of the application, or None if still running.

output: str

Captured stdout content of the application.

output_position: int | None

Current read position in stdout stream (wire header X-Output-Position), or None if not applicable.

status_code: HTTPStatus

HTTP status code from the get_app_output() API request.

appmesh.app_run module

Application run object for remote application execution.

class appmesh.app_run.AppRun(client, app_name, process_uuid)

Bases: object

Application run object for monitoring and retrieving results of a remote application run initiated by run_app_async().

app_name

Name of the application associated with this run.

process_uuid

Unique process UUID from run_app_async() (wire field process_uuid).

wait(stdout_handler=None, timeout=0)

Wait for the asynchronous run to complete with the saved forwarding target restored.

Parameters:

• stdout_handler (Optional[Callable[[str, int], None]]) – optional callback (data, position) -> None invoked with each chunk of stdout. Use print_output_handler for console output.

• timeout (int) – Maximum time to wait in seconds. 0 means wait indefinitely.

Return type:

Optional[int]

Returns:

Exit code if the process finished, or None when timeout elapsed first.

Raises:

• AppMeshConnectionError – On polling/transport failure while waiting.

• AppMeshAppRemovedError – If the app was removed before its exit was observed (TCP/WSS).

Warning

While waiting, the SHARED client’s forward_to is temporarily overridden (see _use_forward_host); concurrent requests are routed to that host.

appmesh.app_run.print_output_handler(data, position)

Pre-built OutputHandler that prints data to stdout.

Return type:

None

appmesh.appmesh_client module

appmesh.client_http module

App Mesh HTTP Client SDK for REST API interactions.

class appmesh.client_http.AppMeshClient(base_url='https://127.0.0.1:6060', ssl_verify=None, ssl_client_cert=None, request_timeout=(60, 300), jwt_token=None, cookie_file=None, auto_refresh_token=False)

Bases: object

Client SDK for interacting with the App Mesh service via REST API.

The AppMeshClient class provides a comprehensive interface for managing and monitoring distributed applications within the App Mesh ecosystem. It enables communication with the App Mesh REST API for operations such as application lifecycle management, monitoring, and configuration.

This client is designed for direct usage in applications that require access to App Mesh services over HTTP-based REST.

- TLS

Supports secure connections between the client and App Mesh service, ensuring encrypted communication.

Type:

Transport Layer Security

- JWT

Provides secure API access with token-based authentication and authorization to enforce fine-grained permissions.

Type:

JSON Web Token) and RBAC (Role-Based Access Control

# Authentication Management

- login()

- logout()

- authenticate()

- renew_token()

- disable_totp()

- get_totp_uri()

- get_totp_secret()

- enable_totp()

# Application Management

- add_app()

- delete_app()

- disable_app()

- enable_app()

- check_app_health()

- get_app_output()

- get_app()

- list_apps()

# Run Application Operations

- run_app_async()

- wait_for_async_run()

- run_app_sync()

- run_task()

- cancel_task()

# Event Subscription (AppMeshClientTCP/AppMeshClientWSS only; see `supports_events`)

- subscribe()

- unsubscribe()

# System Management

- forward_to

- set_config()

- get_config()

- set_log_level()

- get_host_resources()

- get_metrics()

- add_label()

- delete_label()

- list_labels()

# File Management

- download_file()

- upload_file()

# User and Role Management

- add_user()

- delete_user()

- lock_user()

- update_password()

- get_current_user()

- unlock_user()

- list_users()

- get_user_permissions()

- list_permissions()

- delete_role()

- update_role()

- list_roles()

- list_groups()

Example

>>> python -m pip install --upgrade appmesh
>>> from appmesh import AppMeshClient
>>> client = AppMeshClient()
>>> client.login("your-name", "your-password")
>>> client.authenticate("your-token-for-token-login")
>>> response = client.get_app(app_name='ping')

Initialize an App Mesh HTTP client for interacting with the App Mesh server via secure HTTPS.

Parameters:

• base_url (str) – The server’s base URI. Defaults to “https://127.0.0.1:6060”.

• ssl_verify (Union[bool, str, None]) – SSL server verification mode: - None (default): Auto — use the App Mesh CA bundle if installed, otherwise system CAs. - True: Use system CAs. - False: Disable verification (insecure, must be requested explicitly). - str: Path to custom CA or directory (must exist). To include system CAs, combine them into one file (e.g., cat custom_ca.pem /etc/ssl/certs/ca-certificates.crt > combined_ca.pem).

• ssl_client_cert (Union[str, Tuple[str, str], None]) – SSL client certificate file(s): - str: Single PEM file with cert+key - tuple: (cert_path, key_path)

• request_timeout (Tuple[float, float]) – Timeouts (connect_timeout, read_timeout) in seconds. Default (60, 300).

• jwt_token (Optional[str]) – JWT token set directly without server verification (no network call).

• cookie_file (Optional[str]) – Cookie file path for HTTP clients (set this to enable persistent cookie storage).

• auto_refresh_token (bool) – Enable automatic token refresh before expiration (supports App Mesh and Keycloak tokens).

add_app(app, subscribe_events=None)

Register a new application.

subscribe_events only takes effect on a persistent connection (TCP/WSS) and is ignored by the HTTP transport (no demuxer to deliver events to; a RuntimeWarning is emitted). When the daemon creates a subscription, the returned App carries subscription_id.

Return type:

App

add_label(label_name, label_value)

Add a new label.

Return type:

None

add_user(username, user_data)

Add a new user.

Return type:

None

authenticate(token, permission=None, audience=None, update_session=True)

Verify the provided JWT token with the server and optionally update the client session.

Parameters:

• token (str) – JWT token to verify.

• permission (Optional[str]) – Optional permission ID to check (e.g., ‘app-view’, ‘app-delete’).

• audience (Optional[str]) – Optional audience value to verify against the token.

• update_session (bool) – When True, update the current client session with the verified token and persist local token state on success. When False, only verify the provided token and leave local state unchanged.

Return type:

Tuple[bool, str]

Returns:

Tuple of (success, message) where message is the raw response text. (False, message) covers every non-OK status; never raises for a failed verification.

cancel_task(app_name)

Cancel a running task for an App Mesh application.

Parameters:

app_name (str) – Name of the target application (as registered in App Mesh).

Returns:

True if a task existed and was cancelled. False means no task was found (404) or the request failed for another reason (e.g. 401/403); non-404 failures are logged as warnings, never raised.

Return type:

bool

check_app_health(app_name)

Check the health status of an application.

Return type:

bool

close()

Close the client and release resources.

Return type:

None

delete_app(app_name)

Remove an application.

Returns:

True when the app was deleted, False when it did not exist (404). Any other non-OK status is logged and raised as an HTTP error.

Return type:

bool

delete_label(label_name)

Delete a label.

Return type:

None

delete_role(role_name)

Delete a user role.

Return type:

None

delete_user(username)

Delete a user.

Return type:

None

disable_app(app_name)

Disable an application.

Return type:

None

disable_totp(username=None)

Disable 2FA for the specified user.

Parameters:

username (Optional[str]) – Target user name; defaults to “self” (the current user).

Return type:

None

download_file(remote_file, local_file, preserve_permissions=True)

Download a remote file to the local filesystem.

When preserve_permissions is True, POSIX mode/owner/group metadata from App Mesh response headers is applied best-effort on non-Windows platforms.

Return type:

None

enable_app(app_name)

Enable an application.

Return type:

None

enable_totp(totp_code)

Set up 2FA for the current user.

Parameters:

totp_code (str) – TOTP code.

Return type:

None

property forward_to: str

Target host for request forwarding in a cluster.

Supports: - “hostname” or “IP” → uses current service port - “hostname:port” or “IP:port” → uses specified port

Returns:

Target host (e.g., “node” or “node:6060”), or empty string if unset.

Return type:

str

Notes

For JWT sharing across the cluster: - All nodes must use the same JWTSalt and Issuer for JWT settings - If port is omitted, current service port is used

Warning

Shared, per-client state read by every request; AppRun.wait() temporarily overrides it. Use separate client instances for concurrent multi-host access.

get_app(app_name)

Get information about a specific application.

Return type:

App

get_app_output(app_name, stdout_position=0, stdout_index=0, stdout_maxsize=10240, process_uuid='', timeout=0)

Get incremental stdout/stderr output for a running or completed application.

Parameters:

• app_name (str) – the application name

• stdout_position (int) – start read position, 0 means start from beginning.

• stdout_index (int) – index of history process stdout, 0 means get from current running process, the stdout number depends on ‘stdout_cache_size’ of the application.

• stdout_maxsize (int) – max buffer size to read.

• process_uuid (str) – used to get the specified process instance instead of the latest one.

• timeout (int) – long-poll wait time in seconds before returning when no new output is available.

Return type:

AppOutput

Returns:

AppOutput containing response status, payload text, the next read cursor (output_position), and exit_code when the process has already finished.

get_config()

Get the App Mesh configuration in JSON format.

Return type:

Dict[str, Any]

get_current_user()

Get information about the current user.

Return type:

dict

get_host_resources()

Get a report of host resources including CPU, memory, and disk.

Return type:

Dict[str, Any]

get_metrics()

Get Prometheus metrics.

Return type:

str

get_totp_secret()

Return the raw TOTP secret for the current user.

The server responds with a base64-encoded OTP provisioning URI; this helper parses that URI and returns only the secret field. Use get_totp_uri() for the full URI.

Return type:

str

get_totp_uri()

Return the TOTP provisioning URI (otpauth://...) for the current user, decoded from the server’s base64 mfa_uri payload, e.g. for rendering a QR code in an authenticator app.

Return type:

str

get_user_permissions()

Get information about the permissions of the current user.

Return type:

List[str]

list_apps()

Get information about all applications.

Return type:

List[App]

list_groups()

Get information about all user groups.

Return type:

List[str]

list_labels()

Get information about all labels.

Return type:

Dict[str, str]

list_permissions()

Get information about all available permissions.

Return type:

List[str]

list_roles()

Get information about all roles with permission definitions.

Return type:

Dict[str, Dict]

list_users()

Get information about all users.

Return type:

Dict[str, Any]

lock_user(username)

Lock a user.

Return type:

None

login(username, password, totp_code=None, token_expire='P1W', audience=None)

Login with username and password and attach the issued token to this client.

Parameters:

• username (str) – The name of the user.

• password (str) – The password of the user.

• totp_code (Optional[str]) – The TOTP code if enabled for the user.

• token_expire (Union[str, int]) – Token expiration duration. Supports ISO 8601 durations (e.g., ‘P1Y2M3DT4H5M6S’ ‘P1W’).

• audience (Optional[str]) – The audience of the JWT token, should be available by JWT service configuration (default is ‘appmesh-service’).

Return type:

Optional[str]

Returns:

TOTP challenge string if the server responds with HTTP 428 and no code was supplied, otherwise None. On success, the session token/cookie is updated and auto-refresh starts when enabled for this client.

logout()

Logout from the current session.

Returns:

True on success; False when no token was set locally or the server rejected the logoff (logged as a warning). Never raises for a failed logoff.

Return type:

bool

renew_token(token_expire='P1W')

Renew the current JWT token.

Parameters:

token_expire (Union[int, str]) – Token expiration duration (integer seconds or ISO 8601 string).

Return type:

None

run_app_async(app, max_time='P2D', lifecycle='P2DT12H')

Run an application asynchronously on a remote system without blocking the API.

Parameters:

• app (Union[App, str]) – An App instance or a shell command string. - If app is a string, it is treated as a shell command for the remote run, and an App instance is created as: App({“command”: “<command_string>”, “shell”: True}). - If app is an App object, providing only the name attribute (without a command) will run an existing application; otherwise, it is treated as a new application.

• max_time (Union[int, str]) – Maximum runtime for the remote process, after which the daemon kills it (sent as the wire query parameter timeout). Accepts integer seconds or ISO 8601 duration format (e.g., ‘P1Y2M3DT4H5M6S’, ‘P5W’). Defaults to P2D.

• lifecycle (Union[int, str]) – Total retention window for the temporary run app, after which the daemon purges it (including its cached output); must cover max_time plus the time needed to collect results (sent as the wire query parameter lifecycle). Accepts integer seconds or ISO 8601 duration format. Defaults to P2DT12H.

Return type:

AppRun

Returns:

AppRun handle that captures the current forward_to target so later polling can continue against the same cluster node.

run_app_sync(app, max_time='P2D', lifecycle='P2DT12H')

Synchronously run an application remotely, blocking until completion, and return the result.

If ‘app’ is a string, it is treated as a shell command and converted to an App instance. If ‘app’ is App object, the name attribute is used to run an existing application if specified.

Parameters:

• app (Union[App, str]) – An App instance or a shell command string. If a string, an App instance is created as: appmesh.App({“command”: “<command_string>”, “shell”: True})

• max_time (Union[int, str]) – Maximum runtime for the remote process, after which the daemon kills it (sent as the wire query parameter timeout). Accepts integer seconds or ISO 8601 duration format (e.g., ‘P1Y2M3DT4H5M6S’, ‘P5W’).

• lifecycle (Union[int, str]) – Total retention window for the temporary run app, after which the daemon purges it (sent as the wire query parameter lifecycle). Accepts integer seconds or ISO 8601 duration format.

Return type:

Tuple[Optional[int], str]

Returns:

(exit_code, stdout_text). exit_code is None when the server did not return an X-Exit-Code header.

run_task(app_name, data, timeout=300)

Client send an invocation message to a running App Mesh application and wait for result.

This method posts the provided data to the App Mesh service which will forward it to the specified running application instance.

Parameters:

• app_name (str) – Name of the target application (as registered in App Mesh).

• data (str) – Payload to deliver to the application. Typically a string.

• timeout (int) – Maximum time in seconds to wait for a response from the application. Defaults to 300 seconds.

Returns:

The HTTP response body returned by the remote application/service.

Return type:

str

set_config(config)

Update the configuration.

Return type:

Dict[str, Any]

set_log_level(level='DEBUG')

Update the log level.

Return type:

str

set_token(token)

Set a JWT token directly without server-side verification. Use when the token is already known to be valid. For server-side verification, use authenticate() instead.

Parameters:

token (str) – A valid JWT token string. The token is stored in the client’s cookie jar and persisted immediately when cookie_file is configured. The cookie expiry follows the token’s exp claim, falling back to 7 days when absent or undecodable.

Return type:

None

start_token_refresh()

Start background token auto-refresh.

Return type:

None

stop_token_refresh()

Stop background token auto-refresh.

Return type:

None

supports_events = False

unlock_user(username)

Unlock a user.

Return type:

None

update_password(old_password, new_password, username='self')

Change the password of a user.

Return type:

None

update_role(role_name, permission_set)

Update or add a role with defined permissions.

Return type:

None

upload_file(local_file, remote_file, preserve_permissions=True)

Upload a local file to the remote server.

When preserve_permissions is True, the client also sends local POSIX metadata in request headers so the server can recreate permissions/ownership when supported.

Return type:

None

validate_totp(username, challenge, code, token_expire='P1W')

Validate TOTP challenge and obtain a new JWT token.

Parameters:

• username (str) – Username to validate.

• challenge (str) – Challenge string from server.

• code (str) – TOTP code to validate.

• token_expire (Union[int, str]) –

  Token expiration duration, defaults to _DURATION_ONE_WEEK_ISO (1 week). Accepts either:

  • ISO 8601 duration string (e.g., ‘P1Y2M3DT4H5M6S’, ‘P1W’)

  • Numeric value (seconds) for simpler cases.

Return type:

None

wait_for_async_run(run, stdout_handler=None, timeout=0)

Wait for an asynchronous run to finish.

Parameters:

• run (AppRun) – asynchronous run handle returned by run_app_async().

• stdout_handler (Optional[Callable[[str, int], None]]) – optional callback (data, position) -> None invoked with each chunk of remote stdout (print_output_handler prints to console).

• timeout (int) – wait max timeout seconds and return if not finished, 0 means wait until finished

Return type:

Optional[int]

Returns:

Exit code if the process finished, or None when timeout elapsed first. On success, this method also makes a best-effort attempt to delete the temporary run app.

Raises:

AppMeshConnectionError – If polling the app output fails (non-OK response).

appmesh.client_http_oauth module

appmesh.client_tcp module

class appmesh.client_tcp.AppMeshClientTCP(tcp_address=('127.0.0.1', 6059), ssl_verify=None, ssl_client_cert=None, auto_refresh_token=False)

Bases: TransportClientMixin, AppMeshClient

Client SDK for interacting with the App Mesh service over TCP.

The AppMeshClientTCP class extends the functionality of AppMeshClient by offering a TCP-based communication layer for the App Mesh REST API. It overrides the file download and upload methods to support large file transfers with improved performance, leveraging TCP for lower latency and higher throughput compared to HTTP.

This client is suitable for applications requiring efficient data transfers and high-throughput operations within the App Mesh ecosystem, while maintaining compatibility with all other attributes and methods from AppMeshClient.

Inherits all attributes from `AppMeshClient`, including TLS secure connections and JWT-based authentication.

- download_file()

- upload_file()

- Inherits all other methods from `AppMeshClient`, providing a consistent interface for managing applications within App Mesh.

Example

>>> from appmesh import AppMeshClientTCP
>>> client = AppMeshClientTCP()
>>> client.login("your-name", "your-password")
>>> client.download_file("/tmp/os-release", "os-release")

Construct a TCP transport client that reuses the standard App Mesh client API.

Parameters:

• tcp_address (Tuple[str, int]) – Server address as (host, port) tuple, defaults to (“127.0.0.1”, 6059).

• ssl_verify (Union[bool, str, None]) – SSL certificate verification behavior. Can be None, True, False, or a path to CA bundle. - None (default): Auto — use the App Mesh CA bundle if installed, otherwise system CAs - True: Use system CA certificates (e.g., /etc/ssl/certs/ on Linux) - False: Disable verification (insecure, must be requested explicitly) - str: Path to custom CA bundle or directory (must exist)

• ssl_client_cert (Union[str, Tuple[str, str], None]) – SSL client certificate: - str: Path to single PEM with cert+key - tuple: (cert_path, key_path)

Note

TCP connections require an explicit full-chain CA specification for certificate validation, unlike HTTP, which can retrieve intermediate certificates automatically.

close()

Close the connection and release resources.

Return type:

None

download_file(remote_file, local_file, preserve_permissions=True)

Copy a remote file to local through the TCP file-socket side channel.

Parameters:

• remote_file (str) – Remote file path.

• local_file (str) – Local destination path.

• preserve_permissions (bool) – Apply remote file permissions/ownership locally on a best-effort basis.

Return type:

None

upload_file(local_file, remote_file, preserve_permissions=True)

Upload a local file to the remote server through the TCP file-socket side channel.

Parameters:

• local_file (str) – Local file path.

• remote_file (str) – Remote destination path.

• preserve_permissions (bool) – Send local file permissions/ownership metadata when available.

Return type:

None

appmesh.client_wss module

class appmesh.client_wss.AppMeshClientWSS(wss_address=('127.0.0.1', 6058), ssl_verify=None, ssl_client_cert=None, auto_refresh_token=False)

Bases: TransportClientMixin, AppMeshClient

Client SDK for interacting with the App Mesh service over WebSocket Secure (WSS).

The AppMeshClientWSS class extends the functionality of AppMeshClient by offering a WSS-based communication layer for the App Mesh REST API. It overrides the file download and upload methods to support large file transfers with improved performance, leveraging WebSocket for lower latency and higher throughput compared to HTTP.

This client is suitable for applications requiring efficient bidirectional data transfers and high-throughput operations within the App Mesh ecosystem, while maintaining compatibility with all other attributes and methods from AppMeshClient.

Inherits all attributes from `AppMeshClient`, including TLS secure connections and JWT-based authentication.

- download_file()

- upload_file()

- Inherits all other methods from `AppMeshClient`, providing a consistent interface for managing applications within App Mesh.

Example

>>> from appmesh import AppMeshClientWSS
>>> client = AppMeshClientWSS()
>>> client.login("your-name", "your-password")
>>> client.download_file("/tmp/os-release", "os-release")

Construct a WSS transport client that reuses the standard App Mesh client API.

Parameters:

• wss_address (Tuple[str, int]) – Server address as (host, port) tuple, defaults to (“127.0.0.1”, 6058).

• ssl_verify (Union[bool, str, None]) – SSL certificate verification behavior. Can be None, True, False, or a path to CA bundle. - None (default): Auto — use the App Mesh CA bundle if installed, otherwise system CAs - True: Use system CA certificates (e.g., /etc/ssl/certs/ on Linux) - False: Disable verification (insecure, must be requested explicitly) - str: Path to custom CA bundle or directory (must exist)

• ssl_client_cert (Union[str, Tuple[str, str], None]) – SSL client certificate: - str: Path to single PEM with cert+key - tuple: (cert_path, key_path)

Note

WSS connections require an explicit full-chain CA specification for certificate validation, unlike HTTP, which can retrieve intermediate certificates automatically.

close()

Close the connection and release resources.

Return type:

None

download_file(remote_file, local_file, preserve_permissions=True)

Copy a remote file to local through the WSS control channel plus HTTPS data channel.

Parameters:

• remote_file (str) – Remote file path.

• local_file (str) – Local destination path.

• preserve_permissions (bool) – Apply remote file permissions/ownership locally on a best-effort basis.

Return type:

None

upload_file(local_file, remote_file, preserve_permissions=True)

Upload a local file through the WSS control channel plus HTTPS data channel.

Parameters:

• local_file (str) – Local file path.

• remote_file (str) – Remote destination path.

• preserve_permissions (bool) – Send local file permissions/ownership metadata when available.

Return type:

None

appmesh.exceptions module

App Mesh SDK exception hierarchy.

exception appmesh.exceptions.AppMeshAppRemovedError

Bases: AppMeshError

The application was removed before its process exit was observed.

exception appmesh.exceptions.AppMeshAuthError

Bases: AppMeshError

Authentication or authorization error.

exception appmesh.exceptions.AppMeshConnectionError

Bases: AppMeshError

Connection or transport error.

exception appmesh.exceptions.AppMeshError

Bases: Exception

Base exception for all App Mesh SDK errors.

exception appmesh.exceptions.AppMeshProcessSupersededError

Bases: AppMeshError

The current process key was superseded by a newer process instance (HTTP 412).

exception appmesh.exceptions.AppMeshRequestError

Bases: AppMeshError

HTTP request failed.

exception appmesh.exceptions.AppMeshTimeoutError

Bases: AppMeshConnectionError

Receive timeout on an otherwise healthy connection (safe to retry/continue).

appmesh.server_http module

HTTP worker SDK implementation for App Mesh (task fetch/return loop).

class appmesh.server_http.AppMeshWorker(base_url='https://127.0.0.1:6060', ssl_verify=None, ssl_client_cert=None, request_timeout=(60, 300), *, client=None, logger=None)

Bases: object

Worker SDK for an App Mesh application interacting with the local App Mesh REST service over HTTPS.

Despite running inside the managed application, this is a client-side task-loop helper: it polls the daemon for task payloads and returns results.

Build-in runtime environment variables required:

• APP_MESH_PROCESS_KEY

• APP_MESH_APPLICATION_NAME

- fetch_task()

fetch invocation payloads

- send_task_result()

return results to the invoking client

Example

context = appmesh.AppMeshWorker() payload = context.fetch_task() result = do_something_with(payload) context.send_task_result(result)

Initialize a worker-side helper for task fetch/return.

Parameters:

• base_url (str) – The server’s base URI. Defaults to “https://127.0.0.1:6060”.

• ssl_verify (Union[bool, str, None]) – SSL server verification mode (None = auto: App Mesh CA bundle if installed, else system CAs).

• ssl_client_cert (Union[str, Tuple[str, str], None]) – SSL client certificate file(s).

• request_timeout (Tuple[float, float]) – Timeouts (connect_timeout, read_timeout) in seconds.

• client (Optional[AppMeshClient]) – Pre-configured AppMeshClient instance (used by TCP/WSS subclasses so all transports share the same task API). Takes precedence: when provided, the connection parameters (base_url, ssl_verify, ssl_client_cert, request_timeout) are ignored.

• logger (Optional[Logger]) – Optional logger instance.

fetch_task(*, stop_event=None, max_retries=None)

Fetch task data in the currently running App Mesh application process.

Used by an App Mesh application process to obtain the payload from the App Mesh service that a client pushed to it. By default retries indefinitely until successful. If a request fails within 100ms, sleeps briefly before retrying; otherwise retries immediately.

Parameters:

• stop_event (Optional[Event]) – Optional cancellation event checked between attempts; when set, fetching stops and AppMeshError is raised.

• max_retries (Optional[int]) – Optional cap on retries after a failed attempt (N allows up to N + 1 attempts); when exhausted AppMeshError is raised. None (default) retries forever.

Return type:

Union[str, bytes]

Returns:

The payload bytes provided by the invoking client.

Raises:

• AppMeshProcessSupersededError – The daemon reported HTTP 412 — this process key was superseded by a newer process instance; the caller should stop serving.

• AppMeshError – Cancelled via stop_event or max_retries exhausted.

send_task_result(result)

Send the result of a server-side invocation back to the original client.

Used by App Mesh application process to post the result to App Mesh service after processing payload data so the invoking client can retrieve it.

Parameters:

result (Union[str, bytes]) – Result payload to be delivered back to the client exactly as provided.

Return type:

None

appmesh.server_tcp module

class appmesh.server_tcp.AppMeshWorkerTCP(ssl_verify=None, ssl_client_cert=None, tcp_address=('127.0.0.1', 6059), *, logger=None)

Bases: AppMeshWorker

Worker SDK for interacting with the local App Mesh service over TCP (TLS).

Example

>>> worker = AppMeshWorkerTCP(tcp_address=("127.0.0.1", 6059))
>>> payload = worker.fetch_task()

Construct an App Mesh worker TCP object to communicate securely with an App Mesh server over TLS.

Note

Positional order is (ssl_verify, ssl_client_cert, tcp_address) — differs from AppMeshClientTCP (address first); prefer keyword arguments.

Parameters:

• ssl_verify (Union[bool, str, None]) – SSL server verification mode; same semantics as AppMeshClientTCP.

• ssl_client_cert (Union[str, Tuple[str, str], None]) – SSL client certificate file(s); same semantics as AppMeshClientTCP.

• tcp_address (Tuple[str, int]) – Server address as (host, port) tuple, defaults to (“127.0.0.1”, 6059).

• logger (Optional[Logger]) – Optional logger instance.

appmesh.server_wss module

class appmesh.server_wss.AppMeshWorkerWSS(wss_address=('127.0.0.1', 6058), ssl_verify=None, ssl_client_cert=None, *, logger=None)

Bases: AppMeshWorker

Worker SDK for interacting with the local App Mesh service over WebSockets (WSS).

Example

>>> worker = AppMeshWorkerWSS(wss_address=("127.0.0.1", 6058))
>>> payload = worker.fetch_task()

Construct an App Mesh worker WSS object to communicate securely with an App Mesh server over TLS.

Note

Positional order is (wss_address, ssl_verify, ssl_client_cert) — differs from AppMeshWorkerTCP (ssl_verify first); prefer keyword arguments.

Parameters:

• wss_address (Tuple[str, int]) – Server address as (host, port) tuple, defaults to (“127.0.0.1”, 6058).

• ssl_verify (Union[bool, str, None]) – SSL server verification mode; same semantics as AppMeshClientWSS.

• ssl_client_cert (Union[str, Tuple[str, str], None]) – SSL client certificate file(s); same semantics as AppMeshClientWSS.

• logger (Optional[Logger]) – Optional logger instance.

appmesh.subscribe module

Event subscription support for TCP and WSS transports.

class appmesh.subscribe.AppEvent(subscription_id='', event_type='', app_name='', timestamp=0, sequence=0, data=<factory>)

Bases: object

Represents a server-push event notification.

app_name: str = ''

data: Dict[str, Any]

event_type: str = ''

sequence: int = 0

subscription_id: str = ''

timestamp: int = 0

class appmesh.subscribe.SubscriptionResult(subscription_id='', app_name='', events=<factory>)

Bases: object

Server’s response to a subscribe request.

app_name: str = ''

events: list

subscription_id: str = ''

appmesh.tcp_messages module

TCP message classes for HTTP-like communication.

class appmesh.tcp_messages.RequestMessage(uuid='', request_uri='', http_method='', client_addr='', body=b'', headers=<factory>, query=<factory>)

Bases: object

TCP request message for HTTP-like communication.

body: bytes = b''

client_addr: str = ''

headers: Dict[str, str]

http_method: str = ''

query: Dict[str, str]

request_uri: str = ''

serialize()

Serialize request message to bytes.

Return type:

bytes

uuid: str = ''

class appmesh.tcp_messages.ResponseMessage(uuid='', request_uri='', http_status=0, body_msg_type='', body=b'', headers=<factory>)

Bases: object

TCP response message for HTTP-like communication.

body: bytes = b''

body_msg_type: str = ''

classmethod from_bytes(buf)

Deserialize TCP msgpack buffer with proper type conversion.

Return type:

ResponseMessage

headers: Dict[str, str]

http_status: int = 0

request_uri: str = ''

uuid: str = ''

appmesh.tcp_transport module

TCP Transport layer handling socket connections.

class appmesh.tcp_transport.TCPTransport(address, ssl_verify, ssl_client_cert=None)

Bases: object

TCP Transport layer with TLS support.

Initialize TCP transport with TLS configuration.

Parameters:

• address (Tuple[str, int]) – Server address as (host, port) tuple.

• ssl_verify (Union[bool, str]) – SSL server verification mode: - True: Use system CA certificates - False: Disable verification (insecure) - str: Path to custom CA bundle or directory

• ssl_client_cert (Union[str, Tuple[str, str], None]) – SSL client certificate: - str: Path to PEM file with cert and key - tuple: (cert_path, key_path)

Note

TCP connections require an explicit full-chain CA specification for certificate validation, unlike HTTP, which can retrieve intermediate certificates automatically.

TCP_MAX_BLOCK_SIZE = 1073741824

TCP_MESSAGE_HEADER_LENGTH = 8

TCP_MESSAGE_MAGIC = 130484216

close()

Close socket

Return type:

None

connect()

Establish TLS connection to server.

Return type:

None

connected()

Compatibility method; prefer the is_connected property (local socket state only).

Return type:

bool

property is_connected: bool

Whether the transport holds an open socket object. Local state only — does NOT probe the peer, so a server-dropped connection may report True until the next I/O fails.

receive_message()

Receive a message with prefixed header.

Return type:

bytes

Returns:

Message data, or b"" for a zero-length EOF frame (earlier releases returned None; both are falsy).

send_message(data)

Send a message with prefixed header.

Parameters:

data (Union[bytes, bytearray, list, None]) – Message data to send. b"" (empty bytes) is the canonical EOF signal; an empty list or None is also accepted for compatibility.

Return type:

None

appmesh.transport_mixin module

Shared transport logic for TCP and WSS clients.

class appmesh.transport_mixin.TransportClientMixin

Bases: object

Mixin providing shared request/response logic for TCP and WSS transport clients.

Design note: TCP/WSS clients deliberately inherit AppMeshClient rather than wrap it — every REST method funnels through _request_http, so overriding that one choke point with msgpack framing (adapted into a requests.Response) reuses all inherited methods and token/cookie persistence unchanged, at the cost of a mostly idle requests.Session (which the WSS client reuses for its file-transfer HTTPS data channel).

Subclasses must define:

• _transport: the transport object (TCPTransport or WSSTransport)

• _token: the current access token string

• _HTTP_USER_AGENT_TRANSPORT: user agent string for this transport

add_app(app, subscribe_events=None, callback=None)

Register an app, optionally subscribing atomically and wiring a local callback.

Reuses the base add_app for the HTTP round-trip + subscription_id parsing, then registers callback against the local demuxer keyed by the new subscription.

Return type:

App

subscribe(app_name, events=None, callback=None)

Subscribe to app events over the transport connection.

Parameters:

• app_name (str) – Application name, or “*” for all apps.

• events (Optional[list]) – List of event types (e.g. [“START”, “EXIT”, “STDOUT”]).

• callback (Optional[Callable[[AppEvent], None]]) – Function called with AppEvent for each received event.

Return type:

SubscriptionResult

Returns:

SubscriptionResult with subscription_id, app_name, and events.

supports_events = True

unsubscribe(subscription_id)

Remove an event subscription.

Parameters:

subscription_id (str) – The subscription ID returned by subscribe().

Return type:

None

wait_for_async_run(run, stdout_handler=None, timeout=0)

Override: use subscribe-based streaming on TCP/WSS instead of polling.

Subscribes to STDOUT + EXIT + REMOVED, then does a one-shot get_app_output to backfill bytes emitted before the subscribe took effect. Stdout events whose position is already covered by an earlier delivery are deduped (partial overlap → prefix trimmed).

Return type:

Optional[int]

Returns:

Exit code if the process finished, or None when timeout elapsed first.

Raises:

• AppMeshAppRemovedError – If the app was removed before its exit was observed.

• AppMeshConnectionError – If the transport disconnected while waiting, or the daemon delivered an unparseable exit code.

appmesh.wss_transport module

WebSocket Secure (WSS) Transport layer handling WebSocket connections.

class appmesh.wss_transport.WSSTransport(address, ssl_verify, ssl_client_cert=None)

Bases: object

WebSocket Secure (WSS) Transport layer with TLS support using synchronous websocket-client library.

Initialize WebSocket Secure (WSS) transport with TLS configuration.

Parameters:

• address (Tuple[str, int]) – Server address as (host, port) tuple.

• ssl_verify (Union[bool, str]) – SSL server verification mode: - True: Use system CA certificates - False: Disable verification (insecure) - str: Path to custom CA bundle or directory

• ssl_client_cert (Union[str, Tuple[str, str], None]) – SSL client certificate: - str: Path to PEM file with cert and key - tuple: (cert_path, key_path)

Note

This implementation uses synchronous blocking sockets for WebSocket connections. No threading or asyncio is involved for simplicity and reliability.

WSS_CONNECT_TIMEOUT = 30

WSS_MAX_BLOCK_SIZE = 104857600

WSS_MESSAGE_TIMEOUT = 60

close()

Close WebSocket connection.

Return type:

None

connect()

Establish WSS connection to server.

Return type:

None

connected()

Compatibility method; prefer the is_connected property.

Return type:

bool

property is_connected: bool

Whether the websocket-client connected flag reports connected (updated on close/error); does NOT probe the peer with network I/O.

receive_message()

Receive one application message from the WebSocket.

Uses the high-level recv() API so that control frames (PING/PONG/CLOSE) are handled inside websocket-client — in particular, PING is auto-replied with PONG, which is what keeps long-idle subscribe connections alive against server-side idleTimeout. Returns the data as bytearray, or an empty bytearray for EOF / non-data frames (falsy, matching TCPTransport’s b"" EOF convention).

Return type:

Optional[bytearray]

send_message(data)

Send a message over WebSocket.

Parameters:

data (Union[bytes, bytearray, list, None]) – Message data to send. b"" (empty bytes) is the canonical EOF signal; an empty list or None is also accepted for compatibility.

Return type:

None

Note

WebSocket handles message framing automatically, so we don’t need to add a length header. Just send msgpack-serialized data directly.

Module contents

App Mesh SDK package initializer with lazy loading support.

Example

from appmesh import AppMeshClient client = AppMeshClient()

class appmesh.App(data=None)

Bases: object

An application in App Mesh, include all the process attributes, resource limitations, behaviors, and permissions.

Initialize an App instance with optional configuration data.

class Behavior(data=None)

Bases: object

Application error handling behavior, including exit and control behaviors.

class Action(value)

Bases: Enum

Actions for application exit behaviors.

KEEPALIVE = 'keepalive'

REMOVE = 'remove'

RESTART = 'restart'

STANDBY = 'standby'

control

standby), higher priority than default exit behavior

Type:

Exit code specific behavior (e.g, –control 0

Type:

restart –control 1

exit

‘restart’, ‘standby’, ‘keepalive’, ‘remove’.

Type:

Default exit behavior, options

set_control_behavior(control_code, action)

Define behavior for specific exit codes.

Return type:

None

set_exit_behavior(action)

Set default behavior for application exit.

Return type:

None

class DailyLimitation(data=None)

Bases: object

Application availability within a daily time range.

daily_end

00+08).

Type:

End time for application availability (e.g., 09

Type:

00

daily_start

00+08).

Type:

Start time for application availability (e.g., 09

Type:

00

set_daily_range(start, end)

Set the valid daily start and end times.

Return type:

None

class Permission(value)

Bases: Enum

Application permission levels.

DENY = '1'

READ = '2'

WRITE = '3'

class ResourceLimitation(data=None)

Bases: object

Application resource limits, such as CPU and memory usage.

cpu_shares

CPU shares, relative weight of CPU usage.

memory_mb

Physical memory limit in MB.

memory_virt_mb

Virtual memory limit in MB.

command

full command line with arguments

container_id

docker container id

cpu

cpu usage

cron

Whether the interval is specified as a cron expression

description

app description string

docker_image

Docker image for containerized execution

end_time

05’)

Type:

end date time for app (ISO8601 time format, e.g., ‘2020-10-11T10

Type:

22

env

environment variables (e.g., -e env1=value1 -e env2=value2, APP_DOCKER_OPTS is used to input docker run parameters)

fd

file descriptor usage

property group_permission: Permission | None

Typed view of the group-user digit (ones place) of the permission int.

health

0 for healthy, 1 for unhealthy

Type:

health status

health_check_cmd

port/health’, return 0 is health)

Type:

health check script command (e.g., sh -x ‘curl host

property is_enabled: bool | None

True when enabled (1), False when disabled (0), None when unset.

Type:

Typed view of status

last_error

last error message

last_exit_time

last exit time

last_start_time

last start time

memory

memory usage

metadata

metadata string/JSON (input for app, pass to process stdin)

name

app name (unique)

next_start_time

next start time

property others_permission: Permission | None

Typed view of the other-users digit (tens place) of the permission int.

owner

owner name of app mesh user who created the app

permission

2, write:3 (see set_permission(), group_permission/others_permission).

Type:

app user permission, two decimal digits [others][group], each digit deny

Type:

1, read

pid

process id used to attach to the running process

pstree

process tree

register_time

app register time

retention

extra timeout seconds for stopping current process, support ISO 8601 durations (e.g., ‘P1Y2M3DT4H5M6S’ ‘P5W’).

return_code

last process exit code

sec_env

security environment variables, encrypt in server side with app owner’s cipher

session_login

Whether to run the app in session login mode (inheriting the user’s full login environment)

set_env(key, value, secure=False)

Set an environment variable, marking it secure if specified.

Return type:

None

set_permission(group_user, others_user)

Define application permissions based on user roles.

Return type:

None

set_valid_time(start, end)

Define the valid time window for the application.

Return type:

None

shell

Whether run command in shell mode (enables shell syntax such as pipes and compound commands)

start_interval_seconds

start interval seconds for short running app, support integer seconds, ISO 8601 durations and cron expression (e.g., 30, ‘P1Y2M3DT4H5M6S’, ‘P5W’, ‘* */5 * * * *’)

start_time

05’)

Type:

start date time for app (ISO8601 time format, e.g., ‘2020-10-11T09

Type:

22

starts

number of times started

status

1 for enabled, 0 for disabled

Type:

app status

stdout_cache_num

maximum number of stdout log files to retain

stdout_cache_size

number of stdout log files currently retained

subscription_id

subscription id returned by the daemon when add_app is called atomically with subscribe_events on a TCP/WSS client; empty for HTTP or when no subscribe_events was supplied

task_id

current task id

task_status

task status

to_dict()

Convert the application data into a JSON-compatible dictionary, removing empty items. Attributes are serialized under their wire names (user -> pid_user).

Return type:

Dict[str, Any]

user

process OS user name (wire field pid_user); distinct from owner, the App Mesh user who created the app

version

app version

working_dir

working directory

class appmesh.AppEvent(subscription_id='', event_type='', app_name='', timestamp=0, sequence=0, data=<factory>)

Bases: object

Represents a server-push event notification.

app_name: str = ''

data: Dict[str, Any]

event_type: str = ''

sequence: int = 0

subscription_id: str = ''

timestamp: int = 0

exception appmesh.AppMeshAppRemovedError

Bases: AppMeshError

The application was removed before its process exit was observed.

exception appmesh.AppMeshAuthError

Bases: AppMeshError

Authentication or authorization error.

class appmesh.AppMeshClient(base_url='https://127.0.0.1:6060', ssl_verify=None, ssl_client_cert=None, request_timeout=(60, 300), jwt_token=None, cookie_file=None, auto_refresh_token=False)

Bases: object

Client SDK for interacting with the App Mesh service via REST API.

The AppMeshClient class provides a comprehensive interface for managing and monitoring distributed applications within the App Mesh ecosystem. It enables communication with the App Mesh REST API for operations such as application lifecycle management, monitoring, and configuration.

This client is designed for direct usage in applications that require access to App Mesh services over HTTP-based REST.

- TLS

Supports secure connections between the client and App Mesh service, ensuring encrypted communication.

Type:

Transport Layer Security

- JWT

Provides secure API access with token-based authentication and authorization to enforce fine-grained permissions.

Type:

JSON Web Token) and RBAC (Role-Based Access Control

# Authentication Management

- login()

- logout()

- authenticate()

- renew_token()

- disable_totp()

- get_totp_uri()

- get_totp_secret()

- enable_totp()

# Application Management

- add_app()

- delete_app()

- disable_app()

- enable_app()

- check_app_health()

- get_app_output()

- get_app()

- list_apps()

# Run Application Operations

- run_app_async()

- wait_for_async_run()

- run_app_sync()

- run_task()

- cancel_task()

# Event Subscription (AppMeshClientTCP/AppMeshClientWSS only; see `supports_events`)

- subscribe()

- unsubscribe()

# System Management

- forward_to

- set_config()

- get_config()

- set_log_level()

- get_host_resources()

- get_metrics()

- add_label()

- delete_label()

- list_labels()

# File Management

- download_file()

- upload_file()

# User and Role Management

- add_user()

- delete_user()

- lock_user()

- update_password()

- get_current_user()

- unlock_user()

- list_users()

- get_user_permissions()

- list_permissions()

- delete_role()

- update_role()

- list_roles()

- list_groups()

Example

>>> python -m pip install --upgrade appmesh
>>> from appmesh import AppMeshClient
>>> client = AppMeshClient()
>>> client.login("your-name", "your-password")
>>> client.authenticate("your-token-for-token-login")
>>> response = client.get_app(app_name='ping')

Initialize an App Mesh HTTP client for interacting with the App Mesh server via secure HTTPS.

Parameters:

• base_url (str) – The server’s base URI. Defaults to “https://127.0.0.1:6060”.

• ssl_verify (Union[bool, str, None]) – SSL server verification mode: - None (default): Auto — use the App Mesh CA bundle if installed, otherwise system CAs. - True: Use system CAs. - False: Disable verification (insecure, must be requested explicitly). - str: Path to custom CA or directory (must exist). To include system CAs, combine them into one file (e.g., cat custom_ca.pem /etc/ssl/certs/ca-certificates.crt > combined_ca.pem).

• ssl_client_cert (Union[str, Tuple[str, str], None]) – SSL client certificate file(s): - str: Single PEM file with cert+key - tuple: (cert_path, key_path)

• request_timeout (Tuple[float, float]) – Timeouts (connect_timeout, read_timeout) in seconds. Default (60, 300).

• jwt_token (Optional[str]) – JWT token set directly without server verification (no network call).

• cookie_file (Optional[str]) – Cookie file path for HTTP clients (set this to enable persistent cookie storage).

• auto_refresh_token (bool) – Enable automatic token refresh before expiration (supports App Mesh and Keycloak tokens).

add_app(app, subscribe_events=None)

Register a new application.

subscribe_events only takes effect on a persistent connection (TCP/WSS) and is ignored by the HTTP transport (no demuxer to deliver events to; a RuntimeWarning is emitted). When the daemon creates a subscription, the returned App carries subscription_id.

Return type:

App

add_label(label_name, label_value)

Add a new label.

Return type:

None

add_user(username, user_data)

Add a new user.

Return type:

None

authenticate(token, permission=None, audience=None, update_session=True)

Verify the provided JWT token with the server and optionally update the client session.

Parameters:

• token (str) – JWT token to verify.

• permission (Optional[str]) – Optional permission ID to check (e.g., ‘app-view’, ‘app-delete’).

• audience (Optional[str]) – Optional audience value to verify against the token.

• update_session (bool) – When True, update the current client session with the verified token and persist local token state on success. When False, only verify the provided token and leave local state unchanged.

Return type:

Tuple[bool, str]

Returns:

Tuple of (success, message) where message is the raw response text. (False, message) covers every non-OK status; never raises for a failed verification.

cancel_task(app_name)

Cancel a running task for an App Mesh application.

Parameters:

app_name (str) – Name of the target application (as registered in App Mesh).

Returns:

True if a task existed and was cancelled. False means no task was found (404) or the request failed for another reason (e.g. 401/403); non-404 failures are logged as warnings, never raised.

Return type:

bool

check_app_health(app_name)

Check the health status of an application.

Return type:

bool

close()

Close the client and release resources.

Return type:

None

delete_app(app_name)

Remove an application.

Returns:

True when the app was deleted, False when it did not exist (404). Any other non-OK status is logged and raised as an HTTP error.

Return type:

bool

delete_label(label_name)

Delete a label.

Return type:

None

delete_role(role_name)

Delete a user role.

Return type:

None

delete_user(username)

Delete a user.

Return type:

None

disable_app(app_name)

Disable an application.

Return type:

None

disable_totp(username=None)

Disable 2FA for the specified user.

Parameters:

username (Optional[str]) – Target user name; defaults to “self” (the current user).

Return type:

None

download_file(remote_file, local_file, preserve_permissions=True)

Download a remote file to the local filesystem.

When preserve_permissions is True, POSIX mode/owner/group metadata from App Mesh response headers is applied best-effort on non-Windows platforms.

Return type:

None

enable_app(app_name)

Enable an application.

Return type:

None

enable_totp(totp_code)

Set up 2FA for the current user.

Parameters:

totp_code (str) – TOTP code.

Return type:

None

property forward_to: str

Target host for request forwarding in a cluster.

Supports: - “hostname” or “IP” → uses current service port - “hostname:port” or “IP:port” → uses specified port

Returns:

Target host (e.g., “node” or “node:6060”), or empty string if unset.

Return type:

str

Notes

For JWT sharing across the cluster: - All nodes must use the same JWTSalt and Issuer for JWT settings - If port is omitted, current service port is used

Warning

Shared, per-client state read by every request; AppRun.wait() temporarily overrides it. Use separate client instances for concurrent multi-host access.

get_app(app_name)

Get information about a specific application.

Return type:

App

get_app_output(app_name, stdout_position=0, stdout_index=0, stdout_maxsize=10240, process_uuid='', timeout=0)

Get incremental stdout/stderr output for a running or completed application.

Parameters:

• app_name (str) – the application name

• stdout_position (int) – start read position, 0 means start from beginning.

• stdout_index (int) – index of history process stdout, 0 means get from current running process, the stdout number depends on ‘stdout_cache_size’ of the application.

• stdout_maxsize (int) – max buffer size to read.

• process_uuid (str) – used to get the specified process instance instead of the latest one.

• timeout (int) – long-poll wait time in seconds before returning when no new output is available.

Return type:

AppOutput

Returns:

AppOutput containing response status, payload text, the next read cursor (output_position), and exit_code when the process has already finished.

get_config()

Get the App Mesh configuration in JSON format.

Return type:

Dict[str, Any]

get_current_user()

Get information about the current user.

Return type:

dict

get_host_resources()

Get a report of host resources including CPU, memory, and disk.

Return type:

Dict[str, Any]

get_metrics()

Get Prometheus metrics.

Return type:

str

get_totp_secret()

Return the raw TOTP secret for the current user.

The server responds with a base64-encoded OTP provisioning URI; this helper parses that URI and returns only the secret field. Use get_totp_uri() for the full URI.

Return type:

str

get_totp_uri()

Return the TOTP provisioning URI (otpauth://...) for the current user, decoded from the server’s base64 mfa_uri payload, e.g. for rendering a QR code in an authenticator app.

Return type:

str

get_user_permissions()

Get information about the permissions of the current user.

Return type:

List[str]

list_apps()

Get information about all applications.

Return type:

List[App]

list_groups()

Get information about all user groups.

Return type:

List[str]

list_labels()

Get information about all labels.

Return type:

Dict[str, str]

list_permissions()

Get information about all available permissions.

Return type:

List[str]

list_roles()

Get information about all roles with permission definitions.

Return type:

Dict[str, Dict]

list_users()

Get information about all users.

Return type:

Dict[str, Any]

lock_user(username)

Lock a user.

Return type:

None

login(username, password, totp_code=None, token_expire='P1W', audience=None)

Login with username and password and attach the issued token to this client.

Parameters:

• username (str) – The name of the user.

• password (str) – The password of the user.

• totp_code (Optional[str]) – The TOTP code if enabled for the user.

• token_expire (Union[str, int]) – Token expiration duration. Supports ISO 8601 durations (e.g., ‘P1Y2M3DT4H5M6S’ ‘P1W’).

• audience (Optional[str]) – The audience of the JWT token, should be available by JWT service configuration (default is ‘appmesh-service’).

Return type:

Optional[str]

Returns:

TOTP challenge string if the server responds with HTTP 428 and no code was supplied, otherwise None. On success, the session token/cookie is updated and auto-refresh starts when enabled for this client.

logout()

Logout from the current session.

Returns:

True on success; False when no token was set locally or the server rejected the logoff (logged as a warning). Never raises for a failed logoff.

Return type:

bool

renew_token(token_expire='P1W')

Renew the current JWT token.

Parameters:

token_expire (Union[int, str]) – Token expiration duration (integer seconds or ISO 8601 string).

Return type:

None

run_app_async(app, max_time='P2D', lifecycle='P2DT12H')

Run an application asynchronously on a remote system without blocking the API.

Parameters:

• app (Union[App, str]) – An App instance or a shell command string. - If app is a string, it is treated as a shell command for the remote run, and an App instance is created as: App({“command”: “<command_string>”, “shell”: True}). - If app is an App object, providing only the name attribute (without a command) will run an existing application; otherwise, it is treated as a new application.

• max_time (Union[int, str]) – Maximum runtime for the remote process, after which the daemon kills it (sent as the wire query parameter timeout). Accepts integer seconds or ISO 8601 duration format (e.g., ‘P1Y2M3DT4H5M6S’, ‘P5W’). Defaults to P2D.

• lifecycle (Union[int, str]) – Total retention window for the temporary run app, after which the daemon purges it (including its cached output); must cover max_time plus the time needed to collect results (sent as the wire query parameter lifecycle). Accepts integer seconds or ISO 8601 duration format. Defaults to P2DT12H.

Return type:

AppRun

Returns:

AppRun handle that captures the current forward_to target so later polling can continue against the same cluster node.

run_app_sync(app, max_time='P2D', lifecycle='P2DT12H')

Synchronously run an application remotely, blocking until completion, and return the result.

If ‘app’ is a string, it is treated as a shell command and converted to an App instance. If ‘app’ is App object, the name attribute is used to run an existing application if specified.

Parameters:

• app (Union[App, str]) – An App instance or a shell command string. If a string, an App instance is created as: appmesh.App({“command”: “<command_string>”, “shell”: True})

• max_time (Union[int, str]) – Maximum runtime for the remote process, after which the daemon kills it (sent as the wire query parameter timeout). Accepts integer seconds or ISO 8601 duration format (e.g., ‘P1Y2M3DT4H5M6S’, ‘P5W’).

• lifecycle (Union[int, str]) – Total retention window for the temporary run app, after which the daemon purges it (sent as the wire query parameter lifecycle). Accepts integer seconds or ISO 8601 duration format.

Return type:

Tuple[Optional[int], str]

Returns:

(exit_code, stdout_text). exit_code is None when the server did not return an X-Exit-Code header.

run_task(app_name, data, timeout=300)

Client send an invocation message to a running App Mesh application and wait for result.

This method posts the provided data to the App Mesh service which will forward it to the specified running application instance.

Parameters:

• app_name (str) – Name of the target application (as registered in App Mesh).

• data (str) – Payload to deliver to the application. Typically a string.

• timeout (int) – Maximum time in seconds to wait for a response from the application. Defaults to 300 seconds.

Returns:

The HTTP response body returned by the remote application/service.

Return type:

str

set_config(config)

Update the configuration.

Return type:

Dict[str, Any]

set_log_level(level='DEBUG')

Update the log level.

Return type:

str

set_token(token)

Set a JWT token directly without server-side verification. Use when the token is already known to be valid. For server-side verification, use authenticate() instead.

Parameters:

token (str) – A valid JWT token string. The token is stored in the client’s cookie jar and persisted immediately when cookie_file is configured. The cookie expiry follows the token’s exp claim, falling back to 7 days when absent or undecodable.

Return type:

None

start_token_refresh()

Start background token auto-refresh.

Return type:

None

stop_token_refresh()

Stop background token auto-refresh.

Return type:

None

supports_events = False

unlock_user(username)

Unlock a user.

Return type:

None

update_password(old_password, new_password, username='self')

Change the password of a user.

Return type:

None

update_role(role_name, permission_set)

Update or add a role with defined permissions.

Return type:

None

upload_file(local_file, remote_file, preserve_permissions=True)

Upload a local file to the remote server.

When preserve_permissions is True, the client also sends local POSIX metadata in request headers so the server can recreate permissions/ownership when supported.

Return type:

None

validate_totp(username, challenge, code, token_expire='P1W')

Validate TOTP challenge and obtain a new JWT token.

Parameters:

• username (str) – Username to validate.

• challenge (str) – Challenge string from server.

• code (str) – TOTP code to validate.

• token_expire (Union[int, str]) –

  Token expiration duration, defaults to _DURATION_ONE_WEEK_ISO (1 week). Accepts either:

  • ISO 8601 duration string (e.g., ‘P1Y2M3DT4H5M6S’, ‘P1W’)

  • Numeric value (seconds) for simpler cases.

Return type:

None

wait_for_async_run(run, stdout_handler=None, timeout=0)

Wait for an asynchronous run to finish.

Parameters:

• run (AppRun) – asynchronous run handle returned by run_app_async().

• stdout_handler (Optional[Callable[[str, int], None]]) – optional callback (data, position) -> None invoked with each chunk of remote stdout (print_output_handler prints to console).

• timeout (int) – wait max timeout seconds and return if not finished, 0 means wait until finished

Return type:

Optional[int]

Returns:

Exit code if the process finished, or None when timeout elapsed first. On success, this method also makes a best-effort attempt to delete the temporary run app.

Raises:

AppMeshConnectionError – If polling the app output fails (non-OK response).

class appmesh.AppMeshClientTCP(tcp_address=('127.0.0.1', 6059), ssl_verify=None, ssl_client_cert=None, auto_refresh_token=False)

Bases: TransportClientMixin, AppMeshClient

Client SDK for interacting with the App Mesh service over TCP.

The AppMeshClientTCP class extends the functionality of AppMeshClient by offering a TCP-based communication layer for the App Mesh REST API. It overrides the file download and upload methods to support large file transfers with improved performance, leveraging TCP for lower latency and higher throughput compared to HTTP.

This client is suitable for applications requiring efficient data transfers and high-throughput operations within the App Mesh ecosystem, while maintaining compatibility with all other attributes and methods from AppMeshClient.

Inherits all attributes from `AppMeshClient`, including TLS secure connections and JWT-based authentication.

- download_file()

- upload_file()

- Inherits all other methods from `AppMeshClient`, providing a consistent interface for managing applications within App Mesh.

Example

>>> from appmesh import AppMeshClientTCP
>>> client = AppMeshClientTCP()
>>> client.login("your-name", "your-password")
>>> client.download_file("/tmp/os-release", "os-release")

Construct a TCP transport client that reuses the standard App Mesh client API.

Parameters:

• tcp_address (Tuple[str, int]) – Server address as (host, port) tuple, defaults to (“127.0.0.1”, 6059).

• ssl_verify (Union[bool, str, None]) – SSL certificate verification behavior. Can be None, True, False, or a path to CA bundle. - None (default): Auto — use the App Mesh CA bundle if installed, otherwise system CAs - True: Use system CA certificates (e.g., /etc/ssl/certs/ on Linux) - False: Disable verification (insecure, must be requested explicitly) - str: Path to custom CA bundle or directory (must exist)

• ssl_client_cert (Union[str, Tuple[str, str], None]) – SSL client certificate: - str: Path to single PEM with cert+key - tuple: (cert_path, key_path)

Note

TCP connections require an explicit full-chain CA specification for certificate validation, unlike HTTP, which can retrieve intermediate certificates automatically.

close()

Close the connection and release resources.

Return type:

None

download_file(remote_file, local_file, preserve_permissions=True)

Copy a remote file to local through the TCP file-socket side channel.

Parameters:

• remote_file (str) – Remote file path.

• local_file (str) – Local destination path.

• preserve_permissions (bool) – Apply remote file permissions/ownership locally on a best-effort basis.

Return type:

None

upload_file(local_file, remote_file, preserve_permissions=True)

Upload a local file to the remote server through the TCP file-socket side channel.

Parameters:

• local_file (str) – Local file path.

• remote_file (str) – Remote destination path.

• preserve_permissions (bool) – Send local file permissions/ownership metadata when available.

Return type:

None

class appmesh.AppMeshClientWSS(wss_address=('127.0.0.1', 6058), ssl_verify=None, ssl_client_cert=None, auto_refresh_token=False)

Bases: TransportClientMixin, AppMeshClient

Client SDK for interacting with the App Mesh service over WebSocket Secure (WSS).

The AppMeshClientWSS class extends the functionality of AppMeshClient by offering a WSS-based communication layer for the App Mesh REST API. It overrides the file download and upload methods to support large file transfers with improved performance, leveraging WebSocket for lower latency and higher throughput compared to HTTP.

This client is suitable for applications requiring efficient bidirectional data transfers and high-throughput operations within the App Mesh ecosystem, while maintaining compatibility with all other attributes and methods from AppMeshClient.

Inherits all attributes from `AppMeshClient`, including TLS secure connections and JWT-based authentication.

- download_file()

- upload_file()

- Inherits all other methods from `AppMeshClient`, providing a consistent interface for managing applications within App Mesh.

Example

>>> from appmesh import AppMeshClientWSS
>>> client = AppMeshClientWSS()
>>> client.login("your-name", "your-password")
>>> client.download_file("/tmp/os-release", "os-release")

Construct a WSS transport client that reuses the standard App Mesh client API.

Parameters:

• wss_address (Tuple[str, int]) – Server address as (host, port) tuple, defaults to (“127.0.0.1”, 6058).

• ssl_verify (Union[bool, str, None]) – SSL certificate verification behavior. Can be None, True, False, or a path to CA bundle. - None (default): Auto — use the App Mesh CA bundle if installed, otherwise system CAs - True: Use system CA certificates (e.g., /etc/ssl/certs/ on Linux) - False: Disable verification (insecure, must be requested explicitly) - str: Path to custom CA bundle or directory (must exist)

• ssl_client_cert (Union[str, Tuple[str, str], None]) – SSL client certificate: - str: Path to single PEM with cert+key - tuple: (cert_path, key_path)

Note

WSS connections require an explicit full-chain CA specification for certificate validation, unlike HTTP, which can retrieve intermediate certificates automatically.

close()

Close the connection and release resources.

Return type:

None

download_file(remote_file, local_file, preserve_permissions=True)

Copy a remote file to local through the WSS control channel plus HTTPS data channel.

Parameters:

• remote_file (str) – Remote file path.

• local_file (str) – Local destination path.

• preserve_permissions (bool) – Apply remote file permissions/ownership locally on a best-effort basis.

Return type:

None

upload_file(local_file, remote_file, preserve_permissions=True)

Upload a local file through the WSS control channel plus HTTPS data channel.

Parameters:

• local_file (str) – Local file path.

• remote_file (str) – Remote destination path.

• preserve_permissions (bool) – Send local file permissions/ownership metadata when available.

Return type:

None

exception appmesh.AppMeshConnectionError

Bases: AppMeshError

Connection or transport error.

exception appmesh.AppMeshError

Bases: Exception

Base exception for all App Mesh SDK errors.

exception appmesh.AppMeshProcessSupersededError

Bases: AppMeshError

The current process key was superseded by a newer process instance (HTTP 412).

exception appmesh.AppMeshRequestError

Bases: AppMeshError

HTTP request failed.

exception appmesh.AppMeshTimeoutError

Bases: AppMeshConnectionError

Receive timeout on an otherwise healthy connection (safe to retry/continue).

class appmesh.AppMeshWorker(base_url='https://127.0.0.1:6060', ssl_verify=None, ssl_client_cert=None, request_timeout=(60, 300), *, client=None, logger=None)

Bases: object

Worker SDK for an App Mesh application interacting with the local App Mesh REST service over HTTPS.

Despite running inside the managed application, this is a client-side task-loop helper: it polls the daemon for task payloads and returns results.

Build-in runtime environment variables required:

• APP_MESH_PROCESS_KEY

• APP_MESH_APPLICATION_NAME

- fetch_task()

fetch invocation payloads

- send_task_result()

return results to the invoking client

Example

context = appmesh.AppMeshWorker() payload = context.fetch_task() result = do_something_with(payload) context.send_task_result(result)

Initialize a worker-side helper for task fetch/return.

Parameters:

• base_url (str) – The server’s base URI. Defaults to “https://127.0.0.1:6060”.

• ssl_verify (Union[bool, str, None]) – SSL server verification mode (None = auto: App Mesh CA bundle if installed, else system CAs).

• ssl_client_cert (Union[str, Tuple[str, str], None]) – SSL client certificate file(s).

• request_timeout (Tuple[float, float]) – Timeouts (connect_timeout, read_timeout) in seconds.

• client (Optional[AppMeshClient]) – Pre-configured AppMeshClient instance (used by TCP/WSS subclasses so all transports share the same task API). Takes precedence: when provided, the connection parameters (base_url, ssl_verify, ssl_client_cert, request_timeout) are ignored.

• logger (Optional[Logger]) – Optional logger instance.

fetch_task(*, stop_event=None, max_retries=None)

Fetch task data in the currently running App Mesh application process.

Used by an App Mesh application process to obtain the payload from the App Mesh service that a client pushed to it. By default retries indefinitely until successful. If a request fails within 100ms, sleeps briefly before retrying; otherwise retries immediately.

Parameters:

• stop_event (Optional[Event]) – Optional cancellation event checked between attempts; when set, fetching stops and AppMeshError is raised.

• max_retries (Optional[int]) – Optional cap on retries after a failed attempt (N allows up to N + 1 attempts); when exhausted AppMeshError is raised. None (default) retries forever.

Return type:

Union[str, bytes]

Returns:

The payload bytes provided by the invoking client.

Raises:

• AppMeshProcessSupersededError – The daemon reported HTTP 412 — this process key was superseded by a newer process instance; the caller should stop serving.

• AppMeshError – Cancelled via stop_event or max_retries exhausted.

send_task_result(result)

Send the result of a server-side invocation back to the original client.

Used by App Mesh application process to post the result to App Mesh service after processing payload data so the invoking client can retrieve it.

Parameters:

result (Union[str, bytes]) – Result payload to be delivered back to the client exactly as provided.

Return type:

None

class appmesh.AppMeshWorkerTCP(ssl_verify=None, ssl_client_cert=None, tcp_address=('127.0.0.1', 6059), *, logger=None)

Bases: AppMeshWorker

Worker SDK for interacting with the local App Mesh service over TCP (TLS).

Example

>>> worker = AppMeshWorkerTCP(tcp_address=("127.0.0.1", 6059))
>>> payload = worker.fetch_task()

Construct an App Mesh worker TCP object to communicate securely with an App Mesh server over TLS.

Note

Positional order is (ssl_verify, ssl_client_cert, tcp_address) — differs from AppMeshClientTCP (address first); prefer keyword arguments.

Parameters:

• ssl_verify (Union[bool, str, None]) – SSL server verification mode; same semantics as AppMeshClientTCP.

• ssl_client_cert (Union[str, Tuple[str, str], None]) – SSL client certificate file(s); same semantics as AppMeshClientTCP.

• tcp_address (Tuple[str, int]) – Server address as (host, port) tuple, defaults to (“127.0.0.1”, 6059).

• logger (Optional[Logger]) – Optional logger instance.

class appmesh.AppMeshWorkerWSS(wss_address=('127.0.0.1', 6058), ssl_verify=None, ssl_client_cert=None, *, logger=None)

Bases: AppMeshWorker

Worker SDK for interacting with the local App Mesh service over WebSockets (WSS).

Example

>>> worker = AppMeshWorkerWSS(wss_address=("127.0.0.1", 6058))
>>> payload = worker.fetch_task()

Construct an App Mesh worker WSS object to communicate securely with an App Mesh server over TLS.

Note

Positional order is (wss_address, ssl_verify, ssl_client_cert) — differs from AppMeshWorkerTCP (ssl_verify first); prefer keyword arguments.

Parameters:

• wss_address (Tuple[str, int]) – Server address as (host, port) tuple, defaults to (“127.0.0.1”, 6058).

• ssl_verify (Union[bool, str, None]) – SSL server verification mode; same semantics as AppMeshClientWSS.

• ssl_client_cert (Union[str, Tuple[str, str], None]) – SSL client certificate file(s); same semantics as AppMeshClientWSS.

• logger (Optional[Logger]) – Optional logger instance.

class appmesh.AppOutput(status_code, output, output_position, exit_code)

Bases: object

Output information returned by the get_app_output() API.

Includes the application’s stdout, current read position, HTTP status code, and process exit code.

exit_code: int | None

Exit code of the application, or None if still running.

output: str

Captured stdout content of the application.

output_position: int | None

Current read position in stdout stream (wire header X-Output-Position), or None if not applicable.

status_code: HTTPStatus

HTTP status code from the get_app_output() API request.

class appmesh.AppRun(client, app_name, process_uuid)

Bases: object

Application run object for monitoring and retrieving results of a remote application run initiated by run_app_async().

app_name

Name of the application associated with this run.

process_uuid

Unique process UUID from run_app_async() (wire field process_uuid).

wait(stdout_handler=None, timeout=0)

Wait for the asynchronous run to complete with the saved forwarding target restored.

Parameters:

• stdout_handler (Optional[Callable[[str, int], None]]) – optional callback (data, position) -> None invoked with each chunk of stdout. Use print_output_handler for console output.

• timeout (int) – Maximum time to wait in seconds. 0 means wait indefinitely.

Return type:

Optional[int]

Returns:

Exit code if the process finished, or None when timeout elapsed first.

Raises:

• AppMeshConnectionError – On polling/transport failure while waiting.

• AppMeshAppRemovedError – If the app was removed before its exit was observed (TCP/WSS).

Warning

While waiting, the SHARED client’s forward_to is temporarily overridden (see _use_forward_host); concurrent requests are routed to that host.

class appmesh.SubscriptionResult(subscription_id='', app_name='', events=<factory>)

Bases: object

Server’s response to a subscribe request.

app_name: str = ''

events: list

subscription_id: str = ''

appmesh.print_output_handler(data, position)

Pre-built OutputHandler that prints data to stdout.

Return type:

None