miso-client 4.9.0

Python client SDK for AI Fabrix authentication, authorization, and logging

AI Fabrix Miso Client SDK (Python)

The AI Fabrix Miso Client SDK provides authentication, authorization, and logging for Python applications integrated with the AI Fabrix platform.

✨ Benefits

🔐 Enterprise Security

SSO and Federated Identity

• Single Sign-On (SSO) with Keycloak
• OAuth 2.0 and OpenID Connect (OIDC) support
• Multi-factor authentication (MFA) ready
• Social login integration (Google, Microsoft, etc.)

Centralized Access Control

• Role-based access control (RBAC)
• Fine-grained permissions
• Dynamic policy enforcement
• Attribute-based access control (ABAC)

API Security

• JWT token validation
• API key authentication
• Token revocation support
• Secure token storage
• Data encryption/decryption (AES-256-GCM)

📊 Compliance & Audit

ISO 27001 Compliance

• Comprehensive audit trails for all user actions and HTTP requests
• Automatic data masking for all sensitive information in logs
• HTTP request/response audit logging with masked sensitive data
• Data access logging and monitoring
• Security event tracking
• Accountability and non-repudiation
• Configurable sensitive fields via JSON configuration

Regulatory Compliance

• GDPR-ready data protection
• HIPAA-compliant audit logging
• SOC 2 audit trail requirements
• Industry-standard security controls

Audit Capabilities

• Real-time audit event logging
• Immutable audit records
• Forensic analysis support
• Compliance reporting automation

⚡ Performance & Scalability

Intelligent Caching

• Redis-based role and permission caching
• Generic cache service with Redis and in-memory fallback
• Configurable cache TTL (default: 15 minutes)
• Automatic cache invalidation
• Fallback to controller when Redis unavailable

High Availability

• Automatic failover to controller
• Redundant infrastructure support
• Load balancing compatible
• Zero-downtime deployments

Optimized Network

• Efficient API calls with caching
• Batch operations support
• Connection pooling
• Minimal latency

🛠️ Developer Experience

Easy Integration

• Progressive activation (6-step setup)
• Works with any framework (FastAPI, Django, Flask, Starlette)
• Python 3.8+ support with full type hints
• Async/await support throughout

Flexible Configuration

• Environment-based configuration
• Support for dev, test, and production environments
• Docker and Kubernetes ready
• CI/CD friendly

Observability

• Centralized logging with correlation IDs
• Automatic HTTP request/response audit logging (ISO 27001 compliant)
• Debug logging with detailed request/response information (when log_level='debug')
• Performance tracking and metrics
• Error tracking and debugging
• Health monitoring
• Automatic data masking for sensitive information in logs
• Configurable sensitive fields via JSON configuration

---

🚀 Quick Start

Get your application secured in 30 seconds.

Step 1: Install

pip install miso-client

Step 2: Create .env

MISO_CLIENTID=ctrl-dev-my-app
MISO_CLIENTSECRET=your-secret
MISO_CONTROLLER_URL=http://localhost:3000
REDIS_HOST=localhost

Step 3: Use It

from miso_client import MisoClient, load_config

client = MisoClient(load_config())
await client.initialize()

is_valid = await client.validate_token(token)

That's it! You now have authentication, roles, and logging.

→ Full Getting Started Guide

---

Infrastructure Setup

First time? You'll need Keycloak and Miso Controller running.

Use the AI Fabrix Builder:

# Start infrastructure (Postgres, Redis)
aifabrix up

# Install Keycloak for authentication
aifabrix create keycloak --port 8082 --database --template platform
aifabrix build keycloak
aifabrix run keycloak

# Install Miso Controller
aifabrix create miso-controller --port 3000 --database --redis --template platform
aifabrix build miso-controller
aifabrix run miso-controller

→ Infrastructure Guide

Already have Keycloak and Controller? Use the Quick Start above.

---

📚 Documentation

What happens: Your app validates user tokens from Keycloak.

from miso_client import MisoClient, load_config

# Create client (loads from .env automatically)
client = MisoClient(load_config())
await client.initialize()

