feat(ai): add Atlas Cloud provider support

.env.example

@@ -9,7 +9,11 @@ BETTER_AUTH_SECRET=              # openssl rand -hex 32
 BETTER_AUTH_URL=                 # https://trykimu.com (production) or http://localhost:5173 (dev)
 GOOGLE_CLIENT_ID=                # Google OAuth client ID
 GOOGLE_CLIENT_SECRET=            # Google OAuth client secret
+AI_PROVIDER=gemini               # "gemini" (default) or "atlascloud"
 GEMINI_API_KEY=
+ATLASCLOUD_API_KEY=
+ATLASCLOUD_BASE_URL=https://api.atlascloud.ai/v1
+ATLASCLOUD_MODEL=deepseek-ai/deepseek-v4-pro
 
 # NODE_ENV=production              # Set to "production" to disable uvicorn hot-reload
 
@@ -19,4 +23,3 @@ R2_ACCESS_KEY_ID=                # R2 API token access key
 R2_SECRET_ACCESS_KEY=            # R2 API token secret key
 R2_ASSETS_BUCKET=
 R2_RENDERS_BUCKET=
-

backend/ai/provider.py

@@ -0,0 +1,102 @@
+import json
+import os
+from functools import lru_cache
+from typing import Literal
+
+from google import genai
+from openai import OpenAI
+
+from ai.schema import FunctionCallResponse
+from utils import require_env
+
+_GEMINI_MODEL = "gemini-2.5-flash"
+_ATLAS_BASE_URL = "https://api.atlascloud.ai/v1"
+_ATLAS_MODEL = "deepseek-ai/deepseek-v4-pro"
+_ATLAS_MAX_TOKENS = 4096
+
+AIProvider = Literal["gemini", "atlascloud"]
+
+
+def _first_env(*names: str) -> str | None:
+    for name in names:
+        value = os.getenv(name, "").strip()
+        if value:
+            return value
+    return None
+
+
+def _provider_name() -> AIProvider:
+    value = os.getenv("AI_PROVIDER", "gemini").strip().lower()
+    if value == "gemini":
+        return "gemini"
+    if value in {"atlas", "atlas-cloud", "atlascloud"}:
+        return "atlascloud"
+    raise RuntimeError("AI_PROVIDER must be either 'gemini' or 'atlascloud'")
+
+
+@lru_cache
+def _gemini_client() -> genai.Client:
+    return genai.Client(api_key=require_env("GEMINI_API_KEY"))
+
+
+@lru_cache
+def _atlas_client() -> OpenAI:
+    api_key = _first_env("ATLASCLOUD_API_KEY", "ATLAS_CLOUD_API_KEY")
+    if api_key is None:
+        raise RuntimeError("ATLASCLOUD_API_KEY or ATLAS_CLOUD_API_KEY must be set")
+    base_url = _first_env(
+        "ATLASCLOUD_BASE_URL",
+        "ATLAS_CLOUD_BASE_URL",
+        "ATLASCLOUD_API_BASE",
+        "ATLAS_CLOUD_API_BASE",
+    )
+    return OpenAI(api_key=api_key, base_url=base_url or _ATLAS_BASE_URL)
+
+
+def _generate_with_gemini(prompt: str) -> FunctionCallResponse:
+    response = _gemini_client().models.generate_content(
+        model=_GEMINI_MODEL,
+        contents=prompt,
+        config={
+            "response_mime_type": "application/json",
+            "response_schema": FunctionCallResponse,
+        },
+    )
+    return FunctionCallResponse.model_validate(response.parsed)
+
+
+def _generate_with_atlas(prompt: str) -> FunctionCallResponse:
+    model = _first_env("ATLASCLOUD_MODEL", "ATLAS_CLOUD_MODEL") or _ATLAS_MODEL
+    schema = json.dumps(
+        FunctionCallResponse.model_json_schema(),
+        ensure_ascii=False,
+        separators=(",", ":"),
+    )
+    completion = _atlas_client().chat.completions.create(
+        model=model,
+        messages=[
+            {
+                "role": "system",
+                "content": (
+                    "Return only a valid JSON object matching the schema supplied "
+                    "in the user message."
+                ),
+            },
+            {
+                "role": "user",
+                "content": f"{prompt}\n\n## Response JSON schema\n{schema}",
+            },
+        ],
+        response_format={"type": "json_object"},
+        max_tokens=_ATLAS_MAX_TOKENS,
+    )
+    content = completion.choices[0].message.content
+    if content is None:
+        raise ValueError("Atlas Cloud returned an empty response")
+    return FunctionCallResponse.model_validate_json(content)
+
+
+def generate_ai_response(prompt: str) -> FunctionCallResponse:
+    if _provider_name() == "atlascloud":
+        return _generate_with_atlas(prompt)
+    return _generate_with_gemini(prompt)

