Component: Tenant APIs¶

Date: December 16, 2025
Status: Architecture Specification
Component Type: Backend API Services

Table of Contents¶

Overview
Architecture
Tenant Management API
User Management API
CAT Management API
Data Isolation
API Endpoints
Authentication & Authorization
Error Handling
Implementation Details
Testing Strategy
Deployment

Overview¶

The Tenant APIs are FastAPI-based backend services that handle tenant-specific operations, user management, CAT (Conversational AI Technology) management, and data isolation. These APIs enforce tenant boundaries at the database and application level.

Key Responsibilities¶

Tenant Management: CRUD operations for tenants
User Management: Tenant-scoped user operations
CAT Management: Tenant-scoped CAT operations
Data Isolation: Enforce tenant boundaries in data access
Configuration Management: Tenant-specific configurations
Access Control: Tenant-level authorization

API Services¶

Security API (bast-security-api)
Authentication
Authorization
JWT token management
User Management API (bast-user-management-api)
Tenant management
User management (per tenant)
CAT management (per tenant)
Configuration management
CAT API (bast-cat-api)
CAT-specific operations
Chat processing
Conversation management

Architecture¶

High-Level Architecture¶

┌─────────────────────────────────────────────────────────┐
│              BFF Tenant Aware                            │
│  - Tenant ID in headers (X-Tenant-ID)                  │
└────────────────────┬────────────────────────────────────┘
                     │ HTTP Request
                     │ Headers: X-Tenant-ID, X-CAT-Prefix
                     ▼
┌─────────────────────────────────────────────────────────┐
│              Tenant APIs (FastAPI)                      │
│                                                          │
│  ┌──────────────────────────────────────────────────┐  │
│  │  Security API                                    │  │
│  │  - Authentication (tenant-aware)                │  │
│  │  - Authorization (tenant-scoped)                │  │
│  └──────────────────────────────────────────────────┘  │
│                        │                                │
│  ┌──────────────────────────────────────────────────┐  │
│  │  User Management API                              │  │
│  │  - Tenant CRUD                                   │  │
│  │  - User CRUD (per tenant)                        │  │
│  │  - CAT CRUD (per tenant)                         │  │
│  │  - Configuration (per tenant)                    │  │
│  └──────────────────────────────────────────────────┘  │
│                        │                                │
│  ┌──────────────────────────────────────────────────┐  │
│  │  CAT API                                         │  │
│  │  - Chat (per CAT)                                │  │
│  │  - Conversations (per CAT)                       │  │
│  │  - Configuration (per CAT)                        │  │
│  └──────────────────────────────────────────────────┘  │
└────────────────────┬────────────────────────────────────┘
                     │
         ┌───────────┴───────────┐
         │                       │
┌────────▼────────┐    ┌────────▼────────┐
│  MongoDB        │    │  Other DBs      │
│  - auth_db      │    │  - GraphDB       │
│  - config_db    │    │  - VectorDB      │
│  - {prefix}_*   │    │  - CloudStorage  │
└─────────────────┘    └──────────────────┘

Database Structure¶

auth_db (shared)
├── users (collection)
│   └── { user_id, email, organization, tenants: [] }
└── user_memories (collection)

config_db (shared)
├── tenants (collection)
│   └── { tenant_id, name, slug, status, config }
└── cat_config (collection)
    └── { prefix, tenant_id, name, config }

{tenant_id}_cat_data (tenant-specific)
├── logs (collection)
├── sessions (collection)
└── metrics (collection)

{prefix}_cat_data (CAT-specific)
├── logs (collection)
├── sessions (collection)
└── metrics (collection)

Tenant Management API¶

Endpoints¶

Get Tenant¶

GET /api/v1/tenants/{tenant_id}

Response:
{
  "tenant_id": "acme",
  "name": "Acme Corporation",
  "slug": "acme",
  "status": "active",
  "created_at": "2025-01-01T00:00:00Z",
  "updated_at": "2025-01-01T00:00:00Z",
  "subscription": {
    "tier": "standard",
    "expires_at": null
  }
}

Get Tenant Configuration¶

GET /api/v1/tenants/{tenant_id}/config