# Get token from request (helper method)
token = client.get_token(req)

if token:
    is_valid = await client.validate_token(token)
    if is_valid:
        user = await client.get_user(token)
        print('User:', user)

Where to get tokens? Users authenticate via Keycloak, then your app receives JWTs in the Authorization header.

Two-token model: The SDK uses a client token (obtained once from the controller with client id/secret, sent as x-client-token on all API requests) and a user token (sent as Authorization: Bearer <token> for user-scoped operations). Only the client token endpoint receives client id/secret; all other APIs use the client token.

Token exchange (Entra/delegated tokens): If your app has an external token (e.g. Entra ID), call exchange_token(delegated_token) to get a Keycloak token for use with the SDK: result = await client.exchange_token(entra_token) then use result.accessToken for validate_token, get_roles, etc.

Validate application token (client token): To verify a controller-issued application token (e.g. x-client-token from your backend or dataplane), use validate_client_token(token). Results are cached to avoid extra controller calls. Use for dataplane or services that need to check an app token: result = await client.validate_client_token(app_token) then check result.data.authenticated and result.data.application, result.data.expiresAt, etc.

→ Complete authentication example

---

Step 4: Activate RBAC (Roles)

What happens: Check user roles to control access. Roles are cached in Redis for performance.

from miso_client import MisoClient, load_config

# Build on Step 3 - add Redis in .env file
client = MisoClient(load_config())
await client.initialize()

token = client.get_token(req)

# Check if user has role
is_admin = await client.has_role(token, 'admin')
roles = await client.get_roles(token)

# Gate features by role
if is_admin:
    # Show admin panel
    pass

Pro tip: Without Redis, checks go to the controller. Add Redis to cache role lookups (15-minute default TTL).

→ Complete RBAC example
→ AI Fabrix Builder Quick Start

---

Step 5: Activate Logging

What happens: Application logs are sent to the Miso Controller with client token authentication. All HTTP requests are automatically audited with ISO 27001 compliant data masking.

from miso_client import MisoClient, load_config

# Client token is automatically managed - no API key needed
client = MisoClient(load_config())
await client.initialize()

token = client.get_token(req)
user = await client.get_user(token)

# Log messages
await client.log.info('User accessed dashboard', {'userId': user.id if user else None})
await client.log.error('Operation failed', {'error': str(err)})
await client.log.warn('Unusual activity', {'details': '...'})

# HTTP requests are automatically audited
# All sensitive data is automatically masked before logging
result = await client.http_client.get('/api/users')
# This automatically creates an audit log: http.request.GET with masked sensitive data

What happens to logs? They're sent to the Miso Controller for centralized monitoring and analysis. Client token is automatically included. Audit logs are automatically batched using AuditLogQueue for improved performance (configurable via AuditConfig).

Migration note (breaking):

• Indexed context fields are id-based now:
  • sourceKey -> sourceId
  • externalSystemKey -> externalSystemId
  • recordKey -> recordId
• LoggerChain.with_indexed_context() now uses:
  • source_id, external_system_id, record_id
• Logs/audit list methods are id-filter based:
  • list_general_logs(..., application_id=..., source_id=..., external_system_id=..., record_id=...)
  • list_audit_logs(..., application_id=..., source_id=..., external_system_id=..., record_id=...)
  • Application-name list filtering is removed from these two methods.
• Stats/export APIs keep application for compatibility and also accept id-based filters where supported.

ISO 27001 Compliance: All HTTP requests are automatically audited with sensitive data masked. Configure audit logging behavior using AuditConfig:

• Audit Levels: Choose from minimal, standard, detailed, or full (default: detailed)
  • minimal: Only metadata, no masking
  • standard: Metadata + basic context
  • detailed: Full context with request/response sizes (default)
  • full: Complete audit trail with all available data
• Performance Optimizations:
  • Response body truncation based on maxResponseSize configuration (default: 10000 bytes)
  • Size-based masking skip for large objects (prevents performance degradation)
  • Automatic batching via AuditLogQueue reduces HTTP overhead for high-volume logging
