Dynamic Per-User Tool Generation MCP Server

""" Data Models Type-safe data models for slot resolution requests and responses. """ from dataclasses import dataclass, field from enum import Enum from typing import Dict, List, Optional, Any from datetime import datetime import uuid class ResolutionStatus(str, Enum): """Status of slot resolution.""" RESOLVED = "RESOLVED" MULTIPLE_MATCHES = "MULTIPLE_MATCHES" NO_MATCH = "NO_MATCH" LOW_CONFIDENCE = "LOW_CONFIDENCE" INVALID_REQUEST = "INVALID_REQUEST" SERVICE_ERROR = "SERVICE_ERROR" class ResolutionMethod(str, Enum): """Method used for resolution.""" EXACT_MATCH = "exact_match" ALIAS_MATCH = "alias_match" FUZZY_MATCH = "fuzzy_match" FUZZY_MATCH_WITH_GAP = "fuzzy_match_with_gap" MANUAL_SELECTION = "manual_selection" @dataclass class Candidate: """ Represents a candidate entity match. Attributes: id: Entity ID canonical_name: The canonical name of the entity entity_type: Type of entity (e.g., "impact", "user") confidence: Confidence score (0.0 to 1.0) attributes: Additional attributes for disambiguation """ id: str canonical_name: str entity_type: str confidence: float attributes: Dict[str, Any] = field(default_factory=dict) def to_dict(self) -> dict: """Convert to dictionary.""" return { "id": self.id, "canonicalName": self.canonical_name, "entityType": self.entity_type, "confidence": self.confidence, "attributes": self.attributes } @dataclass class ResolvedEntity: """ Represents a successfully resolved entity. Attributes: id: Entity ID canonical_name: The canonical name of the entity entity_type: Type of entity confidence: Confidence score method: Resolution method used attributes: Additional entity attributes """ id: str canonical_name: str entity_type: str confidence: float method: ResolutionMethod attributes: Dict[str, Any] = field(default_factory=dict) def to_dict(self) -> dict: """Convert to dictionary.""" return { "id": self.id, "canonicalName": self.canonical_name, "entityType": self.entity_type, "confidence": self.confidence, "method": self.method.value, "attributes": self.attributes } @dataclass class SlotResolutionRequest: """ Request for slot resolution. Attributes: tenant_id: Tenant identifier for isolation entity_type: Type of entity to search (e.g., "impact", "user") input: User's natural language input context: Optional contextual filters (e.g., {"userType": "technician"}) limit: Maximum number of candidates to return min_score: Minimum confidence score threshold score_gap_delta: Minimum score gap for auto-resolution """ tenant_id: str entity_type: str input: str context: Optional[Dict[str, Any]] = None limit: int = 5 min_score: Optional[float] = None score_gap_delta: Optional[float] = None def __post_init__(self): """Validate request after initialization.""" if not self.tenant_id: raise ValueError("tenant_id is required") if not self.entity_type: raise ValueError("entity_type is required") if not self.input or not self.input.strip(): raise ValueError("input is required and cannot be empty") if self.limit < 1: raise ValueError("limit must be at least 1") @dataclass class SlotResolutionResponse: """ Response from slot resolution. Attributes: status: Resolution status resolved: Resolved entity (if status is RESOLVED) candidates: List of candidate matches (if status is MULTIPLE_MATCHES) input_echo: Original input echoed back normalization: Normalized input error: Error message (if applicable) guidance_text: User guidance text for disambiguation config: Configuration used for resolution trace_id: Unique trace ID for debugging latency_ms: Resolution latency in milliseconds method: Resolution method used """ status: ResolutionStatus input_echo: str normalization: str trace_id: str = field(default_factory=lambda: str(uuid.uuid4())) resolved: Optional[ResolvedEntity] = None candidates: List[Candidate] = field(default_factory=list) error: Optional[str] = None guidance_text: Optional[str] = None config: Dict[str, Any] = field(default_factory=dict) latency_ms: Optional[int] = None method: Optional[ResolutionMethod] = None timestamp: datetime = field(default_factory=datetime.utcnow) def to_dict(self) -> dict: """Convert to dictionary.""" result = { "status": self.status.value, "inputEcho": self.input_echo, "normalization": self.normalization, "traceId": self.trace_id, "timestamp": self.timestamp.isoformat() } if self.resolved: result["resolved"] = self.resolved.to_dict() if self.candidates: result["candidates"] = [c.to_dict() for c in self.candidates] if self.error: result["error"] = self.error if self.guidance_text: result["guidanceText"] = self.guidance_text if self.config: result["config"] = self.config if self.latency_ms is not None: result["latencyMs"] = self.latency_ms if self.method: result["method"] = self.method.value return result @dataclass class DisambiguationRequest: """ Request for user to disambiguate between multiple matches. Attributes: field: Field name that requires disambiguation target_field: Target database field name candidates: List of candidate entities original_input: Original user input """ field: str target_field: str candidates: List[Candidate] original_input: str def to_dict(self) -> dict: """Convert to dictionary.""" return { "field": self.field, "targetField": self.target_field, "candidates": [c.to_dict() for c in self.candidates], "originalInput": self.original_input } @dataclass class RequestTransformationResult: """ Result of request transformation. Attributes: status: Overall status ("READY" or "DISAMBIGUATION_REQUIRED") payload: Transformed request payload with IDs disambiguations: List of fields requiring disambiguation audit_trail: Audit information about resolutions """ status: str payload: Dict[str, Any] disambiguations: List[DisambiguationRequest] = field(default_factory=list) audit_trail: Dict[str, Any] = field(default_factory=dict) def to_dict(self) -> dict: """Convert to dictionary.""" result = { "status": self.status, "payload": self.payload } if self.disambiguations: result["disambiguations"] = [d.to_dict() for d in self.disambiguations] if self.audit_trail: result["auditTrail"] = self.audit_trail return result