fix(ai): bind prompt reads to project token (#433)

* fix(ai): bind prompt reads to project token

* chore(release): bump version to 7.8.7

Author: Radu-Raicea

CHANGELOG.md

@@ -1,3 +1,7 @@
+# 7.9.1 - 2026-02-17
+
+fix(llma): make prompt fetches deterministic by requiring project_api_key and sending it as token query param
+
 # 7.9.0 - 2026-02-17
 
 feat: Support device_id as bucketing identifier for local evaluation

posthog/ai/prompts.py

@@ -54,7 +54,11 @@ class Prompts:
         prompts = Prompts(posthog)
 
         # Or with direct options (no PostHog client needed)
-        prompts = Prompts(personal_api_key='phx_xxx', host='https://us.posthog.com')
+        prompts = Prompts(
+            personal_api_key='phx_xxx',
+            project_api_key='phc_xxx',
+            host='https://us.posthog.com',
+        )
 
         # Fetch with caching and fallback
         template = prompts.get('support-system-prompt', fallback='You are a helpful assistant.')
@@ -72,6 +76,7 @@ def __init__(
         posthog: Optional[Any] = None,
         *,
         personal_api_key: Optional[str] = None,
+        project_api_key: Optional[str] = None,
         host: Optional[str] = None,
         default_cache_ttl_seconds: Optional[int] = None,
     ):
@@ -80,7 +85,8 @@ def __init__(
 
         Args:
             posthog: PostHog client instance (optional if personal_api_key provided)
-            personal_api_key: Direct API key (optional if posthog provided)
+            personal_api_key: Direct personal API key (optional if posthog provided)
+            project_api_key: Direct project API key (optional if posthog provided)
             host: PostHog host (defaults to app endpoint)
             default_cache_ttl_seconds: Default cache TTL (defaults to 300)
         """
@@ -91,11 +97,13 @@ def __init__(
 
         if posthog is not None:
             self._personal_api_key = getattr(posthog, "personal_api_key", None) or ""
+            self._project_api_key = getattr(posthog, "api_key", None) or ""
             self._host = remove_trailing_slash(
                 getattr(posthog, "raw_host", None) or APP_ENDPOINT
             )
         else:
             self._personal_api_key = personal_api_key or ""
+            self._project_api_key = project_api_key or ""
             self._host = remove_trailing_slash(host or APP_ENDPOINT)
 
     def get(
@@ -215,7 +223,7 @@ def _fetch_prompt_from_api(self, name: str) -> str:
         """
         Fetch prompt from PostHog API.
 
-        Endpoint: {host}/api/environments/@current/llm_prompts/name/{encoded_name}/
+        Endpoint: {host}/api/environments/@current/llm_prompts/name/{encoded_name}/?token={encoded_project_api_key}
         Auth: Bearer {personal_api_key}
 
         Args:
@@ -232,9 +240,15 @@ def _fetch_prompt_from_api(self, name: str) -> str:
                 "[PostHog Prompts] personal_api_key is required to fetch prompts. "
                 "Please provide it when initializing the Prompts instance."
             )
+        if not self._project_api_key:
+            raise Exception(
+                "[PostHog Prompts] project_api_key is required to fetch prompts. "
+                "Please provide it when initializing the Prompts instance."
+            )
 
         encoded_name = urllib.parse.quote(name, safe="")
-        url = f"{self._host}/api/environments/@current/llm_prompts/name/{encoded_name}/"
+        encoded_project_api_key = urllib.parse.quote(self._project_api_key, safe="")
+        url = f"{self._host}/api/environments/@current/llm_prompts/name/{encoded_name}/?token={encoded_project_api_key}"
 
         headers = {
             "Authorization": f"Bearer {self._personal_api_key}",

posthog/test/ai/test_prompts.py

@@ -33,11 +33,15 @@ class TestPrompts(unittest.TestCase):
     }
 
     def create_mock_posthog(
-        self, personal_api_key="phx_test_key", host="https://us.posthog.com"
+        self,
+        personal_api_key="phx_test_key",
+        project_api_key="phc_test_key",
+        host="https://us.posthog.com",
     ):
         """Create a mock PostHog client."""
         mock = MagicMock()
         mock.personal_api_key = personal_api_key
+        mock.api_key = project_api_key
         mock.raw_host = host
         return mock
 
@@ -61,7 +65,7 @@ def test_successfully_fetch_a_prompt(self, mock_get_session):
         call_args = mock_get.call_args
         self.assertEqual(
             call_args[0][0],
-            "https://us.posthog.com/api/environments/@current/llm_prompts/name/test-prompt/",
+            "https://us.posthog.com/api/environments/@current/llm_prompts/name/test-prompt/?token=phc_test_key",
         )
         self.assertIn("Authorization", call_args[1]["headers"])
         self.assertEqual(
@@ -235,6 +239,18 @@ def test_throw_when_no_personal_api_key_configured(self):
             "personal_api_key is required to fetch prompts", str(context.exception)
         )
 
+    def test_throw_when_no_project_api_key_configured(self):
+        """Should throw when no project_api_key is configured."""
+        posthog = self.create_mock_posthog(project_api_key=None)
+        prompts = Prompts(posthog)
+
+        with self.assertRaises(Exception) as context:
+            prompts.get("test-prompt")
+
+        self.assertIn(
+            "project_api_key is required to fetch prompts", str(context.exception)
+        )
+
     @patch("posthog.ai.prompts._get_session")
     def test_throw_when_api_returns_invalid_response_format(self, mock_get_session):
         """Should throw when API returns invalid response format."""
@@ -255,15 +271,17 @@ def test_use_custom_host_from_posthog_options(self, mock_get_session):
         mock_get = mock_get_session.return_value.get
         mock_get.return_value = MockResponse(json_data=self.mock_prompt_response)
 
-        posthog = self.create_mock_posthog(host="https://eu.i.posthog.com")
+        posthog = self.create_mock_posthog(host="https://eu.posthog.com")
         prompts = Prompts(posthog)
 
         prompts.get("test-prompt")
 
         call_args = mock_get.call_args
         self.assertTrue(
-            call_args[0][0].startswith("https://eu.i.posthog.com/"),
-            f"Expected URL to start with 'https://eu.i.posthog.com/', got {call_args[0][0]}",
+            call_args[0][0].startswith(
+                "https://eu.posthog.com/api/environments/@current/llm_prompts/name/test-prompt/?token=phc_test_key"
+            ),
+            f"Expected URL to start with 'https://eu.posthog.com/api/environments/@current/llm_prompts/name/test-prompt/?token=phc_test_key', got {call_args[0][0]}",
         )
 
     @patch("posthog.ai.prompts._get_session")
@@ -333,7 +351,7 @@ def test_url_encode_prompt_names_with_special_characters(self, mock_get_session)
         call_args = mock_get.call_args
         self.assertEqual(
             call_args[0][0],
-            "https://us.posthog.com/api/environments/@current/llm_prompts/name/prompt%20with%20spaces%2Fand%2Fslashes/",
+            "https://us.posthog.com/api/environments/@current/llm_prompts/name/prompt%20with%20spaces%2Fand%2Fslashes/?token=phc_test_key",
         )
 
     @patch("posthog.ai.prompts._get_session")
@@ -342,15 +360,17 @@ def test_work_with_direct_options_no_posthog_client(self, mock_get_session):
         mock_get = mock_get_session.return_value.get
         mock_get.return_value = MockResponse(json_data=self.mock_prompt_response)
 
-        prompts = Prompts(personal_api_key="phx_direct_key")
+        prompts = Prompts(
+            personal_api_key="phx_direct_key", project_api_key="phc_direct_key"
+        )
 
         result = prompts.get("test-prompt")
 
         self.assertEqual(result, self.mock_prompt_response["prompt"])
         call_args = mock_get.call_args
         self.assertEqual(
             call_args[0][0],
-            "https://us.posthog.com/api/environments/@current/llm_prompts/name/test-prompt/",
+            "https://us.posthog.com/api/environments/@current/llm_prompts/name/test-prompt/?token=phc_direct_key",
         )
         self.assertEqual(
             call_args[1]["headers"]["Authorization"], "Bearer phx_direct_key"
@@ -363,15 +383,17 @@ def test_use_custom_host_from_direct_options(self, mock_get_session):
         mock_get.return_value = MockResponse(json_data=self.mock_prompt_response)
 
         prompts = Prompts(
-            personal_api_key="phx_direct_key", host="https://eu.posthog.com"
+            personal_api_key="phx_direct_key",
+            project_api_key="phc_direct_key",
+            host="https://eu.posthog.com",
         )
 
         prompts.get("test-prompt")
 
         call_args = mock_get.call_args
         self.assertEqual(
             call_args[0][0],
-            "https://eu.posthog.com/api/environments/@current/llm_prompts/name/test-prompt/",
+            "https://eu.posthog.com/api/environments/@current/llm_prompts/name/test-prompt/?token=phc_direct_key",
         )
 
     @patch("posthog.ai.prompts._get_session")
@@ -385,7 +407,9 @@ def test_use_custom_default_cache_ttl_from_direct_options(
         mock_time.return_value = 1000.0
 
         prompts = Prompts(
-            personal_api_key="phx_direct_key", default_cache_ttl_seconds=60
+            personal_api_key="phx_direct_key",
+            project_api_key="phc_direct_key",
+            default_cache_ttl_seconds=60,
         )
 
         # First call
@@ -486,23 +510,29 @@ def test_handle_multiple_occurrences_of_same_variable(self):
 
     def test_work_with_direct_options_initialization(self):
         """Should work with direct options initialization."""
-        prompts = Prompts(personal_api_key="phx_test_key")
+        prompts = Prompts(
+            personal_api_key="phx_test_key", project_api_key="phc_test_key"
+        )
 
         result = prompts.compile("Hello, {{name}}!", {"name": "World"})
 
         self.assertEqual(result, "Hello, World!")
 
     def test_handle_variables_with_hyphens(self):
         """Should handle variables with hyphens."""
-        prompts = Prompts(personal_api_key="phx_test_key")
+        prompts = Prompts(
+            personal_api_key="phx_test_key", project_api_key="phc_test_key"
+        )
 
         result = prompts.compile("User ID: {{user-id}}", {"user-id": "12345"})
 
         self.assertEqual(result, "User ID: 12345")
 
     def test_handle_variables_with_dots(self):
         """Should handle variables with dots."""
-        prompts = Prompts(personal_api_key="phx_test_key")
+        prompts = Prompts(
+            personal_api_key="phx_test_key", project_api_key="phc_test_key"
+        )
 
         result = prompts.compile("Company: {{company.name}}", {"company.name": "Acme"})
 

posthog/version.py

@@ -1,4 +1,4 @@
-VERSION = "7.9.0"
+VERSION = "7.9.1"
 
 if __name__ == "__main__":
     print(VERSION, end="")  # noqa: T201