• Set log_level='debug' to enable detailed request/response logging (all sensitive data is still masked).

→ Complete logging example
→ Logging Reference

Unified Logging Interface (Recommended)

What happens: The SDK provides a unified logging interface with minimal API (1-3 parameters maximum) and automatic context extraction. This eliminates the need to manually pass Request objects or context dictionaries.

Benefits:

• Minimal API: Maximum 1-3 parameters per logging call
• Automatic Context: Context extracted automatically via contextvars
• Simple Usage: logger.info(message), logger.error(message, error?), logger.audit(action, resource, entity_id?, old_values?, new_values?)
• Framework Agnostic: Works in FastAPI routes, Flask routes, service layers, background jobs
• Zero Configuration: Context automatically available when middleware is used

Quick Start:

FastAPI Setup

from fastapi import FastAPI
from miso_client import get_logger
from miso_client.utils.fastapi_logger_middleware import logger_context_middleware

app = FastAPI()

# Add middleware early in middleware chain (after auth middleware)
app.middleware("http")(logger_context_middleware)

@app.get("/api/users")
async def get_users():
    logger = get_logger()  # Auto-detects context from contextvars
    
    await logger.info("Users list accessed")  # Auto-extracts request context
    
    users = await fetch_users()
    return users

Flask Setup

from flask import Flask
from miso_client import get_logger
from miso_client.utils.flask_logger_middleware import register_logger_context_middleware

app = Flask(__name__)

# Register middleware
register_logger_context_middleware(app)

@app.route("/api/users")
async def get_users():
    logger = get_logger()  # Auto-detects context from contextvars
    
    await logger.info("Users list accessed")  # Auto-extracts request context
    
    users = await fetch_users()
    return users

Service Layer Usage

from miso_client import get_logger

class UserService:
    async def get_user(self, user_id: str):
        logger = get_logger()  # Uses contextvars context if available
        
        await logger.info("Fetching user")  # Auto-extracts context if available
        
        try:
            user = await db.user.find_unique({"id": user_id})
            await logger.audit("ACCESS", "User", user_id)  # Read access audit
            return user
        except Exception as error:
            await logger.error("Failed to fetch user", error)  # Auto-extracts error details
            raise

Background Job Usage

from miso_client import get_logger, set_logger_context

async def background_job():
    # Set context for this async execution context
    set_logger_context({
        "jobId": "job-123",
        "jobType": "sync",
    })
    
    logger = get_logger()
    await logger.info("Background job started")
    
    # All logs in this async context will use the set context
    await process_data()

UnifiedLogger Methods:

• info(message: str) -> None - Log info message
• warn(message: str) -> None - Log warning message (preserved as warn level end-to-end)
• debug(message: str) -> None - Log debug message
• error(message: str, error: Optional[Exception] = None) -> None - Log error message with optional exception
• audit(action: str, resource: str, entity_id?: str, old_values?: Dict, new_values?: Dict) -> None - Log audit event

Context Management:

• get_logger() -> UnifiedLogger - Get logger instance with automatic context detection
• set_logger_context(context: Dict[str, Any]) -> None - Set context manually for background jobs
• clear_logger_context() -> None - Clear context

Async Context Notes:

• Context is stored via contextvars, so it flows through await in the same async call chain.
• Middleware (fastapi_logger_context_middleware / flask_logger_context_middleware) sets and clears context per request.
• For background tasks or separate event loops/threads, call set_logger_context(...) inside that task to ensure context is available.

Context Fields Automatically Extracted:

• ipAddress - Client IP address
• userAgent - User agent string
• correlationId - Request correlation ID
• requestId - Request ID from headers
• userId - Authenticated user ID (from JWT token)
• sessionId - Session ID (from JWT token)
• method - HTTP method
• path - Request path
• hostname - Request hostname
• referer - Referrer URL
• requestSize - Request size in bytes
• applicationId - Application identifier (from JWT token)

---

Step 6: Activate Audit

What happens: Create audit trails for compliance and security monitoring.

