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 the default exit action and per-exit-code actions.

class Action(value)

Bases: Enum

Actions for application exit behaviors.

KEEPALIVE = 'keepalive'

REMOVE = 'remove'

RESTART = 'restart'

STANDBY = 'standby'

exit

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

Type:

Default exit behavior, options

exit_code_actions

standby), higher priority than default exit behavior

Type:

Exit code specific behavior (exit code -> action, e.g., 0

Type:

restart, 1

set_exit_behavior(action)

Set default behavior for application exit.

Return type:

None

set_exit_code_action(exit_code, action)

Define the action for a specific exit code.

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

Total memory plus swap limit in MB; must be at least memory_mb.

command

full command line with arguments

container_id

docker container id

cpu

cpu usage

cron_schedule

cron expression for the start schedule (e.g., ‘* */5 * * * *’); presence means the app runs on a cron schedule

depends_on

list of application names this app depends on; each must be registered, enabled, running and healthy before this app starts (continuously scheduled apps only, Optional[List[str]])

description

app description string

docker_image

Docker image for containerized execution

enabled

True for enabled, False for disabled

Type:

app enable flag

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

interval

integer seconds or ISO 8601 duration (e.g., 30, ‘P1Y2M3DT4H5M6S’, ‘P5W’); use cron_schedule for a cron expression

Type:

start interval for short running app

property is_enabled: bool | None

True when enabled, False when disabled, None when unset.

Type:

Typed view of enabled

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_display_name

response-only owner label for display; never an authorization key

owner_principal_id

immutable principal ID that owns the app (assigned by Engine)

permission

2, write:3. Only the tens digit (others) is evaluated by the daemon for non-owner access; see set_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

return_code

last process exit code

secret_env

security environment variables protected by Engine at rest

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(others_user)

Define the application permission for non-owner principals.

The daemon reads only the others grant (tens digit); the ones digit mirrors it because the daemon never evaluates a group digit.

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_time

05’)

Type:

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

Type:

22

starts

number of times started

stdout_backup_count

maximum number of stdout log files to retain

stdout_file_count

count of rotated stdout log files currently retained

stop_grace_period

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

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 the owning principal

waiting_for

dependency names not yet satisfied at the last scheduler pass (Optional[List[str]], absent when nothing is held)

Type:

read-only

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), *, bearer_token=None, token_provider=None)

Bases: object

HTTP (REST) client for the App Mesh service.

Manages application lifecycle, monitoring, and configuration over HTTPS, with TLS transport and bearer-token authentication.

# Authentication context

- set_token_provider()

- set_bearer_token()

- clear_bearer_token()

- get_auth_config()

- get_current_principal()

# Application Management

- get_app()

- list_apps()

- get_app_output()

- check_app_health()

- add_app()

- delete_app()

- enable_app()

- disable_app()

# 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 & Configuration

- 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()

# Authorization Management

- list_principals()

- update_principal()

- delete_principal()

- get_principal_permissions()

- list_permissions()

- list_roles()

- update_role()

- delete_role()

# Client Lifecycle

- close()

Example

>>> from appmesh import AppMeshClient
>>> client = AppMeshClient(bearer_token="access-token")
>>> app = 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) - None (default): Do not send a client certificate. mTLS is opt-in.

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

• bearer_token (Optional[str]) – Access token to send as an RFC 6750 bearer token. Token acquisition, refresh, persistence, and revocation are handled by OAuthClient.

• token_provider (Optional[TokenProvider]) – Provider that supplies and refreshes access tokens. Mutually exclusive with bearer_token. Refresh credentials remain provider-private.

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

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 pending (208), the application was not found (404), or the request failed for another reason (e.g. 401/403); unexpected 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

clear_bearer_token()

Detach local authentication state without contacting the authentication service.

Return type:

None

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 AppMeshRequestError.

Return type:

bool

delete_label(label_name)

Delete a label.

Return type:

None

delete_principal(principal_id)