Response:
{
  "tenant_id": "acme",
  "branding": {
    "logo": "https://cdn.bast.ai/tenants/acme/logo.png",
    "logo_dark": "https://cdn.bast.ai/tenants/acme/logo-dark.png",
    "colors": {
      "primary": "#2295c4",
      "secondary": "#02736b",
      "accent": "#55ccd9"
    },
    "theme": "auto"
  },
  "features": {
    "enabled": ["ontology", "governance", "analytics"],
    "disabled": ["advanced_ai"],
    "limits": {
      "max_cats": 10,
      "max_users": 100,
      "max_storage": 1000
    }
  },
  "environments": ["dev", "stage", "prod"]
}

Update Tenant Configuration¶

PUT /api/v1/tenants/{tenant_id}/config
Headers: X-Tenant-ID: {tenant_id}, Authorization: Bearer {token}

Request Body:
{
  "branding": {
    "logo": "https://...",
    "colors": {
      "primary": "#2295c4"
    }
  }
}

Response:
{
  "success": true,
  "updated_at": "2025-01-01T00:00:00Z"
}

List Tenants (Admin Only)¶

GET /api/v1/tenants
Headers: Authorization: Bearer {admin_token}

Query Parameters:
- limit: int = 10
- offset: int = 0
- status: str = "active"

Response:
{
  "tenants": [
    {
      "tenant_id": "acme",
      "name": "Acme Corporation",
      "status": "active"
    }
  ],
  "total": 100,
  "limit": 10,
  "offset": 0
}

Implementation Example¶

File: bast-ai/src/apis/user_management/bast_user_management_api/routers/tenant.py

from fastapi import APIRouter, Depends, HTTPException, Header
from pydantic import BaseModel
from typing import Optional
from ..dependencies import get_current_user, verify_tenant_access

router = APIRouter(prefix="/api/v1/tenants", tags=["tenants"])

class TenantConfigUpdate(BaseModel):
    branding: Optional[dict] = None
    features: Optional[dict] = None

@router.get("/{tenant_id}")
async def get_tenant(
    tenant_id: str,
    x_tenant_id: Optional[str] = Header(None, alias="X-Tenant-ID"),
    current_user: dict = Depends(get_current_user)
):
    """Get tenant information."""
    # Verify tenant access
    if x_tenant_id and x_tenant_id != tenant_id:
        raise HTTPException(status_code=403, detail="Tenant ID mismatch")

    # Verify user has access to tenant
    if not await verify_tenant_access(current_user["id"], tenant_id):
        raise HTTPException(status_code=403, detail="Access denied")

    # Load tenant from database
    tenant = await get_tenant_from_db(tenant_id)
    if not tenant:
        raise HTTPException(status_code=404, detail="Tenant not found")

    return tenant

@router.get("/{tenant_id}/config")
async def get_tenant_config(
    tenant_id: str,
    x_tenant_id: Optional[str] = Header(None, alias="X-Tenant-ID"),
    current_user: dict = Depends(get_current_user)
):
    """Get tenant configuration."""
    # Verify access
    if x_tenant_id and x_tenant_id != tenant_id:
        raise HTTPException(status_code=403, detail="Tenant ID mismatch")

    if not await verify_tenant_access(current_user["id"], tenant_id):
        raise HTTPException(status_code=403, detail="Access denied")

    # Load config
    config = await get_tenant_config_from_db(tenant_id)
    return config

@router.put("/{tenant_id}/config")
async def update_tenant_config(
    tenant_id: str,
    config_update: TenantConfigUpdate,
    x_tenant_id: Optional[str] = Header(None, alias="X-Tenant-ID"),
    current_user: dict = Depends(get_current_user)
):
    """Update tenant configuration."""
    # Verify access
    if x_tenant_id and x_tenant_id != tenant_id:
        raise HTTPException(status_code=403, detail="Tenant ID mismatch")

    if not await verify_tenant_access(current_user["id"], tenant_id):
        raise HTTPException(status_code=403, detail="Access denied")

    # Update config
    updated = await update_tenant_config_in_db(tenant_id, config_update.dict(exclude_unset=True))
    return {"success": True, "updated_at": updated}

User Management API¶

Endpoints¶

List Users (Per Tenant)¶

GET /api/v1/tenants/{tenant_id}/users
Headers: X-Tenant-ID: {tenant_id}, Authorization: Bearer {token}

Query Parameters:
- limit: int = 10
- offset: int = 0
- role: str = None