from miso_client import MisoClient, load_config

# Complete configuration (all in .env)
client = MisoClient(load_config())
await client.initialize()

token = client.get_token(req)
is_valid = await client.validate_token(token)
can_edit = await client.has_permission(token, 'edit:content')
user = await client.get_user(token)

# Audit: User actions
await client.log.audit('user.login', 'authentication', {
})

# Audit: Content changes
await client.log.audit('post.created', 'content', {
    'userId': user.id if user else None,
    'postId': 'post-123',
    'postTitle': req.get('body', {}).get('title', ''),
})

# Audit: Permission checks
await client.log.audit('access.denied', 'authorization', {
    'requiredPermission': 'edit:content',
    'resource': 'posts',
})

What to audit: Login/logout, permission checks, content creation/deletion, role changes, sensitive operations.

→ Complete audit example
→ Best Practices

---

Encryption and Caching

What happens: Use encryption for sensitive data and generic caching for improved performance.

from miso_client import MisoClient, load_config

client = MisoClient(load_config())
await client.initialize()

# Encryption (requires ENCRYPTION_KEY in .env)
encrypted = client.encrypt('sensitive-data')
decrypted = client.decrypt(encrypted)
print('Decrypted:', decrypted)

# Generic caching (automatically uses Redis if available, falls back to memory)
await client.cache_set('user:123', {'name': 'John', 'age': 30}, 600)  # 10 minutes TTL
user = await client.cache_get('user:123')
if user:
    print('Cached user:', user)

Configuration:

# Add to .env
ENCRYPTION_KEY=your-32-byte-encryption-key

→ API Reference
→ Cache Methods

---

Testing with API Key

What happens: When API_KEY is set in your .env file, you can authenticate requests using the API key as a bearer token, bypassing OAuth2 authentication. This is useful for testing without setting up Keycloak.

from miso_client import MisoClient, load_config

client = MisoClient(load_config())
await client.initialize()

# Use API_KEY as bearer token (for testing only)
api_key_token = "your-api-key-from-env"
is_valid = await client.validate_token(api_key_token)
# Returns True if token matches API_KEY from .env

user = await client.get_user(api_key_token)
# Returns None (API key auth doesn't provide user info)

Configuration:

# Add to .env for testing
API_KEY=your-test-api-key-here

Important:

• API_KEY authentication bypasses OAuth2 validation completely
• User information methods (get_user(), get_user_info()) return None when using API_KEY
• Token validation returns True if the bearer token matches the configured API_KEY
• This feature is intended for testing and development only

---

🔧 Configuration

from miso_client import MisoClientConfig, RedisConfig, AuditConfig

config = MisoClientConfig(
    controller_url="http://localhost:3000",  # Required: Controller URL
    client_id="ctrl-dev-my-app",              # Required: Client ID
    client_secret="your-secret",              # Required: Client secret
    redis=RedisConfig(                        # Optional: For caching
        host="localhost",
        port=6379,
    ),
    log_level="info",                         # Optional: 'debug' | 'info' | 'warn' | 'error'
                                              # Set to 'debug' for detailed HTTP request/response logging
    api_key="your-test-api-key",              # Optional: API key for testing (bypasses OAuth2)
    cache={                                   # Optional: Cache TTL settings
        "role_ttl": 900,       # Role cache TTL (default: 900s)
        "permission_ttl": 900, # Permission cache TTL (default: 900s)
    },
    audit=AuditConfig(                        # Optional: Audit logging configuration
        enabled=True,                         # Enable/disable audit logging (default: true)
        level="detailed",                     # Audit detail level: 'minimal' | 'standard' | 'detailed' | 'full' (default: 'detailed')
        maxResponseSize=10000,                # Truncate responses larger than this in bytes (default: 10000)
        maxMaskingSize=50000,                 # Skip masking for objects larger than this in bytes (default: 50000)
        batchSize=10,                         # Batch size for queued logs (default: 10)
        batchInterval=100,                    # Flush interval in milliseconds (default: 100)
        skipEndpoints=None                    # Array of endpoint patterns to exclude from audit logging
    )
)

