"""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