add management API

Author: kanghengliu

pyproject.toml

@@ -35,6 +35,7 @@ dependencies = [
     "langchain-anthropic>=0.3.17",
     "matplotlib>=3.10.5",
     "psutil>=6.1.0",
+    "email-validator>=2.2.0",
     "aiohttp>=3.8.0",
     "rich>=13.0.0",
     "requests>=2.31.0",

registry/api/management_routes.py

@@ -0,0 +1,152 @@
+from __future__ import annotations
+
+import logging
+from typing import Annotated
+
+from fastapi import APIRouter, Depends, HTTPException, status
+
+from ..auth.dependencies import nginx_proxied_auth
+from ..schemas.management import (
+    HumanUserRequest,
+    KeycloakUserSummary,
+    M2MAccountRequest,
+    UserDeleteResponse,
+    UserListResponse,
+)
+from ..utils.keycloak_manager import (
+    KeycloakAdminError,
+    create_human_user_account,
+    create_service_account_client,
+    delete_keycloak_user,
+    list_keycloak_groups,
+    list_keycloak_users,
+)
+
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(prefix="/management", tags=["Management API"])
+
+
+def _translate_keycloak_error(exc: KeycloakAdminError) -> HTTPException:
+    """Map Keycloak admin errors to HTTP responses."""
+    detail = str(exc)
+    lowered = detail.lower()
+    status_code = status.HTTP_502_BAD_GATEWAY
+    if any(keyword in lowered for keyword in ("already exists", "not found", "provided")):
+        status_code = status.HTTP_400_BAD_REQUEST
+    return HTTPException(status_code=status_code, detail=detail)
+
+
+def _require_admin(user_context: dict) -> None:
+    if not user_context.get("is_admin"):
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="Administrator permissions are required for this operation",
+        )
+
+
+@router.get("/iam/users", response_model=UserListResponse)
+async def management_list_users(
+    search: str | None = None,
+    limit: int = 500,
+    user_context: Annotated[dict, Depends(nginx_proxied_auth)] = None,
+):
+    """List Keycloak users for administrators."""
+    _require_admin(user_context)
+    try:
+        raw_users = await list_keycloak_users(search=search, max_results=limit)
+    except KeycloakAdminError as exc:
+        raise _translate_keycloak_error(exc) from exc
+
+    summaries = [
+        KeycloakUserSummary(
+            id=user.get("id", ""),
+            username=user.get("username", ""),
+            email=user.get("email"),
+            firstName=user.get("firstName"),
+            lastName=user.get("lastName"),
+            enabled=user.get("enabled", True),
+            groups=user.get("groups", []),
+        )
+        for user in raw_users
+    ]
+    return UserListResponse(users=summaries, total=len(summaries))
+
+
+@router.post("/iam/users/m2m")
+async def management_create_m2m_user(
+    payload: M2MAccountRequest,
+    user_context: Annotated[dict, Depends(nginx_proxied_auth)] = None,
+):
+    """Create a service account client and return its credentials."""
+    _require_admin(user_context)
+    try:
+        result = await create_service_account_client(
+            client_id=payload.name,
+            group_names=payload.groups,
+            description=payload.description,
+        )
+    except KeycloakAdminError as exc:
+        raise _translate_keycloak_error(exc) from exc
+    return result
+
+
+@router.post("/iam/users/human")
+async def management_create_human_user(
+    payload: HumanUserRequest,
+    user_context: Annotated[dict, Depends(nginx_proxied_auth)] = None,
+):
+    """Create a Keycloak human user and assign groups."""
+    _require_admin(user_context)
+    try:
+        user_doc = await create_human_user_account(
+            username=payload.username,
+            email=payload.email,
+            first_name=payload.first_name,
+            last_name=payload.last_name,
+            groups=payload.groups,
+            password=payload.password,
+        )
+    except KeycloakAdminError as exc:
+        raise _translate_keycloak_error(exc) from exc
+
+    return KeycloakUserSummary(
+        id=user_doc.get("id", ""),
+        username=user_doc.get("username", payload.username),
+        email=user_doc.get("email"),
+        firstName=user_doc.get("firstName"),
+        lastName=user_doc.get("lastName"),
+        enabled=user_doc.get("enabled", True),
+        groups=user_doc.get("groups", payload.groups),
+    )
+
+
+@router.delete("/iam/users/{username}", response_model=UserDeleteResponse)
+async def management_delete_user(
+    username: str,
+    user_context: Annotated[dict, Depends(nginx_proxied_auth)] = None,
+):
+    """Delete a Keycloak user by username."""
+    _require_admin(user_context)
+    try:
+        await delete_keycloak_user(username)
+    except KeycloakAdminError as exc:
+        raise _translate_keycloak_error(exc) from exc
+    return UserDeleteResponse(username=username)
+
+
+@router.get("/iam/groups")
+async def management_list_keycloak_groups(
+    user_context: Annotated[dict, Depends(nginx_proxied_auth)] = None,
+):
+    """List raw Keycloak IAM groups (without scopes)."""
+    _require_admin(user_context)
+    try:
+        return await list_keycloak_groups()
+    except Exception as exc:  # noqa: BLE001 - surface upstream failure
+        logger.error("Failed to list Keycloak groups: %s", exc)
+        raise HTTPException(
+            status_code=status.HTTP_502_BAD_GATEWAY,
+            detail="Unable to list Keycloak groups",
+        ) from exc