Recommended: Use load_config() to load from .env file automatically.

ISO 27001 Data Masking Configuration:

Sensitive fields are configured via miso_client/utils/sensitive_fields_config.json. You can customize this by:

1. Setting MISO_SENSITIVE_FIELDS_CONFIG environment variable to point to a custom JSON file
2. Using DataMasker.set_config_path() to set a custom path programmatically

The default configuration includes ISO 27001 compliant sensitive fields:

• Authentication: password, token, secret, key, auth, authorization
• PII: ssn, creditcard, cc, cvv, pin, otp
• Security: apikey, accesstoken, refreshtoken, privatekey, secretkey, cookie, session

Masking JSON documents (for other projects): Use DataMasker to mask sensitive fields in any JSON structure before logging or sending to external systems:

from miso_client import DataMasker

doc = {"user": "john", "password": "secret123", "email": "[e-mail]"}
masked = DataMasker.mask_sensitive_data(doc)
# {"user": "john", "password": "***MASKED***", "email": "[e-mail]"}

Audit Logging Configuration:

Configure audit logging behavior using AuditConfig (see Configuration section above):

• Audit Levels: Control detail level (minimal, standard, detailed, full)
• Response Truncation: Configure maxResponseSize to truncate large responses (default: 10000 bytes)
• Performance: Set maxMaskingSize to skip masking for very large objects (default: 50000 bytes)
• Batching: Configure batchSize and batchInterval for audit log queuing (reduces HTTP overhead)

→ Complete Configuration Reference

---

📚 Read more

• Getting Started - Detailed setup guide
• Backend client-token endpoint - Simple backend code for any app (FastAPI / Flask)
• API Reference - Complete API documentation
• Configuration - Configuration options
• Examples - Framework-specific examples
• Troubleshooting - Common issues and solutions

---

🏗️ Architecture

The SDK consists of five core services:

• AuthService - Token validation and user authentication
• RoleService - Role management with Redis caching
• PermissionService - Fine-grained permissions
• LoggerService - Centralized logging with API key authentication
• RedisService - Caching and queue management (optional)

HTTP Client Architecture

The SDK uses a two-layer HTTP client architecture for ISO 27001 compliance:

• InternalHttpClient - Core HTTP functionality with automatic client token management (internal)
• HttpClient - Public wrapper that adds automatic ISO 27001 compliant audit and debug logging

Features:

• Automatic audit logging for all HTTP requests (http.request.{METHOD})
• Configurable audit levels (minimal, standard, detailed, full) via AuditConfig
• Debug logging when log_level === 'debug' with detailed request/response information
• Automatic data masking using DataMasker before logging (ISO 27001 compliant)
• Sensitive endpoints (/api/logs, /api/auth/token) are excluded from audit logging to prevent infinite loops
• All sensitive data (headers, bodies, query params) is automatically masked before logging
• AuditLogQueue integration for automatic batching of audit logs (reduces HTTP overhead)
• Performance optimizations: response body truncation and size-based masking skip for large objects

ISO 27001 Compliance:

• All request headers are masked (Authorization, x-client-token, Cookie, etc.)
• All request bodies are recursively masked for sensitive fields (password, token, secret, SSN, etc.)
• All response bodies are masked and truncated based on maxResponseSize configuration (default: 10000 bytes)
• Query parameters are automatically masked
• Error messages are masked if they contain sensitive data
• Sensitive fields configuration can be customized via sensitive_fields_config.json
• Configurable audit levels control the detail level of audit logs (minimal, standard, detailed, full)

→ Architecture Details

---

🌐 Setup Your Application

First time setup? Use the AI Fabrix Builder:

1. Create your app:

   aifabrix create myapp --port 3000 --database --language python
   

2. Login to controller:

   aifabrix login
   

3. Register your application:

   aifabrix app register myapp --environment dev
   

4. Start development and then deploy to Docker or Azure.

→ Full Quick Start Guide

---

💡 Next Steps

Learn More

