diff --git a/README.md b/README.md
index 0139b4e..b5da505 100644
--- a/README.md
+++ b/README.md
@@ -1,9 +1,9 @@
 # cloudrift-go
 
-Cloud-agnostic abstraction for **storage**, **messaging**, **document databases**, **cache**, **secrets**, and **pub/sub** — built for Lyzr microservices. The Go port of [cloudrift](../README.md) (Python).
+Cloud-agnostic abstraction for **storage**, **messaging**, **document databases**, **cache**, **secrets**, **pub/sub**, and **email** — built for Lyzr microservices. The Go port of [cloudrift](../README.md) (Python).
 
 - **Context-first.** Every operation takes a `context.Context` and is backed by native Go cloud SDKs (`aws-sdk-go-v2`, `azure-sdk-for-go`, `mongo-driver/v2`, `go-redis/v9`) — connection-pooled, no wrappers.
-- **Drop-in providers.** Same interface across AWS, Azure, and self-hosted backends. Swap `s3` ↔ `azure_blob` (or `sqs` ↔ `azure_bus`, `documentdb` ↔ `cosmos`, `redis` ↔ `elasticache` ↔ `azure_redis`) by changing one string.
+- **Drop-in providers.** Same interface across AWS, Azure, and self-hosted backends. Swap `s3` ↔ `azure_blob` (or `sqs` ↔ `azure_bus`, `documentdb` ↔ `cosmos`, `redis` ↔ `elasticache` ↔ `azure_redis`, `ses` ↔ `azure_acs` ↔ `smtp`) by changing one string.
 - **Multiple auth methods per provider.** Static keys, IAM roles, profiles, managed identity, service principals, SAS tokens, mTLS, IAM auth — pick what your microservice already has.
 
 | Category | Factory | AWS | Azure | Self-hosted |
@@ -12,8 +12,9 @@ Cloud-agnostic abstraction for **storage**, **messaging**, **document databases*
 | Messaging | `messaging.New` | `sqs` | `azure_bus` | — |
 | Document DB | `document.New` | `documentdb` | `cosmos` | — |
 | Cache | `cache.New` | `elasticache` | `azure_redis` | `redis` |
-| Secrets | `secrets.New` | `aws_secrets_manager` | `azure_keyvault` | — |
+| Secrets | `secrets.New` | `aws_secrets_manager` | `azure_keyvault` | `env` / `file` / `memory` |
 | Pub/Sub | `pubsub.New` | `sns` | `azure_eventgrid` | — |
+| Email | `email.New` | `ses` | `azure_acs` | `smtp` |
 
 ---
 
