diff --git a/pyproject.toml b/pyproject.toml
index 748ca4b..d8f0561 100644
--- a/pyproject.toml
+++ b/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",
diff --git a/registry/api/management_routes.py b/registry/api/management_routes.py
new file mode 100644
index 0000000..b27ca1d
--- /dev/null
+++ b/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
diff --git a/registry/main.py b/registry/main.py
index 06ab172..6b5a462 100644
--- a/registry/main.py
+++ b/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"])
 
diff --git a/registry/schemas/management.py b/registry/schemas/management.py
new file mode 100644
index 0000000..38ba908
--- /dev/null
+++ b/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
diff --git a/registry/utils/keycloak_manager.py b/registry/utils/keycloak_manager.py
index 88a02cf..628d576 100644
--- a/registry/utils/keycloak_manager.py
+++ b/registry/utils/keycloak_manager.py
@@ -7,9 +7,10 @@
 
 import os
 import logging
-import httpx
-from typing import Dict, Any, List, Optional
 import base64
+from typing import Dict, Any, List, Optional
+
+import httpx
 
 
 logger = logging.getLogger(__name__)
@@ -21,6 +22,10 @@
 KEYCLOAK_ADMIN_PASSWORD: Optional[str] = os.environ.get("KEYCLOAK_ADMIN_PASSWORD")
 
 
