appmesh package

Submodules

appmesh.app module

Application definition

class appmesh.app.App(data=None)[source]

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)[source]

Bases: object

Application error handling behavior, including exit and control behaviors.

class Action(value)[source]

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)[source]

Define behavior for specific exit codes.

Return type:

None

set_exit_behavior(action)[source]

Set default behavior for application exit.

Return type:

None

class DailyLimitation(data=None)[source]

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)[source]

Set the valid daily start and end times.

Return type:

None

class Permission(value)[source]

Bases: Enum

Application permission levels.

DENY = '1'
READ = '2'
WRITE = '3'
class ResourceLimitation(data=None)[source]

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

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

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

owner

owner name of app mesh user who created the app

permission

1, read:2, write: 3.

Type:

app user permission, value is 2 bit integer

Type:

[group & other], each bit can be deny

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)[source]

Set an environment variable, marking it secure if specified.

Return type:

None

set_permission(group_user, others_user)[source]

Define application permissions based on user roles.

Return type:

None

set_valid_time(start, end)[source]

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()[source]

Convert the application data into a JSON-compatible dictionary, removing empty items.

Return type:

Dict[str, Any]

user

process OS user name

version

app version

working_dir

working directory

appmesh.app_output module

Application output information.

class appmesh.app_output.AppOutput(status_code, output, out_position, exit_code)[source]

Bases: object

Output information returned by the 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.

out_position: int | None

Current read position in stdout stream, or None if not applicable.

output: str

Captured stdout content of the application.

status_code: HTTPStatus

HTTP status code from the app_output() API request.

appmesh.app_run module

Application run object for remote application execution.

class appmesh.app_run.AppRun(client, app_name, process_id)[source]

Bases: object

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

app_name

Name of the application associated with this run.

forward_to()[source]

Context manager to temporarily override the client’s forward_to setting.

Ensures operations during this run use the correct target server, then restores the original setting.

proc_uid

Unique process ID from run_async().

wait(stdout_handler=None, timeout=0)[source]

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 finishes successfully, or None on timeout/polling failure.

appmesh.app_run.print_output_handler(data, position)[source]

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=False, ssl_client_cert=None, request_timeout=(60, 300), jwt_token=None, cookie_file=None, auto_refresh_token=False)[source]

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_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()
# 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]) – SSL server verification mode: - True: Use system CAs. - False: Disable verification (insecure). - str: Path to custom CA or directory. 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)[source]

Register a new application.

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

Return type:

App

add_label(label_name, label_value)[source]

Add a new label.

Return type:

None

add_user(username, user_data)[source]

Add a new user.

Return type:

None

authenticate(token, permission=None, audience=None, update_session=True)[source]

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.

authentication(token, permission=None, audience=None, update_session=True)[source]

Deprecated: Use authenticate() instead.

Return type:

Tuple[bool, str]

cancel_task(app_name)[source]

Cancel a running task for an App Mesh application.

Parameters:

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

Returns:

Task exist and cancelled status.

Return type:

bool

check_app_health(app_name)[source]

Check the health status of an application.

Return type:

bool

close()[source]

Close the client and release resources.

Return type:

None

delete_app(app_name)[source]

Remove an application.

Return type:

bool

delete_label(label_name)[source]

Delete a label.

Return type:

None

delete_role(role_name)[source]

Delete a user role.

Return type:

None

delete_user(username)[source]

Delete a user.

Return type:

None

disable_app(app_name)[source]

Disable an application.

Return type:

None

disable_totp(user='self')[source]

Disable 2FA for the specified user.

Return type:

None

download_file(remote_file, local_file, preserve_permissions=True)[source]

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)[source]

Enable an application.

Return type:

None

enable_totp(totp_code)[source]

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

get_app(app_name)[source]

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)[source]

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 (out_position), and exit_code when the process has already finished.

get_config()[source]

Get the App Mesh configuration in JSON format.

Return type:

Dict[str, Any]

get_current_user()[source]

Get information about the current user.

Return type:

dict

get_host_resources()[source]

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

Return type:

Dict[str, Any]

get_metrics()[source]

Get Prometheus metrics.

Return type:

str

get_totp_secret()[source]

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 for QR-code or authenticator setup.

Return type:

str

get_user_permissions()[source]

Get information about the permissions of the current user.

Return type:

List[str]

list_apps()[source]

Get information about all applications.

Return type:

List[App]

list_groups()[source]

Get information about all user groups.

Return type:

List[str]

list_labels()[source]

Get information about all labels.

Return type:

Dict[str, str]

list_permissions()[source]

Get information about all available permissions.

Return type:

List[str]

list_roles()[source]

Get information about all roles with permission definitions.

Return type:

Dict[str, Dict]

list_users()[source]

Get information about all users.

Return type:

Dict[str, Any]

lock_user(username)[source]

