Metadata-Version: 2.5
Name: 021-python-clients
Version: 0.5.0
Summary: Observed clients for Redis, SQS, QStash, S3, Vercel Blob, HTTP, email, KMS, Secrets Manager, Bedrock and Kubernetes Jobs and AppConfig feature flags, built on 021-python-observability
Project-URL: Repository, https://github.com/from021-io/python-clients
Project-URL: Changelog, https://github.com/from021-io/python-clients/blob/main/CHANGELOG.md
Project-URL: Contract, https://github.com/from021-io/python-clients/blob/main/CONTRACT.md
License-Expression: Apache-2.0
License-File: LICENSE
Keywords: appconfig,bedrock,email,feature-flags,httpx,kms,kubernetes,observability,opentelemetry,qstash,redis,resend,s3,secrets-manager,ses,sqs
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3 :: Only
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Topic :: System :: Monitoring
Classifier: Typing :: Typed
Requires-Python: >=3.12
Requires-Dist: 021-python-observability<0.5,>=0.3
Requires-Dist: httpx<1,>=0.28
Requires-Dist: opentelemetry-api<2,>=1.44
Provides-Extra: aws
Requires-Dist: boto3<2,>=1.40; extra == 'aws'
Requires-Dist: cryptography>=44; extra == 'aws'
Provides-Extra: httpx2
Requires-Dist: httpx2<3,>=2; extra == 'httpx2'
Provides-Extra: redis
Requires-Dist: redis<9,>=6; extra == 'redis'
Provides-Extra: vercel-blob
Requires-Dist: httpx2<3,>=2; extra == 'vercel-blob'
Requires-Dist: vercel-blob<0.6,>=0.5; extra == 'vercel-blob'
Description-Content-Type: text/markdown

# 021-python-clients

