Dynamic Per-User Tool Generation MCP Server

""" Audit Service Handles audit logging for slot resolution decisions. """ import logging import json from datetime import datetime from typing import Dict, Any, Optional from slot_resolution.core.models import SlotResolutionResponse logger = logging.getLogger(__name__) class AuditService: """ Service for auditing slot resolution decisions. Logs resolution decisions for compliance and debugging purposes. """ def __init__(self, log_to_file: bool = False, log_file_path: Optional[str] = None): """ Initialize the audit service. Args: log_to_file: If True, write audit logs to file log_file_path: Path to audit log file """ self.log_to_file = log_to_file self.log_file_path = log_file_path or "slot_resolution_audit.log" logger.info(f"AuditService initialized (log_to_file={log_to_file})") def log_resolution( self, tenant_id: str, user_id: Optional[str], entity_type: str, input_query: str, response: SlotResolutionResponse, latency_ms: int ): """ Log a resolution decision. Args: tenant_id: Tenant identifier user_id: User identifier (if available) entity_type: Type of entity input_query: Original user input response: Resolution response latency_ms: Resolution latency in milliseconds """ audit_record = { "timestamp": datetime.utcnow().isoformat(), "tenant_id": tenant_id, "user_id": user_id, "entity_type": entity_type, "input_query": input_query, "normalized_query": response.normalization, "status": response.status.value, "trace_id": response.trace_id, "latency_ms": latency_ms } # Add resolution details based on status if response.resolved: audit_record["resolved_id"] = response.resolved.id audit_record["resolved_name"] = response.resolved.canonical_name audit_record["confidence"] = response.resolved.confidence audit_record["method"] = response.resolved.method.value if response.candidates: audit_record["candidate_count"] = len(response.candidates) audit_record["top_candidate"] = { "id": response.candidates[0].id, "name": response.candidates[0].canonical_name, "confidence": response.candidates[0].confidence } # Log to standard logger logger.info(f"AUDIT: {json.dumps(audit_record)}") # Optionally log to file if self.log_to_file: self._write_to_file(audit_record) def _write_to_file(self, audit_record: Dict[str, Any]): """Write audit record to file.""" try: with open(self.log_file_path, 'a') as f: f.write(json.dumps(audit_record) + '\n') except Exception as e: logger.error(f"Failed to write audit log to file: {e}")