Secrets resolution

Production agents shouldn’t hard-code API keys; they shouldn’t depend on os.environ either (process-level env vars leak across threads, get logged in crashes, and don’t rotate). The Secrets protocol gives you a pluggable resolver:


from loomflow import Agent, Tuning
from loomflow.security import EnvSecrets, DictSecrets
 
# Default — reads from os.environ. Same behaviour as pre-M10.
agent = Agent("...", model="claude-opus-4-7", tuning=Tuning(secrets=EnvSecrets()))
 
# In-memory for tests or vault-fetched-once-at-startup:
agent = Agent(
    "...",
    model="claude-opus-4-7",
    tuning=Tuning(secrets=DictSecrets({
        "ANTHROPIC_API_KEY": api_key_from_vault,
        "OPENAI_API_KEY": openai_key_from_vault,
    })),
)

Resolution order

Resolution order inside model adapters:

Explicit api_key= argument on the model adapter
secrets.lookup_sync(<ENV_VAR_NAME>) if a Secrets backend is wired
os.environ[<ENV_VAR_NAME>] as the bare fallback

Vault adapter example

Production callers running on AWS / GCP / Vault should write a small custom adapter:


class VaultSecrets:
    """Pulls from HashiCorp Vault, caches into a local dict for
    the constructor-time ``lookup_sync`` path."""
 
    def __init__(self, vault_client, path: str) -> None:
        self._client = vault_client
        self._cache: dict[str, str] = {}
 
    async def resolve(self, ref: str) -> str:
        if ref in self._cache:
            return self._cache[ref]
        secret = await self._client.read(f"secret/data/{ref}")
        value = secret["data"]["value"]
        self._cache[ref] = value
        return value
 
    async def store(self, ref: str, value: str) -> None:
        await self._client.write(f"secret/data/{ref}", value=value)
        self._cache[ref] = value
 
    def redact(self, text: str) -> str:
        from loomflow.security.secrets import _apply_redaction
        return _apply_redaction(text)
 
    def lookup_sync(self, ref: str) -> str | None:
        # Constructor-time path — return whatever's already in the
        # cache; if nothing's there, return None and let the caller
        # fall back to env-vars.
        return self._cache.get(ref)
 
 
# Pre-warm the cache at startup so lookup_sync hits during Agent()
# construction:
async def boot():
    vault = await connect_vault(...)
    secrets = VaultSecrets(vault, path="prod/jeeves")
    await secrets.resolve("ANTHROPIC_API_KEY")
    await secrets.resolve("OPENAI_API_KEY")
 
    agent = Agent("...", model="claude-opus-4-7", tuning=Tuning(secrets=secrets))

Redaction

secrets.redact(text) masks common API-key shapes (OpenAI, Anthropic, AWS access keys, GitHub PATs). Useful inside @agent.before_tool hooks that log tool args, or before payload strings hit the audit log.