Response:
{
  "users": [
    {
      "id": "user123",
      "email": "user@example.com",
      "name": "John Doe",
      "role": "admin",
      "status": "active",
      "created_at": "2025-01-01T00:00:00Z"
    }
  ],
  "total": 50,
  "limit": 10,
  "offset": 0
}

Get User¶

GET /api/v1/tenants/{tenant_id}/users/{user_id}
Headers: X-Tenant-ID: {tenant_id}, Authorization: Bearer {token}

Response:
{
  "id": "user123",
  "email": "user@example.com",
  "name": "John Doe",
  "role": "admin",
  "organization": "acme",
  "tenants": ["acme"],
  "status": "active"
}

Create User¶

POST /api/v1/tenants/{tenant_id}/users
Headers: X-Tenant-ID: {tenant_id}, Authorization: Bearer {token}

Request Body:
{
  "email": "newuser@example.com",
  "name": "Jane Doe",
  "role": "developer",
  "password": "secure_password"
}

Response:
{
  "id": "user456",
  "email": "newuser@example.com",
  "name": "Jane Doe",
  "role": "developer",
  "organization": "acme",
  "status": "active",
  "created_at": "2025-01-01T00:00:00Z"
}

Update User¶

PUT /api/v1/tenants/{tenant_id}/users/{user_id}
Headers: X-Tenant-ID: {tenant_id}, Authorization: Bearer {token}

Request Body:
{
  "name": "Jane Smith",
  "role": "admin"
}

Response:
{
  "id": "user456",
  "email": "newuser@example.com",
  "name": "Jane Smith",
  "role": "admin",
  "updated_at": "2025-01-01T00:00:00Z"
}

Delete User¶

DELETE /api/v1/tenants/{tenant_id}/users/{user_id}
Headers: X-Tenant-ID: {tenant_id}, Authorization: Bearer {token}

Response:
{
  "success": true,
  "deleted_at": "2025-01-01T00:00:00Z"
}

Implementation Example¶

File: bast-ai/src/apis/user_management/bast_user_management_api/routers/users.py

from fastapi import APIRouter, Depends, HTTPException, Header, Query
from pydantic import BaseModel, EmailStr
from typing import Optional, List
from ..dependencies import get_current_user, verify_tenant_access, verify_user_access

router = APIRouter(prefix="/api/v1/tenants/{tenant_id}/users", tags=["users"])

class UserCreate(BaseModel):
    email: EmailStr
    name: str
    role: str
    password: str

class UserUpdate(BaseModel):
    name: Optional[str] = None
    role: Optional[str] = None
    status: Optional[str] = None

@router.get("")
async def list_users(
    tenant_id: str,
    x_tenant_id: Optional[str] = Header(None, alias="X-Tenant-ID"),
    current_user: dict = Depends(get_current_user),
    limit: int = Query(10, ge=1, le=100),
    offset: int = Query(0, ge=0),
    role: Optional[str] = None
):
    """List users for a tenant."""
    # Verify tenant access
    if x_tenant_id and x_tenant_id != tenant_id:
        raise HTTPException(status_code=403, detail="Tenant ID mismatch")

    if not await verify_tenant_access(current_user["id"], tenant_id):
        raise HTTPException(status_code=403, detail="Access denied")

    # Query users from database (tenant-scoped)
    users = await get_users_from_db(
        tenant_id=tenant_id,
        limit=limit,
        offset=offset,
        role=role
    )

    total = await count_users_in_db(tenant_id=tenant_id, role=role)

    return {
        "users": users,
        "total": total,
        "limit": limit,
        "offset": offset
    }

