Unified API: Kaapi Backend as a Proxy

backend/app/core/providers.py

@@ -17,6 +17,7 @@ class Provider(str, Enum):
  ANTHROPIC = "anthropic"
  GOOGLE = "google"
  WEBHOOK_SECRET = "webhook_secret"
+ PROXY = "proxy"
 
 
 @dataclass
@@ -61,6 +62,9 @@ class ProviderConfig:
  Provider.WEBHOOK_SECRET: ProviderConfig(
  required_fields=["webhook_secret"], sensitive_fields=["webhook_secret"]
  ),
+ Provider.PROXY: ProviderConfig(
+ required_fields=["api_key"], sensitive_fields=["api_key"]
+ ),
 }
 
 

backend/app/crud/model_config.py

@@ -7,7 +7,7 @@
 from sqlmodel import Session, select
 
 from app.models import ModelConfig
-from app.models.llm.constants import CompletionType
+from app.models.llm.constants import CompletionType, Provider as ProviderEnum
 from app.models.llm.request import ConfigBlob
 from app.models.model_config import (
  ModelConfigBulkUpdateItem,
@@ -137,6 +137,12 @@ def validate_blob_model_or_raise(session: Session, blob: ConfigBlob) -> None:
  completion = blob.completion
  raw_provider = completion.provider
  completion_type = completion.type
+
+ # Proxy forwards the request to the client's own LLM endpoint — no model
+ # lookup, no provider mapping.
+ if completion_type == ProviderEnum.PROXY.value:
+ return
+
  if raw_provider is None:
  return
 
@@ -145,7 +151,7 @@ def validate_blob_model_or_raise(session: Session, blob: ConfigBlob) -> None:
 
  provider = _normalize_provider(raw_provider)
 
- model_name = (completion.params or {}).get("model")
+ model_name = (completion.params or {}).get("model") or None
  if not model_name:
  raise HTTPException(
  status_code=400,

backend/app/models/llm/request.py

@@ -1,6 +1,6 @@
 from datetime import datetime
 from enum import Enum
-from typing import Annotated, Any, Literal, Union
+from typing import Annotated, Any, Literal, Self, Union
 from uuid import UUID, uuid4
 
 import sqlalchemy as sa
@@ -71,6 +71,8 @@ class TextLLMParams(SQLModel):
 
 
 class STTLLMParams(SQLModel):
+ model_config = {"extra": "forbid"}
+
  model: str = DEFAULT_STT_MODEL
  instructions: str | None = None
  input_language: str | None = "auto"
@@ -88,17 +90,35 @@ class STTLLMParams(SQLModel):
 
 
 class TTSLLMParams(SQLModel):
+ model_config = {"extra": "forbid"}
+
  model: str = DEFAULT_TTS_MODEL
  voice: str = DEFAULT_TTS_VOICE
  language: str | None = None
  response_format: Literal["mp3", "wav", "ogg"] | None = "wav"
 
 
-KaapiLLMParams = Union[
- TextLLMParams,
- STTLLMParams,
- TTSLLMParams,
-]
+class ProxyLLMParams(SQLModel):
+ model_config = {"extra": "forbid"}
+
+ client_llm_url: HttpUrl = Field(
+ ...,
+ description=(
+ "HTTPS URL of the client's own LLM endpoint. Kaapi forwards the "
+ "(guardrail-sanitised) input here and applies output guardrails to the response."
+ ),
+ )
+
+ @model_validator(mode="after")
+ def _require_https(self) -> Self:
+ if self.client_llm_url.scheme != "https":
+ raise ValueError(
+ f"client_llm_url must be HTTPS, got scheme: {self.client_llm_url.scheme}"
+ )
+ return self
+
+
+KaapiLLMParams = Union[TextLLMParams, STTLLMParams, TTSLLMParams, ProxyLLMParams]
 
 
 # Input type models for discriminated union
@@ -313,9 +333,39 @@ def validate_params(self):
  return self
 
 
+class ProxyCompletionConfig(SQLModel):
+ """
+ Proxy completion: Kaapi forwards the (guardrail-sanitised) input to the
+ client's own LLM endpoint and applies output guardrails to the response.
+ No upstream provider is dispatched — `provider` is fixed to "proxy" so
+ the discriminated union can route cleanly.
+ """
+
+ provider: Literal["proxy"] = Field(
+ "proxy",
+ description=(
+ "Discriminator value for the proxy variant. Auto-injected when "
+ "type=proxy; clients may omit it."
+ ),
+ )
+ type: Literal["proxy"] = Field(..., description="Must be 'proxy'.")
+ params: dict[str, Any] = Field(
+ ...,
+ description="Proxy params (client_llm_url, ...)",
+ )
+
+ @model_validator(mode="after")
+ def validate_params(self) -> Self:
+ validated = ProxyLLMParams.model_validate(self.params)
+ # mode="json" coerces HttpUrl → plain str so downstream consumers
+ # (httpx.post, urlparse) get the type they expect from params dict.
+ self.params = validated.model_dump(mode="json", exclude_none=True)
+ return self
+
+
 # Discriminated union for completion configs based on provider field
 CompletionConfig = Annotated[
- Union[NativeCompletionConfig, KaapiCompletionConfig],
+ Union[NativeCompletionConfig, KaapiCompletionConfig, ProxyCompletionConfig],
  Field(discriminator="provider"),
 ]
 
@@ -333,6 +383,24 @@ class ConfigBlob(SQLModel):
 
  completion: CompletionConfig = Field(..., description="Completion configuration")
 
+ @model_validator(mode="before")
+ @classmethod
+ def _default_proxy_provider(cls, data: Any) -> Any:
+ """For `type=proxy`, provider is meaningless to the caller.
+ Inject provider="proxy" so the CompletionConfig discriminator routes
+ to ProxyCompletionConfig without forcing the client to set it."""
+ if not isinstance(data, dict):
+ return data
+ completion = data.get("completion")
+ if (
+ isinstance(completion, dict)
+ and completion.get("type") == Provider.PROXY.value
+ ):
+ existing = completion.get("provider")
+ if existing in (None, Provider.PROXY.value):
+ completion["provider"] = Provider.PROXY.value
+ return data
+
  # used for llm-chain to provide prompt interpolation
  prompt_template: PromptTemplate | None = Field(
  default=None,