• FastAPI Integration - Protect API routes
• Django Middleware - Django integration
• Flask Decorators - Decorator-based auth
• Error Handling - Best practices

---

Structured Error Responses

What happens: The SDK automatically parses structured error responses from the API (RFC 7807-style format) and makes them available through the MisoClientError and ApiErrorException exceptions.

from miso_client import MisoClient, MisoClientError, ApiErrorException, ErrorResponse, load_config, handleApiError

client = MisoClient(load_config())
await client.initialize()

try:
    result = await client.http_client.get("/api/some-endpoint")
except MisoClientError as e:
    # Check if structured error response is available
    if e.error_response:
        print(f"Error Type: {e.error_response.type}")
        print(f"Error Title: {e.error_response.title}")
        print(f"Status Code: {e.error_response.statusCode}")
        print(f"Errors: {e.error_response.errors}")
        print(f"Instance: {e.error_response.instance}")
    else:
        # Fallback to traditional error handling
        print(f"Error: {e.message}")
        print(f"Status Code: {e.status_code}")
        print(f"Error Body: {e.error_body}")

# Using handleApiError() for structured error handling
try:
    response_data = {"errors": ["Validation failed"], "type": "/Errors/Validation", "title": "Validation Error", "statusCode": 422}
    error = handleApiError(response_data, 422, "/api/endpoint")
    # handleApiError() returns ApiErrorException (extends MisoClientError)
    if isinstance(error, ApiErrorException):
        print(f"Structured Error: {error.error_response.title}")
except ApiErrorException as e:
    # ApiErrorException provides better structured error information
    print(f"API Error: {e.error_response.title}")
    print(f"Errors: {e.error_response.errors}")

Error Response Structure:

The ErrorResponse model follows RFC 7807-style format:

{
   "errors": [
      "The user has provided input that the browser is unable to convert.",
      "There are multiple rows in the database for the same value"
   ],
   "type": "/Errors/Bad Input",
   "title": "Bad Request",
   "statusCode": 400,
   "instance": "/OpenApi/rest/Xzy"
}

Features:

• Automatic Parsing: Structured error responses are automatically parsed from HTTP responses
• ApiErrorException: Exception class (extends MisoClientError) for better structured error handling
  • handleApiError() returns ApiErrorException with structured error response support
• Type Safety: Full type hints with Pydantic models for reliable error handling
• Generic Interface: ErrorResponse model can be reused across different applications
• Instance URI: Automatically extracted from request URL if not provided in response

Using ErrorResponse directly:

from miso_client import ErrorResponse

# Create ErrorResponse from dict
error_data = {
    "errors": ["Validation failed"],
    "type": "/Errors/Validation",
    "title": "Validation Error",
    "statusCode": 422,
    "instance": "/api/endpoint"
}
error_response = ErrorResponse(**error_data)

# Access fields
print(error_response.errors)  # ["Validation failed"]
print(error_response.type)    # "/Errors/Validation"
print(error_response.title)   # "Validation Error"
print(error_response.statusCode)  # 422
print(error_response.instance)   # "/api/endpoint"

---

Pagination, Filtering, and Sorting Utilities

What happens: The SDK provides reusable utilities for pagination, filtering, sorting, and error handling that work with any API endpoint.

Pagination

Pagination Parameters:

• page: Page number (1-based, defaults to 1)
• page_size: Number of items per page (defaults to 20)

from miso_client import (
    parsePaginationParams,
    parse_pagination_params,
    createPaginatedListResponse,
    PaginatedListResponse,
)

# Parse pagination from query parameters (returns dict with currentPage/pageSize keys)
params = {"page": "1", "pageSize": "20"}
pagination = parsePaginationParams(params)
# Returns: {"currentPage": 1, "pageSize": 20}

# Parse pagination from query parameters (returns tuple with page/page_size)
page, page_size = parse_pagination_params({"page": "1", "page_size": "20"})
# Returns: (1, 20)

# Create paginated response
items = [{"id": 1}, {"id": 2}]
response = createPaginatedListResponse(
    items,
    totalItems=120,
    currentPage=1,
    pageSize=20,
    type="item"
)

