Secrets resolution

A secrets backend tells model adapters where to get API keys. Use a custom one when keys live in a vault or secret manager rather than in environment variables. Production agents shouldn’t hard-code API keys, and often shouldn’t depend on os.environ either: process-level env vars are visible to all code in the process, 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
 
# The default: reads from os.environ.
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

Models the agent builds from strings use the same backend: the main model, and role models an architecture takes as a string, such as Reflexion(evaluator_model="gpt-4.1-mini") or SelfRefine(critic_model=...). (Before 0.17, architecture role models read the environment directly. In 0.17.0 to 0.17.2, an OpenAI or DeepSeek role-model string still needed its key in the environment when the architecture was built.)

Vault adapter example

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


from loomflow import Agent, Tuning
from loomflow.security import EnvSecrets
 
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._path = path
        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/{self._path}/{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/{self._path}/{ref}", value=value)
        self._cache[ref] = value
 
    def redact(self, text: str) -> str:
        return EnvSecrets().redact(text)   # reuse the built-in key masking
 
    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/loomflow")
    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.