Delete an App Mesh authorization overlay; this never deletes an identity-provider user.

Return type:

None

delete_role(role_name)

Delete an App Mesh authorization role.

Return type:

None

disable_app(app_name)

Disable an application.

Return type:

None

download_file(remote_file, local_file=None, preserve_permissions=False)

Download a remote file to the local filesystem (local_file defaults to the remote basename).

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

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

Every target node must trust the same issuer and App Mesh resource audience. If port is omitted, the 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_file_count’ 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_auth_config()

Return the public OAuth/OIDC configuration advertised by App Mesh.

Return type:

Dict[str, Any]

get_config()

Get the App Mesh configuration in JSON format.

Return type:

Dict[str, Any]

get_current_principal()

Return the verified principal represented by the current bearer token.

Return type:

Dict[str, Any]

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_principal_permissions()

Return effective permissions for the current verified principal.

Return type:

List[str]

list_apps()

Get information about all applications.

Return type:

List[App]

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_principals()

List App Mesh authorization overlays keyed by immutable principal ID.

Return type:

Dict[str, Any]

list_roles()

Get information about all roles with permission definitions.

Return type:

Dict[str, Dict]

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.

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_bearer_token(token)

Attach a caller-owned, non-refreshing access token.

Return type:

None

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_provider(provider)

Attach a refresh-capable provider to this bearer-only Engine client.

Return type:

None

supports_events = False

property token_provider: TokenProvider | None

Return the provider currently attached to this Engine client.

update_principal(principal_id, principal_data)

Create or update an App Mesh authorization overlay for a principal.

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=None, preserve_permissions=False)

Upload a local file to the remote server (remote_file defaults to the local file’s basename).

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

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_tcp module

class appmesh.client_tcp.AppMeshClientTCP(tcp_address=('127.0.0.1', 6059), ssl_verify=None, ssl_client_cert=None, *, bearer_token=None, token_provider=None)

Bases: TransportClientMixin, AppMeshClient

App Mesh client over TCP.

Same API as AppMeshClient but overrides file up/download to use a TCP side channel for faster large-file transfers, and supports event subscription.

# Overridden for the TCP transport

- download_file()

- upload_file()

- add_app()             # subscribe atomically when the app starts

- wait_for_async_run()  # subscribe-based output streaming

- close()

# Event Subscription (TCP/WSS only)

- subscribe()

- unsubscribe()

Inherits all other methods from AppMeshClient.

Example

>>> from appmesh import AppMeshClientTCP
>>> client = AppMeshClientTCP(bearer_token="access-token")
>>> 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)

• bearer_token (Optional[str]) – Caller-owned access token.

• token_provider (Optional[TokenProvider]) – Provider that supplies and refreshes access tokens.

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=None, preserve_permissions=False)

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

Parameters:

• remote_file (str) – Remote file path.

• local_file (Optional[str]) – Local destination path; defaults to the remote file’s basename.

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

Return type:

None

upload_file(local_file, remote_file=None, preserve_permissions=False)

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

Parameters:

• local_file (str) – Local file path.

• remote_file (Optional[str]) – Remote destination path; defaults to the local file’s basename.

• 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, *, bearer_token=None, token_provider=None)

Bases: TransportClientMixin, AppMeshClient

App Mesh client over WebSocket Secure (WSS).

Same API as AppMeshClient but overrides file up/download to use a WSS side channel for faster large-file transfers, and supports event subscription.

# Overridden for the WSS transport

- download_file()

- upload_file()

- add_app()             # subscribe atomically when the app starts

- wait_for_async_run()  # subscribe-based output streaming

- close()

# Event Subscription (TCP/WSS only)

- subscribe()

- unsubscribe()

Inherits all other methods from AppMeshClient.

Example

>>> from appmesh import AppMeshClientWSS
>>> client = AppMeshClientWSS(bearer_token="access-token")
>>> 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)

• bearer_token (Optional[str]) – Caller-owned access token.