Lock a user.

Return type:

None

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

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()[source]

Logout from the current session.

Return type:

bool

renew_token(token_expire='P1W')[source]

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')[source]

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. Accepts integer seconds or ISO 8601 duration format (e.g., ‘P1Y2M3DT4H5M6S’, ‘P5W’). Defaults to P2D.

  • lifecycle (Union[int, str]) – Maximum lifecycle time for the remote process. 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')[source]

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. Accepts integer seconds or ISO 8601 duration format (e.g., ‘P1Y2M3DT4H5M6S’, ‘P5W’).

  • lifecycle (Union[int, str]) – Maximum lifecycle time for the remote process. 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)[source]

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)[source]

Update the configuration.

Return type:

Dict[str, Any]

set_log_level(level='DEBUG')[source]

Update the log level.

Return type:

str

set_token(token)[source]

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.

Return type:

None

start_token_refresh()[source]

Start background token auto-refresh.

Return type:

None

stop_token_refresh()[source]

Stop background token auto-refresh.

Return type:

None

unlock_user(username)[source]

Unlock a user.

Return type:

None

update_password(old_password, new_password, username='self')[source]

Change the password of a user.

Return type:

None

update_role(role_name, permission_set)[source]

Update or add a role with defined permissions.

Return type:

None

upload_file(local_file, remote_file, preserve_permissions=True)[source]

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')[source]

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, print_stdout=True, timeout=0)[source]

Wait for an asynchronous run to finish.

Parameters:

  • run (AppRun) – asyncrized run result from run_async().

  • print_stdout (bool) – print remote stdout to local or not.

  • 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, otherwise None on timeout or polling failure. On success, this method also makes a best-effort attempt to delete the temporary run app.

appmesh.client_http_oauth module

appmesh.client_tcp module

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

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]) – SSL certificate verification behavior. Can be True, False, or a path to CA bundle. - True: Use system CA certificates (e.g., /etc/ssl/certs/ on Linux) - 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 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()[source]

Close the connection and release resources.

Return type:

None

download_file(remote_file, local_file, preserve_permissions=True)[source]

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)[source]

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=False, ssl_client_cert=None, auto_refresh_token=False)[source]

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]) – SSL certificate verification behavior. Can be True, False, or a path to CA bundle. - True: Use system CA certificates (e.g., /etc/ssl/certs/ on Linux) - 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 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()[source]

Close the connection and release resources.

Return type:

None

download_file(remote_file, local_file, preserve_permissions=True)[source]

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)[source]

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.AppMeshAuthError[source]

Bases: AppMeshError

Authentication or authorization error.

exception appmesh.exceptions.AppMeshConnectionError[source]

Bases: AppMeshError

Connection or transport error.

exception appmesh.exceptions.AppMeshError[source]

Bases: Exception

Base exception for all App Mesh SDK errors.

exception appmesh.exceptions.AppMeshRequestError[source]

Bases: AppMeshError

HTTP request failed.

exception appmesh.exceptions.AppMeshTimeoutError[source]

Bases: AppMeshConnectionError

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

appmesh.server_http module

HTTP server SDK implementation for App Mesh.

class appmesh.server_http.AppMeshServer(base_url='https://127.0.0.1:6060', ssl_verify=False, ssl_client_cert=None, request_timeout=(60, 300), *, client=None, logger_=None)[source]

Bases: object

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

Build-in runtime environment variables required:

  • APP_MESH_PROCESS_KEY

  • APP_MESH_APPLICATION_NAME

- task_fetch()

fetch invocation payloads

- task_return()

return results to the invoking client

Example

context = appmesh.AppMeshServer() payload = context.task_fetch() result = do_something_with(payload) context.task_return(result)

Initialize a server-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]) – SSL server verification mode.

  • 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).

  • logger – Optional logger instance.

task_fetch()[source]

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

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

Return type:

Union[str, bytes]

Returns:

The payload bytes provided by the invoking client.

task_return(result)[source]

Return 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.AppMeshServerTCP(ssl_verify=False, ssl_client_cert=None, tcp_address=('127.0.0.1', 6059), *, logger_=None)[source]

Bases: AppMeshServer

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

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

Parameters:

AppMeshClientTCP. (follows the same parameters as)

appmesh.server_wss module

class appmesh.server_wss.AppMeshServerWSS(wss_address=('127.0.0.1', 6058), ssl_verify=False, ssl_client_cert=None, *, logger_=None)[source]

Bases: AppMeshServer

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

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

Parameters:

AppMeshClientWSS. (follows the same parameters as)

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>)[source]

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>)[source]

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>)[source]

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()[source]

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>)[source]

Bases: object

TCP response message for HTTP-like communication.

