diff --git a/app/assets/api/routes.py b/app/assets/api/routes.py
index 30e87a898..d08eeae5a 100644
--- a/app/assets/api/routes.py
+++ b/app/assets/api/routes.py
@@ -1,5 +1,6 @@
 import logging
 import uuid
+import urllib.parse
 from aiohttp import web
 
 from pydantic import ValidationError
@@ -100,3 +101,32 @@ async def get_tags(request: web.Request) -> web.Response:
         owner_id=USER_MANAGER.get_request_user_id(request),
     )
     return web.json_response(result.model_dump(mode="json"))
+
+
+@ROUTES.get(f"/api/assets/{{id:{UUID_RE}}}/content")
+async def download_asset_content(request: web.Request) -> web.Response:
+    # question: do we need disposition? could we just stick with one of these?
+    disposition = request.query.get("disposition", "attachment").lower().strip()
+    if disposition not in {"inline", "attachment"}:
+        disposition = "attachment"
+
+    try:
+        abs_path, content_type, filename = manager.resolve_asset_content_for_download(
+            asset_info_id=str(uuid.UUID(request.match_info["id"])),
+            owner_id=USER_MANAGER.get_request_user_id(request),
+        )
+    except ValueError as ve:
+        return _error_response(404, "ASSET_NOT_FOUND", str(ve))
+    except NotImplementedError as nie:
+        return _error_response(501, "BACKEND_UNSUPPORTED", str(nie))
+    except FileNotFoundError:
+        return _error_response(404, "FILE_NOT_FOUND", "Underlying file not found on disk.")
+
+    quoted = (filename or "").replace("\r", "").replace("\n", "").replace('"', "'")
+    cd = f'{disposition}; filename="{quoted}"; filename*=UTF-8\'\'{urllib.parse.quote(filename)}'
+
+    resp = web.FileResponse(abs_path)
+    resp.content_type = content_type
+    resp.headers["Content-Disposition"] = cd
+    return resp
+
diff --git a/app/assets/database/queries.py b/app/assets/database/queries.py
index 0824c0c2f..086b7010c 100644
--- a/app/assets/database/queries.py
+++ b/app/assets/database/queries.py
@@ -2,7 +2,7 @@ import sqlalchemy as sa
 from collections import defaultdict
 from sqlalchemy import select, exists, func
 from sqlalchemy.orm import Session, contains_eager, noload
-from app.assets.database.models import Asset, AssetInfo, AssetInfoMeta, AssetInfoTag, Tag
+from app.assets.database.models import Asset, AssetInfo, AssetCacheState, AssetInfoMeta, AssetInfoTag, Tag
 from app.assets.helpers import escape_like_prefix, normalize_tags
 from typing import Sequence
 
@@ -208,6 +208,39 @@ def fetch_asset_info_asset_and_tags(
             tags.append(tag_name)
     return first_info, first_asset, tags
 
+def fetch_asset_info_and_asset(
+    session: Session,
+    *,
+    asset_info_id: str,
+    owner_id: str = "",
+) -> tuple[AssetInfo, Asset] | None:
+    stmt = (
+        select(AssetInfo, Asset)
+        .join(Asset, Asset.id == AssetInfo.asset_id)
+        .where(
+            AssetInfo.id == asset_info_id,
+            visible_owner_clause(owner_id),
+        )
+        .limit(1)
+        .options(noload(AssetInfo.tags))
+    )
+    row = session.execute(stmt)
+    pair = row.first()
+    if not pair:
+        return None
+    return pair[0], pair[1]
+
+def list_cache_states_by_asset_id(
+    session: Session, *, asset_id: str
+) -> Sequence[AssetCacheState]:
+    return (
+        session.execute(
+            select(AssetCacheState)
+            .where(AssetCacheState.asset_id == asset_id)
+            .order_by(AssetCacheState.id.asc())
+        )
+    ).scalars().all()
+
 def list_tags_with_usage(
     session: Session,
     prefix: str | None = None,
diff --git a/app/assets/manager.py b/app/assets/manager.py
index 6425e7aa2..1db25be11 100644
--- a/app/assets/manager.py
+++ b/app/assets/manager.py
@@ -1,3 +1,5 @@
+import os
+import mimetypes
 from typing import Sequence
 
 from app.database.db import create_session
@@ -5,6 +7,8 @@ from app.assets.api import schemas_out
 from app.assets.database.queries import (
     asset_exists_by_hash,
     fetch_asset_info_asset_and_tags,
+    fetch_asset_info_and_asset,
+    list_cache_states_by_asset_id,
     list_asset_infos_page,
     list_tags_with_usage,
 )
@@ -24,6 +28,7 @@ def asset_exists(asset_hash: str) -> bool:
         return asset_exists_by_hash(session, asset_hash=asset_hash)
 
 def list_assets(
+    *,
     include_tags: Sequence[str] | None = None,
     exclude_tags: Sequence[str] | None = None,
     name_contains: str | None = None,
@@ -76,7 +81,11 @@ def list_assets(
         has_more=(offset + len(summaries)) < total,
     )
 
-def get_asset(asset_info_id: str, owner_id: str = "") -> schemas_out.AssetDetail:
+def get_asset(
+    *,
+    asset_info_id: str,
+    owner_id: str = "",
+) -> schemas_out.AssetDetail:
     with create_session() as session:
         res = fetch_asset_info_asset_and_tags(session, asset_info_id=asset_info_id, owner_id=owner_id)
         if not res:
@@ -97,6 +106,29 @@ def get_asset(asset_info_id: str, owner_id: str = "") -> schemas_out.AssetDetail
         last_access_time=info.last_access_time,
     )
 
+def resolve_asset_content_for_download(
+    *,
+    asset_info_id: str,
+    owner_id: str = "",
+) -> tuple[str, str, str]:
+    with create_session() as session:
+        pair = fetch_asset_info_and_asset(session, asset_info_id=asset_info_id, owner_id=owner_id)
+        if not pair:
+            raise ValueError(f"AssetInfo {asset_info_id} not found")
+
+        info, asset = pair
+        states = list_cache_states_by_asset_id(session, asset_id=asset.id)
+        abs_path = pick_best_live_path(states)
+        if not abs_path:
+            raise FileNotFoundError
+
+        touch_asset_info_by_id(session, asset_info_id=asset_info_id)
+        session.commit()
+
+    ctype = asset.mime_type or mimetypes.guess_type(info.name or abs_path)[0] or "application/octet-stream"
+    download_name = info.name or os.path.basename(abs_path)
+    return abs_path, ctype, download_name
+
 def list_tags(
     prefix: str | None = None,
     limit: int = 100,