• token_provider (Optional[TokenProvider]) – Provider that supplies and refreshes access tokens.

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=None, preserve_permissions=False)

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

Parameters:

• remote_file (str) – Remote file path.

• local_file (Optional[str]) – Local destination path; defaults to the remote file’s basename.

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

Return type:

None

upload_file(local_file, remote_file=None, preserve_permissions=False)

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

Parameters:

• local_file (str) – Local file path.

• remote_file (Optional[str]) – Remote destination path; defaults to the local file’s basename.

• 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(message, status_code=None)

Bases: AppMeshError

Authentication or authorization error.

status_code distinguishes 401 (the credential itself is rejected) from 403 (the credential is fine but lacks the permission); callers must not treat them alike. None when the error did not come from an HTTP response.

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

exception appmesh.exceptions.AppMeshWorkerRejectedError(message, status_code=400)

Bases: AppMeshError

The daemon permanently rejected a worker task request (HTTP 400).

appmesh.oauth module

Standards-based OAuth 2.0 and OpenID Connect client support.

Identity authentication happens directly at the authentication service. App Mesh receives only the access token as an RFC 6750 bearer token. It never receives passwords, MFA challenges, or directory-management requests.

class appmesh.oauth.OAuthClient(appmesh_client, issuer, access_url=None, client_id=None, audience=None, scopes=None, ssl_verify=True, timeout=None, allow_plain_http=False)

Bases: TokenProvider

Acquire tokens and attach their access token to an AppMeshClient.

The implementation supports OAuth authorization code with PKCE (RFC 7636), device authorization (RFC 8628), refresh tokens, and token revocation (RFC 7009).

authorization_request(redirect_uri, scopes=None, state=None, nonce=None)

Create a browser authorization request using PKCE S256.

The request is retained in memory so complete_authorization_callback() can validate the callback state before exchanging the code. This helper consumes access tokens for Engine API calls; it does not treat ID-token claims as an authenticated identity. Callers that explicitly supply nonce must also pass a standards-compliant ID-token validator when completing the callback.

Return type:

Dict[str, str]

property can_refresh: bool

Return whether the provider can replace the current access token.

clear()

Forget local tokens without making a network request.

Return type:

None

close()

Close local HTTP resources. Call revoke() to revoke tokens.

Return type:

None

complete_authorization_callback(callback_url, id_token_validator=None)

Validate a browser callback and install its access token.

id_token_validator is required only when the authorization request explicitly included a nonce. It must cryptographically validate the ID token according to OIDC (signature, issuer, audience, expiry) and compare its nonce with the supplied expected value. The SDK itself never consumes ID-token identity claims.

Return type:

Dict[str, Any]

device_authorization(scopes=None)

Start RFC 8628 device authorization and return the user-facing prompt data.

Return type:

Dict[str, Any]

exchange_authorization_code(code, redirect_uri, code_verifier)

Low-level code exchange after the caller has independently validated state.

Prefer complete_authorization_callback() for browser callbacks. Do not use this method to consume a nonce-bearing OIDC response without independently validating the ID token.

Return type:

Dict[str, Any]

classmethod from_appmesh(appmesh_client, access_url=None, client_id=None, scopes=None, ssl_verify=True)

Construct from App Mesh’s public /appmesh/auth/config response.

appmesh_client.base_url selects the Engine host. access_url selects how this process reaches the authentication service. The canonical issuer comes from the Engine and must match discovery and token claims.

Return type:

OAuthClient

get_access_token()

Return an access token and refresh it shortly before expiry.

Return type:

Optional[str]

refresh()

Refresh and atomically replace the App Mesh bearer access token.

Return type:

Dict[str, Any]

refresh_access_token(rejected_token=None)

Refresh a token rejected by Engine, coalescing concurrent refreshes.

Return type:

Optional[str]

revoke()

Revoke held refresh and access tokens. Then clear local authentication state.

