Usage#

The purpose of this page is to provide a usage guide for boto3-refresh-session. The following sections cover installation, initialization, refresh methods and behavior, MFA support, client and resource caching, IoT Core X.509 support, and some miscellaneous features. Use these instructions as a guide to get started with boto3-refresh-session, but refer to the API docs for comprehensive technical documentation of all features and parameters. If you have any questions or run into issues, please open an issue on GitHub or reach out to the maintainer directly.

Installation#

boto3-refresh-session is available on PyPI.

You can install extras for IoT support and development dependencies as needed.

# with pip
pip install boto3-refresh-session

# with pip + iot as an extra
pip install boto3-refresh-session[iot]

# with pip + dev dependencies
pip install boto3-refresh-session[dev]

# with pip + all extras
pip install boto3-refresh-session[dev,iot]

# with pip in editable mode (for contributors)
uv pip install -e ".[dev,iot]"

Attention

Versions 7.2.4 through 7.2.14 were deleted from PyPI and GitHub tags due to a critical packaging issue. Users with versions in that range should upgrade to v7.2.15+. Refer to the changelog for more details.

Authorization#

To use boto3-refresh-session, you must have AWS credentials configured locally. Configuring credentials for boto3-refresh-session is the same as for boto3. Refer to the boto3 documentation for more details.

Initialization#

Everything in boto3 is ultimately built on the boto3.session.Session object — including Client and Resource objects. boto3-refresh-session extends this interface while adding automatic temporary credential refresh and some additional features.

Creating a session with boto3_refresh_session.session.RefreshableSession is straightforward. Below is a basic example of creating a Client via STS role assumption.

Tip

You can leave RoleSessionName in assume_role_kwargs blank if you like. boto3-refresh-session will default to “boto3-refresh-session” for you.

Tip

You can also provide assume_role_kwargs as a simple dictionary instead of an boto3_refresh_session.utils.config.AssumeRoleConfig object if you prefer. Both approaches are functionally equivalent!

from boto3_refresh_session import AssumeRoleConfig, RefreshableSession

session = RefreshableSession(
    assume_role_kwargs=AssumeRoleConfig(
        RoleArn="<your-role-arn>",
        RoleSessionName="<your-role-session-name>",
    )
)

s3 = session.client('s3')

boto3_refresh_session.session.RefreshableSession can be initialized exactly like a normal boto3.session.Session object. It accepts every parameter which boto3.session.Session does, in addition to parameters for STS role assumption, refresh behavior, timeout limits, MFA, and cache behavior. To illustrate, below is an example of creating a Client via STS role assumption with custom retry configuration and region specification.

Tip

boto3_refresh_session.session.RefreshableSession also accepts a sts_client_kwargs parameter, which allows you to pass custom parameters to the internal STS client used for role assumption. Like assume_role_kwargs, sts_client_kwargs can be provided as either a dictionary or a boto3_refresh_session.utils.config.STSClientConfig object. Check the STS.Client documentation for available parameters.

from botocore.config import Config
from boto3_refresh_session import AssumeRoleConfig, RefreshableSession

session = RefreshableSession(
    assume_role_kwargs=AssumeRoleConfig(RoleArn="<your-role-arn>"),
    region_name="us-east-1",
)

s3 = session.client('s3', config=Config(retries={"max_attempts": 10}))

Tip

Attributes in assume_role_kwargs and sts_client_kwargs can be accessed using dot-notation or dictionary-style access. Same goes for boto3_refresh_session.utils.config.AssumeRoleConfig and boto3_refresh_session.utils.config.STSClientConfig objects.

Refresh Methods#

boto3_refresh_session.session.RefreshableSession uses STS by default. To be more precise, if method is not specified, it defaults to "sts". If you want to use boto3_refresh_session.session.RefreshableSession with IoT or provide a custom credential provider instead then you must modulate the method parameter accordingly to "iot" or "custom". Those methods require additional parameters; refer to the API docs for more details.

Refresh Behavior#

There are two ways to trigger automatic credential refresh in boto3-refresh-session:

Deferred (default) — Refresh occurs only when credentials are required
Eager — Credentials are refreshed as soon as they expire

Set defer_refresh to False to enable eager refresh:

from boto3_refresh_session import AssumeRoleConfig, RefreshableSession

session = RefreshableSession(
    assume_role_kwargs=AssumeRoleConfig(RoleArn="<your-role-arn>"),
    defer_refresh=False,
)

Eager Refresh Behavior#