Clients for the external services our Python services call, with the observability of
[021-python-observability](https://pypi.org/project/021-python-observability/) built in.
A service uses these instead of calling `redis`, `boto3`, `httpx` or an email SDK directly, so every
dependency call gets the same span, the same metrics under a stable dependency name, and
the same failure type.

```bash
uv add "021-python-clients[redis,aws,httpx2,vercel-blob]"   # pick the extras you use
```

| Module | Extra | What it gives |
| --- | --- | --- |
| `clients_021.http.httpx` | none | `HttpClient`: buffered requests, streaming and full-body telemetry; native `ObservedTransport` |
| `clients_021.http.httpx2` | `httpx2` | `ObservedTransport` for SDKs built on httpx2 (Anthropic, OpenAI) |
| `clients_021.cache.redis` | `redis` | `RedisClient`: JSON with TTL, one-time reads, windowed counters, ping |
| `clients_021.queue.qstash` | none | `QStashClient`: JSON publishing, deduplication, delivery options and forwarded trace context |
| `clients_021.queue.sqs` | `aws` | `SqsClient`: send, receive, delete, visibility, trace carried in the message |
| `clients_021.storage.s3` | `aws` | `S3Client`: put, get, head, delete, presigned URLs |
| `clients_021.storage.vercel_blob` | `vercel-blob` | `VercelBlobClient`: put, get, head and conditional delete for public/private stores |
| `clients_021.email` | none | `EmailMessage`, the `EmailSender` protocol, `ResendSender` (Resend's HTTP API) |
| `clients_021.email.ses` | `aws` | `SesSender`: the same protocol over Amazon SES v2 |
| `clients_021.crypto.kms` | `aws` | `KmsClient`: encrypt, decrypt and data keys bound to an encryption context; AES-256-GCM envelopes |
| `clients_021.secrets.secrets_manager` | `aws` | `SecretsManagerClient`: read a secret by id, version id or stage; the value is never printed |
| `clients_021.compute.kubernetes` | none | `KubernetesJobsClient`: create (idempotent by name), read and delete Jobs in a namespace, and the Secrets they own; in-cluster service account |
| `clients_021.featureflags` | none | `Flag` and the `FeatureFlags` protocol services type against |
| `clients_021.featureflags.appconfig` | `aws` | `AppConfigFlags`: AWS AppConfig feature flags, cached and refreshed on an interval |
| `clients_021.llm.bedrock` | `aws` | `BedrockRuntime`: an observed `bedrock-runtime` client to pass to SDKs such as LangChain's `ChatBedrockConverse`; streams measured to the end, token usage counted |

Every client takes the service's `Observability` (from `setup_observability`) and a
`dependency` name, which becomes the `dependency` label of its metrics.

## Organization

Providers with the same purpose live under one capability package: `email/` contains
Resend and SES; `queue/` contains SQS and QStash; `storage/` contains S3 and Vercel Blob; `cache/`
contains Redis; `crypto/` contains KMS; `secrets/` contains Secrets Manager; `llm/` contains Bedrock; `compute/` contains Kubernetes Jobs; `featureflags/` contains AppConfig. `http/` contains the httpx client and httpx2
transport. Import providers explicitly:

```python
from clients_021.http import HttpClient
from clients_021.http.httpx2 import ObservedTransport  # requires the httpx2 extra
from clients_021.queue.sqs import SqsClient
from clients_021.queue.qstash import QStashClient
from clients_021.storage.s3 import S3Client
from clients_021.cache.redis import RedisClient
from clients_021.crypto.kms import KmsClient
from clients_021.secrets.secrets_manager import SecretsManagerClient
from clients_021.llm.bedrock import BedrockRuntime
from clients_021.compute.kubernetes import KubernetesJobsClient
```

SQS supports polling, receipts and visibility changes; QStash delivers to an HTTP endpoint.
Their provider APIs remain separate inside `queue/`. Email providers share `EmailSender`
because both implement the same send operation. A shared package does not imply that every
provider implements an interchangeable consumption protocol. Importing `queue` or QStash
does not load the optional AWS SDK.

The 0.1.0 imports `clients_021.redis`, `clients_021.s3` and `clients_021.kms` remain
compatible aliases for the same classes and result types. Importing a capability package
does not load its optional SDKs. `clients_021.HttpClient` and `clients_021.http` keep their
published exports; `clients_021.httpx2.ObservedTransport` remains an alias for
`clients_021.http.httpx2.ObservedTransport`.

## What every call gets

- A CLIENT span named `<dependency> <operation>` (`HTTP <dependency>` for HTTP), with the
  system attribute (`db.system.name=redis`, `messaging.system=aws_sqs|resend|aws_ses`,
  `rpc.service=S3|KMS|SecretsManager|BedrockRuntime|AppConfigData`, `rpc.system=vercel-blob`, `rpc.system=kubernetes`;
  Bedrock also `gen_ai.system=aws.bedrock`).
- A counter and a duration histogram: `db.client.operation.*` (Redis, shared with SQL),
  `messaging.client.operation.*` (SQS, QStash, email), `storage.client.operation.*` (S3, Vercel Blob),
  `crypto.client.operation.*` (KMS), `secrets.client.operation.*` (Secrets Manager),
  `llm.client.operation.*` and `llm.client.token.usage` (Bedrock), `compute.client.operation.*`
  (Kubernetes Jobs), `config.client.operation.*` (feature flags),
  `http.client.request.*` (HTTP, and the HTTP request
  under QStash and Resend). Labels are bounded; see [CONTRACT.md](CONTRACT.md).
- Redis, SQS, QStash, S3, Vercel Blob, email, KMS, Secrets Manager, Bedrock and Kubernetes Jobs emit a `dependency.call_failed` warning on failure,
  with dependency, operation and error type only. HTTP/httpx2 also log failed attempts,
  including body failures and cancellation; QStash/Resend suppress the nested HTTP log.
- `DependencyUnavailable(dependency, operation, reason)` when the service timed out,
  refused the connection, throttled or failed with a 5xx. Map it to a 503. Errors that mean
  the request is wrong (access denied, a missing bucket, a wrong Redis type) keep their
  original type: they are bugs, not outages.

Never in this library's spans, metrics or logs: keys, URLs, bodies, message ids or object
names; email addresses, subjects, bodies or tags; secret ids, version ids, stages and
values; model ids, prompts and outputs; namespaces, Job names, manifests and service
account tokens; feature flag names, values and AppConfig identifiers; KMS key ids, encryption contexts,
plaintext, ciphertext or data keys. Use static dependency names such as `identity-provider` or `uploads`; never derive
them from a tenant, user or request. All clients apply observability's identifier rule
(1–96 letters, digits, `_`, `.` or `-`, starting with a letter or digit).

## Examples

```python
from clients_021 import DependencyUnavailable, HttpClient
from clients_021.cache.redis import RedisClient
from clients_021.storage.s3 import S3Client
from clients_021.queue.sqs import SqsClient

redis = RedisClient.from_url(obs, settings.redis_url)
await redis.set_json("session:abc", {"user": "ana"}, ttl_seconds=1800)
used = await redis.incr_window("ratelimit:ip:1.2.3.4:28735", by=1, ttl_seconds=61)

queue = SqsClient.create(obs, settings.queue_url, region="us-east-1")
await queue.send(body, attributes={"type": "notes.created.v1"})
for message in await queue.receive(wait_seconds=20):
    with obs.tracer.start_as_current_span("handle", context=message.trace_context()):
        ...
    await queue.delete(message.receipt)

files = S3Client.create(obs, "service-uploads", region="us-east-1")
await files.put("reports/2026-10.pdf", data, content_type="application/pdf")
url = files.presign_get("reports/2026-10.pdf", expires_seconds=300)

http = HttpClient(obs, timeout_seconds=10)
response = await http.request("GET", jwks_url, dependency="identity-provider")
```

### Vercel Blob

```python
from clients_021.storage.vercel_blob import VercelBlobClient

files = VercelBlobClient(
    obs,
    token=settings.blob_token,
    store_id=settings.blob_store_id,
    access="private",  # Required; must match the store. Public is also supported.
    dependency="documents",
    timeout_seconds=10,
)
try:
    uploaded = await files.put("reports/report.pdf", pdf_bytes, content_type="application/pdf")
    content = await files.get(uploaded.url)  # bytes, or None for a missing object
    info = await files.head(uploaded.url)    # metadata, or None for a missing object
    if info is not None:
        await files.delete(uploaded.url, if_match=info.etag)
finally:
    await files.aclose()
```

Uses the [official Python Blob SDK](https://vercel.com/docs/vercel-blob/python-sdk-reference),
`vercel-blob>=0.5,<0.6`, installed with the `vercel-blob` extra. This is the current functional
SDK API; it does not depend on the earlier `AsyncBlobClient` API or the full `vercel` bundle.
Credentials and store access are explicit; no environment or default SDK session is used
for requests. A read-write token's store must match `store_id` (the `store_` prefix is allowed).

`put` returns `BlobUpload` (the SDK's `PutResult`); `head` returns `BlobInfo` (its `HeadResult`).
These are provider-specific types, not a common S3 protocol. Uploads require a pathname;
reads, metadata and deletes accept a pathname or a Vercel delivery URL. The SDK validates
those URLs and does not follow redirects. `get` buffers the entire body in memory.

Each operation has one attempt and a finite total deadline, including body consumption.
The client reuses its connection pool across calls and closes it through `aclose()`,
including an injected `httpx2.AsyncBaseTransport`. Call `aclose()` only after in-flight calls
finish. Calls after close fail locally. SDK sessions are scoped to each operation so
concurrent clients keep their credentials separate. No trace headers are sent by default.

`allow_overwrite=False` and `add_random_suffix=False` by default. Both can be explicitly
enabled, along with `cache_control_max_age`. A timeout may follow a successful upload:
retrying with a random suffix can create another object; overwriting a stable pathname can
race with other writers. Deleting a missing object succeeds. `if_match` guards a delete
against a changed ETag. CDN caches can retain deleted content briefly.

429, 5xx, transport failures and deadlines become `DependencyUnavailable`. Authentication,
missing stores, precondition failures, invalid inputs and malformed responses retain SDK
exceptions; only a missing object becomes `None` for `get`/`head`. Cancellation propagates.
The storage operation includes the complete request and response body; this client does
not emit an additional HTTP metric. Signed URLs, multipart and streaming APIs stay outside
this initial client.

### QStash

```python
from clients_021.queue.qstash import QStashClient

qstash = QStashClient(
    obs,
    token=settings.qstash_token,
    base_url=settings.qstash_url,  # the API URL for your selected region
    timeout_seconds=10,           # this service's publication request
)
try:
    published = await qstash.publish_json(
        settings.events_url,
        {"type": "notes.created.v1", "data": {"note_id": note_id}},
        deduplication_id=event_id,  # stable across retries of this logical event
        delivery_retries=3,
        delivery_timeout_seconds=60,
        delay_seconds=0,
        label="notes.created.v1",
        forward_headers={"X-Request-ID": request_id},
    )
    # published.message_id and published.deduplicated describe QStash acceptance.
finally:
    await qstash.aclose()
```

Create the client once per service lifecycle; the example closes it explicitly for
completeness. `publish_json` targets one HTTP(S) URL and supports any JSON value, including
`null`. Acknowledgement means queued, not delivered. Deduplication keys are required and
preserved verbatim; QStash rejects keys containing `:`, so this client raises `ValueError`
for those keys before making a request. Use a UUID or a stable hash of a composite business
key instead of replacing punctuation, which can collapse different keys. Reuse the key after an uncertain publication outcome; scope it to the
logical event and destination. QStash's deduplication window is finite, so consumers must
still be idempotent. See [Upstash's deduplication contract](https://upstash.com/docs/qstash/features/deduplication).

The publisher forwards the current trace to the destination by default. Use
`propagate=False` for destinations that should not receive it. Caller-supplied forwarded
trace headers are replaced or removed together. `forward_headers` uses ordinary header
names; the client adds the provider prefix. These headers are sent to QStash for delivery,
so only pass credentials intended for that destination.

The first slice covers publishing, not schedules, batches, URL groups or a web framework's
receiving endpoint. Authenticate incoming deliveries using
[QStash signature verification](https://upstash.com/docs/qstash/howto/signature).

### Email

Behind one protocol, so a service can switch between Resend and SES by configuration:

```python
from clients_021.email import EmailMessage, EmailRejected, EmailSender, ResendSender
from clients_021.email.ses import SesSender

email: EmailSender = (
    ResendSender(obs, settings.resend_api_key)
    if settings.email_provider == "resend"
    else SesSender.create(obs, region="us-east-1", configuration_set="transactional")
)
message = EmailMessage(
    sender="021 <hola@from021.io>",
    to=user.email,
    subject="Your sign-in code",
    text=f"Your code is {code}",
    tags={"category": "sign_in"},
)
try:
    await email.send(message, idempotency_key=f"sign-in:{attempt.id}")
except EmailRejected as rejected:  # the address, the domain or the credentials; not an outage
    log.warning("email.rejected", status=rejected.status, code=rejected.code)
```

`EmailRejected` carries the provider's status and error name (`validation_error`,
`MessageRejected`, ...), never its message. Resend sends each request once; with an
`idempotency_key` a retry after `DependencyUnavailable` within 24 hours cannot send twice.
SES has no idempotency key (the argument is ignored): deduplicate in the caller when a retry
must not send twice. Tags are for provider analytics; never put personal data in them.

### Kubernetes Jobs

```python
from clients_021.compute.kubernetes import KubernetesJobsClient

jobs = KubernetesJobsClient.in_cluster(obs, namespace="agent-runs")
created = await jobs.create({"metadata": {"name": "cmd-4821"}, "spec": job_spec})
status = await jobs.get("cmd-4821")        # JobStatus(phase="running", ...) or None
await jobs.delete("cmd-4821", propagation="Background")
```

No SDK: it calls the `batch/v1` REST API through `HttpClient`. `in_cluster` uses the pod's
service account (CA bundle, namespace, and a token it re-reads at most once a minute because
projected tokens rotate). `create` is idempotent by name: `generateName` is refused, and an
existing Job returns `created=False`, so a caller can repeat it after a timeout. `get`
returns `None` and `delete` returns `False` for a missing Job. POST is sent once; GET and
DELETE retry 502, 503, 504 and network failures once. 429, 5xx, timeouts and connection
failures raise `DependencyUnavailable`; other rejections raise `KubernetesApiError` with the
status code and the API's reason, never its message.

A value the pod needs but that must not be readable in the Job's spec (a run token, say) goes
in a Secret the Job owns:

```python
created = await jobs.create(manifest)      # the container reads env from secretKeyRef
uid = created.uid or (await jobs.get(name)).uid   # None in created when the Job existed
await jobs.create_secret(f"{name}-token", {"token": run_token}, job=name, job_uid=uid)
```

```yaml
env:
  - name: RUN_TOKEN
    valueFrom: { secretKeyRef: { name: cmd-4821-token, key: token } }
```

The Secret is `Opaque` and `immutable`, with an owner reference to the Job's name and uid, so
the garbage collector deletes it with the Job (including through `ttlSecondsAfterFinished`).
The Job is created first because the reference needs its uid; the kubelet keeps retrying a
pod whose Secret does not exist yet. `create_secret` is idempotent by name like `create`: a
409 `AlreadyExists` returns `created=False` and keeps the existing Secret without comparing or
updating it, so a retry must reuse the same value (or `delete_secret` and create again).
`delete_secret` returns `False` when there was none. `blockOwnerDeletion` is not set, so the
caller needs only `create` and `delete` on `secrets` (no `get`, and nothing on
`jobs/finalizers`). Secret names, data keys and values never reach telemetry, logs or error
messages; the request body is still in the chained httpx exception's `request`, so do not log
`__cause__` details.

### Bedrock

```python
from clients_021.llm.bedrock import BedrockRuntime
from langchain_aws import ChatBedrockConverse

runtime = BedrockRuntime.create(obs, region="us-east-1")
model = ChatBedrockConverse(model=settings.inference_profile, client=runtime)
```

`BedrockRuntime` keeps boto3's call signatures for `converse`, `converse_stream`,
`invoke_model`, `invoke_model_with_response_stream` and `count_tokens`, and passes every
other attribute (`meta`, `exceptions`) through to the boto3 client, so an SDK can take it
where it expects one. Calls stay synchronous like boto3's. A streaming call is measured
until its event stream ends; leaving the loop early or calling `close()` ends it as `ok`.
Converse and ConverseStream add the reported input, output and cache tokens to
`llm.client.token.usage`. Throttling (also mid-stream), 5xx, a model still loading and the
stream errors Bedrock documents as retryable raise `DependencyUnavailable`; access denied,
validation errors, a missing model and quota limits keep their botocore type. A retried
model call is billed again.

### Secrets Manager

```python
from clients_021.secrets.secrets_manager import SecretsManagerClient

secrets = SecretsManagerClient.create(obs, region="us-east-1")
secret = await secrets.get("connections/brain")
token = secret.text()            # SecretString; secret.binary() for SecretBinary
previous = await secrets.get("connections/brain", version_stage="AWSPREVIOUS")
```

`get` reads the current version unless a `version_id` or a `version_stage` is given (not
both). The client does not cache: every call is a billed request, so keep a short-lived
copy when the same secret is read often. A missing secret, access denied, a value KMS cannot
decrypt or a secret scheduled for deletion keep their botocore type; throttling, 5xx and
timeouts raise `DependencyUnavailable`. `repr(secret)` never shows the value.

### Feature flags (AWS AppConfig)

Reads [AppConfig feature flags](https://aws.amazon.com/blogs/mt/using-aws-appconfig-feature-flags/)
through the AppConfig Data API and keeps a snapshot in memory; evaluating a flag is a lookup:

```python
from clients_021.featureflags import FeatureFlags          # the protocol, no extra needed
from clients_021.featureflags.appconfig import AppConfigFlags

flags: FeatureFlags = AppConfigFlags.create(
    obs, settings.appconfig_app, settings.appconfig_env, settings.appconfig_profile,
    region="us-east-1",
)
if await flags.is_enabled("new-checkout"):          # unknown flag -> default=False
    ...
limit = await flags.flag("rate-limit")              # Flag(enabled=True, attributes={"rps": 50})
```

The snapshot is refreshed by the first request after `refresh_seconds` (default 45, minimum
15), one poll at a time. If a refresh fails while a snapshot exists, the stale one keeps
serving and the next attempt is 15 seconds later; with no snapshot yet the outage raises
`DependencyUnavailable`. A configuration that is not valid flag JSON raises `ValueError` on
the first load and is ignored with a warning afterwards. Supports plain flags and their
attributes; multi-variant flags and rules need the AppConfig Agent, which evaluates them.
The IAM role needs `appconfig:StartConfigurationSession` and
`appconfig:GetLatestConfiguration` on the configuration profile.

### KMS

The encryption context is required on every call, so a ciphertext only decrypts for the
purpose it was made for:

```python
from clients_021.crypto.kms import KmsClient

keys = KmsClient.create(obs, settings.session_key_arn, region="us-east-1")
context = {"purpose": "refresh-token", "tenant": tenant_id}
stored = await keys.encrypt(refresh_token, context=context)   # up to 4096 bytes
refresh_token = await keys.decrypt(stored, context=context)
envelope = await keys.seal(document, context=context)          # any size
document = await keys.unseal(envelope, context=context)
```

The context is logged by KMS in CloudTrail in plaintext: it names a purpose, never a
secret. `seal` makes one `GenerateDataKey` call and encrypts locally with AES-256-GCM;
`unseal` makes one `Decrypt` call. A wrong context, a tampered ciphertext or another key's
ciphertext keep their botocore type (they are bugs or attacks, not outages); a tampered
envelope raises `cryptography`'s `InvalidTag`. The envelope layout is in
[CONTRACT.md](CONTRACT.md).

### HTTP bodies and streaming

```python
from clients_021.http import HttpClient

http = HttpClient(obs, timeout_seconds=10)
try:
    async with http.stream("GET", download_url, dependency="documents") as response:
        response.raise_for_status()
        async for chunk in response.aiter_bytes():
            await write_chunk(chunk)
finally:
    await http.aclose()
```

These async clients run on asyncio. `request()` returns a buffered `HttpResponse`; `stream()` yields a native httpx response
without buffering and accepts an async iterable as upload `content`. Streaming sends once,
even for GET. The buffered client's configured idempotent retries are unchanged. Close the
response context before closing the client. For SDKs using native httpx, inject
`clients_021.http.httpx.ObservedTransport`; httpx2 SDKs use the transport below.

Each HTTP attempt emits one span, counter and duration after body EOF and connection close,
or early close. This includes upload, download, pauses between chunks and cleanup. Body
errors after 200 headers are recorded as errors. `http.response.body.complete` distinguishes
EOF from partial/unread close; it does not mean the status or cleanup succeeded. Intentional
partial close is INFO `dependency.stream_closed`; exceptions, cancellation and HTTP >= 400
produce WARNING `dependency.call_failed`. No success log, URL, header, body or provider
error message is emitted. `log_failures=False` disables attempt logs when an enclosing
operation owns the warning, as QStash and Resend do.

These duration semantics and the completion label change the 0.2.x telemetry contract;
see [migration notes](CONTRACT.md#migration-from-02x) before updating dashboards or alerts.

### SDKs on httpx2

For an SDK on httpx2:

```python
from anthropic import AsyncAnthropic, DefaultAsyncHttpxClient
from clients_021.http.httpx2 import ObservedTransport

transport = ObservedTransport(obs, dependency="anthropic")
claude = AsyncAnthropic(http_client=DefaultAsyncHttpxClient(transport=transport))
```

It does not inject trace context into third-party API requests unless `propagate=True`.
The transport preserves SDK exceptions so the SDK retains control of retries and error
handling; it does not convert them to `DependencyUnavailable`. Its span stays open through
incremental response reads and close. Use the SDK's stream context manager, or close manual
responses explicitly; abandoning a stream leaves both its connection and span open. Caller
processing keeps its own trace context, including when streams are interleaved or consumed
in another task.

## Service integration

Create clients once during service startup and reuse their connection pools. Keep business
rules in service adapters: queue envelopes, deduplication-key generation, storage integrity and key naming
do not belong in this library. Inject the configured SDK client into the constructor when
you need provider-specific options (for example S3 addressing style or credentials).

Close clients during shutdown, after stopping consumers and waiting for in-flight calls.
`aclose()` closes the underlying client, including an injected one; do not share that SDK
client with another owner. An `AsyncExitStack` makes cleanup explicit:

```python
from contextlib import AsyncExitStack

async with AsyncExitStack() as resources:
    http = HttpClient(obs, timeout_seconds=10)
    resources.push_async_callback(http.aclose)
    redis = RedisClient.from_url(obs, settings.redis_url)
    resources.push_async_callback(redis.aclose)
    files = S3Client.create(obs, "service-uploads", region="us-east-1")
    resources.push_async_callback(files.aclose)
    queue = SqsClient.create(obs, settings.queue_url, region="us-east-1")
    resources.push_async_callback(queue.aclose)
    email = ResendSender(obs, settings.resend_api_key)
    resources.push_async_callback(email.aclose)
    keys = KmsClient.create(obs, settings.session_key_arn, region="us-east-1")
    resources.push_async_callback(keys.aclose)
    secrets = SecretsManagerClient.create(obs, region="us-east-1")
    resources.push_async_callback(secrets.aclose)
    # Run the service here. Stop its background workers before leaving this scope.
```

Retry policies are explicit:

- Buffered HTTP (`request`) retries GET, HEAD, OPTIONS, PUT and DELETE on network failures, timeouts,
  remote protocol/proxy connection failures and 502/503/504. `retries=2` means at most
  three attempts.
  POST and PATCH run once. Invalid URLs/protocols and invalid request construction keep
  their original exception. HTTP status responses are returned for the caller to handle.
- Resend sends each request once (POST); send an `idempotency_key` and retry in the caller.
- AWS factories (SQS, S3, SES, KMS, Secrets Manager, Bedrock, feature flags) use botocore standard retries with at most **three total
  attempts**.
  Injected SDK clients keep their configured policy. SQS consumers must tolerate duplicate
  messages; a timeout does not prove that a send or write was not applied.
  SQS allows ten message attributes including trace context; reserve two for `traceparent`
  and `tracestate`, which belong to the library.
- Redis keeps the injected SDK's retry policy. For non-idempotent operations such as
  `take_json` and `incr_window`, do not configure automatic replay after an ambiguous
  timeout. Authentication and authorization errors keep their Redis exception type.
- QStash sends each publication POST once. `delivery_retries` configures subsequent
  delivery by QStash, not retrying this request. HTTP 429 and 5xx become
  `DependencyUnavailable`; other rejected statuses and invalid acknowledgements raise
  `QStashResponseError` with a status code and no provider body. Delivery timeouts and
  delays use integer seconds and remain subject to the provider plan's limits.
- Vercel Blob sends each operation once, with a finite total deadline (10 seconds by default).
  An injected transport must also avoid automatic retries of uncertain writes.
- httpx2 SDKs own their retries; `ObservedTransport` adds no extra retry loop.

`DependencyUnavailable` signals a transient dependency failure, **not permission to replay
any operation**. Map it at the service boundary and apply the operation's idempotency rules.
Its message and structured fields are safe to log when dependency names are static. Its
chained SDK exception may contain sensitive provider data; do not log exception chains.

S3, SQS, SES, KMS, Secrets Manager and feature flags run the synchronous AWS SDK in worker threads. Cancelling the await does not
stop an SDK request already running in its thread. Timeouts are per SDK/network attempt,
not an overall deadline. Limit consumer concurrency and let in-flight work finish before
closing the clients. Redis window counters require Redis server 7 or newer (`EXPIRE NX`).

HTTP spans now include network body consumption and cleanup. The SDK/application still
owns content decoding, JSON/SSE interpretation and processing errors outside raw-stream I/O.
Timeouts remain per I/O, not a total stream deadline. Cancellation waits for shielded body
cleanup before propagating. Close all responses and consumers before shutdown/flush.
S3 `get` buffers the entire object in memory; use a dedicated streaming adapter for large files.

## Development

```bash
uv sync --all-extras
make check   # ruff, pyright strict, pytest (Redis at REDIS_URL, default redis://127.0.0.1:63799/15)
```

See [CONTRIBUTING.md](CONTRIBUTING.md) for the module layout and the checklist for adding
clients. QStash protocol tests need no external service. To exercise actual delivery and
deduplication, start the [official local server](https://upstash.com/docs/qstash/howto/local-development),
set its `QSTASH_URL` and `QSTASH_TOKEN`, and run `uv run python tests/qstash_local.py`.
That smoke test accepts only a loopback QStash server and binds its destination dynamically.

Kubernetes Jobs tests use httpx's `MockTransport`. Bedrock tests use botocore's Stubber and
fake event streams. SQS, S3, SES, KMS and Secrets
Manager tests run against [moto](https://github.com/getmoto/moto) (plus
botocore's Stubber where moto ignores a field or differs from AWS); Resend tests use httpx's
`MockTransport`; Redis tests need a real Redis. Licensed under Apache-2.0.