Each token is revoked independently: a failure on one (network error or non-2xx) is recorded in the return value but does not prevent the other from being revoked.

Return type:

bool

property tokens: Dict[str, Any]

Return a copy of the in-memory token response.

wait_for_device_authorization(device, on_prompt=None)

Poll until a device request is approved, denied, or expires.

Return type:

Dict[str, Any]

exception appmesh.oauth.OAuthError(message, status_code=None)

Bases: AppMeshAuthError

The authentication service rejected a request or returned an invalid response.

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.token_provider module

Access-token provider contracts for App Mesh SDK clients.

Providers own token acquisition and refresh. Engine clients consume only the resulting access token and never receive passwords, refresh tokens, or OAuth authorization responses.

class appmesh.token_provider.StaticAccessTokenProvider(token)

Bases: TokenProvider

In-memory provider for a caller-supplied access token.

clear()

Forget provider-owned in-memory token state, if any.

Return type:

None

get_access_token()

Return a current access token, refreshing before expiry when possible.

Return type:

Optional[str]

class appmesh.token_provider.TokenProvider

Bases: object

Provide a usable access token to an App Mesh Engine client.

get_access_token may refresh proactively when the current token is near expiry. refresh_access_token is called at most once after the Engine rejects a provider-managed token with HTTP 401. Implementations should use rejected_token to avoid duplicate refreshes when requests race.

Providers keep refresh credentials private; only an access token crosses this boundary into the Engine client.

property can_refresh: bool

Whether this provider can replace a rejected or expiring token.

clear()

Forget provider-owned in-memory token state, if any.

Return type:

None

abstractmethod get_access_token()

Return a current access token, refreshing before expiry when possible.

Return type:

Optional[str]

refresh_access_token(rejected_token=None)

Replace a rejected token and return the new access token.

Return type:

Optional[str]

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 bearer-token injection 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)

• _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.worker_http module

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

class appmesh.worker_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 transient failures indefinitely; HTTP 400/412 stop the loop. 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.

• AppMeshWorkerRejectedError – The daemon reported HTTP 400 — the worker request is incompatible or invalid and retrying cannot recover it.

• 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.worker_tcp module

class appmesh.worker_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.worker_wss module

class appmesh.worker_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.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.

set_bearer_token(token)

Set the bearer used on the next WebSocket upgrade.

An established session keeps its pinned identity; a refreshed token is used automatically after reconnect.

Return type:

None

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 the default exit action and per-exit-code actions.

class Action(value)

Bases: Enum

Actions for application exit behaviors.

KEEPALIVE = 'keepalive'

REMOVE = 'remove'

RESTART = 'restart'

STANDBY = 'standby'

exit

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

Type:

Default exit behavior, options

exit_code_actions

standby), higher priority than default exit behavior

Type:

Exit code specific behavior (exit code -> action, e.g., 0

Type:

restart, 1

set_exit_behavior(action)

Set default behavior for application exit.

Return type:

None

set_exit_code_action(exit_code, action)

Define the action for a specific exit code.

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

Total memory plus swap limit in MB; must be at least memory_mb.

command

full command line with arguments

container_id

docker container id

cpu

cpu usage

cron_schedule

cron expression for the start schedule (e.g., ‘* */5 * * * *’); presence means the app runs on a cron schedule

depends_on

list of application names this app depends on; each must be registered, enabled, running and healthy before this app starts (continuously scheduled apps only, Optional[List[str]])

description

app description string

docker_image

Docker image for containerized execution

enabled

True for enabled, False for disabled

Type:

app enable flag

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

interval

integer seconds or ISO 8601 duration (e.g., 30, ‘P1Y2M3DT4H5M6S’, ‘P5W’); use cron_schedule for a cron expression

Type:

start interval for short running app

property is_enabled: bool | None

True when enabled, False when disabled, None when unset.

Type:

Typed view of enabled

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_display_name

response-only owner label for display; never an authorization key

owner_principal_id

immutable principal ID that owns the app (assigned by Engine)

permission

2, write:3. Only the tens digit (others) is evaluated by the daemon for non-owner access; see set_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

return_code

last process exit code

secret_env

security environment variables protected by Engine at rest

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(others_user)

Define the application permission for non-owner principals.

The daemon reads only the others grant (tens digit); the ones digit mirrors it because the daemon never evaluates a group digit.

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_time

05’)

