datarobot_genai.drtools.files_api.common_utils

common_utils

Shared helpers for the Files API tool modules.

FilesApiLocalSettings

Bases: DataRobotAppFrameworkBaseSettings

Settings governing local-disk access for the Files API tools.

Resolves from env vars, .env, file secrets, and MLOPS_RUNTIME_PARAM_ runtime parameters by field name (files_api_local_allowed_roots reads FILES_API_LOCAL_ALLOWED_ROOTS). Defined here, rather than in the drmcp config, so the drtools layer stays independent of the server package.

Source code in datarobot_genai/drtools/files_api/common_utils.py

class FilesApiLocalSettings(DataRobotAppFrameworkBaseSettings):
    """Settings governing local-disk access for the Files API tools.

    Resolves from env vars, ``.env``, file secrets, and ``MLOPS_RUNTIME_PARAM_``
    runtime parameters by field name (``files_api_local_allowed_roots`` reads
    ``FILES_API_LOCAL_ALLOWED_ROOTS``). Defined here, rather than in the drmcp
    config, so the drtools layer stays independent of the server package.
    """

    files_api_local_allowed_roots: str = Field(
        default="",
        description=(
            "Comma-separated absolute directories that file_upload may read from. "
            "Empty (default) disables all local-disk access."
        ),
    )

    model_config = SettingsConfigDict(
        env_file=".env",
        extra="ignore",
        env_ignore_empty=True,
    )

ensure_local_path_allowed

ensure_local_path_allowed(local_path: str) -> Path

Validate local_path against the allowlist and return its resolved real path.

Resolves .. and symlinks before checking so a path cannot escape an allowed root. Raises :class:ToolError when local access is disabled (no roots configured) or the path falls outside every allowed root.

Source code in datarobot_genai/drtools/files_api/common_utils.py

def ensure_local_path_allowed(local_path: str) -> Path:
    """Validate ``local_path`` against the allowlist and return its resolved real path.

    Resolves ``..`` and symlinks before checking so a path cannot escape an
    allowed root. Raises :class:`ToolError` when local access is disabled (no
    roots configured) or the path falls outside every allowed root.
    """
    roots = _allowed_local_roots()
    if not roots:
        raise ToolError(
            "Local filesystem access is disabled. Set FILES_API_LOCAL_ALLOWED_ROOTS to a "
            "comma-separated list of allowed base directories to enable file_upload.",
            kind=ToolErrorKind.VALIDATION,
        )
    resolved = Path(local_path).expanduser().resolve()
    if not _is_within_roots(resolved, roots):
        allowed = ", ".join(str(root) for root in roots)
        raise ToolError(
            f"Local path {local_path!r} is outside the allowed directories ({allowed}).",
            kind=ToolErrorKind.VALIDATION,
        )
    return resolved

get_store

get_store() -> FileSystemStore

Return the filesystem store. Indirection keeps tools easy to patch in tests.

Source code in datarobot_genai/drtools/files_api/common_utils.py

def get_store() -> FileSystemStore:
    """Return the filesystem store. Indirection keeps tools easy to patch in tests."""
    return DataRobotFileSystemStore()

require_file_path

require_file_path(path: str | None, name: str = 'path') -> str

Validate a path that must reference something under a catalog item, not the root.

Source code in datarobot_genai/drtools/files_api/common_utils.py

def require_file_path(path: str | None, name: str = "path") -> str:
    """Validate a path that must reference something under a catalog item, not the root."""
    cleaned = require_path(path, name)
    if is_root(cleaned):
        raise ToolError(
            f"Argument validation error: '{name}' must reference a path under a catalog item "
            f"(e.g. dr://<catalog_id>/file.txt), not the filesystem root.",
            kind=ToolErrorKind.VALIDATION,
        )
    return cleaned

resolve_local_sources async

resolve_local_sources(local_path: str, *, recursive: bool, maxdepth: int | None) -> list[tuple[str, int]]

Expand a local file/dir/glob to a manifest of (path, size) tuples.

Mirrors the expansion the upload backend performs, so callers can validate that something will actually be uploaded and report an accurate file count and byte total. Runs in a worker thread.

Source code in datarobot_genai/drtools/files_api/common_utils.py

async def resolve_local_sources(
    local_path: str, *, recursive: bool, maxdepth: int | None
) -> list[tuple[str, int]]:
    """Expand a local file/dir/glob to a manifest of (path, size) tuples.

    Mirrors the expansion the upload backend performs, so callers can validate
    that something will actually be uploaded and report an accurate file count
    and byte total. Runs in a worker thread.
    """
    cleaned = require_path(local_path, "local_path")
    ensure_local_path_allowed(cleaned)
    roots = _allowed_local_roots()

    def _expand() -> list[tuple[str, int]]:
        if os.path.isdir(cleaned) and not recursive:
            raise ToolError(
                f"Local path {cleaned!r} is a directory; set recursive=True "
                "to upload its contents.",
                kind=ToolErrorKind.VALIDATION,
            )
        lfs = LocalFileSystem()
        try:
            candidates = lfs.expand_path(cleaned, recursive=recursive, maxdepth=maxdepth)
        except FileNotFoundError:
            # No file or glob match — report uniformly via the caller's empty-manifest path.
            return []

        manifest: list[tuple[str, int]] = []
        for path in candidates:
            if not os.path.isfile(path):
                continue
            # Re-check each expanded path so a symlink cannot escape the allowlist.
            if not _is_within_roots(Path(path).resolve(), roots):
                raise ToolError(
                    f"Local path {path!r} resolves outside the allowed directories.",
                    kind=ToolErrorKind.VALIDATION,
                )
            manifest.append((path, os.path.getsize(path)))
        return manifest

    return await asyncio.to_thread(_expand)

decode_content

decode_content(content: str, encoding: str) -> bytes

Decode tool-supplied content to bytes per encoding ('utf-8' or 'base64').

Source code in datarobot_genai/drtools/files_api/common_utils.py

def decode_content(content: str, encoding: str) -> bytes:
    """Decode tool-supplied ``content`` to bytes per ``encoding`` ('utf-8' or 'base64')."""
    if encoding == "utf-8":
        return content.encode("utf-8")
    if encoding == "base64":
        try:
            return base64.b64decode(content, validate=True)
        except (binascii.Error, ValueError) as exc:
            raise ToolError(
                f"Argument validation error: 'content' is not valid base64: {exc}",
                kind=ToolErrorKind.VALIDATION,
            ) from exc
    raise ToolError(
        "Argument validation error: 'encoding' must be 'utf-8' or 'base64'.",
        kind=ToolErrorKind.VALIDATION,
    )