# Response structure:
# {
#   "meta": {
#     "totalItems": 120,
#     "currentPage": 1,
#     "pageSize": 20,
#     "type": "item"
#   },
#   "data": [{"id": 1}, {"id": 2}]
# }

Filtering

Filter Operators: eq, neq, in, nin, gt, lt, gte, lte, contains, like

Filter Format: field:op:value (e.g., status:eq:active)

from miso_client import FilterBuilder, parse_filter_params, build_query_string

# Dynamic filter building with FilterBuilder
filter_builder = FilterBuilder() \
    .add('status', 'eq', 'active') \
    .add('region', 'in', ['eu', 'us']) \
    .add('created_at', 'gte', '2024-01-01')

# Get query string
query_string = filter_builder.to_query_string()
# Returns: "filter=status:eq:active&filter=region:in:eu,us&filter=created_at:gte:2024-01-01"

# Parse existing filter parameters
params = {'filter': ['status:eq:active', 'region:in:eu,us']}
filters = parse_filter_params(params)
# Returns: [FilterOption(field='status', op='eq', value='active'), ...]

# Use with HTTP client
response = await client.http_client.get_with_filters(
    '/api/items',
    filter_builder=filter_builder
)

Building Complete Filter Queries:

from miso_client import FilterQuery, FilterOption, build_query_string

# Create filter query with filters, sort, pagination, and fields
filter_query = FilterQuery(
    filters=[
        FilterOption(field='status', op='eq', value='active'),
        FilterOption(field='region', op='in', value=['eu', 'us'])
    ],
    sort=['-updated_at', 'created_at'],
    page=1,
    pageSize=20,  # Note: camelCase for API compatibility
    fields=['id', 'name', 'status']
)

# Build query string
query_string = build_query_string(filter_query)

Sorting

Sort Format: -field for descending, field for ascending (e.g., -updated_at, created_at)

from miso_client import parse_sort_params, build_sort_string, SortOption

# Parse sort parameters
params = {'sort': '-updated_at'}
sort_options = parse_sort_params(params)
# Returns: [SortOption(field='updated_at', order='desc')]

# Parse multiple sorts
params = {'sort': ['-updated_at', 'created_at']}
sort_options = parse_sort_params(params)
# Returns: [
#   SortOption(field='updated_at', order='desc'),
#   SortOption(field='created_at', order='asc')
# ]

# Build sort string
sort_options = [
    SortOption(field='updated_at', order='desc'),
    SortOption(field='created_at', order='asc')
]
sort_string = build_sort_string(sort_options)
# Returns: "-updated_at,created_at"

Combined Usage

Pagination + Filter + Sort:

from miso_client import (
    FilterBuilder,
    FilterQuery,
    build_query_string,
    parsePaginationParams,
    parse_pagination_params,
)

# Build filters
filter_builder = FilterBuilder() \
    .add('status', 'eq', 'active') \
    .add('region', 'in', ['eu', 'us'])

# Parse pagination
params = {'page': '1', 'pageSize': '20'}
# Using camelCase function (returns dict)
pagination = parsePaginationParams(params)
current_page = pagination['currentPage']
page_size = pagination['pageSize']

# Using snake_case function (returns tuple)
page, page_size = parse_pagination_params({"page": "1", "page_size": "25"})

# Create complete query
filter_query = FilterQuery(
    filters=filter_builder.build(),
    sort=['-updated_at'],
    page=current_page,
    pageSize=page_size
)

# Build query string
query_string = build_query_string(filter_query)

# Use with HTTP client
response = await client.http_client.get_with_filters(
    '/api/items',
    filter_builder=filter_builder,
    params={'page': current_page, 'pageSize': page_size}
)

Or use pagination helper:

# Get paginated response
response = await client.http_client.get_paginated(
    '/api/items',
    page=1,
    page_size=20
)

# Response is automatically parsed as PaginatedListResponse
print(response.meta.totalItems)  # 120
print(response.meta.currentPage)  # 1
print(len(response.data))  # 25