Type:

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

Type:

22

starts

number of times started

stdout_backup_count

maximum number of stdout log files to retain

stdout_file_count

count of rotated stdout log files currently retained

stop_grace_period

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

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 the owning principal

waiting_for

dependency names not yet satisfied at the last scheduler pass (Optional[List[str]], absent when nothing is held)

Type:

read-only

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(message, status_code=None)

Bases: AppMeshError

Authentication or authorization error.

status_code distinguishes 401 (the credential itself is rejected) from 403 (the credential is fine but lacks the permission); callers must not treat them alike. None when the error did not come from an HTTP response.

class appmesh.AppMeshClient(base_url='https://127.0.0.1:6060', ssl_verify=None, ssl_client_cert=None, request_timeout=(60, 300), *, bearer_token=None, token_provider=None)

Bases: object

HTTP (REST) client for the App Mesh service.

Manages application lifecycle, monitoring, and configuration over HTTPS, with TLS transport and bearer-token authentication.

# Authentication context

- set_token_provider()

- set_bearer_token()

- clear_bearer_token()

- get_auth_config()

- get_current_principal()

# Application Management

- get_app()

- list_apps()

- get_app_output()

- check_app_health()

- add_app()

- delete_app()

- enable_app()

- disable_app()

# 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 & Configuration

- 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()

# Authorization Management

- list_principals()

- update_principal()

- delete_principal()

- get_principal_permissions()

- list_permissions()

- list_roles()

- update_role()

- delete_role()

# Client Lifecycle

- close()

Example

>>> from appmesh import AppMeshClient
>>> client = AppMeshClient(bearer_token="access-token")
>>> app = 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) - None (default): Do not send a client certificate. mTLS is opt-in.

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

• bearer_token (Optional[str]) – Access token to send as an RFC 6750 bearer token. Token acquisition, refresh, persistence, and revocation are handled by OAuthClient.

• token_provider (Optional[TokenProvider]) – Provider that supplies and refreshes access tokens. Mutually exclusive with bearer_token. Refresh credentials remain provider-private.

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

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 pending (208), the application was not found (404), or the request failed for another reason (e.g. 401/403); unexpected 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

clear_bearer_token()

Detach local authentication state without contacting the authentication service.

Return type:

None

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 AppMeshRequestError.

Return type:

bool

delete_label(label_name)

Delete a label.

Return type:

None

delete_principal(principal_id)

Delete an App Mesh authorization overlay; this never deletes an identity-provider user.

Return type:

None

delete_role(role_name)

Delete an App Mesh authorization role.

Return type:

None

disable_app(app_name)

Disable an application.

Return type:

None

download_file(remote_file, local_file=None, preserve_permissions=False)

Download a remote file to the local filesystem (local_file defaults to the remote basename).

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

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

Every target node must trust the same issuer and App Mesh resource audience. If port is omitted, the 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_file_count’ 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_auth_config()

Return the public OAuth/OIDC configuration advertised by App Mesh.

Return type:

Dict[str, Any]

get_config()

Get the App Mesh configuration in JSON format.

Return type:

Dict[str, Any]

get_current_principal()

Return the verified principal represented by the current bearer token.

Return type:

Dict[str, Any]

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_principal_permissions()

Return effective permissions for the current verified principal.

Return type:

List[str]

list_apps()

Get information about all applications.

Return type:

List[App]

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_principals()

List App Mesh authorization overlays keyed by immutable principal ID.

Return type:

Dict[str, Any]

list_roles()

Get information about all roles with permission definitions.

Return type:

Dict[str, Dict]

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.

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_bearer_token(token)

Attach a caller-owned, non-refreshing access token.

Return type:

None

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_provider(provider)

Attach a refresh-capable provider to this bearer-only Engine client.

Return type:

None

supports_events = False

property token_provider: TokenProvider | None

Return the provider currently attached to this Engine client.

update_principal(principal_id, principal_data)

Create or update an App Mesh authorization overlay for a principal.

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=None, preserve_permissions=False)

Upload a local file to the remote server (remote_file defaults to the local file’s basename).

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

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, *, bearer_token=None, token_provider=None)

Bases: TransportClientMixin, AppMeshClient

App Mesh client over TCP.

Same API as AppMeshClient but overrides file up/download to use a TCP side channel for faster large-file transfers, and supports event subscription.

# Overridden for the TCP transport

- download_file()

- upload_file()

- add_app()             # subscribe atomically when the app starts

- wait_for_async_run()  # subscribe-based output streaming

- close()

# Event Subscription (TCP/WSS only)

- subscribe()

- unsubscribe()

Inherits all other methods from AppMeshClient.

Example

>>> from appmesh import AppMeshClientTCP
>>> client = AppMeshClientTCP(bearer_token="access-token")
>>> 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)

• bearer_token (Optional[str]) – Caller-owned access token.

• token_provider (Optional[TokenProvider]) – Provider that supplies and refreshes access tokens.

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=None, preserve_permissions=False)

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

Parameters:

• remote_file (str) – Remote file path.

• local_file (Optional[str]) – Local destination path; defaults to the remote file’s basename.

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

Return type:

None

upload_file(local_file, remote_file=None, preserve_permissions=False)

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

Parameters:

• local_file (str) – Local file path.

• remote_file (Optional[str]) – Remote destination path; defaults to the local file’s basename.

• 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, *, bearer_token=None, token_provider=None)

Bases: TransportClientMixin, AppMeshClient

App Mesh client over WebSocket Secure (WSS).

Same API as AppMeshClient but overrides file up/download to use a WSS side channel for faster large-file transfers, and supports event subscription.

# Overridden for the WSS transport

- download_file()

- upload_file()

- add_app()             # subscribe atomically when the app starts

- wait_for_async_run()  # subscribe-based output streaming

- close()

# Event Subscription (TCP/WSS only)

- subscribe()

- unsubscribe()

Inherits all other methods from AppMeshClient.

Example

>>> from appmesh import AppMeshClientWSS
>>> client = AppMeshClientWSS(bearer_token="access-token")
>>> 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)

• bearer_token (Optional[str]) – Caller-owned access token.

• token_provider (Optional[TokenProvider]) – Provider that supplies and refreshes access tokens.

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=None, preserve_permissions=False)

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

Parameters:

• remote_file (str) – Remote file path.

• local_file (Optional[str]) – Local destination path; defaults to the remote file’s basename.

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

Return type:

None

upload_file(local_file, remote_file=None, preserve_permissions=False)

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

Parameters:

• local_file (str) – Local file path.

• remote_file (Optional[str]) – Remote destination path; defaults to the local file’s basename.

• 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 transient failures indefinitely; HTTP 400/412 stop the loop. 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.

• AppMeshWorkerRejectedError – The daemon reported HTTP 400 — the worker request is incompatible or invalid and retrying cannot recover it.

• 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

exception appmesh.AppMeshWorkerRejectedError(message, status_code=400)

Bases: AppMeshError

The daemon permanently rejected a worker task request (HTTP 400).

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.OAuthClient(appmesh_client, issuer, access_url=None, client_id=None, audience=None, scopes=None, ssl_verify=True, timeout=None, allow_plain_http=False)

Bases: TokenProvider

Acquire tokens and attach their access token to an AppMeshClient.

The implementation supports OAuth authorization code with PKCE (RFC 7636), device authorization (RFC 8628), refresh tokens, and token revocation (RFC 7009).

