# src/fctr_okta_mcp/utils/response_compressor.py
"""
Response Compressor for Okta MCP Server
Reduces context size by removing unnecessary fields from Okta API responses
while preserving the data structure the LLM needs for code generation.
Typical savings: 30-50% reduction in token usage
"""
from typing import Any, Dict, List, Optional, Set
from copy import deepcopy
# Fields to ALWAYS remove at ANY level (HAL metadata, pagination, internal)
FIELDS_TO_REMOVE: Set[str] = {
"_links", # HAL pagination/self links - ALWAYS discard
"link_header", # Raw Link header from response
"pages", # Page count metadata
"total_items", # Already obvious from len(data)
"limited_by_max_results", # Internal flag
"transitioningToStatus", # Internal state
}
def compress_response(response: Dict[str, Any], aggressive: bool = False) -> Dict[str, Any]:
"""
Compress an Okta API response for LLM consumption.
Args:
response: The raw response dict from make_request()
aggressive: If True, also truncates long string values
Returns:
Compressed response with same structure but reduced size
"""
if not isinstance(response, dict):
return response
result = {}
for key, value in response.items():
# Skip fields we don't need at any level
if key in FIELDS_TO_REMOVE:
continue
# Recursively compress all nested structures
if isinstance(value, dict):
result[key] = _compress_object(value, aggressive)
elif isinstance(value, list):
result[key] = [_compress_object(item, aggressive) for item in value]
else:
result[key] = value
return result
def _compress_object(obj: Any, aggressive: bool = False) -> Any:
"""Recursively compress an object (user, group, etc.)."""
if isinstance(obj, dict):
compressed = {}
for key, value in obj.items():
# Skip fields we don't need at any level
if key in FIELDS_TO_REMOVE:
continue
# Recursively compress nested dicts/lists
if isinstance(value, dict):
compressed[key] = _compress_object(value, aggressive)
elif isinstance(value, list):
compressed[key] = [_compress_object(item, aggressive) for item in value]
elif aggressive and isinstance(value, str) and len(value) > 100:
# Truncate very long strings in aggressive mode
compressed[key] = value[:100] + "..."
else:
compressed[key] = value
return compressed
elif isinstance(obj, list):
return [_compress_object(item, aggressive) for item in obj]
else:
return obj
def get_compression_stats(original: Dict[str, Any], compressed: Dict[str, Any]) -> Dict[str, Any]:
"""
Calculate compression statistics for debugging.
Returns dict with original_size, compressed_size, and savings_percent
"""
import json
original_str = json.dumps(original, default=str)
compressed_str = json.dumps(compressed, default=str)
original_size = len(original_str)
compressed_size = len(compressed_str)
savings = original_size - compressed_size
savings_percent = (savings / original_size * 100) if original_size > 0 else 0
return {
"original_chars": original_size,
"compressed_chars": compressed_size,
"savings_chars": savings,
"savings_percent": round(savings_percent, 1)
}
