Activity caching: keys, storage, and expiry

Use @activity to reuse the result of an async function that your application calls, such as a slow lookup. The SDK stores the JSON-serializable result and returns it for the same arguments until the entry expires. @activity runs only in your process and does not publish a Tool.

Decide what to cache

Cache a result that stays correct for its whole lifetime and that your application can safely reuse. A cache is not an external Effect ledger: do not rely on it to stop an external action from running twice. For a state-changing operation, record the attempt in your own durable store, as Secure integration describes.

Declare a cached function

This excerpt runs inside an open SDK context:

from datetime import timedelta

from dualeai import activity


@activity(sdk=sdk, cache_ttl=timedelta(minutes=10), max_retries=0)
async def normalize_sku(sku: str) -> str:
    return sku.strip().upper()


normalized = await normalize_sku(" sample-001 ")

The same arguments can reuse a cached result. If you omit cache_ttl or set it to None, an entry lives for one hour. By default, the SDK varies each lifetime by up to 10% in either direction to spread expiry times.

Check. Call the function twice with the same arguments. The second call returns the stored result without running the function body again.

Choose keys and invalidate them

The key depends on the function name and the repr() of its arguments, not on the function body. Use distinct function names and stable arguments. Include a version argument, or clear the cache, when a code change makes old results invalid. Do not rely on caching a None return value: the SDK also uses None to mean that no entry exists.

Choose the storage

The SDK opens its cache on first use. It tries Redis, then falls back to SQLite after a connection failure. The cache keys carry a prefix derived from the original API token, not from DUALEAI_TENANT_ID; replacing that token selects a new prefix. Protect the storage for the data your functions return. SDK configuration lists the Redis and SQLite settings.

To choose the storage explicitly, import CacheConfig and create_cache_backend from dualeai.cache and assign the created cache to sdk.cache before the first activity. The factory’s tenant_id argument is the prefix for cache keys; it does not configure Platform access. This excerpt chooses SQLite directly and turns off lifetime variation:

from dualeai.cache import CacheConfig, create_cache_backend

sdk.cache = await create_cache_backend(
    tenant_id="my-application-cache",
    redis_url=None,
    sqlite_path="activity-cache.db",
    config=CacheConfig(ttl_jitter_enabled=False),
)

The SDK closes the assigned cache during cleanup. Choose a separate key prefix for data that must not share cached results.

Set retries and the timeout

max_retries sets how many extra attempts follow an ordinary exception. It defaults to three. Enable retries only when repeating the operation is safe.

The job_timeout argument of DualeAISDK limits the wait for scheduled activity work, including the time it waits to run and its retries. Cache reads, cache writes, and any wait to enter a full queue are outside that timer, so the whole call can take longer. It is separate from a Task’s absolute deadline. Blocking synchronous work on the event loop can also delay timeout handling.

When that wait exceeds job_timeout, the call raises the built-in TimeoutError. Treat any external Effect of the function as uncertain: cancellation cannot roll back an action that already completed.

Clear and expire entries

The lower-level sdk.execute_activity(func, cache_ttl, max_retries, *args, **kwargs) also accepts synchronous functions, which run on the event loop; offload blocking work yourself. Unlike the decorator, cache_ttl=None on this method stores the entry without expiry.

Decorators gives the @activity signature, and BackpressureConfig sets how many activities run and wait at once.