With eager refresh enabled (i.e. defer_refresh=False), credentials are refreshed according to two settings: advisory and mandatory timeouts. The so-called “advisory” and “mandatory” timeouts are concepts created by botocore to manage credential expiration. By default in botocore, the advisory timeout is set to 15 minutes before expiration and the mandatory timeout is set to 10 minutes before expiration. botocore will attempt to refresh credentials when the advisory timeout is reached, but will force a refresh when the mandatory timeout is reached. This behavior is inherited by boto3-refresh-session when eager refresh is enabled, and the advisory_timeout and mandatory_timeout parameters can be adjusted as needed.

Tip

The vast majority of use cases will not require modification of advisory or mandatory timeouts, but they are available for edge cases where precise control over refresh timing is necessary.

MFA#

If your role assumption requires MFA, you must provide a token provider via mfa_token_provider. It accepts a callable (recommended), a list[str] of command arguments (also recommended), or a command str (least recommended).

Callable: Called during each refresh. Arguments are passed via mfa_token_provider_kwargs. You are responsible for writing the callable.
list[str] | str: Treated as a CLI command and executed via subprocess.run. Keyword arguments passed via mfa_token_provider_kwargs are forwarded to subprocess.run.

Below are examples for a callable and direct CLI invocation.

Tip

Be sure to provide SerialNumber in assume_role_kwargs when using MFA. If you provide mfa_token_provider, any TokenCode you set in assume_role_kwargs will be ignored and overwritten on each refresh. If you run into latency issues, pass a botocore.config.Config with retries to the internal STS client via sts_client_kwargs.

Note

Token output is parsed to the last 6-digit numeric token found in stdout when mfa_token_provider is a CLI command. The token must be a standalone 6-digit string (adjacent characters cause an error).

from boto3_refresh_session import AssumeRoleConfig, RefreshableSession
import subprocess
from typing import Sequence

# example MFA token provider as a callable
def mfa_token_provider(cmd: Sequence[str], timeout: float) -> str:
    p = subprocess.run(
        list(cmd),
        check=False,
        capture_output=True,
        text=True,
        timeout=timeout,
    )
    return (p.stdout or "").strip()


session = RefreshableSession(
    assume_role_kwargs=AssumeRoleConfig(
        RoleArn="<your-role-arn>",
        RoleSessionName="<your-role-session-name>",
        SerialNumber="arn:aws:iam::<account-id>:mfa/<device-name>",
    ),
    mfa_token_provider=mfa_token_provider,
    mfa_token_provider_kwargs={
        # "ykman oath code --single AWS-prod" is equivalent to the cmd below
        "cmd": ["ykman", "oath", "code", "--single", "AWS-prod"],
        "timeout": 3.0,
    },
)

# or, pass a CLI command directly (list[str] or str)
session = RefreshableSession(
    assume_role_kwargs=AssumeRoleConfig(
        RoleArn="<your-role-arn>",
        RoleSessionName="<your-role-session-name>",
        SerialNumber="arn:aws:iam::<account-id>:mfa/<device-name>",
    ),
    mfa_token_provider=["ykman", "oath", "code", "--single", "AWS-prod"],
    mfa_token_provider_kwargs={"timeout": 20},
)

Warning

It is highly recommended to use mfa_token_provider instead of passing TokenCode directly. Without mfa_token_provider, you must refresh TokenCode yourself, which will be cumbersome.

Warning

Erroneous TokenCode values (i.e. non 6-digit numeric strings) will raise errors during construction of assume_role_kwargs. Ensure your token provider returns valid tokens.

Warning

For security, the stdout, stderr, shell, executable, and preexec_fn subprocess.run parameters are blocked and will raise an error if provided in mfa_token_provider_kwargs.

Client and Resource Caching#

boto3-refresh-session uses boto3-client-cache for both client and resource caching. Calls to session.client(...) and session.resource(...) are cached by default and will return the same object for equivalent parameters.

Note

Caches are available on the cache attribute:

session.cache.client["LRU"]
session.cache.client["LFU"]
session.cache.resource["LRU"]
session.cache.resource["LFU"]

Direct lookups use boto3_client_cache.cache.ClientCacheKey and boto3_client_cache.cache.ResourceCacheKey.

Tip

The default eviction policy is "LRU". You can opt into "LFU" per call by passing eviction_policy="LFU" to client(...) or resource(...).

Tip

You can control cache size per policy with max_size on client(...) and resource(...). If omitted, each cache defaults to a maximum size of 10.

Warning

For STS operations, STS.Client objects are added to the cache automatically. Therefore, if you are using STS for credential refresh, be mindful of cache size and eviction policy to avoid unintended evictions of the internal STS client.