@@ -126,13 +127,23 @@ q, err := messaging.New(ctx, "azure_bus", messaging.Config{
 
 **Operations**:
 
+Bodies are **raw bytes** (binary-safe — JSON, OTLP protobuf, base64, anything), with optional string attributes (SQS MessageAttributes / Service Bus ApplicationProperties):
+
 ```go
-id, err := q.Send(ctx, map[string]any{"action": "process", "id": 42}, 0)
-ids, err := q.SendBatch(ctx, []map[string]any{{"n": 1}, {"n": 2}})
+// Raw bytes + attributes
+id, err := q.Send(ctx, []byte("…otlp bytes…"), map[string]string{"encoding": "otlp_proto"}, 0)
+ids, err := q.SendBatch(ctx, []messaging.OutgoingMessage{
+    {Body: []byte("a"), Attributes: map[string]string{"k": "v"}},
+    {Body: []byte("b")},
+})
+
+// JSON convenience for object payloads
+id, err = messaging.SendJSON(ctx, q, map[string]any{"action": "process", "id": 42}, nil, 0)
 
 msgs, err := q.Receive(ctx, 10, 20*time.Second) // long-poll
 for _, m := range msgs {
-    handleJob(m.Body)
+    handleJob(m.Body)            // []byte; m.Attributes is map[string]string
+    obj, _ := m.JSON()           // decode body as a JSON object when applicable
     err = q.Delete(ctx, m.ReceiptHandle)              // ack
     // or: q.DeadLetter(ctx, m.ReceiptHandle, "bad payload")
 }
@@ -142,6 +153,8 @@ err = q.Purge(ctx)
 err = q.Close(ctx)
 ```
 
+For cross-account access, set `Config.RoleARN` (+ optional `ExternalID`) on `messaging.New`/`storage.New` — the AWS client assumes the role via STS on top of the base credentials.
+
 > **Dead-lettering:** Azure Service Bus dead-letters natively. SQS has no per-message dead-letter API, so the backend emulates it: it re-sends the message body to the DLQ (from `Config.DLQURL` or the queue's RedrivePolicy) and deletes the original.
 
 ---
@@ -252,6 +265,12 @@ sm, err := secrets.New(ctx, "aws_secrets_manager", secrets.Config{Region: "us-ea
 kv, err := secrets.New(ctx, "azure_keyvault", secrets.Config{
     VaultURL: "https://myvault.vault.azure.net"}) // managed identity
 
+// Non-cloud backends — same interface, no secrets manager required (dev, on-prem, CI, tests):
+env, err := secrets.New(ctx, "env", secrets.Config{Prefix: "SECRET_"})        // read SECRET_<name> env vars
+file, err := secrets.New(ctx, "file", secrets.Config{FilePath: "/run/secrets.json"}) // JSON {name: value}
+mem, err := secrets.New(ctx, "memory", secrets.Config{                         // in-memory (alias "local")
+    Mapping: map[string]string{"db-password": "s3cret"}})
+
 val, err := sm.GetSecret(ctx, "db-password")
 cfg, err := sm.GetSecretJSON(ctx, "db-config")
 err = sm.SetSecret(ctx, "db-password", "s3cret")
@@ -279,6 +298,67 @@ ids, err := ps.PublishBatch(ctx, topicARN, []pubsub.BatchMessage{
 
 ---
 
+## Email
+
+```go
+import "github.com/LYZR-OSS/cloudrift-go/email"
+
+// AWS SES (SESv2)
+em, err := email.New(ctx, "ses", email.Config{
+    Region: "us-east-1", DefaultFrom: "noreply@example.com"})                  // IAM role / env
+em, err := email.New(ctx, "ses", email.Config{AWSAccessKeyID: "AKIA...",       // static keys
+    AWSSecretAccessKey: "...", Region: "us-east-1", DefaultFrom: "noreply@example.com"})
+em, err := email.New(ctx, "ses", email.Config{ProfileName: "dev",              // ~/.aws profile
+    Region: "us-east-1", DefaultFrom: "noreply@example.com"})
+
+// Azure Communication Services (talks to the ACS REST API; no Go SDK needed)
+em, err := email.New(ctx, "azure_acs", email.Config{                           // access-key signing
+    ConnectionString: "endpoint=https://x.communication.azure.com;accesskey=...",
+    DefaultFrom: "DoNotReply@example.com"})
+em, err := email.New(ctx, "azure_acs", email.Config{                           // managed identity
+    Endpoint: "https://x.communication.azure.com", DefaultFrom: "DoNotReply@example.com"})
+em, err := email.New(ctx, "azure_acs", email.Config{                           // service principal
+    Endpoint: "https://x.communication.azure.com", TenantID: "...",
+    ClientID: "...", ClientSecret: "...", DefaultFrom: "DoNotReply@example.com"})
+
+// Raw SMTP (SendGrid, Mailgun, Postmark, Office365, MailHog, ...)
+em, err := email.New(ctx, "smtp", email.Config{Host: "smtp.sendgrid.net",      // STARTTLS, port 587 (default)
+    Username: "apikey", Password: "...", DefaultFrom: "noreply@example.com"})
+em, err := email.New(ctx, "smtp", email.Config{Mode: "tls", Host: "smtp.example.com", // implicit TLS
+    Port: 465, Username: "user", Password: "pw", DefaultFrom: "noreply@example.com"})
+em, err := email.New(ctx, "smtp", email.Config{Mode: "plaintext",              // MailHog / Mailpit (dev)
+    Host: "localhost", Port: 1025, DefaultFrom: "noreply@example.test"})
+```
+
+**Operations** — same on every backend. Build an `EmailMessage`; `From` falls back to `Config.DefaultFrom` when empty, and at least one of `BodyText` / `BodyHTML` must be set:
+
+```go
+id, err := em.Send(ctx, email.EmailMessage{
+    To:       []string{"alice@example.com"},
+    Subject:  "Welcome",
+    BodyText: "Plain text body",
+    BodyHTML: "<p>HTML body</p>",
+    Cc:       []string{"bob@example.com"},
+    Bcc:      []string{"audit@example.com"}, // envelope-only, never in headers
+    ReplyTo:  []string{"support@example.com"},
+    Attachments: []email.Attachment{{
+        Filename: "welcome.pdf", Content: pdfBytes, ContentType: "application/pdf"}},
+    Headers: map[string]string{"X-Campaign": "welcome-v2"},
+})
+
+ids, err := em.SendBatch(ctx, []email.EmailMessage{ // loops Send by default
+    {To: []string{"alice@example.com"}, Subject: "hi",  BodyText: "hi"},
+    {To: []string{"bob@example.com"},   Subject: "hi2", BodyHTML: "<b>hi2</b>"},
+})
+
+ok := em.HealthCheck(ctx)
+err = em.Close(ctx)
+```
+
+> **Default sender.** Each backend takes a `DefaultFrom` in its config; calls that leave `EmailMessage.From` empty fall back to it. SES requires the sender (address or domain) to be verified; ACS requires the sending domain to be linked to the resource.
+
+---
+
 ## Errors
 
 All backends translate provider-native errors into one hierarchy under `core`, built on wrapped sentinel errors — match with `errors.Is` at any level:
@@ -294,7 +374,7 @@ case errors.Is(err, core.ErrCloudRift):       // any cloudrift error
 }
 ```
 
-Specific errors per category: `ErrObjectNotFound`/`ErrStoragePermission`, `ErrQueueNotFound`/`ErrMessageSend`, `ErrDocumentConnection`, `ErrCacheConnection`/`ErrCacheKeyNotFound`, `ErrSecretNotFound`/`ErrSecretPermission`, `ErrTopicNotFound`/`ErrPublish`. Operations a provider cannot honor return `core.ErrNotImplemented`. The document package returns a native `*mongo.Client`, so operation errors come from the MongoDB driver directly.
+Specific errors per category: `ErrObjectNotFound`/`ErrStoragePermission`, `ErrQueueNotFound`/`ErrMessageSend`, `ErrDocumentConnection`, `ErrCacheConnection`/`ErrCacheKeyNotFound`, `ErrSecretNotFound`/`ErrSecretPermission`, `ErrTopicNotFound`/`ErrPublish`, `ErrEmailSend`/`ErrRecipientRejected`/`ErrSenderUnverified`/`ErrEmailThrottled`. Operations a provider cannot honor return `core.ErrNotImplemented`. The document package returns a native `*mongo.Client`, so operation errors come from the MongoDB driver directly.
 
 ---
 
diff --git a/VERSION b/VERSION
index 3eefcb9..7dea76e 100644
--- a/VERSION
+++ b/VERSION
@@ -1 +1 @@
-1.0.0
+1.0.1
diff --git a/core/errors.go b/core/errors.go
index 0c0be55..1fd570b 100644
--- a/core/errors.go
+++ b/core/errors.go
@@ -32,6 +32,7 @@ var (
 	ErrCache     = fmt.Errorf("%w: cache", ErrCloudRift)
 	ErrSecret    = fmt.Errorf("%w: secret", ErrCloudRift)
 	ErrPubSub    = fmt.Errorf("%w: pubsub", ErrCloudRift)
+	ErrEmail     = fmt.Errorf("%w: email", ErrCloudRift)
 )
 
 // Storage errors.
@@ -71,6 +72,14 @@ var (
 	ErrPublish       = fmt.Errorf("%w: publish failed", ErrPubSub)
 )
 
+// Email errors.
+var (
+	ErrEmailSend         = fmt.Errorf("%w: send failed", ErrEmail)
+	ErrRecipientRejected = fmt.Errorf("%w: recipient rejected", ErrEmail)
+	ErrSenderUnverified  = fmt.Errorf("%w: sender unverified", ErrEmail)
+	ErrEmailThrottled    = fmt.Errorf("%w: throttled", ErrEmail)
+)
+
 // Ptr returns a pointer to v. Useful for optional config fields like
 // cache.Config.TLS (*bool), where nil means "use the provider default".
 func Ptr[T any](v T) *T { return &v }
diff --git a/email/azureacs.go b/email/azureacs.go
new file mode 100644
index 0000000..90a1613
--- /dev/null
+++ b/email/azureacs.go
@@ -0,0 +1,317 @@
+package email
+
+import (
+	"bytes"
+	"context"
+	"crypto/hmac"
+	"crypto/sha256"
+	"encoding/base64"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"net/url"
+	"strings"
+	"time"
+
+	"github.com/Azure/azure-sdk-for-go/sdk/azcore"
+	"github.com/Azure/azure-sdk-for-go/sdk/azcore/policy"
+	"github.com/Azure/azure-sdk-for-go/sdk/azidentity"
+
+	"github.com/LYZR-OSS/cloudrift-go/core"
+)
+
+const acsAPIVersion = "2023-03-31"
+
+// acsTokenScope is the Entra scope for ACS data-plane access.
+var acsTokenScope = []string{"https://communication.azure.com/.default"}
+
+// AzureACSBackend is the Azure Communication Services email backend.
+//
+// There is no official Go SDK for ACS email, so this backend talks to the ACS
+// REST API directly. Two auth modes are supported, chosen by NewAzureACS:
+//   - access key (from ConnectionString or Endpoint) → HMAC-SHA256 request
+//     signing
+//   - Entra (service principal / managed identity) → Bearer token
+type AzureACSBackend struct {
+	endpoint    string
+	defaultFrom string
+	httpClient  *http.Client
+
+	// Access-key auth.
+	accessKey []byte // decoded HMAC key; nil when using token auth
+
+	// Token auth.
+	cred azcore.TokenCredential // nil when using access-key auth
+}
+
+var _ Backend = (*AzureACSBackend)(nil)
+
+// NewAzureACS constructs an ACS email backend. Auth is inferred from Config:
+//   - ConnectionString set → access-key signing (endpoint + key parsed from it)
+//   - ClientSecret set → service principal (TenantID + ClientID + ClientSecret)
+//   - otherwise → managed identity (Endpoint required)
+func NewAzureACS(cfg Config) (*AzureACSBackend, error) {
+	b := &AzureACSBackend{
+		endpoint:    cfg.Endpoint,
+		defaultFrom: cfg.DefaultFrom,
+		httpClient:  http.DefaultClient,
+	}
+
+	if cfg.ConnectionString != "" {
+		endpoint, key, err := parseACSConnectionString(cfg.ConnectionString)
+		if err != nil {
+			return nil, err
+		}
+		b.endpoint = endpoint
+		decoded, err := base64.StdEncoding.DecodeString(key)
+		if err != nil {
+			return nil, fmt.Errorf("%w: ACS access key is not valid base64: %w", core.ErrEmail, err)
+		}
+		b.accessKey = decoded
+		return b, nil
+	}
+
+	if b.endpoint == "" {
+		return nil, fmt.Errorf("%w: ACS requires Endpoint or ConnectionString", core.ErrEmail)
+	}
+
+	var cred azcore.TokenCredential
+	var err error
+	if cfg.ClientSecret != "" {
+		cred, err = azidentity.NewClientSecretCredential(cfg.TenantID, cfg.ClientID, cfg.ClientSecret, nil)
+	} else {
+		var miOpts *azidentity.ManagedIdentityCredentialOptions
+		if cfg.ClientID != "" {
+			miOpts = &azidentity.ManagedIdentityCredentialOptions{ID: azidentity.ClientID(cfg.ClientID)}
+		}
+		cred, err = azidentity.NewManagedIdentityCredential(miOpts)
+	}
+	if err != nil {
+		return nil, fmt.Errorf("%w: %w", core.ErrEmail, err)
+	}
+	b.cred = cred
+	return b, nil
+}
+
+// acsMessage is the JSON request body for POST /emails:send.
+type acsMessage struct {
+	SenderAddress string            `json:"senderAddress"`
+	Content       acsContent        `json:"content"`
+	Recipients    acsRecipients     `json:"recipients"`
+	Attachments   []acsAttachment   `json:"attachments,omitempty"`
+	ReplyTo       []acsAddress      `json:"replyTo,omitempty"`
+	Headers       map[string]string `json:"headers,omitempty"`
+}
+
+type acsContent struct {
+	Subject   string `json:"subject"`
+	PlainText string `json:"plainText,omitempty"`
+	HTML      string `json:"html,omitempty"`
+}
+
+type acsRecipients struct {
+	To  []acsAddress `json:"to"`
+	Cc  []acsAddress `json:"cc,omitempty"`
+	Bcc []acsAddress `json:"bcc,omitempty"`
+}
+
+type acsAddress struct {
+	Address string `json:"address"`
+}
+
+type acsAttachment struct {
+	Name            string `json:"name"`
+	ContentType     string `json:"contentType"`
+	ContentInBase64 string `json:"contentInBase64"`
+}
+
+func addresses(addrs []string) []acsAddress {
+	if len(addrs) == 0 {
+		return nil
+	}
+	out := make([]acsAddress, len(addrs))
+	for i, a := range addrs {
+		out[i] = acsAddress{Address: a}
+	}
+	return out
+}
+
+func (b *AzureACSBackend) Send(ctx context.Context, msg EmailMessage) (string, error) {
+	sender, err := resolveSender(msg, b.defaultFrom)
+	if err != nil {
+		return "", err
+	}
+
+	body := acsMessage{
+		SenderAddress: sender,
+		Content:       acsContent{Subject: msg.Subject, PlainText: msg.BodyText, HTML: msg.BodyHTML},
+		Recipients: acsRecipients{
+			To:  addresses(msg.To),
+			Cc:  addresses(msg.Cc),
+			Bcc: addresses(msg.Bcc),
+		},
+		ReplyTo: addresses(msg.ReplyTo),
+		Headers: msg.Headers,
+	}
+	for _, att := range msg.Attachments {
+		body.Attachments = append(body.Attachments, acsAttachment{
+			Name:            att.Filename,
+			ContentType:     att.contentType(),
+			ContentInBase64: base64.StdEncoding.EncodeToString(att.Content),
+		})
+	}
+
+	payload, err := json.Marshal(body)
+	if err != nil {
+		return "", fmt.Errorf("%w: marshaling ACS request: %w", core.ErrEmailSend, err)
+	}
+
+	endpoint := strings.TrimRight(b.endpoint, "/")
+	reqURL := fmt.Sprintf("%s/emails:send?api-version=%s", endpoint, acsAPIVersion)
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, reqURL, bytes.NewReader(payload))
+	if err != nil {
+		return "", fmt.Errorf("%w: %w", core.ErrEmailSend, err)
+	}
+	req.Header.Set("Content-Type", "application/json")
+
+	if err := b.authorize(ctx, req, payload); err != nil {
+		return "", err
+	}
+
+	resp, err := b.httpClient.Do(req)
+	if err != nil {
+		return "", fmt.Errorf("%w: %w", core.ErrEmailSend, err)
+	}
+	defer resp.Body.Close()
+	respBody, _ := io.ReadAll(resp.Body)
+
+	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
+		return "", mapACSErr(resp.StatusCode, string(respBody))
+	}
+
+	// The send is async (typically 202). Prefer the JSON operation id, then
+	// the request id header.
+	var parsed struct {
+		ID string `json:"id"`
+	}
+	if err := json.Unmarshal(respBody, &parsed); err == nil && parsed.ID != "" {
+		return parsed.ID, nil
+	}
+	if id := resp.Header.Get("x-ms-request-id"); id != "" {
+		return id, nil
+	}
+	if loc := resp.Header.Get("Operation-Location"); loc != "" {
+		return loc, nil
+	}
+	return "", nil
+}
+
+func (b *AzureACSBackend) SendBatch(ctx context.Context, msgs []EmailMessage) ([]string, error) {
+	return sendBatchLoop(ctx, b, msgs)
+}
+
+func (b *AzureACSBackend) HealthCheck(ctx context.Context) bool {
+	// Validate that the configured auth path can produce a credential. A full
+	// reachability probe would require sending an email, so we keep it cheap.
+	if b.cred != nil {
+		_, err := b.cred.GetToken(ctx, policy.TokenRequestOptions{Scopes: acsTokenScope})
+		return err == nil
+	}
+	return b.endpoint != "" && len(b.accessKey) > 0
+}
+
+// Close is a no-op: ACS uses the shared HTTP transport managed by Go.
+func (b *AzureACSBackend) Close(ctx context.Context) error { return nil }
+
+// authorize attaches the appropriate auth header to req for the given body.
+func (b *AzureACSBackend) authorize(ctx context.Context, req *http.Request, body []byte) error {
+	if b.accessKey != nil {
+		signACSRequest(req, body, b.accessKey)
+		return nil
+	}
+	token, err := b.cred.GetToken(ctx, policy.TokenRequestOptions{Scopes: acsTokenScope})
+	if err != nil {
+		return fmt.Errorf("%w: acquiring ACS token: %w", core.ErrEmailSend, err)
+	}
+	req.Header.Set("Authorization", "Bearer "+token.Token)
+	return nil
+}
+
+// signACSRequest signs req with the ACS HMAC-SHA256 scheme in place.
+func signACSRequest(req *http.Request, body []byte, key []byte) {
+	now := time.Now().UTC().Format(http.TimeFormat) // RFC1123 GMT
+	contentHash := acsContentHash(body)
+
+	pathAndQuery := req.URL.Path
+	if req.URL.RawQuery != "" {
+		pathAndQuery += "?" + req.URL.RawQuery
+	}
+	host := req.URL.Host
+
+	stringToSign := strings.Join([]string{
+		req.Method,
+		pathAndQuery,
+		now + ";" + host + ";" + contentHash,
+	}, "\n")
+
+	mac := hmac.New(sha256.New, key)
+	mac.Write([]byte(stringToSign))
+	signature := base64.StdEncoding.EncodeToString(mac.Sum(nil))
+
+	req.Header.Set("x-ms-date", now)
+	req.Header.Set("Date", now)
+	req.Header.Set("x-ms-content-sha256", contentHash)
+	req.Header.Set("Authorization",
+		"HMAC-SHA256 SignedHeaders=x-ms-date;host;x-ms-content-sha256&Signature="+signature)
+}
+
+// acsContentHash returns base64(SHA256(body)).
+func acsContentHash(body []byte) string {
+	sum := sha256.Sum256(body)
+	return base64.StdEncoding.EncodeToString(sum[:])
+}
+
+// parseACSConnectionString extracts the endpoint and base64 access key from a
+// connection string of the form "endpoint=https://...;accesskey=BASE64KEY".
+func parseACSConnectionString(cs string) (endpoint, accessKey string, err error) {
+	for _, part := range strings.Split(cs, ";") {
+		part = strings.TrimSpace(part)
+		if part == "" {
+			continue
+		}
+		k, v, ok := strings.Cut(part, "=")
+		if !ok {
+			continue
+		}
+		switch strings.ToLower(strings.TrimSpace(k)) {
+		case "endpoint":
+			endpoint = strings.TrimSpace(v)
+		case "accesskey":
+			accessKey = strings.TrimSpace(v)
+		}
+	}
+	if endpoint == "" || accessKey == "" {
+		return "", "", fmt.Errorf("%w: ACS connection string must contain endpoint= and accesskey=", core.ErrEmail)
+	}
+	if _, perr := url.Parse(endpoint); perr != nil {
+		return "", "", fmt.Errorf("%w: invalid ACS endpoint: %w", core.ErrEmail, perr)
+	}
+	return endpoint, accessKey, nil
+}
+
+func mapACSErr(status int, body string) error {
+	switch status {
+	case http.StatusTooManyRequests:
+		return fmt.Errorf("%w: ACS %d: %s", core.ErrEmailThrottled, status, body)
+	case http.StatusBadRequest, http.StatusForbidden:
+		low := strings.ToLower(body)
+		if strings.Contains(low, "sender") || strings.Contains(low, "domain") {
+			return fmt.Errorf("%w: ACS %d: %s", core.ErrSenderUnverified, status, body)
+		}
+		if strings.Contains(low, "recipient") {
+			return fmt.Errorf("%w: ACS %d: %s", core.ErrRecipientRejected, status, body)
+		}
+	}
+	return fmt.Errorf("%w: ACS %d: %s", core.ErrEmailSend, status, body)
+}
diff --git a/email/email.go b/email/email.go
new file mode 100644
index 0000000..8fe75c3
--- /dev/null
+++ b/email/email.go
@@ -0,0 +1,152 @@
+// Package email provides a provider-neutral interface over transactional
+// email backends: AWS SES (SESv2), Azure Communication Services (ACS) email,
+// and raw SMTP.
+//
+// Construct a backend once at service startup via New (or NewSES /
+// NewAzureACS / NewSMTP) and reuse it — backends hold long-lived clients.
+// Release sockets at shutdown with Close.
+//
+// Build an EmailMessage and pass it to Send. From falls back to the backend's
+// DefaultFrom when empty; at least one of BodyText / BodyHTML must be set.
+package email
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/LYZR-OSS/cloudrift-go/core"
+)
+
+// Attachment is an email attachment.
+//
+// Content is the raw payload bytes. ContentType is used directly in the MIME /
+// provider request — pick the right one ("application/pdf", "image/png", ...)
+// so the recipient's mail client renders it correctly. An empty ContentType
+// defaults to "application/octet-stream".
+type Attachment struct {
+	Filename    string
+	Content     []byte
+	ContentType string
+}
+
+// EmailMessage is an outbound email.
+//
+// From falls back to the backend's DefaultFrom when empty. At least one of
+// BodyText / BodyHTML must be set.
+type EmailMessage struct {
+	To          []string
+	Subject     string
+	BodyText    string
+	BodyHTML    string
+	From        string
+	Cc          []string
+	Bcc         []string
+	ReplyTo     []string
+	Attachments []Attachment
+	Headers     map[string]string
+}
+
+// Backend is the provider-neutral transactional email interface.
+type Backend interface {
+	// Send sends a single email and returns the provider message ID.
+	Send(ctx context.Context, msg EmailMessage) (string, error)
+	// SendBatch sends multiple emails and returns their message IDs.
+	SendBatch(ctx context.Context, msgs []EmailMessage) ([]string, error)
+	// HealthCheck returns true if the email backend is reachable (never errors).
+	HealthCheck(ctx context.Context) bool
+	// Close releases the underlying client and sockets.
+	Close(ctx context.Context) error
+}
+
+// Config carries the union of provider options. Only the fields relevant to
+// the chosen provider are read; the factory routes to the appropriate auth
+// method based on which credential fields are set.
+type Config struct {
+	// DefaultFrom is the sender used when EmailMessage.From is empty.
+	DefaultFrom string
+
+	// AWS SES (SESv2).
+	Region             string // default "us-east-1"
+	AWSAccessKeyID     string
+	AWSSecretAccessKey string
+	AWSSessionToken    string
+	ProfileName        string
+	EndpointURL        string // custom endpoint (LocalStack, ...)
+
+	// Azure Communication Services email.
+	ConnectionString string // "endpoint=https://...;accesskey=BASE64KEY"
+	Endpoint         string // https://<resource>.communication.azure.com
+	TenantID         string
+	ClientID         string // service principal app ID, or user-assigned MI client ID
+	ClientSecret     string
+
+	// SMTP.
+	Host     string
+	Port     int    // default depends on Mode (587 starttls, 465 tls, 25 plaintext)
+	Username string
+	Password string
+	Mode     string // "starttls" (default), "tls", "plaintext"
+}
+
+// New instantiates an email backend.
+//
+// provider is "ses", "azure_acs", or "smtp". The auth method is inferred from
+// which credential fields are set, exactly as in the Python library:
+//
+//	New(ctx, "ses", Config{Region: "us-east-1", DefaultFrom: "noreply@example.com"})            // IAM role / env
+//	New(ctx, "ses", Config{AWSAccessKeyID: "...", AWSSecretAccessKey: "...", DefaultFrom: "..."})
+//	New(ctx, "azure_acs", Config{ConnectionString: "endpoint=https://...;accesskey=...", DefaultFrom: "..."})
+//	New(ctx, "azure_acs", Config{Endpoint: "https://...communication.azure.com", DefaultFrom: "..."}) // managed identity
+//	New(ctx, "smtp", Config{Host: "smtp.sendgrid.net", Username: "apikey", Password: "...", DefaultFrom: "..."})
+//	New(ctx, "smtp", Config{Mode: "tls", Host: "smtp.example.com", Port: 465, Username: "u", Password: "p", DefaultFrom: "..."})
+func New(ctx context.Context, provider string, cfg Config) (Backend, error) {
+	switch provider {
+	case "ses":
+		return NewSES(ctx, cfg)
+	case "azure_acs":
+		return NewAzureACS(cfg)
+	case "smtp":
+		return NewSMTP(cfg)
+	}
+	return nil, fmt.Errorf("%w: unknown email provider %q (choose 'ses', 'azure_acs', or 'smtp')",
+		core.ErrEmail, provider)
+}
+
+// contentType returns the attachment's content type, defaulting to
+// "application/octet-stream" when unset.
+func (a Attachment) contentType() string {
+	if a.ContentType == "" {
+		return "application/octet-stream"
+	}
+	return a.ContentType
+}
+
+// resolveSender returns msg.From, falling back to defaultFrom, and validates
+// that a sender and at least one body part are present.
+func resolveSender(msg EmailMessage, defaultFrom string) (string, error) {
+	sender := msg.From
+	if sender == "" {
+		sender = defaultFrom
+	}
+	if sender == "" {
+		return "", fmt.Errorf("%w: no sender address: set EmailMessage.From or Config.DefaultFrom",
+			core.ErrEmail)
+	}
+	if msg.BodyText == "" && msg.BodyHTML == "" {
+		return "", fmt.Errorf("%w: Send requires BodyText and/or BodyHTML", core.ErrEmail)
+	}
+	return sender, nil
+}
+
+// sendBatchLoop is the default SendBatch implementation: it loops Send.
+func sendBatchLoop(ctx context.Context, b Backend, msgs []EmailMessage) ([]string, error) {
+	ids := make([]string, 0, len(msgs))
+	for _, msg := range msgs {
+		id, err := b.Send(ctx, msg)
+		if err != nil {
+			return nil, err
+		}
+		ids = append(ids, id)
+	}
+	return ids, nil
+}
diff --git a/email/email_test.go b/email/email_test.go
new file mode 100644
index 0000000..cd7ebc1
--- /dev/null
+++ b/email/email_test.go
@@ -0,0 +1,186 @@
+package email
+
+import (
+	"context"
+	"errors"
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/LYZR-OSS/cloudrift-go/core"
+)
+
+func TestNewUnknownProvider(t *testing.T) {
+	_, err := New(context.Background(), "mailchimp", Config{})
+	if !errors.Is(err, core.ErrEmail) {
+		t.Fatalf("err = %v; want core.ErrEmail", err)
+	}
+}
+
+func TestFactoryRouting(t *testing.T) {
+	ctx := context.Background()
+
+	// The SES "from profile" path eagerly loads the named shared-config
+	// profile, so point the AWS SDK at a temp credentials file that defines
+	// "dev" — keeps the routing assertion hermetic (no machine AWS config).
+	credsFile := filepath.Join(t.TempDir(), "credentials")
+	if err := os.WriteFile(credsFile, []byte("[dev]\naws_access_key_id = AKIATEST\naws_secret_access_key = testsecret\nregion = us-east-1\n"), 0o600); err != nil {
+		t.Fatal(err)
+	}
+	t.Setenv("AWS_SHARED_CREDENTIALS_FILE", credsFile)
+	t.Setenv("AWS_CONFIG_FILE", filepath.Join(t.TempDir(), "config-empty"))
+	tests := []struct {
+		name     string
+		provider string
+		cfg      Config
+		wantType any
+		wantErr  bool
+	}{
+		{"ses iam role", "ses", Config{Region: "us-east-1", DefaultFrom: "a@b.com"}, &SESBackend{}, false},
+		{"ses access key", "ses", Config{AWSAccessKeyID: "AKIA", AWSSecretAccessKey: "secret"}, &SESBackend{}, false},
+		{"ses profile", "ses", Config{ProfileName: "dev"}, &SESBackend{}, false},
+		{"acs connection string", "azure_acs", Config{ConnectionString: "endpoint=https://x.communication.azure.com;accesskey=" + b64key}, &AzureACSBackend{}, false},
+		{"acs service principal", "azure_acs", Config{Endpoint: "https://x.communication.azure.com", TenantID: "t", ClientID: "c", ClientSecret: "s"}, &AzureACSBackend{}, false},
+		{"acs managed identity", "azure_acs", Config{Endpoint: "https://x.communication.azure.com"}, &AzureACSBackend{}, false},
+		{"acs missing endpoint", "azure_acs", Config{}, nil, true},
+		{"smtp default starttls", "smtp", Config{Host: "smtp.example.com"}, &SMTPBackend{}, false},
+		{"smtp tls", "smtp", Config{Mode: "tls", Host: "smtp.example.com"}, &SMTPBackend{}, false},
+		{"smtp plaintext", "smtp", Config{Mode: "plaintext", Host: "localhost"}, &SMTPBackend{}, false},
+		{"smtp bad mode", "smtp", Config{Mode: "ssl", Host: "smtp.example.com"}, nil, true},
+		{"smtp missing host", "smtp", Config{}, nil, true},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			b, err := New(ctx, tt.provider, tt.cfg)
+			if tt.wantErr {
+				if err == nil {
+					t.Fatalf("expected error, got backend %T", b)
+				}
+				if !errors.Is(err, core.ErrEmail) {
+					t.Fatalf("err = %v; want core.ErrEmail", err)
+				}
+				return
+			}
+			if err != nil {
+				t.Fatalf("unexpected error: %v", err)
+			}
+			switch tt.wantType.(type) {
+			case *SESBackend:
+				if _, ok := b.(*SESBackend); !ok {
+					t.Fatalf("got %T; want *SESBackend", b)
+				}
+			case *AzureACSBackend:
+				if _, ok := b.(*AzureACSBackend); !ok {
+					t.Fatalf("got %T; want *AzureACSBackend", b)
+				}
+			case *SMTPBackend:
+				if _, ok := b.(*SMTPBackend); !ok {
+					t.Fatalf("got %T; want *SMTPBackend", b)
+				}
+			}
+		})
+	}
+}
+
+func TestACSAuthMode(t *testing.T) {
+	b, err := NewAzureACS(Config{ConnectionString: "endpoint=https://x.communication.azure.com;accesskey=" + b64key})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if b.accessKey == nil {
+		t.Fatal("connection-string backend should use access-key auth")
+	}
+	if b.cred != nil {
+		t.Fatal("connection-string backend should not have a token credential")
+	}
+
+	b2, err := NewAzureACS(Config{Endpoint: "https://x.communication.azure.com", TenantID: "t", ClientID: "c", ClientSecret: "s"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if b2.accessKey != nil {
+		t.Fatal("service-principal backend should not have an access key")
+	}
+	if b2.cred == nil {
+		t.Fatal("service-principal backend should have a token credential")
+	}
+}
+
+func TestSMTPPortDefaults(t *testing.T) {
+	tests := []struct {
+		mode string
+		want int
+	}{
+		{"", 587},
+		{"starttls", 587},
+		{"tls", 465},
+		{"plaintext", 25},
+	}
+	for _, tt := range tests {
+		b, err := NewSMTP(Config{Host: "h", Mode: tt.mode})
+		if err != nil {
+			t.Fatalf("mode %q: %v", tt.mode, err)
+		}
+		if b.port != tt.want {
+			t.Fatalf("mode %q: port = %d; want %d", tt.mode, b.port, tt.want)
+		}
+	}
+	// Explicit port wins.
+	b, err := NewSMTP(Config{Host: "h", Mode: "tls", Port: 2525})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if b.port != 2525 {
+		t.Fatalf("explicit port = %d; want 2525", b.port)
+	}
+}
+
+func TestAttachmentContentTypeDefault(t *testing.T) {
+	a := Attachment{Filename: "f.bin", Content: []byte("x")}
+	if got := a.contentType(); got != "application/octet-stream" {
+		t.Fatalf("default content type = %q; want application/octet-stream", got)
+	}
+	a2 := Attachment{Filename: "f.pdf", ContentType: "application/pdf"}
+	if got := a2.contentType(); got != "application/pdf" {
+		t.Fatalf("content type = %q; want application/pdf", got)
+	}
+}
+
+func TestResolveSender(t *testing.T) {
+	tests := []struct {
+		name       string
+		msg        EmailMessage
+		def        string
+		wantSender string
+		wantErr    bool
+	}{
+		{"from wins", EmailMessage{From: "a@x.com", BodyText: "hi"}, "b@x.com", "a@x.com", false},
+		{"falls back to default", EmailMessage{BodyText: "hi"}, "b@x.com", "b@x.com", false},
+		{"no sender at all", EmailMessage{BodyText: "hi"}, "", "", true},
+		{"no body", EmailMessage{From: "a@x.com"}, "", "", true},
+		{"html only ok", EmailMessage{From: "a@x.com", BodyHTML: "<p>hi</p>"}, "", "a@x.com", false},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got, err := resolveSender(tt.msg, tt.def)
+			if tt.wantErr {
+				if err == nil {
+					t.Fatal("expected error")
+				}
+				if !errors.Is(err, core.ErrEmail) {
+					t.Fatalf("err = %v; want core.ErrEmail", err)
+				}
+				return
+			}
+			if err != nil {
+				t.Fatalf("unexpected error: %v", err)
+			}
+			if got != tt.wantSender {
+				t.Fatalf("sender = %q; want %q", got, tt.wantSender)
+			}
+		})
+	}
+}
+
+// b64key is a deterministic base64-encoded 32-byte key used across tests.
+const b64key = "MDEyMzQ1Njc4OWFiY2RlZjAxMjM0NTY3ODlhYmNkZWY="
diff --git a/email/mime.go b/email/mime.go
new file mode 100644
index 0000000..27237e8
--- /dev/null
+++ b/email/mime.go
@@ -0,0 +1,203 @@
+package email
+
+import (
+	"bytes"
+	"encoding/base64"
+	"fmt"
+	"mime"
+	"mime/multipart"
+	"mime/quotedprintable"
+	"net/textproto"
+	"strings"
+)
+
+// buildMIME builds a raw RFC 5322 MIME message.
+//
+// Structure:
+//   - text-only / html-only  → a single text/plain or text/html body part
+//   - text + html            → multipart/alternative
+//   - any attachments        → multipart/mixed wrapping the body part(s)
+//
+// Bcc is intentionally NOT written to the headers — Bcc recipients belong only
+// in the SMTP envelope. The caller is responsible for the envelope.
+//
+// messageID, when non-empty, is written as the Message-ID header.
+func buildMIME(sender string, msg EmailMessage, messageID string) ([]byte, error) {
+	var buf bytes.Buffer
+
+	writeHeader(&buf, "From", sender)
+	if len(msg.To) > 0 {
+		writeHeader(&buf, "To", strings.Join(msg.To, ", "))
+	}
+	if len(msg.Cc) > 0 {
+		writeHeader(&buf, "Cc", strings.Join(msg.Cc, ", "))
+	}
+	if len(msg.ReplyTo) > 0 {
+		writeHeader(&buf, "Reply-To", strings.Join(msg.ReplyTo, ", "))
+	}
+	writeHeader(&buf, "Subject", msg.Subject)
+	if messageID != "" {
+		writeHeader(&buf, "Message-ID", messageID)
+	}
+	for k, v := range msg.Headers {
+		// Skip headers we manage ourselves to avoid duplicates.
+		switch textproto.CanonicalMIMEHeaderKey(k) {
+		case "From", "To", "Cc", "Reply-To", "Subject", "Message-Id",
+			"Mime-Version", "Content-Type", "Content-Transfer-Encoding":
+			continue
+		}
+		writeHeader(&buf, k, v)
+	}
+	buf.WriteString("MIME-Version: 1.0\r\n")
+
+	if len(msg.Attachments) > 0 {
+		return buildMixed(&buf, msg)
+	}
+	return buildBody(&buf, msg)
+}
+
+// buildBody writes the body directly into the message headers (no attachments).
+func buildBody(buf *bytes.Buffer, msg EmailMessage) ([]byte, error) {
+	if msg.BodyText != "" && msg.BodyHTML != "" {
+		mw := multipart.NewWriter(buf)
+		writeHeader(buf, "Content-Type", "multipart/alternative; boundary="+mw.Boundary())
+		buf.WriteString("\r\n")
+		if err := writeAlternative(mw, msg); err != nil {
+			return nil, err
+		}
+		return buf.Bytes(), nil
+	}
+
+	ctype := "text/plain; charset=UTF-8"
+	body := msg.BodyText
+	if msg.BodyHTML != "" {
+		ctype = "text/html; charset=UTF-8"
+		body = msg.BodyHTML
+	}
+	writeHeader(buf, "Content-Type", ctype)
+	buf.WriteString("Content-Transfer-Encoding: quoted-printable\r\n\r\n")
+	if err := writeQP(buf, body); err != nil {
+		return nil, err
+	}
+	return buf.Bytes(), nil
+}
+
+// buildMixed wraps the body part(s) and attachments in multipart/mixed.
+func buildMixed(buf *bytes.Buffer, msg EmailMessage) ([]byte, error) {
+	mw := multipart.NewWriter(buf)
+	writeHeader(buf, "Content-Type", "multipart/mixed; boundary="+mw.Boundary())
+	buf.WriteString("\r\n")
+
+	if err := writeBodyPart(mw, msg); err != nil {
+		return nil, err
+	}
+
+	for _, att := range msg.Attachments {
+		h := make(textproto.MIMEHeader)
+		h.Set("Content-Type", att.contentType())
+		h.Set("Content-Transfer-Encoding", "base64")
+		h.Set("Content-Disposition", fmt.Sprintf("attachment; filename=%q", att.Filename))
+		pw, err := mw.CreatePart(h)
+		if err != nil {
+			return nil, err
+		}
+		if err := writeBase64(pw, att.Content); err != nil {
+			return nil, err
+		}
+	}
+
+	if err := mw.Close(); err != nil {
+		return nil, err
+	}
+	return buf.Bytes(), nil
+}
+
+// writeBodyPart writes the body (single part or nested multipart/alternative)
+// as one part of an enclosing multipart writer.
+func writeBodyPart(mw *multipart.Writer, msg EmailMessage) error {
+	if msg.BodyText != "" && msg.BodyHTML != "" {
+		var inner bytes.Buffer
+		altW := multipart.NewWriter(&inner)
+		h := make(textproto.MIMEHeader)
+		h.Set("Content-Type", "multipart/alternative; boundary="+altW.Boundary())
+		pw, err := mw.CreatePart(h)
+		if err != nil {
+			return err
+		}
+		if err := writeAlternative(altW, msg); err != nil {
+			return err
+		}
+		_, err = pw.Write(inner.Bytes())
+		return err
+	}
+
+	ctype := "text/plain; charset=UTF-8"
+	body := msg.BodyText
+	if msg.BodyHTML != "" {
+		ctype = "text/html; charset=UTF-8"
+		body = msg.BodyHTML
+	}
+	h := make(textproto.MIMEHeader)
+	h.Set("Content-Type", ctype)
+	h.Set("Content-Transfer-Encoding", "quoted-printable")
+	pw, err := mw.CreatePart(h)
+	if err != nil {
+		return err
+	}
+	return writeQP(pw, body)
+}
+
+// writeAlternative writes a text/plain then text/html part into a
+// multipart/alternative writer and closes it.
+func writeAlternative(mw *multipart.Writer, msg EmailMessage) error {
+	for _, part := range []struct{ ctype, body string }{
+		{"text/plain; charset=UTF-8", msg.BodyText},
+		{"text/html; charset=UTF-8", msg.BodyHTML},
+	} {
+		h := make(textproto.MIMEHeader)
+		h.Set("Content-Type", part.ctype)
+		h.Set("Content-Transfer-Encoding", "quoted-printable")
+		pw, err := mw.CreatePart(h)
+		if err != nil {
+			return err
+		}
+		if err := writeQP(pw, part.body); err != nil {
+			return err
+		}
+	}
+	return mw.Close()
+}
+
+func writeHeader(buf *bytes.Buffer, key, value string) {
+	buf.WriteString(textproto.CanonicalMIMEHeaderKey(key))
+	buf.WriteString(": ")
+	// RFC 2047-encode values with non-ASCII so headers stay 7-bit clean.
+	buf.WriteString(mime.QEncoding.Encode("UTF-8", value))
+	buf.WriteString("\r\n")
+}
+
+func writeQP(w interface{ Write([]byte) (int, error) }, body string) error {
+	qp := quotedprintable.NewWriter(w)
+	if _, err := qp.Write([]byte(body)); err != nil {
+		return err
+	}
+	return qp.Close()
+}
+
+func writeBase64(w interface{ Write([]byte) (int, error) }, content []byte) error {
+	const lineLen = 76
+	enc := base64.StdEncoding.EncodeToString(content)
+	for i := 0; i < len(enc); i += lineLen {
+		end := i + lineLen
+		if end > len(enc) {
+			end = len(enc)
+		}
+		if _, err := w.Write([]byte(enc[i:end])); err != nil {
+			return err
+		}
+		if _, err := w.Write([]byte("\r\n")); err != nil {
+			return err
+		}
+	}
+	return nil
+}
diff --git a/email/ses.go b/email/ses.go
new file mode 100644
index 0000000..dd85a84
--- /dev/null
+++ b/email/ses.go
@@ -0,0 +1,141 @@
+package email
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"strings"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	awsconfig "github.com/aws/aws-sdk-go-v2/config"
+	"github.com/aws/aws-sdk-go-v2/credentials"
+	"github.com/aws/aws-sdk-go-v2/service/sesv2"
+	sesv2types "github.com/aws/aws-sdk-go-v2/service/sesv2/types"
+	"github.com/aws/smithy-go"
+
+	"github.com/LYZR-OSS/cloudrift-go/core"
+)
+
+// SESBackend is the AWS SES (SESv2) email backend.
+//
+// A single client is held for the lifetime of the backend. Authentication is
+// chosen by NewSES based on which Config fields are set: static access key,
+// named profile, or the default AWS credential chain.
+type SESBackend struct {
+	client      *sesv2.Client
+	defaultFrom string
+}
+
+var _ Backend = (*SESBackend)(nil)
+
+// NewSES constructs an SES backend. Region defaults to "us-east-1".
+// cfg.EndpointURL points the client at a custom endpoint (LocalStack).
+func NewSES(ctx context.Context, cfg Config) (*SESBackend, error) {
+	region := cfg.Region
+	if region == "" {
+		region = "us-east-1"
+	}
+	loadOpts := []func(*awsconfig.LoadOptions) error{awsconfig.WithRegion(region)}
+	if cfg.AWSAccessKeyID != "" {
+		loadOpts = append(loadOpts, awsconfig.WithCredentialsProvider(
+			credentials.NewStaticCredentialsProvider(cfg.AWSAccessKeyID, cfg.AWSSecretAccessKey, cfg.AWSSessionToken)))
+	} else if cfg.ProfileName != "" {
+		loadOpts = append(loadOpts, awsconfig.WithSharedConfigProfile(cfg.ProfileName))
+	}
+	awsCfg, err := awsconfig.LoadDefaultConfig(ctx, loadOpts...)
+	if err != nil {
+		return nil, fmt.Errorf("%w: %w", core.ErrEmail, err)
+	}
+	client := sesv2.NewFromConfig(awsCfg, func(o *sesv2.Options) {
+		if cfg.EndpointURL != "" {
+			o.BaseEndpoint = aws.String(cfg.EndpointURL)
+		}
+	})
+	return &SESBackend{client: client, defaultFrom: cfg.DefaultFrom}, nil
+}
+
+func (b *SESBackend) Send(ctx context.Context, msg EmailMessage) (string, error) {
+	sender, err := resolveSender(msg, b.defaultFrom)
+	if err != nil {
+		return "", err
+	}
+
+	dest := &sesv2types.Destination{
+		ToAddresses:  msg.To,
+		CcAddresses:  msg.Cc,
+		BccAddresses: msg.Bcc,
+	}
+
+	var content *sesv2types.EmailContent
+	if len(msg.Attachments) > 0 || len(msg.Headers) > 0 {
+		raw, err := buildMIME(sender, msg, "")
+		if err != nil {
+			return "", fmt.Errorf("%w: building MIME message: %w", core.ErrEmailSend, err)
+		}
+		content = &sesv2types.EmailContent{
+			Raw: &sesv2types.RawMessage{Data: raw},
+		}
+	} else {
+		body := &sesv2types.Body{}
+		if msg.BodyText != "" {
+			body.Text = &sesv2types.Content{Data: aws.String(msg.BodyText), Charset: aws.String("UTF-8")}
+		}
+		if msg.BodyHTML != "" {
+			body.Html = &sesv2types.Content{Data: aws.String(msg.BodyHTML), Charset: aws.String("UTF-8")}
+		}
+		content = &sesv2types.EmailContent{
+			Simple: &sesv2types.Message{
+				Subject: &sesv2types.Content{Data: aws.String(msg.Subject), Charset: aws.String("UTF-8")},
+				Body:    body,
+			},
+		}
+	}
+
+	resp, err := b.client.SendEmail(ctx, &sesv2.SendEmailInput{
+		FromEmailAddress: aws.String(sender),
+		Destination:      dest,
+		Content:          content,
+		ReplyToAddresses: msg.ReplyTo,
+	})
+	if err != nil {
+		return "", mapSESErr(err)
+	}
+	return aws.ToString(resp.MessageId), nil
+}
+
+func (b *SESBackend) SendBatch(ctx context.Context, msgs []EmailMessage) ([]string, error) {
+	return sendBatchLoop(ctx, b, msgs)
+}
+
+func (b *SESBackend) HealthCheck(ctx context.Context) bool {
+	_, err := b.client.GetAccount(ctx, &sesv2.GetAccountInput{})
+	return err == nil
+}
+
+// Close is a no-op: the aws-sdk-go-v2 client shares the default HTTP
+// transport, which Go manages.
+func (b *SESBackend) Close(ctx context.Context) error { return nil }
+
+func mapSESErr(err error) error {
+	var ae smithy.APIError
+	if errors.As(err, &ae) {
+		switch ae.ErrorCode() {
+		case "MessageRejected":
+			return fmt.Errorf("%w: %w", core.ErrRecipientRejected, err)
+		case "MailFromDomainNotVerified", "MailFromDomainNotVerifiedException",
+			"FromEmailAddressNotVerified", "AccountSendingPausedException":
+			return fmt.Errorf("%w: %w", core.ErrSenderUnverified, err)
+		case "Throttling", "TooManyRequestsException", "SendingPausedException":
+			return fmt.Errorf("%w: %w", core.ErrEmailThrottled, err)
+		}
+	}
+	// Fall back to message inspection for cases the code alone misses.
+	low := strings.ToLower(err.Error())
+	switch {
+	case strings.Contains(low, "not verified"):
+		return fmt.Errorf("%w: %w", core.ErrSenderUnverified, err)
+	case strings.Contains(low, "throttl"):
+		return fmt.Errorf("%w: %w", core.ErrEmailThrottled, err)
+	}
+	return fmt.Errorf("%w: %w", core.ErrEmailSend, err)
+}
diff --git a/email/smtp.go b/email/smtp.go
new file mode 100644
index 0000000..5885b5e
--- /dev/null
+++ b/email/smtp.go
@@ -0,0 +1,213 @@
+package email
+
+import (
+	"context"
+	"crypto/rand"
+	"crypto/tls"
+	"encoding/hex"
+	"errors"
+	"fmt"
+	"net"
+	"net/smtp"
+	"net/textproto"
+	"strings"
+
+	"github.com/LYZR-OSS/cloudrift-go/core"
+)
+
+// SMTP transport modes.
+const (
+	modeStartTLS  = "starttls"
+	modeTLS       = "tls"
+	modePlaintext = "plaintext"
+)
+
+// SMTPBackend is a raw SMTP backend (SendGrid, Mailgun, Postmark, Office365,
+// MailHog, ...).
+//
+// A fresh connection is opened per Send. SMTP servers commonly drop idle
+// connections and transactional volumes don't benefit from pooling.
+type SMTPBackend struct {
+	host        string
+	port        int
+	mode        string
+	username    string
+	password    string
+	defaultFrom string
+}
+
+var _ Backend = (*SMTPBackend)(nil)
+
+// NewSMTP constructs an SMTP backend. Mode defaults to "starttls". Port
+// defaults by mode: 587 (starttls), 465 (tls), 25 (plaintext).
+func NewSMTP(cfg Config) (*SMTPBackend, error) {
+	mode := cfg.Mode
+	if mode == "" {
+		mode = modeStartTLS
+	}
+	if mode != modeStartTLS && mode != modeTLS && mode != modePlaintext {
+		return nil, fmt.Errorf("%w: unknown SMTP mode %q (choose 'starttls', 'tls', or 'plaintext')",
+			core.ErrEmail, mode)
+	}
+	if cfg.Host == "" {
+		return nil, fmt.Errorf("%w: SMTP host is required", core.ErrEmail)
+	}
+	port := cfg.Port
+	if port == 0 {
+		switch mode {
+		case modeTLS:
+			port = 465
+		case modePlaintext:
+			port = 25
+		default:
+			port = 587
+		}
+	}
+	return &SMTPBackend{
+		host:        cfg.Host,
+		port:        port,
+		mode:        mode,
+		username:    cfg.Username,
+		password:    cfg.Password,
+		defaultFrom: cfg.DefaultFrom,
+	}, nil
+}
+
+func (b *SMTPBackend) Send(ctx context.Context, msg EmailMessage) (string, error) {
+	sender, err := resolveSender(msg, b.defaultFrom)
+	if err != nil {
+		return "", err
+	}
+
+	messageID := newMessageID(sender)
+	raw, err := buildMIME(sender, msg, messageID)
+	if err != nil {
+		return "", fmt.Errorf("%w: building MIME message: %w", core.ErrEmailSend, err)
+	}
+
+	// Bcc recipients go in the envelope, never in the headers.
+	recipients := make([]string, 0, len(msg.To)+len(msg.Cc)+len(msg.Bcc))
+	recipients = append(recipients, msg.To...)
+	recipients = append(recipients, msg.Cc...)
+	recipients = append(recipients, msg.Bcc...)
+
+	if err := b.deliver(ctx, sender, recipients, raw); err != nil {
+		return "", err
+	}
+	return messageID, nil
+}
+
+func (b *SMTPBackend) SendBatch(ctx context.Context, msgs []EmailMessage) ([]string, error) {
+	return sendBatchLoop(ctx, b, msgs)
+}
+
+// deliver opens a connection per the mode, authenticates if credentials are
+// set, and writes the message.
+func (b *SMTPBackend) deliver(ctx context.Context, sender string, recipients []string, raw []byte) error {
+	client, err := b.dial(ctx)
+	if err != nil {
+		return mapSMTPErr(err)
+	}
+	defer client.Close()
+
+	if b.username != "" {
+		auth := smtp.PlainAuth("", b.username, b.password, b.host)
+		if err := client.Auth(auth); err != nil {
+			return mapSMTPErr(err)
+		}
+	}
+	if err := client.Mail(sender); err != nil {
+		return mapSMTPErr(err)
+	}
+	for _, rcpt := range recipients {
+		if err := client.Rcpt(rcpt); err != nil {
+			return mapSMTPErr(err)
+		}
+	}
+	w, err := client.Data()
+	if err != nil {
+		return mapSMTPErr(err)
+	}
+	if _, err := w.Write(raw); err != nil {
+		return mapSMTPErr(err)
+	}
+	if err := w.Close(); err != nil {
+		return mapSMTPErr(err)
+	}
+	return client.Quit()
+}
+
+// dial establishes the SMTP connection and performs the TLS handshake or
+// STARTTLS upgrade per the configured mode.
+func (b *SMTPBackend) dial(ctx context.Context) (*smtp.Client, error) {
+	addr := net.JoinHostPort(b.host, fmt.Sprint(b.port))
+	var dialer net.Dialer
+
+	if b.mode == modeTLS {
+		conn, err := tls.DialWithDialer(&dialer, "tcp", addr, &tls.Config{ServerName: b.host})
+		if err != nil {
+			return nil, err
+		}
+		return smtp.NewClient(conn, b.host)
+	}
+
+	conn, err := dialer.DialContext(ctx, "tcp", addr)
+	if err != nil {
+		return nil, err
+	}
+	client, err := smtp.NewClient(conn, b.host)
+	if err != nil {
+		conn.Close()
+		return nil, err
+	}
+	if b.mode == modeStartTLS {
+		if err := client.StartTLS(&tls.Config{ServerName: b.host}); err != nil {
+			client.Close()
+			return nil, err
+		}
+	}
+	return client, nil
+}
+
+func (b *SMTPBackend) HealthCheck(ctx context.Context) bool {
+	client, err := b.dial(ctx)
+	if err != nil {
+		return false
+	}
+	defer client.Close()
+	if err := client.Noop(); err != nil {
+		return false
+	}
+	return client.Quit() == nil
+}
+
+// Close is a no-op: SMTPBackend opens a fresh connection per Send and holds no
+// long-lived sockets.
+func (b *SMTPBackend) Close(ctx context.Context) error { return nil }
+
+// newMessageID generates an RFC 5322 Message-ID using the sender's domain (or
+// "localhost" when it cannot be derived).
+func newMessageID(sender string) string {
+	domain := "localhost"
+	if i := strings.LastIndex(sender, "@"); i >= 0 && i < len(sender)-1 {
+		domain = sender[i+1:]
+		domain = strings.Trim(domain, "<> ")
+	}
+	var rnd [16]byte
+	_, _ = rand.Read(rnd[:])
+	return fmt.Sprintf("<%s@%s>", hex.EncodeToString(rnd[:]), domain)
+}
+
+func mapSMTPErr(err error) error {
+	// net/smtp surfaces SMTP protocol replies as *textproto.Error.
+	var protoErr *textproto.Error
+	if errors.As(err, &protoErr) {
+		switch protoErr.Code {
+		case 421, 450, 451, 452:
+			return fmt.Errorf("%w: %w", core.ErrEmailThrottled, err)
+		case 550, 551, 553:
+			return fmt.Errorf("%w: %w", core.ErrRecipientRejected, err)
+		}
+	}
+	return fmt.Errorf("%w: %w", core.ErrEmailSend, err)
+}
diff --git a/go.mod b/go.mod
index bb0e0b6..6c480ef 100644
--- a/go.mod
+++ b/go.mod
@@ -15,6 +15,7 @@ require (
 	github.com/aws/aws-sdk-go-v2/credentials v1.19.23
 	github.com/aws/aws-sdk-go-v2/service/s3 v1.103.3
 	github.com/aws/aws-sdk-go-v2/service/secretsmanager v1.42.3
+	github.com/aws/aws-sdk-go-v2/service/sesv2 v1.62.4
 	github.com/aws/aws-sdk-go-v2/service/sns v1.40.1
 	github.com/aws/aws-sdk-go-v2/service/sqs v1.44.0
 	github.com/aws/smithy-go v1.27.2
diff --git a/go.sum b/go.sum
index 5c10730..c62f3c1 100644
--- a/go.sum
+++ b/go.sum
@@ -54,6 +54,8 @@ github.com/aws/aws-sdk-go-v2/service/s3 v1.103.3 h1:JRseEu/vIDMaWis4bSw0QbXL+cvI
 github.com/aws/aws-sdk-go-v2/service/s3 v1.103.3/go.mod h1:77ZAgynvx1txMvDG8gGWoWkO1augYDxkp9JElWFgjQU=
 github.com/aws/aws-sdk-go-v2/service/secretsmanager v1.42.3 h1:L9gPLf3sFH1/ao3oB2QZcaX1xGYi8hj+WJlsf3/dN+M=
 github.com/aws/aws-sdk-go-v2/service/secretsmanager v1.42.3/go.mod h1:9DKRlwDCw2OUDlyCIFcQCroL5M0mQTUU9qW8JEDcXmI=
+github.com/aws/aws-sdk-go-v2/service/sesv2 v1.62.4 h1:E+hfJiJ/ZAx5SmnpDIejKpgqpd0TqoWqcmBGmLhe7PU=
+github.com/aws/aws-sdk-go-v2/service/sesv2 v1.62.4/go.mod h1:pPfcguG4y24CV1iQeejOJ8MRkTMFqHtxc8/Lmklv4gY=
 github.com/aws/aws-sdk-go-v2/service/signin v1.1.5 h1:6Xt6Ztjkwdia/7EtEaG7ki/qZUYlCcd7tGUotQed1QE=
 github.com/aws/aws-sdk-go-v2/service/signin v1.1.5/go.mod h1:LxYujSTLPRlp2vTtcUO/+1ilrew8ytt6SvQyOgejzFQ=
 github.com/aws/aws-sdk-go-v2/service/sns v1.40.1 h1:DLrOlgom0+OYnNaSiVxAXtR6obPjVlbmD+7w5wim9sc=
diff --git a/messaging/azurebus.go b/messaging/azurebus.go
index 063be2b..53becd5 100644
--- a/messaging/azurebus.go
+++ b/messaging/azurebus.go
@@ -3,7 +3,6 @@ package messaging
 import (
 	"context"
 	"encoding/hex"
-	"encoding/json"
 	"errors"
 	"fmt"
 	"sync"
@@ -97,11 +96,7 @@ func NewAzureServiceBus(cfg Config) (*AzureServiceBusBackend, error) {
 	return b, nil
 }
 
-func (b *AzureServiceBusBackend) Send(ctx context.Context, message map[string]any, delay time.Duration) (string, error) {
-	body, err := json.Marshal(message)
-	if err != nil {
-		return "", fmt.Errorf("%w: %w", core.ErrMessageSend, err)
-	}
+func (b *AzureServiceBusBackend) Send(ctx context.Context, body []byte, attributes map[string]string, delay time.Duration) (string, error) {
 	sender, err := b.client.NewSender(b.queueName, nil)
 	if err != nil {
 		return "", b.mapErr(err, core.ErrMessageSend)
@@ -109,7 +104,7 @@ func (b *AzureServiceBusBackend) Send(ctx context.Context, message map[string]an
 	defer sender.Close(ctx)
 
 	id := newID()
-	msg := &azservicebus.Message{Body: body, MessageID: &id}
+	msg := &azservicebus.Message{Body: body, MessageID: &id, ApplicationProperties: appProps(attributes)}
 	if delay > 0 {
 		t := time.Now().UTC().Add(delay)
 		msg.ScheduledEnqueueTime = &t
@@ -120,7 +115,7 @@ func (b *AzureServiceBusBackend) Send(ctx context.Context, message map[string]an
 	return id, nil
 }
 
-func (b *AzureServiceBusBackend) SendBatch(ctx context.Context, messages []map[string]any) ([]string, error) {
+func (b *AzureServiceBusBackend) SendBatch(ctx context.Context, messages []OutgoingMessage) ([]string, error) {
 	sender, err := b.client.NewSender(b.queueName, nil)
 	if err != nil {
 		return nil, b.mapErr(err, core.ErrMessageSend)
@@ -133,13 +128,9 @@ func (b *AzureServiceBusBackend) SendBatch(ctx context.Context, messages []map[s
 	}
 	ids := make([]string, 0, len(messages))
 	for _, m := range messages {
-		body, err := json.Marshal(m)
-		if err != nil {
-			return nil, fmt.Errorf("%w: %w", core.ErrMessageSend, err)
-		}
 		id := newID()
 		ids = append(ids, id)
-		if err := batch.AddMessage(&azservicebus.Message{Body: body, MessageID: &id}, nil); err != nil {
+		if err := batch.AddMessage(&azservicebus.Message{Body: m.Body, MessageID: &id, ApplicationProperties: appProps(m.Attributes)}, nil); err != nil {
 			return nil, fmt.Errorf("%w: %w", core.ErrMessageSend, err)
 		}
 	}
@@ -149,6 +140,19 @@ func (b *AzureServiceBusBackend) SendBatch(ctx context.Context, messages []map[s
 	return ids, nil
 }
 
+// appProps converts a string attribute map into Service Bus
+// ApplicationProperties. Returns nil for an empty map.
+func appProps(attributes map[string]string) map[string]any {
+	if len(attributes) == 0 {
+		return nil
+	}
+	out := make(map[string]any, len(attributes))
+	for k, v := range attributes {
+		out[k] = v
+	}
+	return out
+}
+
 func (b *AzureServiceBusBackend) Receive(ctx context.Context, maxMessages int, waitTime time.Duration) ([]Message, error) {
 	receiver, err := b.client.NewReceiverForQueue(b.queueName, nil)
 	if err != nil {
@@ -180,10 +184,6 @@ func (b *AzureServiceBusBackend) Receive(ctx context.Context, maxMessages int, w
 		b.pending[token] = &sbPending{receiver: receiver, message: m}
 		tokens[token] = struct{}{}
 
-		var body map[string]any
-		if err := json.Unmarshal(m.Body, &body); err != nil {
-			return nil, fmt.Errorf("%w: decoding message body: %w", core.ErrMessaging, err)
-		}
 		attrs := map[string]string{}
 		if m.SequenceNumber != nil {
 			attrs["sequence_number"] = fmt.Sprint(*m.SequenceNumber)
@@ -191,9 +191,21 @@ func (b *AzureServiceBusBackend) Receive(ctx context.Context, maxMessages int, w
 		if m.EnqueuedTime != nil {
 			attrs["enqueued_time"] = m.EnqueuedTime.String()
 		}
+		// User attributes (ApplicationProperties) take priority over the
+		// system metadata above.
+		for k, v := range m.ApplicationProperties {
+			if v == nil {
+				continue
+			}
+			if s, ok := v.(string); ok {
+				attrs[k] = s
+			} else {
+				attrs[k] = fmt.Sprint(v)
+			}
+		}
 		messages = append(messages, Message{
 			ID:            m.MessageID,
-			Body:          body,
+			Body:          m.Body,
 			ReceiptHandle: token,
 			Attributes:    attrs,
 		})
diff --git a/messaging/messaging.go b/messaging/messaging.go
index 2639fb6..69b2306 100644
--- a/messaging/messaging.go
+++ b/messaging/messaging.go
@@ -10,6 +10,7 @@ import (
 	"context"
 	"crypto/rand"
 	"encoding/hex"
+	"encoding/json"
 	"fmt"
 	"time"
 
@@ -19,20 +20,43 @@ import (
 // Message is a received queue message.
 type Message struct {
 	ID string
-	// Body is the JSON-decoded message payload.
-	Body map[string]any
+	// Body is the raw message payload bytes (binary-safe). Use Message.JSON()
+	// to decode a JSON object body.
+	Body []byte
 	// ReceiptHandle acknowledges the message via Delete or DeadLetter.
 	ReceiptHandle string
-	Attributes    map[string]string
+	// Attributes are the message's user attributes — SQS MessageAttributes /
+	// Azure Service Bus ApplicationProperties — stringified.
+	Attributes map[string]string
 }
 
-// Backend is the provider-neutral queue interface.
+// JSON decodes the message body as a JSON object. Convenience for the common
+// case where the body is JSON (e.g. produced via SendJSON).
+func (m Message) JSON() (map[string]any, error) {
+	var out map[string]any
+	if err := json.Unmarshal(m.Body, &out); err != nil {
+		return nil, fmt.Errorf("%w: decoding message body: %w", core.ErrMessaging, err)
+	}
+	return out, nil
+}
+
+// OutgoingMessage is one message for SendBatch: raw body bytes plus optional
+// user attributes.
+type OutgoingMessage struct {
+	Body       []byte
+	Attributes map[string]string
+}
+
+// Backend is the provider-neutral queue interface. Bodies are raw bytes
+// (binary-safe) so any payload — JSON, OTLP protobuf, base64 — round-trips
+// unchanged. Use SendJSON / Message.JSON for the JSON-object convenience path.
 type Backend interface {
-	// Send sends a message (JSON-encoded). delay postpones visibility.
-	// Returns the message ID.
-	Send(ctx context.Context, message map[string]any, delay time.Duration) (string, error)
+	// Send sends a raw message body with optional user attributes (SQS
+	// MessageAttributes / Service Bus ApplicationProperties). delay postpones
+	// visibility. Returns the message ID.
+	Send(ctx context.Context, body []byte, attributes map[string]string, delay time.Duration) (string, error)
 	// SendBatch sends multiple messages. Returns their message IDs.
-	SendBatch(ctx context.Context, messages []map[string]any) ([]string, error)
+	SendBatch(ctx context.Context, messages []OutgoingMessage) ([]string, error)
 	// Receive fetches up to maxMessages, long-polling up to waitTime.
 	Receive(ctx context.Context, maxMessages int, waitTime time.Duration) ([]Message, error)
 	// Delete acknowledges a message by its receipt handle.
@@ -66,6 +90,11 @@ type Config struct {
 	AWSSessionToken    string
 	ProfileName        string
 	EndpointURL        string // custom endpoint (LocalStack, ...)
+	// RoleARN, when set, is assumed via STS on top of the base credentials
+	// (static keys / profile / default chain). ExternalID is passed in the
+	// AssumeRole call when set. Enables cross-account SQS access.
+	RoleARN    string
+	ExternalID string
 	// DLQURL is the dead-letter queue URL for SQS DeadLetter emulation. If
 	// empty it is resolved lazily from the source queue's RedrivePolicy.
 	DLQURL string
@@ -100,6 +129,16 @@ func New(ctx context.Context, provider string, cfg Config) (Backend, error) {
 		core.ErrMessaging, provider)
 }
 
+// SendJSON is a thin convenience over Send for the common case of a JSON-object
+// payload: it marshals message and sends it with the given attributes/delay.
+func SendJSON(ctx context.Context, b Backend, message map[string]any, attributes map[string]string, delay time.Duration) (string, error) {
+	body, err := json.Marshal(message)
+	if err != nil {
+		return "", fmt.Errorf("%w: %w", core.ErrMessageSend, err)
+	}
+	return b.Send(ctx, body, attributes, delay)
+}
+
 // newID returns a random RFC 4122 v4 UUID string (used where the provider SDK
 // does not assign client-side message IDs).
 func newID() string {
diff --git a/messaging/sqs.go b/messaging/sqs.go
index c02096f..6f8d3f4 100644
--- a/messaging/sqs.go
+++ b/messaging/sqs.go
@@ -13,8 +13,10 @@ import (
 	"github.com/aws/aws-sdk-go-v2/aws"
 	awsconfig "github.com/aws/aws-sdk-go-v2/config"
 	"github.com/aws/aws-sdk-go-v2/credentials"
+	"github.com/aws/aws-sdk-go-v2/credentials/stscreds"
 	"github.com/aws/aws-sdk-go-v2/service/sqs"
 	"github.com/aws/aws-sdk-go-v2/service/sqs/types"
+	"github.com/aws/aws-sdk-go-v2/service/sts"
 	"github.com/aws/smithy-go"
 
 	"github.com/LYZR-OSS/cloudrift-go/core"
@@ -60,6 +62,14 @@ func NewSQS(ctx context.Context, cfg Config) (*AWSSQSBackend, error) {
 	if err != nil {
 		return nil, fmt.Errorf("%w: %w", core.ErrMessaging, err)
 	}
+	if cfg.RoleARN != "" {
+		awsCfg.Credentials = aws.NewCredentialsCache(stscreds.NewAssumeRoleProvider(
+			sts.NewFromConfig(awsCfg), cfg.RoleARN, func(o *stscreds.AssumeRoleOptions) {
+				if cfg.ExternalID != "" {
+					o.ExternalID = aws.String(cfg.ExternalID)
+				}
+			}))
+	}
 	client := sqs.NewFromConfig(awsCfg, func(o *sqs.Options) {
 		if cfg.EndpointURL != "" {
 			o.BaseEndpoint = aws.String(cfg.EndpointURL)
@@ -73,15 +83,12 @@ func NewSQS(ctx context.Context, cfg Config) (*AWSSQSBackend, error) {
 	}, nil
 }
 
-func (b *AWSSQSBackend) Send(ctx context.Context, message map[string]any, delay time.Duration) (string, error) {
-	body, err := json.Marshal(message)
-	if err != nil {
-		return "", fmt.Errorf("%w: %w", core.ErrMessageSend, err)
-	}
+func (b *AWSSQSBackend) Send(ctx context.Context, body []byte, attributes map[string]string, delay time.Duration) (string, error) {
 	resp, err := b.client.SendMessage(ctx, &sqs.SendMessageInput{
-		QueueUrl:     aws.String(b.queueURL),
-		MessageBody:  aws.String(string(body)),
-		DelaySeconds: int32(delay / time.Second),
+		QueueUrl:          aws.String(b.queueURL),
+		MessageBody:       aws.String(string(body)),
+		DelaySeconds:      int32(delay / time.Second),
+		MessageAttributes: sqsAttrs(attributes),
 	})
 	if err != nil {
 		return "", b.mapErr(err)
@@ -89,16 +96,13 @@ func (b *AWSSQSBackend) Send(ctx context.Context, message map[string]any, delay
 	return aws.ToString(resp.MessageId), nil
 }
 
-func (b *AWSSQSBackend) SendBatch(ctx context.Context, messages []map[string]any) ([]string, error) {
+func (b *AWSSQSBackend) SendBatch(ctx context.Context, messages []OutgoingMessage) ([]string, error) {
 	entries := make([]types.SendMessageBatchRequestEntry, 0, len(messages))
 	for i, msg := range messages {
-		body, err := json.Marshal(msg)
-		if err != nil {
-			return nil, fmt.Errorf("%w: %w", core.ErrMessageSend, err)
-		}
 		entries = append(entries, types.SendMessageBatchRequestEntry{
-			Id:          aws.String(strconv.Itoa(i)),
-			MessageBody: aws.String(string(body)),
+			Id:                aws.String(strconv.Itoa(i)),
+			MessageBody:       aws.String(string(msg.Body)),
+			MessageAttributes: sqsAttrs(msg.Attributes),
 		})
 	}
 	resp, err := b.client.SendMessageBatch(ctx, &sqs.SendMessageBatchInput{
@@ -128,6 +132,7 @@ func (b *AWSSQSBackend) Receive(ctx context.Context, maxMessages int, waitTime t
 		MaxNumberOfMessages:         int32(min(maxMessages, 10)),
 		WaitTimeSeconds:             int32(waitTime / time.Second),
 		MessageSystemAttributeNames: []types.MessageSystemAttributeName{"All"},
+		MessageAttributeNames:       []string{"All"},
 	})
 	if err != nil {
 		return nil, b.mapErr(err)
@@ -138,17 +143,18 @@ func (b *AWSSQSBackend) Receive(ctx context.Context, maxMessages int, waitTime t
 	for _, m := range resp.Messages {
 		rawBody := aws.ToString(m.Body)
 		b.pending[aws.ToString(m.ReceiptHandle)] = rawBody
-		var body map[string]any
-		if err := json.Unmarshal([]byte(rawBody), &body); err != nil {
-			return nil, fmt.Errorf("%w: decoding message body: %w", core.ErrMessaging, err)
-		}
-		attrs := make(map[string]string, len(m.Attributes))
+		// User attributes (MessageAttributes) take priority; system attributes
+		// (SentTimestamp, etc.) are merged underneath.
+		attrs := make(map[string]string, len(m.Attributes)+len(m.MessageAttributes))
 		for k, v := range m.Attributes {
 			attrs[k] = v
 		}
+		for k, v := range m.MessageAttributes {
+			attrs[k] = aws.ToString(v.StringValue)
+		}
 		messages = append(messages, Message{
 			ID:            aws.ToString(m.MessageId),
-			Body:          body,
+			Body:          []byte(rawBody),
 			ReceiptHandle: aws.ToString(m.ReceiptHandle),
 			Attributes:    attrs,
 		})
@@ -156,6 +162,22 @@ func (b *AWSSQSBackend) Receive(ctx context.Context, maxMessages int, waitTime t
 	return messages, nil
 }
 
+// sqsAttrs converts a string attribute map into SQS MessageAttributeValues
+// (all String type). Returns nil for an empty map so SDK validation is skipped.
+func sqsAttrs(attributes map[string]string) map[string]types.MessageAttributeValue {
+	if len(attributes) == 0 {
+		return nil
+	}
+	out := make(map[string]types.MessageAttributeValue, len(attributes))
+	for k, v := range attributes {
+		out[k] = types.MessageAttributeValue{
+			DataType:    aws.String("String"),
+			StringValue: aws.String(v),
+		}
+	}
+	return out
+}
+
 func (b *AWSSQSBackend) Delete(ctx context.Context, receiptHandle string) error {
 	_, err := b.client.DeleteMessage(ctx, &sqs.DeleteMessageInput{
 		QueueUrl:      aws.String(b.queueURL),
diff --git a/secrets/local.go b/secrets/local.go
new file mode 100644
index 0000000..a6fd26c
--- /dev/null
+++ b/secrets/local.go
@@ -0,0 +1,258 @@
+package secrets
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"os"
+	"strings"
+
+	"github.com/LYZR-OSS/cloudrift-go/core"
+)
+
+// EnvSecretBackend reads and writes secrets from process environment variables.
+//
+// A secret named "db" maps to the environment variable "{prefix}db" — the
+// prefix lets you namespace secrets, e.g. "SECRET_". This backend is useful for
+// local development, containers, and CI where secrets arrive as env vars.
+type EnvSecretBackend struct {
+	prefix string
+}
+
+var _ Backend = (*EnvSecretBackend)(nil)
+
+// NewEnvSecrets constructs an environment-variable secret backend. prefix
+// namespaces secret names onto env vars; pass "" for no namespace.
+func NewEnvSecrets(prefix string) *EnvSecretBackend {
+	return &EnvSecretBackend{prefix: prefix}
+}
+
+func (b *EnvSecretBackend) key(name string) string {
+	return b.prefix + name
+}
+
+func (b *EnvSecretBackend) GetSecret(_ context.Context, name string) (string, error) {
+	v, ok := os.LookupEnv(b.key(name))
+	if !ok {
+		return "", fmt.Errorf("%w: %s: not found in environment", core.ErrSecretNotFound, name)
+	}
+	return v, nil
+}
+
+func (b *EnvSecretBackend) GetSecretJSON(ctx context.Context, name string) (map[string]any, error) {
+	return getSecretJSON(ctx, b, name)
+}
+
+func (b *EnvSecretBackend) SetSecret(_ context.Context, name, value string) error {
+	if err := os.Setenv(b.key(name), value); err != nil {
+		return fmt.Errorf("%w: %s: %w", core.ErrSecret, name, err)
+	}
+	return nil
+}
+
+func (b *EnvSecretBackend) DeleteSecret(_ context.Context, name string) error {
+	if err := os.Unsetenv(b.key(name)); err != nil {
+		return fmt.Errorf("%w: %s: %w", core.ErrSecret, name, err)
+	}
+	return nil
+}
+
+// ListSecrets scans the environment for variables under the constructor prefix,
+// strips that prefix, and returns names further filtered by the prefix arg.
+func (b *EnvSecretBackend) ListSecrets(_ context.Context, prefix string) ([]string, error) {
+	var names []string
+	for _, kv := range os.Environ() {
+		k, _, _ := strings.Cut(kv, "=")
+		if !strings.HasPrefix(k, b.prefix) {
+			continue
+		}
+		name := k[len(b.prefix):]
+		if strings.HasPrefix(name, prefix) {
+			names = append(names, name)
+		}
+	}
+	return names, nil
+}
+
+func (b *EnvSecretBackend) HealthCheck(_ context.Context) bool { return true }
+
+// Close is a no-op.
+func (b *EnvSecretBackend) Close(_ context.Context) error { return nil }
+
+// MappingSecretBackend holds secrets in an in-memory map. Useful for tests and
+// dev seeding; nothing is persisted.
+type MappingSecretBackend struct {
+	store map[string]string
+}
+
+var _ Backend = (*MappingSecretBackend)(nil)
+
+// NewMappingSecrets constructs an in-memory secret backend, copying the given
+// mapping (nil is fine).
+func NewMappingSecrets(mapping map[string]string) *MappingSecretBackend {
+	store := make(map[string]string, len(mapping))
+	for k, v := range mapping {
+		store[k] = v
+	}
+	return &MappingSecretBackend{store: store}
+}
+
+func (b *MappingSecretBackend) GetSecret(_ context.Context, name string) (string, error) {
+	v, ok := b.store[name]
+	if !ok {
+		return "", fmt.Errorf("%w: %s", core.ErrSecretNotFound, name)
+	}
+	return v, nil
+}
+
+func (b *MappingSecretBackend) GetSecretJSON(ctx context.Context, name string) (map[string]any, error) {
+	return getSecretJSON(ctx, b, name)
+}
+
+func (b *MappingSecretBackend) SetSecret(_ context.Context, name, value string) error {
+	b.store[name] = value
+	return nil
+}
+
+func (b *MappingSecretBackend) DeleteSecret(_ context.Context, name string) error {
+	delete(b.store, name)
+	return nil
+}
+
+func (b *MappingSecretBackend) ListSecrets(_ context.Context, prefix string) ([]string, error) {
+	var names []string
+	for n := range b.store {
+		if strings.HasPrefix(n, prefix) {
+			names = append(names, n)
+		}
+	}
+	return names, nil
+}
+
+func (b *MappingSecretBackend) HealthCheck(_ context.Context) bool { return true }
+
+// Close is a no-op.
+func (b *MappingSecretBackend) Close(_ context.Context) error { return nil }
+
+// FileSecretBackend persists secrets in a JSON file mapping name → value. Writes
+// are atomic via a temp file plus os.Rename. The on-disk format is a single JSON
+// object of string values; store structured data by serializing it to a string.
+type FileSecretBackend struct {
+	path string
+}
+
+var _ Backend = (*FileSecretBackend)(nil)
+
+// NewFileSecrets constructs a file-backed secret store at path. The file is
+// created on first write; a missing file reads as empty.
+func NewFileSecrets(path string) *FileSecretBackend {
+	return &FileSecretBackend{path: path}
+}
+
+// load reads the backing file. A missing file yields an empty map; an
+// unreadable file or a non-object payload is wrapped as core.ErrSecret.
+func (b *FileSecretBackend) load() (map[string]string, error) {
+	raw, err := os.ReadFile(b.path)
+	if err != nil {
+		if os.IsNotExist(err) {
+			return map[string]string{}, nil
+		}
+		return nil, fmt.Errorf("%w: secret file %q is unreadable: %w", core.ErrSecret, b.path, err)
+	}
+	var data map[string]string
+	if err := json.Unmarshal(raw, &data); err != nil {
+		return nil, fmt.Errorf("%w: secret file %q must contain a JSON object: %w", core.ErrSecret, b.path, err)
+	}
+	if data == nil {
+		data = map[string]string{}
+	}
+	return data, nil
+}
+
+// save atomically writes data via a temp file plus os.Rename.
+func (b *FileSecretBackend) save(data map[string]string) error {
+	raw, err := json.Marshal(data)
+	if err != nil {
+		return fmt.Errorf("%w: %w", core.ErrSecret, err)
+	}
+	tmp := b.path + ".tmp"
+	if err := os.WriteFile(tmp, raw, 0o600); err != nil {
+		return fmt.Errorf("%w: %w", core.ErrSecret, err)
+	}
+	if err := os.Rename(tmp, b.path); err != nil {
+		return fmt.Errorf("%w: %w", core.ErrSecret, err)
+	}
+	return nil
+}
+
+func (b *FileSecretBackend) GetSecret(_ context.Context, name string) (string, error) {
+	data, err := b.load()
+	if err != nil {
+		return "", err
+	}
+	v, ok := data[name]
+	if !ok {
+		return "", fmt.Errorf("%w: %s: not found in %s", core.ErrSecretNotFound, name, b.path)
+	}
+	return v, nil
+}
+
+func (b *FileSecretBackend) GetSecretJSON(ctx context.Context, name string) (map[string]any, error) {
+	return getSecretJSON(ctx, b, name)
+}
+
+func (b *FileSecretBackend) SetSecret(_ context.Context, name, value string) error {
+	data, err := b.load()
+	if err != nil {
+		return err
+	}
+	data[name] = value
+	return b.save(data)
+}
+
+func (b *FileSecretBackend) DeleteSecret(_ context.Context, name string) error {
+	data, err := b.load()
+	if err != nil {
+		return err
+	}
+	delete(data, name)
+	return b.save(data)
+}
+
+func (b *FileSecretBackend) ListSecrets(_ context.Context, prefix string) ([]string, error) {
+	data, err := b.load()
+	if err != nil {
+		return nil, err
+	}
+	var names []string
+	for n := range data {
+		if strings.HasPrefix(n, prefix) {
+			names = append(names, n)
+		}
+	}
+	return names, nil
+}
+
+// HealthCheck returns true if the backing file loads without error (a missing
+// file is healthy — it just hasn't been written yet).
+func (b *FileSecretBackend) HealthCheck(_ context.Context) bool {
+	_, err := b.load()
+	return err == nil
+}
+
+// Close is a no-op.
+func (b *FileSecretBackend) Close(_ context.Context) error { return nil }
+
+// getSecretJSON fetches a secret via the backend and parses its value as a JSON
+// object. Invalid JSON is wrapped as core.ErrSecret.
+func getSecretJSON(ctx context.Context, b Backend, name string) (map[string]any, error) {
+	raw, err := b.GetSecret(ctx, name)
+	if err != nil {
+		return nil, err
+	}
+	var out map[string]any
+	if err := json.Unmarshal([]byte(raw), &out); err != nil {
+		return nil, fmt.Errorf("%w: secret %q is not valid JSON: %w", core.ErrSecret, name, err)
+	}
+	return out, nil
+}
diff --git a/secrets/local_test.go b/secrets/local_test.go
new file mode 100644
index 0000000..04d0dcb
--- /dev/null
+++ b/secrets/local_test.go
@@ -0,0 +1,287 @@
+package secrets
+
+import (
+	"context"
+	"errors"
+	"os"
+	"path/filepath"
+	"sort"
+	"testing"
+
+	"github.com/LYZR-OSS/cloudrift-go/core"
+)
+
+func TestEnvSecretBackend(t *testing.T) {
+	ctx := context.Background()
+	// Isolate env mutations to this test.
+	t.Setenv("SECRET_db", "")
+	os.Unsetenv("SECRET_db")
+
+	b := NewEnvSecrets("SECRET_")
+
+	if err := b.SetSecret(ctx, "db", "postgres://x"); err != nil {
+		t.Fatalf("SetSecret: %v", err)
+	}
+	got, err := b.GetSecret(ctx, "db")
+	if err != nil {
+		t.Fatalf("GetSecret: %v", err)
+	}
+	if got != "postgres://x" {
+		t.Fatalf("GetSecret = %q; want %q", got, "postgres://x")
+	}
+	// Confirm the env var is namespaced by the constructor prefix.
+	if os.Getenv("SECRET_db") != "postgres://x" {
+		t.Fatalf("env SECRET_db = %q; want %q", os.Getenv("SECRET_db"), "postgres://x")
+	}
+
+	if err := b.SetSecret(ctx, "api_key", "k"); err != nil {
+		t.Fatalf("SetSecret: %v", err)
+	}
+
+	// ListSecrets("") returns all names under the constructor prefix, stripped.
+	names, err := b.ListSecrets(ctx, "")
+	if err != nil {
+		t.Fatalf("ListSecrets: %v", err)
+	}
+	if !contains(names, "db") || !contains(names, "api_key") {
+		t.Fatalf("ListSecrets = %v; want db and api_key", names)
+	}
+
+	// ListSecrets with a filter prefix narrows further.
+	filtered, err := b.ListSecrets(ctx, "api")
+	if err != nil {
+		t.Fatalf("ListSecrets: %v", err)
+	}
+	if len(filtered) != 1 || filtered[0] != "api_key" {
+		t.Fatalf("ListSecrets(api) = %v; want [api_key]", filtered)
+	}
+
+	if err := b.DeleteSecret(ctx, "db"); err != nil {
+		t.Fatalf("DeleteSecret: %v", err)
+	}
+	if _, err := b.GetSecret(ctx, "db"); !errors.Is(err, core.ErrSecretNotFound) {
+		t.Fatalf("GetSecret after delete err = %v; want ErrSecretNotFound", err)
+	}
+
+	if _, err := b.GetSecret(ctx, "missing"); !errors.Is(err, core.ErrSecretNotFound) {
+		t.Fatalf("GetSecret missing err = %v; want ErrSecretNotFound", err)
+	}
+
+	if !b.HealthCheck(ctx) {
+		t.Fatal("HealthCheck = false; want true")
+	}
+	if err := b.Close(ctx); err != nil {
+		t.Fatalf("Close: %v", err)
+	}
+}
+
+func TestMappingSecretBackend(t *testing.T) {
+	ctx := context.Background()
+
+	// Constructor copies the input map (mutating the source must not leak).
+	seed := map[string]string{"db": "v1"}
+	b := NewMappingSecrets(seed)
+	seed["db"] = "mutated"
+
+	got, err := b.GetSecret(ctx, "db")
+	if err != nil {
+		t.Fatalf("GetSecret: %v", err)
+	}
+	if got != "v1" {
+		t.Fatalf("GetSecret = %q; want v1 (input map should have been copied)", got)
+	}
+
+	if err := b.SetSecret(ctx, "api", "k"); err != nil {
+		t.Fatalf("SetSecret: %v", err)
+	}
+	names, err := b.ListSecrets(ctx, "")
+	if err != nil {
+		t.Fatalf("ListSecrets: %v", err)
+	}
+	sort.Strings(names)
+	if len(names) != 2 || names[0] != "api" || names[1] != "db" {
+		t.Fatalf("ListSecrets = %v; want [api db]", names)
+	}
+
+	if err := b.DeleteSecret(ctx, "db"); err != nil {
+		t.Fatalf("DeleteSecret: %v", err)
+	}
+	if _, err := b.GetSecret(ctx, "db"); !errors.Is(err, core.ErrSecretNotFound) {
+		t.Fatalf("GetSecret after delete err = %v; want ErrSecretNotFound", err)
+	}
+
+	// GetSecretJSON happy path.
+	if err := b.SetSecret(ctx, "cfg", `{"a":1,"b":"x"}`); err != nil {
+		t.Fatalf("SetSecret: %v", err)
+	}
+	m, err := b.GetSecretJSON(ctx, "cfg")
+	if err != nil {
+		t.Fatalf("GetSecretJSON: %v", err)
+	}
+	if m["b"] != "x" {
+		t.Fatalf("GetSecretJSON[b] = %v; want x", m["b"])
+	}
+
+	// GetSecretJSON invalid JSON.
+	if err := b.SetSecret(ctx, "bad", "not json"); err != nil {
+		t.Fatalf("SetSecret: %v", err)
+	}
+	if _, err := b.GetSecretJSON(ctx, "bad"); !errors.Is(err, core.ErrSecret) {
+		t.Fatalf("GetSecretJSON invalid err = %v; want ErrSecret", err)
+	}
+
+	// GetSecretJSON missing.
+	if _, err := b.GetSecretJSON(ctx, "missing"); !errors.Is(err, core.ErrSecretNotFound) {
+		t.Fatalf("GetSecretJSON missing err = %v; want ErrSecretNotFound", err)
+	}
+
+	// Nil mapping is fine.
+	nb := NewMappingSecrets(nil)
+	if err := nb.SetSecret(ctx, "x", "y"); err != nil {
+		t.Fatalf("SetSecret on nil-seeded backend: %v", err)
+	}
+
+	if !b.HealthCheck(ctx) {
+		t.Fatal("HealthCheck = false; want true")
+	}
+	if err := b.Close(ctx); err != nil {
+		t.Fatalf("Close: %v", err)
+	}
+}
+
+func TestFileSecretBackend(t *testing.T) {
+	ctx := context.Background()
+	path := filepath.Join(t.TempDir(), "secrets.json")
+
+	b := NewFileSecrets(path)
+
+	// HealthCheck is true even before the file exists.
+	if !b.HealthCheck(ctx) {
+		t.Fatal("HealthCheck on missing file = false; want true")
+	}
+
+	// Set → get round trip.
+	if err := b.SetSecret(ctx, "db", "postgres://x"); err != nil {
+		t.Fatalf("SetSecret: %v", err)
+	}
+	got, err := b.GetSecret(ctx, "db")
+	if err != nil {
+		t.Fatalf("GetSecret: %v", err)
+	}
+	if got != "postgres://x" {
+		t.Fatalf("GetSecret = %q; want postgres://x", got)
+	}
+
+	if err := b.SetSecret(ctx, "api", "k"); err != nil {
+		t.Fatalf("SetSecret: %v", err)
+	}
+
+	// Persistence: a fresh backend instance over the same path sees the value.
+	b2 := NewFileSecrets(path)
+	got, err = b2.GetSecret(ctx, "db")
+	if err != nil {
+		t.Fatalf("GetSecret (fresh instance): %v", err)
+	}
+	if got != "postgres://x" {
+		t.Fatalf("GetSecret (fresh instance) = %q; want postgres://x", got)
+	}
+
+	// List with prefix.
+	names, err := b2.ListSecrets(ctx, "ap")
+	if err != nil {
+		t.Fatalf("ListSecrets: %v", err)
+	}
+	if len(names) != 1 || names[0] != "api" {
+		t.Fatalf("ListSecrets(ap) = %v; want [api]", names)
+	}
+
+	// Delete.
+	if err := b.DeleteSecret(ctx, "db"); err != nil {
+		t.Fatalf("DeleteSecret: %v", err)
+	}
+	if _, err := b.GetSecret(ctx, "db"); !errors.Is(err, core.ErrSecretNotFound) {
+		t.Fatalf("GetSecret after delete err = %v; want ErrSecretNotFound", err)
+	}
+
+	// Missing key.
+	if _, err := b.GetSecret(ctx, "nope"); !errors.Is(err, core.ErrSecretNotFound) {
+		t.Fatalf("GetSecret missing err = %v; want ErrSecretNotFound", err)
+	}
+
+	// Corrupt file content → ErrSecret.
+	corrupt := filepath.Join(t.TempDir(), "corrupt.json")
+	if err := os.WriteFile(corrupt, []byte("not json"), 0o600); err != nil {
+		t.Fatalf("write corrupt: %v", err)
+	}
+	cb := NewFileSecrets(corrupt)
+	if _, err := cb.GetSecret(ctx, "x"); !errors.Is(err, core.ErrSecret) {
+		t.Fatalf("GetSecret corrupt err = %v; want ErrSecret", err)
+	}
+	if cb.HealthCheck(ctx) {
+		t.Fatal("HealthCheck on corrupt file = true; want false")
+	}
+
+	// Non-object JSON → ErrSecret.
+	arr := filepath.Join(t.TempDir(), "arr.json")
+	if err := os.WriteFile(arr, []byte("[1,2,3]"), 0o600); err != nil {
+		t.Fatalf("write arr: %v", err)
+	}
+	ab := NewFileSecrets(arr)
+	if _, err := ab.GetSecret(ctx, "x"); !errors.Is(err, core.ErrSecret) {
+		t.Fatalf("GetSecret array-json err = %v; want ErrSecret", err)
+	}
+
+	if err := b.Close(ctx); err != nil {
+		t.Fatalf("Close: %v", err)
+	}
+}
+
+func TestNewFactoryLocalProviders(t *testing.T) {
+	ctx := context.Background()
+
+	tests := []struct {
+		provider string
+		cfg      Config
+		want     any
+	}{
+		{"env", Config{Prefix: "SECRET_"}, (*EnvSecretBackend)(nil)},
+		{"file", Config{FilePath: filepath.Join(t.TempDir(), "s.json")}, (*FileSecretBackend)(nil)},
+		{"memory", Config{Mapping: map[string]string{"a": "b"}}, (*MappingSecretBackend)(nil)},
+		{"local", Config{}, (*MappingSecretBackend)(nil)},
+	}
+	for _, tt := range tests {
+		t.Run(tt.provider, func(t *testing.T) {
+			b, err := New(ctx, tt.provider, tt.cfg)
+			if err != nil {
+				t.Fatalf("New(%q): %v", tt.provider, err)
+			}
+			switch tt.want.(type) {
+			case *EnvSecretBackend:
+				if _, ok := b.(*EnvSecretBackend); !ok {
+					t.Fatalf("New(%q) type = %T; want *EnvSecretBackend", tt.provider, b)
+				}
+			case *FileSecretBackend:
+				if _, ok := b.(*FileSecretBackend); !ok {
+					t.Fatalf("New(%q) type = %T; want *FileSecretBackend", tt.provider, b)
+				}
+			case *MappingSecretBackend:
+				if _, ok := b.(*MappingSecretBackend); !ok {
+					t.Fatalf("New(%q) type = %T; want *MappingSecretBackend", tt.provider, b)
+				}
+			}
+		})
+	}
+
+	if _, err := New(ctx, "bogus", Config{}); !errors.Is(err, core.ErrSecret) {
+		t.Fatalf("New(bogus) err = %v; want ErrSecret", err)
+	}
+}
+
+func contains(s []string, v string) bool {
+	for _, x := range s {
+		if x == v {
+			return true
+		}
+	}
+	return false
+}
diff --git a/secrets/secrets.go b/secrets/secrets.go
index 4e0c2ef..0a50ae1 100644
--- a/secrets/secrets.go
+++ b/secrets/secrets.go
@@ -1,8 +1,10 @@
 // Package secrets provides a provider-neutral interface over cloud secret
-// stores: AWS Secrets Manager and Azure Key Vault.
+// stores (AWS Secrets Manager, Azure Key Vault) plus non-cloud backends for
+// local development, self-hosted deployments, CI, and tests: environment
+// variables, a JSON file, or an in-memory mapping.
 //
-// Construct a backend once at service startup via New (or
-// NewAWSSecretsManager / NewAzureKeyVault) and reuse it.
+// Construct a backend once at service startup via New (or one of the concrete
+// constructors) and reuse it.
 package secrets
 
 import (
@@ -47,12 +49,19 @@ type Config struct {
 	TenantID     string
 	ClientID     string // service principal app ID, or user-assigned MI client ID
 	ClientSecret string
+
+	// Local (env / file / memory).
+	Prefix   string            // env: namespace prefix, secret "db" → env "{Prefix}db"
+	FilePath string            // file: path to the JSON {name: value} store
+	Mapping  map[string]string // memory/local: initial in-memory secrets
 }
 
 // New instantiates a secret management backend.
 //
-// provider is "aws_secrets_manager" or "azure_keyvault". The auth method is
-// inferred from which credential fields are set, exactly as in the Python
+// provider is "aws_secrets_manager", "azure_keyvault", or a non-cloud source —
+// "env" (environment variables), "file" (a JSON file), or "memory"/"local"
+// (in-memory mapping, mainly dev/tests). For the cloud providers the auth method
+// is inferred from which credential fields are set, exactly as in the Python
 // library:
 //
 //	New(ctx, "aws_secrets_manager", Config{Region: "us-east-1"})  // IAM role / env
@@ -60,13 +69,22 @@ type Config struct {
 //	New(ctx, "azure_keyvault", Config{VaultURL: "https://myvault.vault.azure.net"})
 //	New(ctx, "azure_keyvault", Config{VaultURL: "...", TenantID: "...",
 //	    ClientID: "...", ClientSecret: "..."})
+//	New(ctx, "env", Config{Prefix: "SECRET_"})            // read SECRET_<name> env vars
+//	New(ctx, "file", Config{FilePath: "/run/secrets.json"}) // JSON {name: value}
+//	New(ctx, "memory", Config{Mapping: map[string]string{"db": "..."}}) // in-memory
 func New(ctx context.Context, provider string, cfg Config) (Backend, error) {
 	switch provider {
 	case "aws_secrets_manager":
 		return NewAWSSecretsManager(ctx, cfg)
 	case "azure_keyvault":
 		return NewAzureKeyVault(cfg)
+	case "env":
+		return NewEnvSecrets(cfg.Prefix), nil
+	case "file":
+		return NewFileSecrets(cfg.FilePath), nil
+	case "memory", "local":
+		return NewMappingSecrets(cfg.Mapping), nil
 	}
-	return nil, fmt.Errorf("%w: unknown secrets provider %q (choose 'aws_secrets_manager' or 'azure_keyvault')",
+	return nil, fmt.Errorf("%w: unknown secrets provider %q (choose 'aws_secrets_manager', 'azure_keyvault', 'env', 'file', or 'memory')",
 		core.ErrSecret, provider)
 }
diff --git a/storage/s3.go b/storage/s3.go
index bf49441..5539a9a 100644
--- a/storage/s3.go
+++ b/storage/s3.go
@@ -14,7 +14,9 @@ import (
 	awshttp "github.com/aws/aws-sdk-go-v2/aws/transport/http"
 	awsconfig "github.com/aws/aws-sdk-go-v2/config"
 	"github.com/aws/aws-sdk-go-v2/credentials"
+	"github.com/aws/aws-sdk-go-v2/credentials/stscreds"
 	"github.com/aws/aws-sdk-go-v2/service/s3"
+	"github.com/aws/aws-sdk-go-v2/service/sts"
 	"github.com/aws/smithy-go"
 
 	"github.com/LYZR-OSS/cloudrift-go/core"
@@ -53,6 +55,14 @@ func NewS3(ctx context.Context, cfg Config) (*AWSS3Backend, error) {
 	if err != nil {
 		return nil, fmt.Errorf("%w: %w", core.ErrStorage, err)
 	}
+	if cfg.RoleARN != "" {
+		awsCfg.Credentials = aws.NewCredentialsCache(stscreds.NewAssumeRoleProvider(
+			sts.NewFromConfig(awsCfg), cfg.RoleARN, func(o *stscreds.AssumeRoleOptions) {
+				if cfg.ExternalID != "" {
+					o.ExternalID = aws.String(cfg.ExternalID)
+				}
+			}))
+	}
 	client := s3.NewFromConfig(awsCfg, func(o *s3.Options) {
 		if cfg.EndpointURL != "" {
 			o.BaseEndpoint = aws.String(cfg.EndpointURL)
diff --git a/storage/storage.go b/storage/storage.go
index 20a75ef..adf60e5 100644
--- a/storage/storage.go
+++ b/storage/storage.go
@@ -68,6 +68,10 @@ type Config struct {
 	AWSSessionToken    string
 	ProfileName        string
 	EndpointURL        string // custom endpoint (LocalStack, MinIO, ...)
+	// RoleARN, when set, is assumed via STS on top of the base credentials.
+	// ExternalID is passed in the AssumeRole call when set. Cross-account S3.
+	RoleARN    string
+	ExternalID string
 
 	// Azure Blob.
 	Container        string