body: bytes = b''
body_msg_type: str = ''
classmethod from_bytes(buf)[source]

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)[source]

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 = 104857600
TCP_MESSAGE_HEADER_LENGTH = 8
TCP_MESSAGE_MAGIC = 130484216
close()[source]

Close socket

Return type:

None

connect()[source]

Establish TLS connection to server.

Return type:

None

connected()[source]

Socket is connected or not

Return type:

bool

receive_message()[source]

Receive a message with prefixed header.

Return type:

Optional[bytearray]

Returns:

Message data, or None for EOF signal.

send_message(data)[source]

Send a message with prefixed header.

Parameters:

data (Union[bytes, bytearray, list]) – Message data to send, or empty list for EOF signal.

Return type:

None

appmesh.transport_mixin module

Shared transport logic for TCP and WSS clients.

class appmesh.transport_mixin.TransportClientMixin[source]

Bases: object

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

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)[source]

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)[source]

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.

unsubscribe(subscription_id)[source]

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)[source]

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]

appmesh.wss_transport module

WebSocket Secure (WSS) Transport layer handling WebSocket connections.

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

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()[source]

Close WebSocket connection.

Return type:

None

connect()[source]

Establish WSS connection to server.

Return type:

None

connected()[source]

Check if WebSocket is connected.

Return type:

bool

receive_message()[source]

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.

Return type:

Optional[bytearray]

send_message(data)[source]

Send a message over WebSocket.

Parameters:

data (Union[bytes, bytearray, list]) – Message data to send, or empty list for EOF signal.

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)[source]

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)[source]

Bases: object

Application error handling behavior, including exit and control behaviors.

class Action(value)[source]

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)[source]

Define behavior for specific exit codes.

Return type:

None

set_exit_behavior(action)[source]

Set default behavior for application exit.

Return type:

None

class DailyLimitation(data=None)[source]

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)[source]

Set the valid daily start and end times.

Return type:

None

class Permission(value)[source]

Bases: Enum

Application permission levels.

DENY = '1'
READ = '2'
WRITE = '3'
class ResourceLimitation(data=None)[source]

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

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

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

owner

owner name of app mesh user who created the app

permission

1, read:2, write: 3.

Type:

app user permission, value is 2 bit integer

Type:

[group & other], each bit can be deny

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)[source]

Set an environment variable, marking it secure if specified.

Return type:

None

set_permission(group_user, others_user)[source]

Define application permissions based on user roles.

Return type:

None

set_valid_time(start, end)[source]

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()[source]

Convert the application data into a JSON-compatible dictionary, removing empty items.

Return type:

Dict[str, Any]

user

process OS user name

version

app version

working_dir

working directory

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

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.AppMeshAuthError[source]

Bases: AppMeshError

Authentication or authorization error.

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

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_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()
# 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]) – SSL server verification mode: - True: Use system CAs. - False: Disable verification (insecure). - str: Path to custom CA or directory. 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)[source]

Register a new application.

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

Return type:

App

add_label(label_name, label_value)[source]

Add a new label.

Return type:

None

add_user(username, user_data)[source]

Add a new user.

Return type:

None

authenticate(token, permission=None, audience=None, update_session=True)[source]

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.

authentication(token, permission=None, audience=None, update_session=True)[source]

Deprecated: Use authenticate() instead.

Return type:

Tuple[bool, str]

cancel_task(app_name)[source]

Cancel a running task for an App Mesh application.

Parameters:

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

Returns:

Task exist and cancelled status.

Return type:

bool

check_app_health(app_name)[source]

Check the health status of an application.

Return type:

bool

close()[source]

Close the client and release resources.

Return type:

None

delete_app(app_name)[source]

Remove an application.

Return type:

bool

delete_label(label_name)[source]

Delete a label.

Return type:

None

delete_role(role_name)[source]

Delete a user role.

Return type:

None

delete_user(username)[source]

Delete a user.

Return type:

None

disable_app(app_name)[source]

Disable an application.

Return type:

None

disable_totp(user='self')[source]

Disable 2FA for the specified user.

Return type:

None

download_file(remote_file, local_file, preserve_permissions=True)[source]

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)[source]

Enable an application.

Return type:

None

enable_totp(totp_code)[source]

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

get_app(app_name)[source]

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)[source]

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 (out_position), and exit_code when the process has already finished.

get_config()[source]

Get the App Mesh configuration in JSON format.

Return type:

Dict[str, Any]

get_current_user()[source]

Get information about the current user.

Return type:

dict

get_host_resources()[source]

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

Return type:

Dict[str, Any]

get_metrics()[source]

Get Prometheus metrics.

Return type:

str

get_totp_secret()[source]

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 for QR-code or authenticator setup.

Return type:

str

get_user_permissions()[source]

Get information about the permissions of the current user.

Return type:

List[str]

list_apps()[source]

Get information about all applications.

Return type:

List[App]

list_groups()[source]