from boto3_client_cache import ClientCacheKey, ResourceCacheKey
from boto3_refresh_session import AssumeRoleConfig, RefreshableSession

session = RefreshableSession(
    assume_role_kwargs=AssumeRoleConfig(RoleArn="<your-role-arn>")
)

# client caching (default LRU)
s3_client_1 = session.client("s3", region_name="us-east-1")
s3_client_2 = session.client("s3", region_name="us-east-1")
assert s3_client_1 is s3_client_2

# resource caching with LFU policy
s3_resource_1 = session.resource(
    "s3", region_name="us-east-1", eviction_policy="LFU", max_size=25
)
s3_resource_2 = session.resource(
    "s3", region_name="us-east-1", eviction_policy="LFU"
)
assert s3_resource_1 is s3_resource_2

# direct cache lookups
client_key = ClientCacheKey("s3", region_name="us-east-1")
cached_client = session.cache.client["LRU"].get(client_key)

resource_key = ResourceCacheKey("s3", region_name="us-east-1")
cached_resource = session.cache.resource["LFU"].get(resource_key)

assert cached_client is s3_client_1
assert cached_resource is s3_resource_1

IoT Core X.509#

Note

This section requires that you have installed boto3-refresh-session with the IoT extra. If you have not done so, please reinstall using:

pip install boto3-refresh-session[iot]

AWS IoT Core can vend temporary AWS credentials through the credentials provider when you connect with an X.509 certificate and a role alias. boto3-refresh-session makes this flow seamless by automatically refreshing credentials over mTLS. boto3-refresh-session supports both PEM files and PKCS#11 modules for private key storage. For additional information on the exact parameters that boto3_refresh_session.session.RefreshableSession takes for IoT, check the docs for boto3_refresh_session.methods.iot.x509.IOTX509RefreshableSession.

For PEM files:

from boto3_refresh_session import RefreshableSession
session = RefreshableSession(
    method="iot",
    endpoint="<your-credentials-endpoint>.credentials.iot.<region>.amazonaws.com",
    role_alias="<your-role-alias>",
    certificate="/path/to/certificate.pem",
    private_key="/path/to/private-key.pem",
    thing_name="<your-thing-name>",       # optional, if used in policies
    duration_seconds=3600,                # optional, capped by role alias
    region_name="us-east-1",
)

For PKCS#11:

from boto3_refresh_session import RefreshableSession

session = brs.RefreshableSession(
    method="iot",
    endpoint="<your-credentials-endpoint>.credentials.iot.<region>.amazonaws.com",
    role_alias="<your-role-alias>",
    certificate="/path/to/certificate.pem",
    pkcs11={
        "pkcs11_lib": "/usr/local/lib/softhsm/libsofthsm2.so",
        "user_pin": "1234",
        "slot_id": 0,
        "token_label": "MyToken",
        "private_key_label": "MyKey",
    },
    thing_name="<your-thing-name>",
    region_name="us-east-1",
)

For MQTT operations:

from awscrt.mqtt.QoS import AT_LEAST_ONCE
from boto3_refresh_session import RefreshableSession

conn = session.mqtt(
endpoint="<your endpoint>-ats.iot.<region>.amazonaws.com",
client_id="<your thing name or client ID>",
)
conn.connect()
conn.connect().result()
conn.publish(topic="foo/bar", payload=b"hi", qos=AT_LEAST_ONCE)
conn.disconnect().result()

Miscellaneous#

To see which identity the session is currently using, call the whoami method:

print(session.whoami())

Tip

The value returned by whoami when method="custom" is not especially informative. This is because custom credential providers vary widely, which this library cannot infer or anticipate in advance. Users employing method="custom" should implement their own identity verification logic as needed.

To return the currently active temporary security credentials, call the refreshable_credentials method or credentials attribute.

# refreshable_credentials method
print(session.refreshable_credentials())

# credentials property
print(session.credentials)

If you wish to make boto3_refresh_session.session.RefreshableSession globally available in your application without needing to pass it around explicitly, cleverly update the default session in boto3 like so:

from boto3 import DEFAULT_SESSION, client
from boto3_refresh_session import RefreshableSession

DEFAULT_SESSION = RefreshableSession(
    assume_role_kwargs={
        "RoleArn": "<your-role-arn>",
        "RoleSessionName": "<your-role-session-name>",
    },
    ...
)
s3 = client("s3")  # uses DEFAULT_SESSION under the hood

Usage#

Installation#

Authorization#

Initialization#

Refresh Methods#

Refresh Behavior#

Eager Refresh Behavior#

MFA#

Client and Resource Caching#

IoT Core X.509#

Miscellaneous#

This Page