@router.get("/{user_id}")
async def get_user(
    tenant_id: str,
    user_id: str,
    x_tenant_id: Optional[str] = Header(None, alias="X-Tenant-ID"),
    current_user: dict = Depends(get_current_user)
):
    """Get user information."""
    # Verify tenant access
    if x_tenant_id and x_tenant_id != tenant_id:
        raise HTTPException(status_code=403, detail="Tenant ID mismatch")

    if not await verify_tenant_access(current_user["id"], tenant_id):
        raise HTTPException(status_code=403, detail="Access denied")

    # Load user (verify belongs to tenant)
    user = await get_user_from_db(user_id, tenant_id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")

    return user

@router.post("")
async def create_user(
    tenant_id: str,
    user_data: UserCreate,
    x_tenant_id: Optional[str] = Header(None, alias="X-Tenant-ID"),
    current_user: dict = Depends(get_current_user)
):
    """Create a new user for a tenant."""
    # Verify tenant access
    if x_tenant_id and x_tenant_id != tenant_id:
        raise HTTPException(status_code=403, detail="Tenant ID mismatch")

    if not await verify_tenant_access(current_user["id"], tenant_id):
        raise HTTPException(status_code=403, detail="Access denied")

    # Check tenant limits
    user_count = await count_users_in_db(tenant_id=tenant_id)
    tenant = await get_tenant_from_db(tenant_id)
    if user_count >= tenant["limits"]["max_users"]:
        raise HTTPException(status_code=403, detail="User limit reached")

    # Create user
    new_user = await create_user_in_db(
        tenant_id=tenant_id,
        email=user_data.email,
        name=user_data.name,
        role=user_data.role,
        password=user_data.password
    )

    return new_user

CAT Management API¶

Endpoints¶

List CATs (Per Tenant)¶

GET /api/v1/tenants/{tenant_id}/cats
Headers: X-Tenant-ID: {tenant_id}, Authorization: Bearer {token}

Response:
{
  "cats": [
    {
      "id": "cat123",
      "prefix": "advisoruat",
      "name": "Advisor UAT",
      "description": "Advisor chatbot for UAT",
      "status": "active",
      "created_at": "2025-01-01T00:00:00Z"
    }
  ],
  "total": 5
}

Get CAT¶

GET /api/v1/tenants/{tenant_id}/cats/{cat_prefix}
Headers: X-Tenant-ID: {tenant_id}, Authorization: Bearer {token}

Response:
{
  "id": "cat123",
  "prefix": "advisoruat",
  "name": "Advisor UAT",
  "description": "Advisor chatbot for UAT",
  "status": "active",
  "config": {
    "model": "gpt-4",
    "temperature": 0.7
  },
  "created_at": "2025-01-01T00:00:00Z"
}

Create CAT¶

POST /api/v1/tenants/{tenant_id}/cats
Headers: X-Tenant-ID: {tenant_id}, Authorization: Bearer {token}

Request Body:
{
  "prefix": "newcat",
  "name": "New CAT",
  "description": "New chatbot",
  "config": {
    "model": "gpt-4",
    "temperature": 0.7
  }
}

Response:
{
  "id": "cat456",
  "prefix": "newcat",
  "name": "New CAT",
  "status": "active",
  "created_at": "2025-01-01T00:00:00Z"
}

Implementation Example¶

File: bast-ai/src/apis/user_management/bast_user_management_api/routers/cats.py

from fastapi import APIRouter, Depends, HTTPException, Header
from pydantic import BaseModel
from typing import Optional
from ..dependencies import get_current_user, verify_tenant_access

router = APIRouter(prefix="/api/v1/tenants/{tenant_id}/cats", tags=["cats"])

class CATCreate(BaseModel):
    prefix: str
    name: str
    description: Optional[str] = None
    config: Optional[dict] = None

@router.get("")
async def list_cats(
    tenant_id: str,
    x_tenant_id: Optional[str] = Header(None, alias="X-Tenant-ID"),
    current_user: dict = Depends(get_current_user)
):
    """List CATs for a tenant."""
    # Verify tenant access
    if x_tenant_id and x_tenant_id != tenant_id:
        raise HTTPException(status_code=403, detail="Tenant ID mismatch")

    if not await verify_tenant_access(current_user["id"], tenant_id):
        raise HTTPException(status_code=403, detail="Access denied")

    # Query CATs from database (tenant-scoped)
    cats = await get_cats_from_db(tenant_id=tenant_id)

    return {
        "cats": cats,
        "total": len(cats)
    }

@router.get("/{cat_prefix}")
async def get_cat(
    tenant_id: str,
    cat_prefix: str,
    x_tenant_id: Optional[str] = Header(None, alias="X-Tenant-ID"),
    current_user: dict = Depends(get_current_user)
):
    """Get CAT information."""
    # Verify tenant access
    if x_tenant_id and x_tenant_id != tenant_id:
        raise HTTPException(status_code=403, detail="Tenant ID mismatch")

    if not await verify_tenant_access(current_user["id"], tenant_id):
        raise HTTPException(status_code=403, detail="Access denied")

    # Load CAT (verify belongs to tenant)
    cat = await get_cat_from_db(cat_prefix, tenant_id)
    if not cat:
        raise HTTPException(status_code=404, detail="CAT not found")

    return cat

@router.post("")
async def create_cat(
    tenant_id: str,
    cat_data: CATCreate,
    x_tenant_id: Optional[str] = Header(None, alias="X-Tenant-ID"),
    current_user: dict = Depends(get_current_user)
):
    """Create a new CAT for a tenant."""
    # Verify tenant access
    if x_tenant_id and x_tenant_id != tenant_id:
        raise HTTPException(status_code=403, detail="Tenant ID mismatch")

    if not await verify_tenant_access(current_user["id"], tenant_id):
        raise HTTPException(status_code=403, detail="Access denied")

    # Check tenant limits
    cat_count = await count_cats_in_db(tenant_id=tenant_id)
    tenant = await get_tenant_from_db(tenant_id)
    if cat_count >= tenant["limits"]["max_cats"]:
        raise HTTPException(status_code=403, detail="CAT limit reached")

    # Check prefix uniqueness
    if await cat_prefix_exists(cat_data.prefix):
        raise HTTPException(status_code=409, detail="CAT prefix already exists")

    # Create CAT
    new_cat = await create_cat_in_db(
        tenant_id=tenant_id,
        prefix=cat_data.prefix,
        name=cat_data.name,
        description=cat_data.description,
        config=cat_data.config
    )

    return new_cat

Data Isolation¶

Database-Level Isolation¶

MongoDB Collections¶

# Tenant-scoped collections
{tenant_id}_users = db[f"{tenant_id}_users"]
{tenant_id}_cats = db[f"{tenant_id}_cats"]

# CAT-scoped collections
{prefix}_cat_data_logs = db[f"{prefix}_cat_data"]["logs"]
{prefix}_cat_data_sessions = db[f"{prefix}_cat_data"]["sessions"]

# Shared collections
auth_db_users = client["auth_db"]["users"]
config_db_tenants = client["config_db"]["tenants"]
config_db_cat_config = client["config_db"]["cat_config"]

Query Filtering¶

async def get_users_from_db(tenant_id: str, **filters):
    """Get users scoped to tenant."""
    collection = db[f"{tenant_id}_users"]
    query = {"tenant_id": tenant_id, **filters}
    return await collection.find(query).to_list(length=None)

async def get_cat_from_db(cat_prefix: str, tenant_id: str):
    """Get CAT scoped to tenant."""
    collection = db["config_db"]["cat_config"]
    return await collection.find_one({
        "prefix": cat_prefix,
        "tenant_id": tenant_id
    })

API Endpoints¶

Complete Endpoint List¶

Tenant Management¶

Method	Endpoint	Description	Auth Required
GET	`/api/v1/tenants/{tenant_id}`	Get tenant	Yes
GET	`/api/v1/tenants/{tenant_id}/config`	Get tenant config	Yes
PUT	`/api/v1/tenants/{tenant_id}/config`	Update tenant config	Yes
GET	`/api/v1/tenants`	List tenants (admin)	Admin

User Management¶

Method	Endpoint	Description	Auth Required
GET	`/api/v1/tenants/{tenant_id}/users`	List users	Yes
GET	`/api/v1/tenants/{tenant_id}/users/{user_id}`	Get user	Yes
POST	`/api/v1/tenants/{tenant_id}/users`	Create user	Yes
PUT	`/api/v1/tenants/{tenant_id}/users/{user_id}`	Update user	Yes
DELETE	`/api/v1/tenants/{tenant_id}/users/{user_id}`	Delete user	Yes

CAT Management¶

Method	Endpoint	Description	Auth Required
GET	`/api/v1/tenants/{tenant_id}/cats`	List CATs	Yes
GET	`/api/v1/tenants/{tenant_id}/cats/{cat_prefix}`	Get CAT	Yes
POST	`/api/v1/tenants/{tenant_id}/cats`	Create CAT	Yes
PUT	`/api/v1/tenants/{tenant_id}/cats/{cat_prefix}`	Update CAT	Yes
DELETE	`/api/v1/tenants/{tenant_id}/cats/{cat_prefix}`	Delete CAT	Yes

Authentication & Authorization¶

Dependency Functions¶

File: bast-ai/src/apis/user_management/bast_user_management_api/dependencies.py

from fastapi import Depends, HTTPException, Header
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from typing import Optional
import jwt

security = HTTPBearer()

async def get_current_user(
    credentials: HTTPAuthorizationCredentials = Depends(security),
    x_tenant_id: Optional[str] = Header(None, alias="X-Tenant-ID")
) -> dict:
    """Verify JWT token and return user."""
    token = credentials.credentials

    try:
        decoded = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
        user = {
            "id": decoded["id"],
            "email": decoded["email"],
            "organization": decoded.get("organization"),
            "tenants": decoded.get("tenants", [])
        }
        return user
    except jwt.ExpiredSignatureError:
        raise HTTPException(status_code=401, detail="Token expired")
    except jwt.InvalidTokenError:
        raise HTTPException(status_code=401, detail="Invalid token")

async def verify_tenant_access(user_id: str, tenant_id: str) -> bool:
    """Verify user has access to tenant."""
    user = await get_user_from_db(user_id)

    # Check if user belongs to tenant
    if user["organization"] == tenant_id:
        return True

    # Check if user has tenant in tenants list
    if tenant_id in user.get("tenants", []):
        return True

    return False

async def verify_user_access(
    current_user: dict,
    target_user_id: str,
    tenant_id: str
) -> bool:
    """Verify user can access another user."""
    # Users can access themselves
    if current_user["id"] == target_user_id:
        return True

    # Admins can access all users in their tenant
    if current_user.get("role") == "admin":
        return await verify_tenant_access(current_user["id"], tenant_id)

    return False

Error Handling¶

Standard Error Responses¶

# 400 Bad Request
{
  "error": "bad_request",
  "message": "Invalid request parameters",
  "details": {}
}

# 401 Unauthorized
{
  "error": "unauthorized",
  "message": "Authentication required"
}

# 403 Forbidden
{
  "error": "forbidden",
  "message": "Access denied for tenant 'acme'"
}

# 404 Not Found
{
  "error": "not_found",
  "message": "Tenant 'acme' not found"
}

# 409 Conflict
{
  "error": "conflict",
  "message": "CAT prefix 'advisoruat' already exists"
}

# 500 Internal Server Error
{
  "error": "internal_error",
  "message": "Internal server error"
}

Implementation Details¶

Step-by-Step Implementation¶

Phase 1: Tenant Endpoints (Week 1-2)¶

Create Tenant Router
Create routers/tenant.py
Implement GET /tenants/{tenant_id}
Implement GET /tenants/{tenant_id}/config
Implement PUT /tenants/{tenant_id}/config
Database Functions
Create tenant database functions
Implement tenant validation
Add tenant caching

Phase 2: User Endpoints (Week 3-4)¶

Create User Router
Create routers/users.py
Implement CRUD operations
Add tenant scoping
Authorization
Implement tenant access verification
Add role-based access control
Add user limits checking

Phase 3: CAT Endpoints (Week 5-6)¶

Create CAT Router
Create routers/cats.py
Implement CRUD operations
Add tenant scoping
CAT Configuration
Integrate with config_db
Add CAT validation
Add prefix uniqueness check

Testing Strategy¶

Unit Tests¶

# test_tenant_api.py
def test_get_tenant():
    response = client.get("/api/v1/tenants/acme", headers={
        "X-Tenant-ID": "acme",
        "Authorization": f"Bearer {token}"
    })
    assert response.status_code == 200
    assert response.json()["tenant_id"] == "acme"

def test_tenant_access_denied():
    response = client.get("/api/v1/tenants/other", headers={
        "X-Tenant-ID": "other",
        "Authorization": f"Bearer {token}"
    })
    assert response.status_code == 403

Integration Tests¶

# test_tenant_integration.py
async def test_tenant_user_flow():
    # Create tenant
    tenant = await create_tenant("test")

    # Create user
    user = await create_user(tenant["tenant_id"], "user@test.com")

    # Verify user belongs to tenant
    assert user["organization"] == tenant["tenant_id"]

Deployment¶

Environment Variables¶

# Database
MONGODB_URI=mongodb://localhost:27017
MONGODB_AUTH_DB=auth_db
MONGODB_CONFIG_DB=config_db

# Security
JWT_SECRET=your-secret-key
JWT_ALGORITHM=HS256

# API
API_PORT=8001
API_HOST=0.0.0.0

Health Check¶

@router.get("/health")
async def health_check():
    return {
        "status": "ok",
        "service": "user-management-api",
        "tenant_aware": True,
        "database": await check_database_connection()
    }

Document created: December 16, 2025
Component: Tenant APIs