Get information about all user groups.

Return type:

List[str]

list_labels()[source]

Get information about all labels.

Return type:

Dict[str, str]

list_permissions()[source]

Get information about all available permissions.

Return type:

List[str]

list_roles()[source]

Get information about all roles with permission definitions.

Return type:

Dict[str, Dict]

list_users()[source]

Get information about all users.

Return type:

Dict[str, Any]

lock_user(username)[source]

Lock a user.

Return type:

None

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

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()[source]

Logout from the current session.

Return type:

bool

renew_token(token_expire='P1W')[source]

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')[source]

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. Accepts integer seconds or ISO 8601 duration format (e.g., ‘P1Y2M3DT4H5M6S’, ‘P5W’). Defaults to P2D.

  • lifecycle (Union[int, str]) – Maximum lifecycle time for the remote process. 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')[source]

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. Accepts integer seconds or ISO 8601 duration format (e.g., ‘P1Y2M3DT4H5M6S’, ‘P5W’).

  • lifecycle (Union[int, str]) – Maximum lifecycle time for the remote process. 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)[source]

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)[source]

Update the configuration.

Return type:

Dict[str, Any]

set_log_level(level='DEBUG')[source]

Update the log level.

Return type:

str

set_token(token)[source]

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.

Return type:

None

start_token_refresh()[source]

Start background token auto-refresh.

Return type:

None

stop_token_refresh()[source]

Stop background token auto-refresh.

Return type:

None

unlock_user(username)[source]

Unlock a user.

Return type:

None

update_password(old_password, new_password, username='self')[source]

Change the password of a user.

Return type:

None

update_role(role_name, permission_set)[source]

Update or add a role with defined permissions.

Return type:

None

upload_file(local_file, remote_file, preserve_permissions=True)[source]

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')[source]

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, print_stdout=True, timeout=0)[source]

Wait for an asynchronous run to finish.

Parameters:

  • run (AppRun) – asyncrized run result from run_async().

  • print_stdout (bool) – print remote stdout to local or not.

  • 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, otherwise None on timeout or polling failure. On success, this method also makes a best-effort attempt to delete the temporary run app.

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

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]) – SSL certificate verification behavior. Can be True, False, or a path to CA bundle. - True: Use system CA certificates (e.g., /etc/ssl/certs/ on Linux) - 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 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()[source]

Close the connection and release resources.

Return type:

None

download_file(remote_file, local_file, preserve_permissions=True)[source]

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)[source]

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=False, ssl_client_cert=None, auto_refresh_token=False)[source]

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]) – SSL certificate verification behavior. Can be True, False, or a path to CA bundle. - True: Use system CA certificates (e.g., /etc/ssl/certs/ on Linux) - 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 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()[source]

Close the connection and release resources.

Return type:

None

download_file(remote_file, local_file, preserve_permissions=True)[source]

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)[source]

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[source]

Bases: AppMeshError

Connection or transport error.

exception appmesh.AppMeshError[source]

Bases: Exception

Base exception for all App Mesh SDK errors.

exception appmesh.AppMeshRequestError[source]

Bases: AppMeshError

HTTP request failed.

class appmesh.AppMeshServer(base_url='https://127.0.0.1:6060', ssl_verify=False, ssl_client_cert=None, request_timeout=(60, 300), *, client=None, logger_=None)[source]

Bases: object

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

Build-in runtime environment variables required:

  • APP_MESH_PROCESS_KEY

  • APP_MESH_APPLICATION_NAME

- task_fetch()

fetch invocation payloads

- task_return()

return results to the invoking client

Example

context = appmesh.AppMeshServer() payload = context.task_fetch() result = do_something_with(payload) context.task_return(result)

Initialize a server-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]) – SSL server verification mode.

  • 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).

  • logger – Optional logger instance.

task_fetch()[source]

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

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

Return type:

Union[str, bytes]

Returns:

The payload bytes provided by the invoking client.

task_return(result)[source]

Return 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.AppMeshServerTCP(ssl_verify=False, ssl_client_cert=None, tcp_address=('127.0.0.1', 6059), *, logger_=None)[source]

Bases: AppMeshServer

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

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

Parameters:

AppMeshClientTCP. (follows the same parameters as)

class appmesh.AppMeshServerWSS(wss_address=('127.0.0.1', 6058), ssl_verify=False, ssl_client_cert=None, *, logger_=None)[source]

Bases: AppMeshServer

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

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

Parameters:

AppMeshClientWSS. (follows the same parameters as)

exception appmesh.AppMeshTimeoutError[source]

Bases: AppMeshConnectionError

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

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

Bases: object

Server’s response to a subscribe request.

app_name: str = ''
events: list
subscription_id: str = ''
appmesh.print_output_handler(data, position)[source]

Pre-built OutputHandler that prints data to stdout.

Return type:

None