Coverage for .venv/lib/python3.13/site-packages/litellm/proxy/prompts/prompt_endpoints.py: 54%
386 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 12:01 +0000
1"""
2CRUD ENDPOINTS FOR PROMPTS
3"""
5import tempfile
6from collections.abc import Awaitable, Mapping, Sequence
7from datetime import datetime
8from pathlib import Path
9from typing import TYPE_CHECKING, Final, Protocol, cast
11from fastapi import (
12 APIRouter,
13 Depends,
14 File,
15 HTTPException,
16 Request,
17 Response,
18 UploadFile,
19)
20from pydantic import BaseModel
22from litellm._logging import verbose_proxy_logger
23from litellm.proxy._types import (
24 CommonProxyErrors,
25 LitellmUserRoles,
26 UserAPIKeyAuth,
27 user_api_key_has_admin_view,
28)
29from litellm.proxy.auth.auth_utils import is_request_body_safe
30from litellm.proxy.auth.user_api_key_auth import user_api_key_auth
31from litellm.proxy.common_utils.path_utils import safe_filename
32from litellm.proxy.prompts.prompt_registry import (
33 DEFAULT_PROMPT_ENVIRONMENT,
34 get_base_prompt_id,
35 get_version_number,
36 prompt_environment_or_default,
37)
38from litellm.repositories.table_repositories import PromptRepository
39from litellm.types.prompts.init_prompts import (
40 ListPromptsResponse,
41 PromptInfo,
42 PromptInfoResponse,
43 PromptLiteLLMParams,
44 PromptSpec,
45 PromptTemplateBase,
46)
47from litellm.types.proxy.prompt_endpoints import TestPromptRequest
49if TYPE_CHECKING: 49 ↛ 50line 49 didn't jump to line 50 because the condition on line 49 was never true
50 from litellm.proxy.prompts.prompt_registry import InMemoryPromptRegistry
51 from litellm.proxy.utils import PrismaClient
53router: Final = APIRouter()
56class _PromptRow(Protocol):
57 @property
58 def id(self) -> str: ... 58 ↛ exitline 58 didn't return from function 'id' because
59 @property
60 def prompt_id(self) -> str: ... 60 ↛ exitline 60 didn't return from function 'prompt_id' because
61 @property
62 def version(self) -> int: ... 62 ↛ exitline 62 didn't return from function 'version' because
63 @property
64 def environment(self) -> str: ... 64 ↛ exitline 64 didn't return from function 'environment' because
65 @property
66 def created_by(self) -> str | None: ... 66 ↛ exitline 66 didn't return from function 'created_by' because
67 @property
68 def created_at(self) -> "datetime": ... 68 ↛ exitline 68 didn't return from function 'created_at' because
69 @property
70 def updated_at(self) -> "datetime": ... 70 ↛ exitline 70 didn't return from function 'updated_at' because
71 @property
72 def litellm_params(self) -> str | Mapping[str, object]: ... 72 ↛ exitline 72 didn't return from function 'litellm_params' because
73 @property
74 def prompt_info(self) -> str | Mapping[str, object] | None: ... 74 ↛ exitline 74 didn't return from function 'prompt_info' because
76 def model_dump(self) -> Mapping[str, object]: ... 76 ↛ exitline 76 didn't return from function 'model_dump' because
79class _PromptRowData(BaseModel):
80 prompt_id: str
81 version: int = 1
82 environment: str = "development"
83 created_by: str | None = None
84 litellm_params: str | Mapping[str, object] | None = None
85 prompt_info: str | Mapping[str, object] | None = None
86 created_at: datetime | None = None
87 updated_at: datetime | None = None
90class _PromptTableActions(Protocol):
91 def find_many( 91 ↛ exitline 91 didn't return from function 'find_many' because
92 self,
93 *,
94 where: Mapping[str, str | int],
95 order: Mapping[str, str] = ...,
96 take: int = ...,
97 distinct: Sequence[str] = ...,
98 ) -> Awaitable[Sequence[_PromptRow]]: ...
100 def create(self, *, data: Mapping[str, str | int | None]) -> Awaitable[_PromptRow]: ... 100 ↛ exitline 100 didn't return from function 'create' because
102 def update(self, *, where: Mapping[str, str | int], data: Mapping[str, str]) -> Awaitable[_PromptRow | None]: ... 102 ↛ exitline 102 didn't return from function 'update' because
104 def delete_many(self, *, where: Mapping[str, str]) -> Awaitable[int]: ... 104 ↛ exitline 104 didn't return from function 'delete_many' because
107def _prompt_table(prisma_client: "PrismaClient") -> _PromptTableActions:
108 return PromptRepository(prisma_client).table
111def get_latest_prompt_versions(prompts: list[PromptSpec]) -> list[PromptSpec]:
112 """
113 Filter prompts down to the latest version per (base prompt id, environment).
114 """
115 sorted_prompts: Final = sorted(prompts, key=lambda prompt: get_version_number(prompt_id=prompt.prompt_id))
116 latest_prompts: Final = {
117 (get_base_prompt_id(prompt_id=prompt.prompt_id), prompt_environment_or_default(prompt.environment)): prompt
118 for prompt in sorted_prompts
119 }
120 return list(latest_prompts.values())
123async def get_next_version_for_prompt(
124 prisma_client: "PrismaClient", prompt_id: str, environment: str = DEFAULT_PROMPT_ENVIRONMENT
125) -> int:
126 """
127 Get the next version number for a prompt in a specific environment.
129 Args:
130 prisma_client: Prisma database client
131 prompt_id: Base prompt ID
132 environment: The environment to check versions for
134 Returns:
135 Next version number (1 if no versions exist, max_version + 1 otherwise)
136 """
137 existing_prompts: Final = await _prompt_table(prisma_client).find_many(
138 where={"prompt_id": prompt_id, "environment": environment}
139 )
141 if existing_prompts:
142 max_version: Final = max(p.version for p in existing_prompts)
143 return max_version + 1
144 else:
145 return 1
148def create_versioned_prompt_spec(db_prompt: _PromptRow) -> PromptSpec:
149 """
150 Helper function to create a PromptSpec with versioned prompt_id from a DB prompt entry.
152 Args:
153 db_prompt: The DB prompt object (from prisma)
155 Returns:
156 PromptSpec with versioned prompt_id (e.g., "chat_prompt.v1")
157 """
158 import json
160 from litellm.types.prompts.init_prompts import PromptLiteLLMParams
162 row: Final = _PromptRowData.model_validate(db_prompt.model_dump())
164 litellm_params_data: Final = row.litellm_params
165 litellm_params_dict: Final[Mapping[str, object] | None] = (
166 json.loads(litellm_params_data) if isinstance(litellm_params_data, str) else litellm_params_data
167 )
168 litellm_params: Final = PromptLiteLLMParams.model_validate(litellm_params_dict)
170 prompt_info_data: Final = row.prompt_info
171 if prompt_info_data: 171 ↛ 177line 171 didn't jump to line 177 because the condition on line 171 was always true
172 prompt_info_dict: Final[Mapping[str, object]] = (
173 json.loads(prompt_info_data) if isinstance(prompt_info_data, str) else prompt_info_data
174 )
175 prompt_info = PromptInfo.model_validate(prompt_info_dict)
176 else:
177 prompt_info = PromptInfo(prompt_type="db")
179 versioned_prompt_id: Final = f"{row.prompt_id}.v{row.version}"
181 return PromptSpec(
182 prompt_id=versioned_prompt_id,
183 litellm_params=litellm_params,
184 prompt_info=prompt_info,
185 created_at=row.created_at,
186 updated_at=row.updated_at,
187 version=row.version,
188 environment=row.environment,
189 created_by=row.created_by,
190 )
193class Prompt(BaseModel):
194 prompt_id: str
195 litellm_params: PromptLiteLLMParams
196 prompt_info: PromptInfo | None = None
199AMBIGUOUS_PROMPT_DATA_ERROR: Final = (
200 "litellm_params.prompt_id cannot be combined with prompt_data keyed by template name. "
201 'Send a flat template, prompt_data={"content": "...", "metadata": {...}}, together with litellm_params.prompt_id, '
202 'or send prompt_data={"<template_id>": {"content": "...", "metadata": {...}}} without litellm_params.prompt_id.'
203)
206def is_ambiguous_keyed_prompt_data(litellm_params: PromptLiteLLMParams) -> bool:
207 extra_fields: Final = litellm_params.model_extra or {}
208 prompt_data: Final = extra_fields.get("prompt_data")
209 if not litellm_params.prompt_id or not isinstance(prompt_data, dict): 209 ↛ 211line 209 didn't jump to line 211 because the condition on line 209 was always true
210 return False
211 return bool(prompt_data) and "content" not in prompt_data
214class PatchPromptRequest(BaseModel):
215 litellm_params: PromptLiteLLMParams | None = None
216 prompt_info: PromptInfo | None = None
219@router.get(
220 "/prompts/list",
221 tags=["Prompt Management"],
222 dependencies=[Depends(user_api_key_auth)],
223 response_model=ListPromptsResponse,
224)
225async def list_prompts(
226 environment: str | None = None,
227 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
228):
229 """
230 List the prompts that are available on the proxy server
232 👉 [Prompt docs](https://docs.litellm.ai/docs/proxy/prompt_management)
234 Example Request:
235 ```bash
236 curl -X GET "http://localhost:4000/prompts/list" -H "Authorization: Bearer <your_api_key>"
237 ```
239 Example Response:
240 ```json
241 {
242 "prompts": [
243 {
244 "prompt_id": "my_prompt_id",
245 "litellm_params": {
246 "prompt_id": "my_prompt_id",
247 "prompt_integration": "dotprompt",
248 "prompt_directory": "/path/to/prompts"
249 },
250 "prompt_info": {
251 "prompt_type": "config"
252 },
253 "created_at": "2023-11-09T12:34:56.789Z",
254 "updated_at": "2023-11-09T12:34:56.789Z"
255 }
256 ]
257 }
258 ```
259 """
260 from litellm.proxy.prompts.prompt_registry import IN_MEMORY_PROMPT_REGISTRY
262 # check key metadata for prompts
263 key_metadata: Final = user_api_key_dict.metadata
264 if key_metadata is not None: 264 ↛ 292line 264 didn't jump to line 292 because the condition on line 264 was always true
265 prompts: Final = cast(list[str] | None, key_metadata.get("prompts", None))
266 if prompts is not None: 266 ↛ 267line 266 didn't jump to line 267 because the condition on line 266 was never true
267 allowed_prompt_ids: Final = frozenset(prompts)
268 allowed_prompts: Final = [
269 spec
270 for spec in IN_MEMORY_PROMPT_REGISTRY.IN_MEMORY_PROMPTS.values()
271 if spec.prompt_id in allowed_prompt_ids
272 or get_base_prompt_id(prompt_id=spec.prompt_id) in allowed_prompt_ids
273 ]
274 all_prompts = get_latest_prompt_versions(prompts=allowed_prompts)
275 if environment:
276 all_prompts = [p for p in all_prompts if p.environment == environment]
277 prompt_list: Final = []
278 for original_prompt in all_prompts:
279 # Create a copy with base prompt_id (without version suffix)
280 prompt_copy = PromptSpec(
281 prompt_id=get_base_prompt_id(prompt_id=original_prompt.prompt_id),
282 litellm_params=original_prompt.litellm_params,
283 prompt_info=original_prompt.prompt_info,
284 created_at=original_prompt.created_at,
285 updated_at=original_prompt.updated_at,
286 environment=original_prompt.environment,
287 created_by=original_prompt.created_by,
288 )
289 prompt_list.append(prompt_copy)
290 return ListPromptsResponse(prompts=prompt_list)
291 # check if user is proxy admin - show all prompts
292 if user_api_key_has_admin_view(user_api_key_dict): 292 ↛ 313line 292 didn't jump to line 313 because the condition on line 292 was always true
293 # Get all prompts and filter to show only the latest version of each
294 all_prompts = list(IN_MEMORY_PROMPT_REGISTRY.IN_MEMORY_PROMPTS.values())
295 if environment:
296 all_prompts = [p for p in all_prompts if p.environment == environment]
297 latest_prompts: Final = get_latest_prompt_versions(prompts=all_prompts)
298 # Create copies with base prompt_id (without version suffix) for display
299 prompts_for_display: Final = []
300 for original_prompt in latest_prompts: 300 ↛ 301line 300 didn't jump to line 301 because the loop on line 300 never started
301 prompt_copy = PromptSpec(
302 prompt_id=get_base_prompt_id(prompt_id=original_prompt.prompt_id),
303 litellm_params=original_prompt.litellm_params,
304 prompt_info=original_prompt.prompt_info,
305 created_at=original_prompt.created_at,
306 updated_at=original_prompt.updated_at,
307 environment=original_prompt.environment,
308 created_by=original_prompt.created_by,
309 )
310 prompts_for_display.append(prompt_copy)
311 return ListPromptsResponse(prompts=prompts_for_display)
312 else:
313 return ListPromptsResponse(prompts=[])
316@router.get(
317 "/prompts/{prompt_id}/versions",
318 tags=["Prompt Management"],
319 dependencies=[Depends(user_api_key_auth)],
320 response_model=ListPromptsResponse,
321)
322async def get_prompt_versions(
323 prompt_id: str,
324 environment: str | None = None,
325 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
326):
327 """
328 Get all versions of a specific prompt by base prompt ID
330 👉 [Prompt docs](https://docs.litellm.ai/docs/proxy/prompt_management)
332 Example Request:
333 ```bash
334 curl -X GET "http://localhost:4000/prompts/jack_success/versions" \\
335 -H "Authorization: Bearer <your_api_key>"
336 ```
338 Example Response:
339 ```json
340 {
341 "prompts": [
342 {
343 "prompt_id": "jack_success.v1",
344 "litellm_params": {...},
345 "prompt_info": {"prompt_type": "db"},
346 "created_at": "2023-11-09T12:34:56.789Z",
347 "updated_at": "2023-11-09T12:34:56.789Z"
348 },
349 {
350 "prompt_id": "jack_success.v2",
351 "litellm_params": {...},
352 "prompt_info": {"prompt_type": "db"},
353 "created_at": "2023-11-09T13:45:12.345Z",
354 "updated_at": "2023-11-09T13:45:12.345Z"
355 }
356 ]
357 }
358 ```
359 """
360 from litellm.proxy.prompts.prompt_registry import IN_MEMORY_PROMPT_REGISTRY
361 from litellm.proxy.proxy_server import prisma_client
363 # Only allow proxy admins to view version history
364 if not user_api_key_has_admin_view(user_api_key_dict): 364 ↛ 365line 364 didn't jump to line 365 because the condition on line 364 was never true
365 raise HTTPException(status_code=403, detail="Only proxy admins can view prompt versions")
367 base_prompt_id: Final = get_base_prompt_id(prompt_id=prompt_id)
369 # Query DB for versions
370 versioned_prompts: Final = []
371 if prisma_client is not None: 371 ↛ 395line 371 didn't jump to line 395 because the condition on line 371 was always true
372 where_clause: Final[dict[str, str]] = {"prompt_id": base_prompt_id}
373 if environment:
374 where_clause["environment"] = environment
375 db_prompts: Final = await _prompt_table(prisma_client).find_many(
376 where=where_clause,
377 order={"version": "desc"},
378 )
379 for db_prompt in db_prompts: 379 ↛ 380line 379 didn't jump to line 380 because the loop on line 379 never started
380 spec = create_versioned_prompt_spec(db_prompt=db_prompt)
381 versioned_prompts.append(
382 PromptSpec(
383 prompt_id=base_prompt_id,
384 litellm_params=spec.litellm_params,
385 prompt_info=spec.prompt_info,
386 created_at=spec.created_at,
387 updated_at=spec.updated_at,
388 version=get_version_number(prompt_id=spec.prompt_id),
389 environment=spec.environment,
390 created_by=spec.created_by,
391 )
392 )
393 else:
394 # Fallback: in-memory registry (no DB)
395 all_prompts: Final = list(IN_MEMORY_PROMPT_REGISTRY.IN_MEMORY_PROMPTS.values())
396 prompt_versions: Final = [
397 prompt
398 for prompt in all_prompts
399 if get_base_prompt_id(prompt_id=prompt.prompt_id) == base_prompt_id
400 and (environment is None or prompt.environment == environment)
401 ]
402 for prompt in prompt_versions:
403 version_number = get_version_number(prompt_id=prompt.prompt_id)
404 versioned_prompts.append(
405 PromptSpec(
406 prompt_id=base_prompt_id,
407 litellm_params=prompt.litellm_params,
408 prompt_info=prompt.prompt_info,
409 created_at=prompt.created_at,
410 updated_at=prompt.updated_at,
411 version=version_number,
412 environment=prompt.environment,
413 created_by=prompt.created_by,
414 )
415 )
416 versioned_prompts.sort(key=lambda p: p.version or 1, reverse=True)
418 if not versioned_prompts: 418 ↛ 421line 418 didn't jump to line 421 because the condition on line 418 was always true
419 raise HTTPException(status_code=404, detail=f"No versions found for prompt ID {base_prompt_id}")
421 return ListPromptsResponse(prompts=versioned_prompts)
424def _get_prompt_template(prompt_spec: PromptSpec, base_prompt_id: str) -> PromptTemplateBase | None:
425 """Resolve the raw prompt template from dotprompt content or the in-memory registry."""
426 from litellm.proxy.prompts.prompt_registry import IN_MEMORY_PROMPT_REGISTRY
428 try:
429 dotprompt_content: Final = prompt_spec.litellm_params.dotprompt_content
430 if dotprompt_content:
431 from litellm.integrations.dotprompt import (
432 _get_prompt_data_from_dotprompt_content,
433 )
435 parsed: Final = _get_prompt_data_from_dotprompt_content(dotprompt_content)
436 if parsed:
437 return PromptTemplateBase(
438 litellm_prompt_id=base_prompt_id,
439 content=parsed.get("content", ""),
440 metadata=parsed.get("metadata"),
441 )
442 else:
443 prompt_callback: Final = IN_MEMORY_PROMPT_REGISTRY.get_prompt_callback_for_prompt(prompt=prompt_spec)
444 if prompt_callback is not None:
445 integration_name: Final = prompt_callback.integration_name
446 if integration_name == "dotprompt":
447 from litellm.integrations.dotprompt.dotprompt_manager import (
448 DotpromptManager,
449 )
451 if isinstance(prompt_callback, DotpromptManager):
452 template: Final = prompt_callback.prompt_manager.get_all_prompts_as_json()
453 if template is not None and len(template) == 1:
454 template_id: Final = list(template.keys())[0]
455 return PromptTemplateBase(
456 litellm_prompt_id=template_id,
457 content=template[template_id]["content"],
458 metadata=template[template_id]["metadata"],
459 )
460 except Exception:
461 pass
462 return None
465@router.get(
466 "/prompts/{prompt_id}",
467 tags=["Prompt Management"],
468 dependencies=[Depends(user_api_key_auth)],
469 response_model=PromptInfoResponse,
470)
471@router.get(
472 "/prompts/{prompt_id}/info",
473 tags=["Prompt Management"],
474 dependencies=[Depends(user_api_key_auth)],
475 response_model=PromptInfoResponse,
476)
477async def get_prompt_info(
478 prompt_id: str,
479 environment: str | None = None,
480 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
481):
482 """
483 Get detailed information about a specific prompt by ID, including prompt content
485 👉 [Prompt docs](https://docs.litellm.ai/docs/proxy/prompt_management)
487 Example Request:
488 ```bash
489 curl -X GET "http://localhost:4000/prompts/my_prompt_id/info" \\
490 -H "Authorization: Bearer <your_api_key>"
491 ```
493 Example Response:
494 ```json
495 {
496 "prompt_id": "my_prompt_id",
497 "litellm_params": {
498 "prompt_id": "my_prompt_id",
499 "prompt_integration": "dotprompt",
500 "prompt_directory": "/path/to/prompts"
501 },
502 "prompt_info": {
503 "prompt_type": "config"
504 },
505 "created_at": "2023-11-09T12:34:56.789Z",
506 "updated_at": "2023-11-09T12:34:56.789Z",
507 "content": "System: You are a helpful assistant.\n\nUser: {{user_message}}"
508 }
509 ```
510 """
511 from litellm.proxy.prompts.prompt_registry import IN_MEMORY_PROMPT_REGISTRY
512 from litellm.proxy.proxy_server import prisma_client
514 ## CHECK IF USER HAS ACCESS TO PROMPT
515 prompts: list[str] | None = None
516 if user_api_key_dict.metadata is not None: 516 ↛ 520line 516 didn't jump to line 520 because the condition on line 516 was always true
517 prompts = cast(list[str] | None, user_api_key_dict.metadata.get("prompts", None))
518 if prompts is not None and prompt_id not in prompts: 518 ↛ 519line 518 didn't jump to line 519 because the condition on line 518 was never true
519 raise HTTPException(status_code=400, detail=f"Prompt {prompt_id} not found")
520 if not user_api_key_has_admin_view(user_api_key_dict): 520 ↛ 521line 520 didn't jump to line 521 because the condition on line 520 was never true
521 raise HTTPException(
522 status_code=403,
523 detail=f"You are not authorized to access this prompt. Your role - {user_api_key_dict.user_role}, Your key's prompts - {prompts}",
524 )
526 base_prompt_id: Final = get_base_prompt_id(prompt_id=prompt_id)
528 # Query all environments this prompt exists in (lightweight: distinct on environment)
529 all_environments: list[str] = []
530 if prisma_client is not None: 530 ↛ 540line 530 didn't jump to line 540 because the condition on line 530 was always true
531 all_prompt_rows: Final = await _prompt_table(prisma_client).find_many(
532 where={"prompt_id": base_prompt_id},
533 distinct=["environment"],
534 )
535 all_environments = sorted(set(row.environment for row in all_prompt_rows if row.environment))
537 # If environment is specified, find the version in that environment from DB
538 # If prompt_id has a version suffix (e.g., "testprompt.v2"), fetch that specific version
539 # Otherwise fetch the latest version in that environment
540 prompt_spec = None
541 requested_version: Final = get_version_number(prompt_id=prompt_id) if prompt_id != base_prompt_id else None
542 if environment and prisma_client is not None:
543 where_clause: Final[dict[str, str | int]] = {
544 "prompt_id": base_prompt_id,
545 "environment": environment,
546 }
547 if requested_version is not None: 547 ↛ 548line 547 didn't jump to line 548 because the condition on line 547 was never true
548 where_clause["version"] = requested_version
549 env_prompts: Final = await _prompt_table(prisma_client).find_many(
550 where=where_clause,
551 order={"version": "desc"},
552 take=1,
553 )
554 if env_prompts: 554 ↛ 555line 554 didn't jump to line 555 because the condition on line 554 was never true
555 prompt_spec = create_versioned_prompt_spec(db_prompt=env_prompts[0])
557 if prompt_spec is None: 557 ↛ 562line 557 didn't jump to line 562 because the condition on line 557 was always true
558 prompt_spec = IN_MEMORY_PROMPT_REGISTRY.resolve_prompt_spec(
559 prompt_id, version=requested_version, environment=environment
560 )
562 if prompt_spec is None: 562 ↛ 569line 562 didn't jump to line 569 because the condition on line 562 was always true
563 raise HTTPException(
564 status_code=400,
565 detail=f"Prompt {prompt_id} not found" + (f" in environment {environment}" if environment else ""),
566 )
568 # Extract version number from the prompt_id
569 version_number: Final = get_version_number(prompt_id=prompt_spec.prompt_id)
571 # Create a copy of the prompt spec with the base prompt ID (stripped of version)
572 prompt_spec_response: Final = PromptSpec(
573 prompt_id=get_base_prompt_id(prompt_id=prompt_spec.prompt_id),
574 litellm_params=prompt_spec.litellm_params,
575 prompt_info=prompt_spec.prompt_info,
576 created_at=prompt_spec.created_at,
577 updated_at=prompt_spec.updated_at,
578 version=version_number,
579 environment=prompt_spec.environment,
580 created_by=prompt_spec.created_by,
581 )
583 # Get prompt content
584 prompt_template: Final = _get_prompt_template(prompt_spec, base_prompt_id)
586 return PromptInfoResponse(
587 prompt_spec=prompt_spec_response,
588 raw_prompt_template=prompt_template,
589 environments=all_environments,
590 )
593@router.post(
594 "/prompts",
595 tags=["Prompt Management"],
596 dependencies=[Depends(user_api_key_auth)],
597)
598async def create_prompt(
599 request: Prompt,
600 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
601):
602 """
603 Create a new prompt
605 👉 [Prompt docs](https://docs.litellm.ai/docs/proxy/prompt_management)
607 Example Request:
608 ```bash
609 curl -X POST "http://localhost:4000/prompts" \\
610 -H "Authorization: Bearer <your_api_key>" \\
611 -H "Content-Type: application/json" \\
612 -d '{
613 "prompt_id": "my_prompt",
614 "litellm_params": {
615 "prompt_id": "my_prompt",
616 "prompt_integration": "dotprompt",
617 "prompt_data": {"content": "This is a prompt", "metadata": {"model": "gpt-4"}}
618 },
619 "prompt_info": {
620 "prompt_type": "config"
621 }
622 }'
623 ```
624 """
626 from litellm.proxy.prompts.prompt_registry import IN_MEMORY_PROMPT_REGISTRY
627 from litellm.proxy.proxy_server import prisma_client
629 # Only allow proxy admins to create prompts
630 if user_api_key_dict.user_role is None or ( 630 ↛ 634line 630 didn't jump to line 634 because the condition on line 630 was never true
631 user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN
632 and user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN.value
633 ):
634 raise HTTPException(status_code=403, detail="Only proxy admins can create prompts")
636 if prisma_client is None: 636 ↛ 637line 636 didn't jump to line 637 because the condition on line 636 was never true
637 raise HTTPException(status_code=500, detail=CommonProxyErrors.db_not_connected_error.value)
639 if is_ambiguous_keyed_prompt_data(request.litellm_params): 639 ↛ 640line 639 didn't jump to line 640 because the condition on line 639 was never true
640 raise HTTPException(status_code=400, detail=AMBIGUOUS_PROMPT_DATA_ERROR)
642 try:
643 # Extract environment from request
644 environment: Final = (
645 request.prompt_info.environment
646 if request.prompt_info and request.prompt_info.environment
647 else DEFAULT_PROMPT_ENVIRONMENT
648 )
650 # Get next version number
651 new_version: Final = await get_next_version_for_prompt(
652 prisma_client=prisma_client,
653 prompt_id=request.prompt_id,
654 environment=environment,
655 )
657 # Store prompt in db with version
658 prompt_db_entry: Final = await _prompt_table(prisma_client).create(
659 data={
660 "prompt_id": request.prompt_id,
661 "version": new_version,
662 "environment": environment,
663 "created_by": user_api_key_dict.user_id,
664 "litellm_params": request.litellm_params.model_dump_json(),
665 "prompt_info": (
666 request.prompt_info.model_dump_json()
667 if request.prompt_info
668 else PromptInfo(prompt_type="db").model_dump_json()
669 ),
670 }
671 )
673 # Create versioned prompt spec
674 prompt_spec: Final = create_versioned_prompt_spec(db_prompt=prompt_db_entry)
676 # Initialize the prompt
677 initialized_prompt = IN_MEMORY_PROMPT_REGISTRY.initialize_prompt(prompt=prompt_spec, config_file_path=None)
679 if initialized_prompt is None:
680 raise HTTPException(status_code=500, detail="Failed to initialize prompt")
682 return initialized_prompt
684 except Exception as e:
685 verbose_proxy_logger.exception("Error creating prompt: %s", e)
686 raise HTTPException(status_code=500, detail=str(e))
689@router.put(
690 "/prompts/{prompt_id}",
691 tags=["Prompt Management"],
692 dependencies=[Depends(user_api_key_auth)],
693)
694async def update_prompt(
695 prompt_id: str,
696 request: Prompt,
697 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
698):
699 """
700 Update an existing prompt
702 👉 [Prompt docs](https://docs.litellm.ai/docs/proxy/prompt_management)
704 Example Request:
705 ```bash
706 curl -X PUT "http://localhost:4000/prompts/my_prompt_id" \\
707 -H "Authorization: Bearer <your_api_key>" \\
708 -H "Content-Type: application/json" \\
709 -d '{
710 "prompt_id": "my_prompt",
711 "litellm_params": {
712 "prompt_id": "my_prompt",
713 "prompt_integration": "dotprompt",
714 "prompt_directory": "/path/to/prompts"
715 },
716 "prompt_info": {
717 "prompt_type": "config"
718 }
719 }
720 }'
721 ```
722 """
723 from litellm.proxy.prompts.prompt_registry import IN_MEMORY_PROMPT_REGISTRY
724 from litellm.proxy.proxy_server import prisma_client
726 # Only allow proxy admins to update prompts
727 if user_api_key_dict.user_role is None or ( 727 ↛ 731line 727 didn't jump to line 731 because the condition on line 727 was never true
728 user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN
729 and user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN.value
730 ):
731 raise HTTPException(status_code=403, detail="Only proxy admins can update prompts")
733 if prisma_client is None: 733 ↛ 734line 733 didn't jump to line 734 because the condition on line 733 was never true
734 raise HTTPException(status_code=500, detail=CommonProxyErrors.db_not_connected_error.value)
736 if is_ambiguous_keyed_prompt_data(request.litellm_params): 736 ↛ 737line 736 didn't jump to line 737 because the condition on line 736 was never true
737 raise HTTPException(status_code=400, detail=AMBIGUOUS_PROMPT_DATA_ERROR)
739 try:
740 # Strip version suffix from prompt_id if present (e.g., "jack_success.v1" -> "jack_success")
741 base_prompt_id: Final = get_base_prompt_id(prompt_id=prompt_id)
743 # Extract environment from request
744 environment: Final = (
745 request.prompt_info.environment
746 if request.prompt_info and request.prompt_info.environment
747 else DEFAULT_PROMPT_ENVIRONMENT
748 )
750 # Check if any version of this prompt exists (in any environment)
751 existing_prompts = await _prompt_table(prisma_client).find_many(where={"prompt_id": base_prompt_id})
753 if not existing_prompts: 753 ↛ 759line 753 didn't jump to line 759 because the condition on line 753 was always true
754 raise HTTPException(
755 status_code=404,
756 detail=f"Prompt with ID {base_prompt_id} not found",
757 )
759 if IN_MEMORY_PROMPT_REGISTRY.has_config_prompt(base_prompt_id=base_prompt_id):
760 raise HTTPException(
761 status_code=400,
762 detail="Cannot update config prompts.",
763 )
765 # Get next version number (UPDATE creates a new version)
766 new_version: Final = await get_next_version_for_prompt(
767 prisma_client=prisma_client,
768 prompt_id=base_prompt_id,
769 environment=environment,
770 )
772 # Store new version in db
773 prompt_db_entry: Final = await _prompt_table(prisma_client).create(
774 data={
775 "prompt_id": base_prompt_id,
776 "version": new_version,
777 "environment": environment,
778 "created_by": user_api_key_dict.user_id,
779 "litellm_params": request.litellm_params.model_dump_json(),
780 "prompt_info": (
781 request.prompt_info.model_dump_json()
782 if request.prompt_info
783 else PromptInfo(prompt_type="db").model_dump_json()
784 ),
785 }
786 )
788 # Create versioned prompt spec
789 prompt_spec: Final = create_versioned_prompt_spec(db_prompt=prompt_db_entry)
791 # Initialize the new version
792 initialized_prompt = IN_MEMORY_PROMPT_REGISTRY.initialize_prompt(prompt=prompt_spec, config_file_path=None)
794 if initialized_prompt is None:
795 raise HTTPException(status_code=500, detail="Failed to update prompt")
797 return initialized_prompt
799 except HTTPException as e:
800 raise e
801 except Exception as e:
802 verbose_proxy_logger.exception("Error updating prompt: %s", e)
803 raise HTTPException(status_code=500, detail=str(e))
806@router.delete(
807 "/prompts/{prompt_id}",
808 tags=["Prompt Management"],
809 dependencies=[Depends(user_api_key_auth)],
810)
811async def delete_prompt(
812 prompt_id: str,
813 environment: str | None = None,
814 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
815):
816 """
817 Delete a prompt
819 👉 [Prompt docs](https://docs.litellm.ai/docs/proxy/prompt_management)
821 Example Request:
822 ```bash
823 curl -X DELETE "http://localhost:4000/prompts/my_prompt_id" \\
824 -H "Authorization: Bearer <your_api_key>"
825 ```
827 Example Response:
828 ```json
829 {
830 "message": "Prompt my_prompt_id deleted successfully"
831 }
832 ```
833 """
834 from litellm.proxy.prompts.prompt_registry import IN_MEMORY_PROMPT_REGISTRY
835 from litellm.proxy.proxy_server import prisma_client
837 # Only allow proxy admins to delete prompts
838 if user_api_key_dict.user_role is None or ( 838 ↛ 842line 838 didn't jump to line 842 because the condition on line 838 was never true
839 user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN
840 and user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN.value
841 ):
842 raise HTTPException(status_code=403, detail="Only proxy admins can delete prompts")
844 if prisma_client is None: 844 ↛ 845line 844 didn't jump to line 845 because the condition on line 844 was never true
845 raise HTTPException(status_code=500, detail=CommonProxyErrors.db_not_connected_error.value)
847 try:
848 base_prompt_id: Final = get_base_prompt_id(prompt_id=prompt_id)
849 existing_prompt: Final = IN_MEMORY_PROMPT_REGISTRY.resolve_prompt_spec(prompt_id, environment=environment)
851 if existing_prompt is None: 851 ↛ 854line 851 didn't jump to line 854 because the condition on line 851 was always true
852 raise HTTPException(status_code=404, detail=f"Prompt with ID {prompt_id} not found")
854 if IN_MEMORY_PROMPT_REGISTRY.has_config_prompt(base_prompt_id=base_prompt_id):
855 raise HTTPException(
856 status_code=400,
857 detail="Cannot delete config prompts.",
858 )
860 delete_where: Final[dict[str, str]] = {
861 "prompt_id": base_prompt_id,
862 **({"environment": environment} if environment else {}),
863 }
864 await _prompt_table(prisma_client).delete_many(where=delete_where)
865 IN_MEMORY_PROMPT_REGISTRY.delete_prompts_by_base_id(
866 base_prompt_id=base_prompt_id, environment=environment or None
867 )
869 env_msg: Final = f" from {environment}" if environment else ""
870 return {"message": f"Prompt {base_prompt_id} deleted successfully{env_msg}"}
872 except HTTPException as e:
873 raise e
874 except Exception as e:
875 verbose_proxy_logger.exception("Error deleting prompt: %s", e)
876 raise HTTPException(status_code=500, detail=str(e))
879def _reload_prompt_in_registry(registry: "InMemoryPromptRegistry", updated_prompt_spec: PromptSpec) -> PromptSpec:
880 initialized: Final = registry.reload_prompt(prompt=updated_prompt_spec)
881 if initialized is None:
882 raise HTTPException(status_code=500, detail="Failed to patch prompt")
883 return initialized
886@router.patch(
887 "/prompts/{prompt_id}",
888 tags=["Prompt Management"],
889 dependencies=[Depends(user_api_key_auth)],
890)
891async def patch_prompt(
892 prompt_id: str,
893 request: PatchPromptRequest,
894 environment: str | None = None,
895 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
896):
897 """
898 Partially update an existing prompt
900 👉 [Prompt docs](https://docs.litellm.ai/docs/proxy/prompt_management)
902 This endpoint allows updating specific fields of a prompt without sending the entire object.
903 Only the following fields can be updated:
904 - litellm_params: LiteLLM parameters for the prompt
905 - prompt_info: Additional information about the prompt
907 Example Request:
908 ```bash
909 curl -X PATCH "http://localhost:4000/prompts/my_prompt_id" \\
910 -H "Authorization: Bearer <your_api_key>" \\
911 -H "Content-Type: application/json" \\
912 -d '{
913 "prompt_info": {
914 "prompt_type": "db"
915 }
916 }'
917 ```
918 """
920 from litellm.proxy.prompts.prompt_registry import IN_MEMORY_PROMPT_REGISTRY
921 from litellm.proxy.proxy_server import prisma_client
923 # Only allow proxy admins to patch prompts
924 if user_api_key_dict.user_role is None or ( 924 ↛ 928line 924 didn't jump to line 928 because the condition on line 924 was never true
925 user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN
926 and user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN.value
927 ):
928 raise HTTPException(status_code=403, detail="Only proxy admins can patch prompts")
930 if prisma_client is None: 930 ↛ 931line 930 didn't jump to line 931 because the condition on line 930 was never true
931 raise HTTPException(status_code=500, detail=CommonProxyErrors.db_not_connected_error.value)
933 if request.litellm_params is not None and is_ambiguous_keyed_prompt_data(request.litellm_params): 933 ↛ 934line 933 didn't jump to line 934 because the condition on line 933 was never true
934 raise HTTPException(status_code=400, detail=AMBIGUOUS_PROMPT_DATA_ERROR)
936 try:
937 # Resolve the target row: find the latest version in the given environment
938 base_prompt_id: Final = get_base_prompt_id(prompt_id=prompt_id)
939 env: Final = prompt_environment_or_default(environment)
940 requested_version: Final = get_version_number(prompt_id=prompt_id) if prompt_id != base_prompt_id else None
942 # Build query to find the exact row by composite unique key
943 find_where: Final[dict[str, str | int]] = {
944 "prompt_id": base_prompt_id,
945 "environment": env,
946 }
947 if requested_version is not None: 947 ↛ 948line 947 didn't jump to line 948 because the condition on line 947 was never true
948 find_where["version"] = requested_version
950 db_rows: Final = await _prompt_table(prisma_client).find_many(
951 where=find_where,
952 order={"version": "desc"},
953 take=1,
954 )
955 if not db_rows: 955 ↛ 961line 955 didn't jump to line 961 because the condition on line 955 was always true
956 raise HTTPException(
957 status_code=404,
958 detail=f"Prompt with ID {base_prompt_id} not found in environment {env}",
959 )
961 target_row: Final = db_rows[0]
963 if IN_MEMORY_PROMPT_REGISTRY.has_config_prompt(base_prompt_id=base_prompt_id):
964 raise HTTPException(
965 status_code=400,
966 detail="Cannot update config prompts.",
967 )
969 current_spec: Final = create_versioned_prompt_spec(db_prompt=target_row)
971 updated_litellm_params: Final = (
972 request.litellm_params if request.litellm_params is not None else current_spec.litellm_params
973 )
975 updated_prompt_info: Final = (
976 request.prompt_info if request.prompt_info is not None else current_spec.prompt_info
977 )
979 # Build update data dict
980 update_data: Final[dict[str, str]] = {
981 "litellm_params": updated_litellm_params.model_dump_json(),
982 "prompt_info": updated_prompt_info.model_dump_json(),
983 }
984 if user_api_key_dict.user_id:
985 update_data["created_by"] = user_api_key_dict.user_id
987 # Update by primary key (id) to target exactly one row
988 updated_prompt_db_entry: Final = await _prompt_table(prisma_client).update(
989 where={"id": target_row.id},
990 data=update_data,
991 )
993 if updated_prompt_db_entry is None:
994 raise HTTPException(
995 status_code=404,
996 detail=f"Prompt with ID {base_prompt_id} not found in environment {env}",
997 )
999 updated_prompt_spec: Final = create_versioned_prompt_spec(db_prompt=updated_prompt_db_entry)
1001 return _reload_prompt_in_registry(IN_MEMORY_PROMPT_REGISTRY, updated_prompt_spec)
1003 except HTTPException as e:
1004 raise e
1005 except Exception as e:
1006 verbose_proxy_logger.exception("Error patching prompt: %s", e)
1007 raise HTTPException(status_code=500, detail=str(e))
1010@router.post(
1011 "/prompts/test",
1012 tags=["Prompt Management"],
1013 dependencies=[Depends(user_api_key_auth)],
1014)
1015async def test_prompt(
1016 request: TestPromptRequest,
1017 fastapi_request: Request,
1018 fastapi_response: Response,
1019 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
1020):
1021 """
1022 Test a prompt by rendering it with variables and executing an LLM call.
1024 This endpoint allows testing prompts before saving them to the database.
1025 The response is always streamed.
1027 👉 [Prompt docs](https://docs.litellm.ai/docs/proxy/prompt_management)
1029 Example Request:
1030 ```bash
1031 curl -X POST "http://localhost:4000/prompts/test" \\
1032 -H "Authorization: Bearer <your_api_key>" \\
1033 -H "Content-Type: application/json" \\
1034 -d '{
1035 "dotprompt_content": "---\\nmodel: gpt-4o\\ntemperature: 0.7\\n---\\n\\nUser: Hello {{name}}",
1036 "prompt_variables": {
1037 "name": "World"
1038 }
1039 }'
1040 ```
1041 """
1042 from pydantic import BaseModel
1044 from litellm.integrations.dotprompt.dotprompt_manager import DotpromptManager
1045 from litellm.integrations.dotprompt.prompt_manager import (
1046 PromptManager,
1047 PromptTemplate,
1048 )
1049 from litellm.proxy.common_request_processing import ProxyBaseLLMRequestProcessing
1050 from litellm.proxy.proxy_server import (
1051 general_settings,
1052 llm_router,
1053 proxy_config,
1054 proxy_logging_obj,
1055 select_data_generator,
1056 user_api_base,
1057 user_max_tokens,
1058 user_model,
1059 user_request_timeout,
1060 user_temperature,
1061 version,
1062 )
1064 try:
1065 # Parse the dotprompt content and create PromptTemplate
1066 prompt_manager: Final = PromptManager()
1067 frontmatter, template_content = prompt_manager._parse_frontmatter(content=request.dotprompt_content)
1069 # Create PromptTemplate to leverage existing parameter extraction logic
1070 template: Final = PromptTemplate(content=template_content, metadata=frontmatter, template_id="test_prompt")
1072 # Extract model from template
1073 if not template.model: 1073 ↛ 1077line 1073 didn't jump to line 1077 because the condition on line 1073 was always true
1074 raise HTTPException(status_code=400, detail="Model is required in dotprompt metadata")
1076 # Always render the template to extract system messages and other metadata
1077 variables: Final = request.prompt_variables or {}
1078 rendered_content: Final = prompt_manager.jinja_env.from_string(template_content).render(**variables)
1080 # Convert rendered content to messages using DotpromptManager's method
1081 dotprompt_manager: Final = DotpromptManager()
1082 rendered_messages: Final = dotprompt_manager._convert_to_messages(rendered_content=rendered_content)
1084 if not rendered_messages:
1085 raise HTTPException(status_code=400, detail="No messages found in rendered prompt")
1087 # If conversation history is provided, use it but preserve system messages
1088 if request.conversation_history:
1089 # Extract system messages from rendered prompt
1090 system_messages: Final = [msg for msg in rendered_messages if msg.get("role") == "system"]
1091 # Use conversation history for user/assistant messages
1092 messages = system_messages + request.conversation_history
1093 else:
1094 messages = rendered_messages
1096 # Use PromptTemplate's optional_params which already extracts all parameters
1097 optional_params: Final = template.optional_params.copy()
1099 # Always stream the response
1100 optional_params["stream"] = True
1102 # Build request data for chat completion
1103 data: Final = {
1104 "model": template.model,
1105 "messages": messages,
1106 }
1107 data.update(optional_params)
1109 is_request_body_safe(
1110 request_body=data,
1111 general_settings=general_settings,
1112 llm_router=llm_router,
1113 model=data.get("model", ""),
1114 )
1116 # Use ProxyBaseLLMRequestProcessing to go through all proxy logic
1117 base_llm_response_processor: Final = ProxyBaseLLMRequestProcessing(data=data)
1118 result: Final[object] = await base_llm_response_processor.base_process_llm_request(
1119 request=fastapi_request,
1120 fastapi_response=fastapi_response,
1121 user_api_key_dict=user_api_key_dict,
1122 route_type="acompletion",
1123 proxy_logging_obj=proxy_logging_obj,
1124 llm_router=llm_router,
1125 general_settings=general_settings,
1126 proxy_config=proxy_config,
1127 select_data_generator=select_data_generator,
1128 model=None,
1129 user_model=user_model,
1130 user_temperature=user_temperature,
1131 user_request_timeout=user_request_timeout,
1132 user_max_tokens=user_max_tokens,
1133 user_api_base=user_api_base,
1134 version=version,
1135 )
1137 if isinstance(result, BaseModel):
1138 return result.model_dump(exclude_none=True, exclude_unset=True)
1139 else:
1140 return result
1142 except HTTPException as e:
1143 raise e
1144 except ValueError as e:
1145 raise HTTPException(status_code=400, detail=str(e))
1146 except Exception as e:
1147 verbose_proxy_logger.exception("Error testing prompt: %s", e)
1148 raise HTTPException(status_code=500, detail=str(e))
1151@router.post(
1152 "/utils/dotprompt_json_converter",
1153 tags=["prompts", "utils"],
1154 dependencies=[Depends(user_api_key_auth)],
1155)
1156async def convert_prompt_file_to_json(
1157 file: UploadFile = File(...),
1158 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
1159) -> Mapping[str, object]:
1160 """
1161 Convert a .prompt file to JSON format.
1163 This endpoint accepts a .prompt file upload and returns the equivalent JSON representation
1164 that can be stored in a database or used programmatically.
1166 Returns the JSON structure with 'content' and 'metadata' fields.
1167 """
1168 global general_settings
1169 from litellm.integrations.dotprompt.prompt_manager import PromptManager
1171 # Validate file extension
1172 if not file.filename or not file.filename.endswith(".prompt"): 1172 ↛ 1175line 1172 didn't jump to line 1175 because the condition on line 1172 was always true
1173 raise HTTPException(status_code=400, detail="File must have .prompt extension")
1175 temp_file_path = None
1176 try:
1177 # Read file content
1178 file_content: Final = await file.read()
1180 # Create temporary file — use safe_filename to prevent path traversal
1181 temp_file_path = Path(tempfile.mkdtemp()) / safe_filename(file.filename)
1182 temp_file_path.write_bytes(file_content)
1184 # Create a PromptManager instance just for conversion
1185 prompt_manager: Final = PromptManager()
1187 # Convert to JSON
1188 json_data: Final = prompt_manager.prompt_file_to_json(temp_file_path)
1190 # Extract prompt ID from filename
1191 prompt_id: Final = temp_file_path.stem
1193 return {
1194 "prompt_id": prompt_id,
1195 "json_data": json_data,
1196 }
1198 except Exception as e:
1199 raise HTTPException(status_code=500, detail=f"Error converting prompt file: {e}")
1201 finally:
1202 # Clean up temp file
1203 if temp_file_path and temp_file_path.exists():
1204 temp_file_path.unlink()
1205 # Also try to remove the temp directory if it's empty
1206 try:
1207 temp_file_path.parent.rmdir()
1208 except OSError:
1209 pass # Directory not empty or other error