Lars
53047b6c16
All checks were successful
Deploy Development / deploy (push) Successful in 38s
Test Suite / pytest-backend (push) Successful in 18s
Test Suite / lint-backend (push) Successful in 1s
Test Suite / compose-smoke (push) Has been skipped
Test Suite / k6 /api/health Baseline (push) Successful in 18s
Test Suite / playwright-smoke (push) Successful in 24s
Migration 005, Registry-Sync, Placeholder-Validation, capability-geschuetzte API, Tests und Abschlussbericht v0.1. Co-authored-by: Cursor <cursoragent@cursor.com>
75 lines
2.4 KiB
Python
75 lines
2.4 KiB
Python
"""Prompt registry and rendering API."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from capabilities import require_capability_ctx
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
from prompt_context import merge_context_values
|
|
from prompt_registry import list_prompt_definitions, load_active_prompt_version, load_prompt_definition
|
|
from prompt_rendering import PromptRenderError, render_prompt_single
|
|
from pydantic import BaseModel, Field
|
|
from tenant_context import TenantContext
|
|
|
|
router = APIRouter(prefix="/api/prompts", tags=["prompts"])
|
|
|
|
|
|
class RenderPromptRequest(BaseModel):
|
|
values: dict[str, Any] = Field(default_factory=dict)
|
|
use_context: bool = True
|
|
|
|
|
|
@router.get("")
|
|
def list_prompts(_ctx=Depends(require_capability_ctx("kairo.prompt.registry.read"))):
|
|
return list_prompt_definitions()
|
|
|
|
|
|
@router.get("/{prompt_key}")
|
|
def get_prompt(prompt_key: str, _ctx=Depends(require_capability_ctx("kairo.prompt.registry.read"))):
|
|
definition = load_prompt_definition(prompt_key)
|
|
if not definition:
|
|
raise HTTPException(status_code=404, detail="Prompt nicht gefunden")
|
|
version = load_active_prompt_version(definition)
|
|
return {
|
|
"definition": definition,
|
|
"active_version": version,
|
|
}
|
|
|
|
|
|
@router.post("/{prompt_key}/render")
|
|
def render_prompt(
|
|
prompt_key: str,
|
|
body: RenderPromptRequest,
|
|
ctx: TenantContext = Depends(require_capability_ctx("kairo.prompt.render")),
|
|
):
|
|
definition = load_prompt_definition(prompt_key)
|
|
if not definition:
|
|
raise HTTPException(status_code=404, detail="Prompt nicht gefunden")
|
|
|
|
if body.use_context and definition["context_kind"] in (
|
|
"kairo.tenant_context",
|
|
"kairo.actor_context",
|
|
):
|
|
if not ctx.tenant_id:
|
|
raise HTTPException(status_code=403, detail="Aktiver Tenant erforderlich für diesen Prompt")
|
|
|
|
values = body.values
|
|
if body.use_context:
|
|
values = merge_context_values(ctx, definition["context_kind"], body.values)
|
|
|
|
try:
|
|
return render_prompt_single(
|
|
prompt_key,
|
|
values=values,
|
|
user_id=ctx.user_id,
|
|
log_execution=True,
|
|
)
|
|
except PromptRenderError as exc:
|
|
status = 400
|
|
if exc.code == "not_found":
|
|
status = 404
|
|
elif exc.code == "mode_not_implemented":
|
|
status = 501
|
|
raise HTTPException(status_code=status, detail=str(exc)) from exc
|