authorization_request(redirect_uri, scopes=None, state=None, nonce=None)

Create a browser authorization request using PKCE S256.

The request is retained in memory so complete_authorization_callback() can validate the callback state before exchanging the code. This helper consumes access tokens for Engine API calls; it does not treat ID-token claims as an authenticated identity. Callers that explicitly supply nonce must also pass a standards-compliant ID-token validator when completing the callback.

Return type:

Dict[str, str]

property can_refresh: bool

Return whether the provider can replace the current access token.

clear()

Forget local tokens without making a network request.

Return type:

None

close()

Close local HTTP resources. Call revoke() to revoke tokens.

Return type:

None

complete_authorization_callback(callback_url, id_token_validator=None)

Validate a browser callback and install its access token.

id_token_validator is required only when the authorization request explicitly included a nonce. It must cryptographically validate the ID token according to OIDC (signature, issuer, audience, expiry) and compare its nonce with the supplied expected value. The SDK itself never consumes ID-token identity claims.

Return type:

Dict[str, Any]

device_authorization(scopes=None)

Start RFC 8628 device authorization and return the user-facing prompt data.

Return type:

Dict[str, Any]

exchange_authorization_code(code, redirect_uri, code_verifier)

Low-level code exchange after the caller has independently validated state.

Prefer complete_authorization_callback() for browser callbacks. Do not use this method to consume a nonce-bearing OIDC response without independently validating the ID token.

Return type:

Dict[str, Any]

classmethod from_appmesh(appmesh_client, access_url=None, client_id=None, scopes=None, ssl_verify=True)

Construct from App Mesh’s public /appmesh/auth/config response.

appmesh_client.base_url selects the Engine host. access_url selects how this process reaches the authentication service. The canonical issuer comes from the Engine and must match discovery and token claims.

Return type:

OAuthClient

get_access_token()

Return an access token and refresh it shortly before expiry.

Return type:

Optional[str]

refresh()

Refresh and atomically replace the App Mesh bearer access token.

Return type:

Dict[str, Any]

refresh_access_token(rejected_token=None)

Refresh a token rejected by Engine, coalescing concurrent refreshes.

Return type:

Optional[str]

revoke()

Revoke held refresh and access tokens. Then clear local authentication state.

Each token is revoked independently: a failure on one (network error or non-2xx) is recorded in the return value but does not prevent the other from being revoked.

Return type:

bool

property tokens: Dict[str, Any]

Return a copy of the in-memory token response.

wait_for_device_authorization(device, on_prompt=None)

Poll until a device request is approved, denied, or expires.

Return type:

Dict[str, Any]

exception appmesh.OAuthError(message, status_code=None)

Bases: AppMeshAuthError

The authentication service rejected a request or returned an invalid response.

class appmesh.StaticAccessTokenProvider(token)

Bases: TokenProvider

In-memory provider for a caller-supplied access token.

clear()

Forget provider-owned in-memory token state, if any.

Return type:

None

get_access_token()

Return a current access token, refreshing before expiry when possible.

Return type:

Optional[str]

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 = ''

class appmesh.TokenProvider

Bases: object

Provide a usable access token to an App Mesh Engine client.

get_access_token may refresh proactively when the current token is near expiry. refresh_access_token is called at most once after the Engine rejects a provider-managed token with HTTP 401. Implementations should use rejected_token to avoid duplicate refreshes when requests race.

Providers keep refresh credentials private; only an access token crosses this boundary into the Engine client.

property can_refresh: bool

Whether this provider can replace a rejected or expiring token.

clear()

Forget provider-owned in-memory token state, if any.

Return type:

None

abstractmethod get_access_token()

Return a current access token, refreshing before expiry when possible.

Return type:

Optional[str]

refresh_access_token(rejected_token=None)

Replace a rejected token and return the new access token.

Return type:

Optional[str]

appmesh.print_output_handler(data, position)

Pre-built OutputHandler that prints data to stdout.

Return type:

None