Metadata Filter Integration

Working with /metadata/filter endpoint:

# Get metadata filters from endpoint
metadata_response = await client.http_client.post(
    "/api/v1/metadata/filter",
    {"documentStorageKey": "my-doc-storage"}
)

# Convert AccessFieldFilter to FilterBuilder
filter_builder = FilterBuilder()
for access_filter in metadata_response.mandatoryFilters:
    filter_builder.add(access_filter.field, 'in', access_filter.values)

# Use with query utilities
query_string = filter_builder.to_query_string()

# Apply to API requests
response = await client.http_client.get_with_filters(
    '/api/items',
    filter_builder=filter_builder
)

Features:

• camelCase Convention: Pagination and error utilities use camelCase to match TypeScript SDK
  • parsePaginationParams() - Returns dict with currentPage/pageSize keys
  • parse_pagination_params() - Returns tuple (page, page_size) (snake_case, Python convention)
  • createMetaObject() - Creates Meta objects with camelCase fields
  • applyPaginationToArray() - Applies pagination to arrays
  • createPaginatedListResponse() - Creates paginated list responses
  • transformError() - Transforms error dictionaries to ErrorResponse objects
  • handleApiError() - Creates ApiErrorException from API error responses
• Type Safety: Full type hints with Pydantic models
• Dynamic Filtering: FilterBuilder supports method chaining for complex filters
• Local Testing: apply_filters() and applyPaginationToArray() for local filtering/pagination in tests
• URL Encoding: Automatic URL encoding for field names and values

---

Common Tasks

Add authentication middleware (FastAPI):

from fastapi import Depends, HTTPException, Security
from fastapi.security import HTTPBearer
from miso_client import MisoClient

security = HTTPBearer()
client = MisoClient(load_config())

async def get_current_user(credentials = Security(security)):
    token = credentials.credentials
    is_valid = await client.validate_token(token)
    if not is_valid:
        raise HTTPException(status_code=401, detail="Invalid token")
    return await client.get_user(token)

Protect routes by role (FastAPI):

@app.get('/admin')
async def admin_panel(user = Depends(get_current_user), credentials = Security(security)):
    token = credentials.credentials
    is_admin = await client.has_role(token, 'admin')
    if not is_admin:
        raise HTTPException(status_code=403, detail="Forbidden")
    
    # Admin only code
    return {"message": "Admin panel"}

Use environment variables:

MISO_CLIENTID=ctrl-dev-my-app
MISO_CLIENTSECRET=your-secret
MISO_CONTROLLER_URL=http://localhost:3000
REDIS_HOST=localhost
REDIS_PORT=6379
MISO_LOG_LEVEL=info
API_KEY=your-test-api-key  # Optional: For testing (bypasses OAuth2)

---

🐛 Troubleshooting

"Cannot connect to controller"
→ Verify controllerUrl is correct and accessible
→ Check network connectivity

"Redis connection failed"
→ SDK falls back to controller-only mode (slower but works)
→ Fix: aifabrix up to start Redis

"Client token fetch failed"
→ Check MISO_CLIENTID and MISO_CLIENTSECRET are correct
→ Verify credentials are configured in controller
→ Ensure ENCRYPTION_KEY environment variable is set (required for encryption service)

"Token validation fails"
→ Ensure Keycloak is running and configured correctly
→ Verify token is from correct Keycloak instance
→ Check that python-dotenv is installed if using .env files

→ More Help

---

📦 Installation

# pip
pip install miso-client

# Development mode
pip install -e .

# With dev dependencies
pip install "miso-client[dev]"

---

🔗 Links

• GitHub Repository: https://github.com/esystemsdev/aifabrix-miso-client-python
• PyPI Package: https://pypi.org/project/miso-client/
• Builder Documentation: https://github.com/esystemsdev/aifabrix-builder
• Issues: https://github.com/esystemsdev/aifabrix-miso-client-python/issues

---

📄 License

This project is licensed under the MIT License - see the LICENSE file for details.

---

Made with ❤️ by eSystems Nordic Ltd.