+class KeycloakAdminError(RuntimeError):
+    """Raised when Keycloak admin API operations fail."""
+
+
 async def _get_keycloak_admin_token() -> str:
     """
     Get admin access token from Keycloak for Admin API calls.
@@ -69,6 +74,52 @@ async def _get_keycloak_admin_token() -> str:
         raise Exception(f"Failed to authenticate with Keycloak: {e}") from e
 
 
+def _auth_headers(token: str, content_type: Optional[str] = "application/json") -> Dict[str, str]:
+    """Build auth headers for Keycloak admin API."""
+    headers = {"Authorization": f"Bearer {token}"}
+    if content_type:
+        headers["Content-Type"] = content_type
+    return headers
+
+
+async def _get_group_name_map(
+    client: httpx.AsyncClient,
+    token: str,
+) -> Dict[str, str]:
+    """Return mapping of Keycloak group name to ID."""
+    groups_url = f"{KEYCLOAK_ADMIN_URL}/admin/realms/{KEYCLOAK_REALM}/groups"
+    response = await client.get(groups_url, headers=_auth_headers(token, None))
+    response.raise_for_status()
+    groups = response.json()
+    return {group.get("name"): group.get("id") for group in groups if group.get("id")}
+
+
+async def _find_client_uuid(
+    client: httpx.AsyncClient,
+    token: str,
+    client_id: str,
+) -> Optional[str]:
+    """Look up a client UUID by clientId."""
+    clients_url = f"{KEYCLOAK_ADMIN_URL}/admin/realms/{KEYCLOAK_REALM}/clients"
+    response = await client.get(
+        clients_url,
+        headers=_auth_headers(token, None),
+        params={"clientId": client_id},
+    )
+    response.raise_for_status()
+    clients = response.json()
+    if clients:
+        return clients[0].get("id")
+    return None
+
+
+def _extract_resource_id(location_header: Optional[str]) -> Optional[str]:
+    """Extract trailing resource ID from a Location header."""
+    if not location_header:
+        return None
+    return location_header.rstrip("/").split("/")[-1]
+
+
 async def create_keycloak_group(
     group_name: str,
     description: str = ""
@@ -290,3 +341,351 @@ async def group_exists_in_keycloak(
         return True
     except Exception:
         return False
+
+
+def _normalize_group_list(groups: List[str]) -> List[str]:
+    """Clean and validate incoming group list."""
+    normalized = [group.strip() for group in groups if group and group.strip()]
+    if not normalized:
+        raise KeycloakAdminError("At least one group must be provided")
+    return normalized
+
+
+async def _assign_user_to_groups_by_name(
+    client: httpx.AsyncClient,
+    token: str,
+    user_id: str,
+    groups: List[str],
+) -> None:
+    """Assign a Keycloak user/service account to a set of groups."""
+    if not groups:
+        return
+
+    name_map = await _get_group_name_map(client, token)
+    for group_name in groups:
+        group_id = name_map.get(group_name)
+        if not group_id:
+            raise KeycloakAdminError(f"Group '{group_name}' not found in Keycloak")
+
+        assign_url = f"{KEYCLOAK_ADMIN_URL}/admin/realms/{KEYCLOAK_REALM}/users/{user_id}/groups/{group_id}"
+        response = await client.put(assign_url, headers=_auth_headers(token, None))
+        if response.status_code not in (204, 409):
+            logger.error("Failed assigning user %s to group %s: %s", user_id, group_name, response.text)
+            raise KeycloakAdminError(
+                f"Failed to assign group '{group_name}' (HTTP {response.status_code})"
+            )
+
+
+async def _get_user_groups(
+    client: httpx.AsyncClient,
+    token: str,
+    user_id: str,
+) -> List[str]:
+    """Fetch group names for a given Keycloak user."""
+    groups_url = f"{KEYCLOAK_ADMIN_URL}/admin/realms/{KEYCLOAK_REALM}/users/{user_id}/groups"
+    response = await client.get(groups_url, headers=_auth_headers(token, None))
+    response.raise_for_status()
+    groups = response.json()
+    return [group.get("name") for group in groups if group.get("name")]
+
+
+async def _get_user_by_username(
+    client: httpx.AsyncClient,
+    token: str,
+    username: str,
+) -> Optional[Dict[str, Any]]:
+    """Look up a user in Keycloak by username."""
+    users_url = f"{KEYCLOAK_ADMIN_URL}/admin/realms/{KEYCLOAK_REALM}/users"
+    response = await client.get(
+        users_url,
+        headers=_auth_headers(token, None),
+        params={"username": username},
+    )
+    response.raise_for_status()
+    matches = response.json()
+    for user in matches:
+        if user.get("username") == username:
+            return user
+    return None
+
+
+async def _get_user_by_id(
+    client: httpx.AsyncClient,
+    token: str,
+    user_id: str,
+) -> Dict[str, Any]:
+    """Fetch a user document by ID."""
+    user_url = f"{KEYCLOAK_ADMIN_URL}/admin/realms/{KEYCLOAK_REALM}/users/{user_id}"
+    response = await client.get(user_url, headers=_auth_headers(token, None))
+    response.raise_for_status()
+    return response.json()
+
+
+async def _ensure_client(
+    client: httpx.AsyncClient,
+    token: str,
+    client_id: str,
+    description: Optional[str],
+) -> str:
+    """Create the client if it does not yet exist and return UUID."""
+    existing_uuid = await _find_client_uuid(client, token, client_id)
+    if existing_uuid:
+        return existing_uuid
+
+    payload = {
+        "clientId": client_id,
+        "name": client_id,
+        "description": description or f"Service account for {client_id}",
+        "enabled": True,
+        "clientAuthenticatorType": "client-secret",
+        "serviceAccountsEnabled": True,
+        "standardFlowEnabled": False,
+        "directAccessGrantsEnabled": False,
+        "publicClient": False,
+        "bearerOnly": False,
+        "protocol": "openid-connect",
+    }
+
+    clients_url = f"{KEYCLOAK_ADMIN_URL}/admin/realms/{KEYCLOAK_REALM}/clients"
+    response = await client.post(clients_url, headers=_auth_headers(token), json=payload)
+    if response.status_code not in (201, 204):
+        logger.error("Failed to create client %s: %s", client_id, response.text)
+        raise KeycloakAdminError(
+            f"Failed to create service account client '{client_id}' (HTTP {response.status_code})"
+        )
+
+    created_id = _extract_resource_id(response.headers.get("Location"))
+    if created_id:
+        return created_id
+
+    client_uuid = await _find_client_uuid(client, token, client_id)
+    if not client_uuid:
+        raise KeycloakAdminError(f"Unable to resolve client ID for '{client_id}' after creation")
+    return client_uuid
+
+
+async def _ensure_groups_mapper(
+    client: httpx.AsyncClient,
+    token: str,
+    client_uuid: str,
+) -> None:
+    """Ensure the standard groups protocol mapper exists for the client."""
+    mapper_url = f"{KEYCLOAK_ADMIN_URL}/admin/realms/{KEYCLOAK_REALM}/clients/{client_uuid}/protocol-mappers/models"
+    response = await client.get(mapper_url, headers=_auth_headers(token, None))
+    response.raise_for_status()
+
+    mappers = response.json()
+    if any(mapper.get("name") == "groups" for mapper in mappers):
+        return
+
+    mapper_payload = {
+        "name": "groups",
+        "protocol": "openid-connect",
+        "protocolMapper": "oidc-group-membership-mapper",
+        "consentRequired": False,
+        "config": {
+            "full.path": "false",
+            "id.token.claim": "true",
+            "access.token.claim": "true",
+            "claim.name": "groups",
+            "userinfo.token.claim": "true",
+        },
+    }
+
+    create_response = await client.post(
+        mapper_url, headers=_auth_headers(token), json=mapper_payload
+    )
+    if create_response.status_code not in (201, 409):
+        logger.error(
+            "Failed to create groups mapper for client %s: %s",
+            client_uuid,
+            create_response.text,
+        )
+        raise KeycloakAdminError(
+            f"Failed to create groups mapper (HTTP {create_response.status_code})"
+        )
+
+
+async def _get_service_account_user_id(
+    client: httpx.AsyncClient,
+    token: str,
+    client_uuid: str,
+) -> str:
+    """Return the user ID of the service account backing a client."""
+    sa_url = f"{KEYCLOAK_ADMIN_URL}/admin/realms/{KEYCLOAK_REALM}/clients/{client_uuid}/service-account-user"
+    response = await client.get(sa_url, headers=_auth_headers(token, None))
+    response.raise_for_status()
+    data = response.json()
+    user_id = data.get("id")
+    if not user_id:
+        raise KeycloakAdminError("Unable to determine service account user ID")
+    return user_id
+
+
+async def _get_client_secret_value(
+    client: httpx.AsyncClient,
+    token: str,
+    client_uuid: str,
+) -> str:
+    """Fetch the client secret value for the specified client."""
+    secret_url = f"{KEYCLOAK_ADMIN_URL}/admin/realms/{KEYCLOAK_REALM}/clients/{client_uuid}/client-secret"
+    response = await client.get(secret_url, headers=_auth_headers(token, None))
+    response.raise_for_status()
+    data = response.json()
+    secret_value = data.get("value")
+    if not secret_value:
+        raise KeycloakAdminError("Keycloak did not return a client secret value")
+    return secret_value
+
+
+async def _set_initial_password(
+    client: httpx.AsyncClient,
+    token: str,
+    user_id: str,
+    password: str,
+    temporary: bool = False,
+) -> None:
+    """Set the initial password for a created user."""
+    password_url = f"{KEYCLOAK_ADMIN_URL}/admin/realms/{KEYCLOAK_REALM}/users/{user_id}/reset-password"
+    payload = {
+        "type": "password",
+        "value": password,
+        "temporary": temporary,
+    }
+    response = await client.put(password_url, headers=_auth_headers(token), json=payload)
+    if response.status_code != 204:
+        logger.error("Failed to set initial password for user %s: %s", user_id, response.text)
+        raise KeycloakAdminError(
+            f"Failed to set password (HTTP {response.status_code})"
+        )
+
+
+async def create_service_account_client(
+    client_id: str,
+    group_names: List[str],
+    description: Optional[str] = None,
+) -> Dict[str, Any]:
+    """
+    Create or update a service account client with group assignments.
+
+    Returns:
+        Dict with client_id, client_uuid, service_account_user_id, client_secret, and groups.
+    """
+    normalized_groups = _normalize_group_list(group_names)
+    admin_token = await _get_keycloak_admin_token()
+
+    async with httpx.AsyncClient(timeout=10.0) as client:
+        client_uuid = await _ensure_client(client, admin_token, client_id, description)
+        await _ensure_groups_mapper(client, admin_token, client_uuid)
+        service_account_user_id = await _get_service_account_user_id(client, admin_token, client_uuid)
+        await _assign_user_to_groups_by_name(client, admin_token, service_account_user_id, normalized_groups)
+        client_secret = await _get_client_secret_value(client, admin_token, client_uuid)
+
+    logger.info("Configured service account client '%s' with groups: %s", client_id, normalized_groups)
+    return {
+        "client_id": client_id,
+        "client_uuid": client_uuid,
+        "service_account_user_id": service_account_user_id,
+        "client_secret": client_secret,
+        "groups": normalized_groups,
+    }
+
+
+async def create_human_user_account(
+    username: str,
+    email: str,
+    first_name: str,
+    last_name: str,
+    groups: List[str],
+    password: Optional[str] = None,
+) -> Dict[str, Any]:
+    """
+    Create a human Keycloak user and assign groups.
+    """
+    normalized_groups = _normalize_group_list(groups)
+    admin_token = await _get_keycloak_admin_token()
+
+    async with httpx.AsyncClient(timeout=10.0) as client:
+        existing = await _get_user_by_username(client, admin_token, username)
+        if existing:
+            raise KeycloakAdminError(f"User '{username}' already exists")
+
+        user_payload = {
+            "username": username,
+            "email": email,
+            "firstName": first_name,
+            "lastName": last_name,
+            "enabled": True,
+            "emailVerified": False,
+        }
+
+        users_url = f"{KEYCLOAK_ADMIN_URL}/admin/realms/{KEYCLOAK_REALM}/users"
+        response = await client.post(users_url, headers=_auth_headers(admin_token), json=user_payload)
+        if response.status_code not in (201, 204):
+            logger.error("Failed to create user %s: %s", username, response.text)
+            raise KeycloakAdminError(f"Failed to create user '{username}' (HTTP {response.status_code})")
+
+        created_id = _extract_resource_id(response.headers.get("Location"))
+        if not created_id:
+            new_user = await _get_user_by_username(client, admin_token, username)
+            if not new_user:
+                raise KeycloakAdminError(f"Unable to resolve new user ID for '{username}'")
+            created_id = new_user.get("id")
+
+        if password:
+            await _set_initial_password(client, admin_token, created_id, password)
+
+        await _assign_user_to_groups_by_name(client, admin_token, created_id, normalized_groups)
+        user_doc = await _get_user_by_id(client, admin_token, created_id)
+        user_doc["groups"] = normalized_groups
+
+    logger.info("Created Keycloak user '%s' with groups: %s", username, normalized_groups)
+    return user_doc
+
+
+async def delete_keycloak_user(username: str) -> bool:
+    """Delete a Keycloak user by username."""
+    admin_token = await _get_keycloak_admin_token()
+
+    async with httpx.AsyncClient(timeout=10.0) as client:
+        user = await _get_user_by_username(client, admin_token, username)
+        if not user:
+            raise KeycloakAdminError(f"User '{username}' not found")
+
+        user_id = user.get("id")
+        delete_url = f"{KEYCLOAK_ADMIN_URL}/admin/realms/{KEYCLOAK_REALM}/users/{user_id}"
+        response = await client.delete(delete_url, headers=_auth_headers(admin_token, None))
+        if response.status_code != 204:
+            logger.error("Failed to delete user %s: %s", username, response.text)
+            raise KeycloakAdminError(f"Failed to delete user '{username}' (HTTP {response.status_code})")
+
+    logger.info("Deleted Keycloak user '%s'", username)
+    return True
+
+
+async def list_keycloak_users(
+    search: Optional[str] = None,
+    max_results: int = 500,
+    include_groups: bool = True,
+) -> List[Dict[str, Any]]:
+    """List users in the Keycloak realm."""
+    admin_token = await _get_keycloak_admin_token()
+    params: Dict[str, Any] = {"max": max_results}
+    if search:
+        params["search"] = search
+
+    users_url = f"{KEYCLOAK_ADMIN_URL}/admin/realms/{KEYCLOAK_REALM}/users"
+    async with httpx.AsyncClient(timeout=10.0) as client:
+        response = await client.get(users_url, headers=_auth_headers(admin_token, None), params=params)
+        response.raise_for_status()
+        users = response.json()
+
+        if include_groups:
+            for user in users:
+                user_id = user.get("id")
+                if not user_id:
+                    user["groups"] = []
+                    continue
+                user["groups"] = await _get_user_groups(client, admin_token, user_id)
+
+        return users