credentials

credentials

Credential persistence for tools and channels.

Stores credentials in ~/.openjarvis/credentials.toml with 0o600 permissions. Thread-safe writes via lock. Sets os.environ on save for immediate effect.

Functions

load_credentials

load_credentials(path: Path | None = None) -> dict[str, dict[str, str]]

Load credentials from TOML file.

Source code in src/openjarvis/core/credentials.py

def load_credentials(path: Path | None = None) -> dict[str, dict[str, str]]:
    """Load credentials from TOML file."""
    p = Path(path) if path else _default_path()
    if not p.exists():
        return {}
    with open(p, "rb") as f:
        return tomllib.load(f)

save_credential

save_credential(tool_name: str, key: str, value: str, *, path: Path | None = None) -> None

Save a single credential key, validate, write file, and set os.environ.

Source code in src/openjarvis/core/credentials.py

def save_credential(
    tool_name: str,
    key: str,
    value: str,
    *,
    path: Path | None = None,
) -> None:
    """Save a single credential key, validate, write file, and set os.environ."""
    _validate_credential_key(tool_name, key)
    stripped = value.strip()
    if not stripped:
        raise ValueError("Credential value must not be empty")

    p = Path(path) if path else _default_path()
    with _LOCK:
        creds = load_credentials(path=p)
        if tool_name not in creds:
            creds[tool_name] = {}
        creds[tool_name][key] = stripped
        _write_credentials(creds, p)

    os.environ[key] = stripped

delete_credential

delete_credential(tool_name: str, key: str, *, path: Path | None = None) -> None

Delete a persisted credential and remove it from the running process.

Source code in src/openjarvis/core/credentials.py

def delete_credential(
    tool_name: str,
    key: str,
    *,
    path: Path | None = None,
) -> None:
    """Delete a persisted credential and remove it from the running process."""
    _validate_credential_key(tool_name, key)
    p = Path(path) if path else _default_path()
    with _LOCK:
        creds = load_credentials(path=p)
        tool_creds = creds.get(tool_name)
        if tool_creds is not None:
            tool_creds.pop(key, None)
            if not tool_creds:
                creds.pop(tool_name, None)
            _write_credentials(creds, p)

    os.environ.pop(key, None)

get_credential_status

get_credential_status(tool_name: str) -> dict[str, bool]

Return {KEY: bool} for each required key indicating if set in env.

Source code in src/openjarvis/core/credentials.py

def get_credential_status(tool_name: str) -> dict[str, bool]:
    """Return {KEY: bool} for each required key indicating if set in env."""
    keys = TOOL_CREDENTIALS.get(tool_name, [])
    return {k: bool(os.environ.get(k)) for k in keys}

inject_credentials

inject_credentials(path: Path | None = None) -> None

Load credentials.toml and inject into os.environ. Call at server startup.

Source code in src/openjarvis/core/credentials.py

def inject_credentials(path: Path | None = None) -> None:
    """Load credentials.toml and inject into os.environ. Call at server startup."""
    creds = load_credentials(path=path)
    for _tool, kvs in creds.items():
        for k, v in kvs.items():
            if k not in os.environ:
                os.environ[k] = v

get_tool_credential

get_tool_credential(tool_name: str, key: str, *, path: Path | None = None) -> str | None

Read a single credential without polluting os.environ.

Falls back to os.environ if the key is not in credentials.toml, for backward compatibility with Docker env var workflows.

Source code in src/openjarvis/core/credentials.py

def get_tool_credential(
    tool_name: str,
    key: str,
    *,
    path: Path | None = None,
) -> str | None:
    """Read a single credential without polluting ``os.environ``.

    Falls back to ``os.environ`` if the key is not in credentials.toml,
    for backward compatibility with Docker env var workflows.
    """
    creds = load_credentials(path=path)
    tool_creds = creds.get(tool_name, {})
    value = tool_creds.get(key)
    if value is not None:
        return value
    return os.environ.get(key) or None