backend/ai/routes.py

@@ -4,24 +4,18 @@
 from typing import Any
 
 from fastapi import APIRouter, Depends, HTTPException, status
-from google import genai
 from pydantic import BaseModel, ConfigDict, Field
 
+from ai.provider import generate_ai_response
 from ai.schema import FunctionCallResponse
 from auth.routes import get_current_user
 from auth.schema import SessionUser
 from db import get_db_pool
-from utils import require_env
 
 logger = logging.getLogger(__name__)
 
 router = APIRouter(tags=["ai"])
 
-_GEMINI_MODEL = "gemini-2.5-flash"
-
-GEMINI_API_KEY: str = require_env("GEMINI_API_KEY")
-gemini_client: genai.Client = genai.Client(api_key=GEMINI_API_KEY)
-
 _MAX_MESSAGE_LENGTH = 20_000
 _MAX_HISTORY_ITEMS = 50
 # Permissive caps — large enough for real projects (hundreds of scrubbers / media items) without
@@ -115,7 +109,7 @@ async def process_ai_message(
 ) -> FunctionCallResponse:
     await _enforce_rate_limit(user.user_id)
 
-    # Bound the serialized payload before forwarding to Gemini to cap token spend.
+    # Bound the serialized payload before forwarding to the AI provider to cap spend.
     timeline_json = json.dumps(request.timeline_state or {}, ensure_ascii=False)
     if len(timeline_json) > _MAX_TIMELINE_BYTES:
         raise HTTPException(
@@ -193,15 +187,7 @@ async def process_ai_message(
 """
 
     try:
-        response = gemini_client.models.generate_content(
-            model=_GEMINI_MODEL,
-            contents=prompt,
-            config={
-                "response_mime_type": "application/json",
-                "response_schema": FunctionCallResponse,
-            },
-        )
-        return FunctionCallResponse.model_validate(response.parsed)
+        return generate_ai_response(prompt)
     except ValueError as exc:
         # Don't include user content (timeline / messages) in logs — log the type only.
         logger.warning("AI response validation failed: %s", type(exc).__name__)

backend/pyproject.toml

@@ -12,6 +12,7 @@ dependencies = [
     "asyncpg>=0.31.0",
     "fastapi[standard]>=0.115.13",
     "google-genai>=1.22.0",
+    "openai>=2.48.0",
     "python-dotenv>=1.0.0",
     "python-multipart>=0.0.22",
 ]

backend/tests/test_ai_provider.py

@@ -0,0 +1,61 @@
+import os
+import unittest
+from unittest.mock import MagicMock, patch
+
+from ai.provider import generate_ai_response
+
+
+class AIProviderTest(unittest.TestCase):
+    @patch("ai.provider._gemini_client")
+    def test_gemini_remains_the_default_provider(
+        self, gemini_client: MagicMock
+    ) -> None:
+        gemini_client.return_value.models.generate_content.return_value.parsed = {
+            "function_call": None,
+            "assistant_message": "Hello",
+        }
+
+        with patch.dict(os.environ, {"AI_PROVIDER": "gemini"}):
+            response = generate_ai_response("hello")
+
+        self.assertEqual(response.assistant_message, "Hello")
+        gemini_client.return_value.models.generate_content.assert_called_once()
+
+    @patch("ai.provider._atlas_client")
+    def test_atlascloud_uses_openai_compatible_json_responses(
+        self, atlas_client: MagicMock
+    ) -> None:
+        choice = MagicMock()
+        choice.message.content = (
+            '{"function_call":null,"assistant_message":"Hello from Atlas"}'
+        )
+        atlas_client.return_value.chat.completions.create.return_value.choices = [
+            choice
+        ]
+
+        with patch.dict(
+            os.environ,
+            {
+                "AI_PROVIDER": "atlascloud",
+                "ATLASCLOUD_MODEL": "qwen/qwen3.5-flash",
+            },
+        ):
+            response = generate_ai_response("hello")
+
+        self.assertEqual(response.assistant_message, "Hello from Atlas")
+        call = atlas_client.return_value.chat.completions.create.call_args.kwargs
+        self.assertEqual(call["model"], "qwen/qwen3.5-flash")
+        self.assertEqual(call["response_format"], {"type": "json_object"})
+        self.assertEqual(call["max_tokens"], 4096)
+        self.assertIn("Response JSON schema", call["messages"][1]["content"])
+
+    def test_unknown_provider_is_rejected(self) -> None:
+        with (
+            patch.dict(os.environ, {"AI_PROVIDER": "unknown"}),
+            self.assertRaisesRegex(RuntimeError, "AI_PROVIDER"),
+        ):
+            generate_ai_response("hello")
+
+
+if __name__ == "__main__":
+    unittest.main()