registry/main.py

@@ -24,6 +24,7 @@
 from registry.api.wellknown_routes import router as wellknown_router
 from registry.api.registry_routes import router as registry_router
 from registry.api.agent_routes import router as agent_router
+from registry.api.management_routes import router as management_router
 from registry.health.routes import router as health_router
 
 # Import auth dependencies
@@ -201,6 +202,7 @@ async def lifespan(app: FastAPI):
 app.include_router(auth_router, prefix="/api/auth", tags=["Authentication"])
 app.include_router(servers_router, prefix="/api", tags=["Server Management"])
 app.include_router(agent_router, prefix="/api", tags=["Agent Management"])
+app.include_router(management_router, prefix="/api")
 app.include_router(search_router, prefix="/api/search", tags=["Semantic Search"])
 app.include_router(health_router, prefix="/api/health", tags=["Health Monitoring"])
 

registry/schemas/management.py

@@ -0,0 +1,54 @@
+from __future__ import annotations
+
+from typing import List, Optional
+
+from pydantic import BaseModel, EmailStr, Field
+
+
+class M2MAccountRequest(BaseModel):
+    """Payload for creating a Keycloak service account client."""
+
+    name: str = Field(..., min_length=1)
+    groups: List[str] = Field(..., min_length=1)
+    description: Optional[str] = None
+
+
+class HumanUserRequest(BaseModel):
+    """Payload for creating a Keycloak human user."""
+
+    username: str = Field(..., min_length=1)
+    email: EmailStr
+    first_name: str = Field(..., min_length=1, alias="firstname")
+    last_name: str = Field(..., min_length=1, alias="lastname")
+    groups: List[str] = Field(..., min_length=1)
+    password: Optional[str] = Field(
+        None, description="Initial password (optional, generated elsewhere)"
+    )
+
+    model_config = {"populate_by_name": True}
+
+
+class UserDeleteResponse(BaseModel):
+    """Standard response returned when a Keycloak user is deleted."""
+
+    username: str
+    deleted: bool = True
+
+
+class KeycloakUserSummary(BaseModel):
+    """Subset of Keycloak user information exposed through the API."""
+
+    id: str
+    username: str
+    email: Optional[str] = None
+    firstName: Optional[str] = None
+    lastName: Optional[str] = None
+    enabled: bool = True
+    groups: List[str] = Field(default_factory=list)
+
+
+class UserListResponse(BaseModel):
+    """Wrapper for list users endpoint."""
+
+    users: List[KeycloakUserSummary] = Field(default_factory=list)
+    total: int