""" DungeonMaster AI - MCP Integration Models Pydantic models for enhanced tool results and protocols for game state access. """ from __future__ import annotations from datetime import datetime from enum import Enum from typing import Protocol, runtime_checkable from pydantic import BaseModel, Field class ConnectionState(str, Enum): """Connection state for MCP server.""" DISCONNECTED = "disconnected" CONNECTING = "connecting" CONNECTED = "connected" ERROR = "error" RECONNECTING = "reconnecting" class CircuitBreakerState(str, Enum): """Circuit breaker states for connection management.""" CLOSED = "closed" # Normal operation OPEN = "open" # Rejecting requests due to failures HALF_OPEN = "half_open" # Testing if service recovered class RollType(str, Enum): """Types of dice rolls for formatting.""" STANDARD = "standard" ATTACK = "attack" DAMAGE = "damage" SAVE = "save" CHECK = "check" INITIATIVE = "initiative" # ============================================================================= # Protocols for loose coupling # ============================================================================= @runtime_checkable class GameStateProtocol(Protocol): """ Protocol for game state access. This interface allows tool wrappers to update game state without depending on the full GameState implementation (coming in Phase 4). """ session_id: str in_combat: bool party: list[str] recent_events: list[dict[str, object]] def add_event( self, event_type: str, description: str, data: dict[str, object], ) -> None: """Add an event to recent events.""" ... def get_character(self, character_id: str) -> dict[str, object] | None: """Get character data from cache.""" ... def update_character_cache( self, character_id: str, data: dict[str, object], ) -> None: """Update character data in cache.""" ... def set_combat_state(self, combat_state: dict[str, object] | None) -> None: """Set or clear combat state.""" ... # ============================================================================= # Base Result Models # ============================================================================= class FormattedResult(BaseModel): """ Base class for formatted tool results. All tool wrappers return results in this format to provide multiple output formats for different consumers. """ raw_result: dict[str, object] = Field( default_factory=dict, description="Original MCP tool result", ) chat_display: str = Field( default="", description="Markdown formatted for chat display", ) ui_data: dict[str, object] = Field( default_factory=dict, description="Structured data for UI components", ) voice_narration: str = Field( default="", description="TTS-friendly plain text for voice narration", ) # Side effect tracking state_updated: bool = Field( default=False, description="Whether game state was updated", ) events_logged: list[str] = Field( default_factory=list, description="List of event types logged to session", ) ui_updates_needed: list[str] = Field( default_factory=list, description="UI components that need refresh", ) # ============================================================================= # Dice Roll Models # ============================================================================= class DiceRollResult(FormattedResult): """Enhanced dice roll result with game context.""" notation: str = Field( default="", description="Dice notation (e.g., '2d6+3')", ) individual_rolls: list[int] = Field( default_factory=list, description="Individual dice results", ) modifier: int = Field( default=0, description="Modifier applied to roll", ) total: int = Field( default=0, description="Total roll result", ) roll_type: RollType = Field( default=RollType.STANDARD, description="Type of roll for context", ) is_critical: bool = Field( default=False, description="Natural 20 on d20", ) is_fumble: bool = Field( default=False, description="Natural 1 on d20", ) success: bool | None = Field( default=None, description="Success/failure for checks with DC", ) dc: int | None = Field( default=None, description="Difficulty class if applicable", ) reason: str = Field( default="", description="Reason for the roll", ) timestamp: datetime = Field( default_factory=datetime.now, description="When the roll occurred", ) # ============================================================================= # HP Change Models # ============================================================================= class HPChangeResult(FormattedResult): """Enhanced HP modification result with death handling.""" character_id: str = Field( default="", description="Character ID", ) character_name: str = Field( default="Unknown", description="Character name for display", ) previous_hp: int = Field( default=0, description="HP before modification", ) current_hp: int = Field( default=0, description="HP after modification", ) max_hp: int = Field( default=1, description="Maximum HP", ) change_amount: int = Field( default=0, description="Absolute value of HP change", ) is_damage: bool = Field( default=False, description="True if damage, False if healing", ) damage_type: str | None = Field( default=None, description="Type of damage if applicable", ) is_unconscious: bool = Field( default=False, description="Character is at 0 HP or below", ) requires_death_save: bool = Field( default=False, description="Character needs to make death saves", ) is_dead: bool = Field( default=False, description="Character died (massive damage)", ) is_bloodied: bool = Field( default=False, description="Character is at 50% HP or below", ) # ============================================================================= # Combat State Models # ============================================================================= class CombatantInfo(BaseModel): """Information about a combatant in initiative order.""" id: str = Field(description="Combatant ID") name: str = Field(description="Combatant name") initiative: int = Field(description="Initiative roll result") is_player: bool = Field(default=False, description="Is this a player character") is_current: bool = Field(default=False, description="Is it this combatant's turn") hp_current: int = Field(default=0, description="Current HP") hp_max: int = Field(default=1, description="Maximum HP") hp_percent: float = Field(default=100.0, description="HP percentage") conditions: list[str] = Field(default_factory=list, description="Active conditions") status: str = Field(default="healthy", description="Status string") class CombatStateResult(FormattedResult): """Enhanced combat state result.""" action: str = Field( default="", description="Combat action: start, end, next_turn, etc.", ) round_number: int | None = Field( default=None, description="Current combat round", ) current_combatant: str | None = Field( default=None, description="Name of current combatant", ) current_combatant_is_player: bool = Field( default=False, description="Whether current combatant is a player", ) turn_order: list[CombatantInfo] = Field( default_factory=list, description="Initiative order with combatant info", ) combat_ended: bool = Field( default=False, description="Whether combat has ended", ) # ============================================================================= # Connection Status Model # ============================================================================= class MCPConnectionStatus(BaseModel): """Status information for MCP connection.""" state: ConnectionState = Field( default=ConnectionState.DISCONNECTED, description="Current connection state", ) is_available: bool = Field( default=False, description="Whether MCP is available for use", ) url: str = Field( default="", description="MCP server URL", ) last_successful_call: datetime | None = Field( default=None, description="When the last successful call was made", ) consecutive_failures: int = Field( default=0, description="Number of consecutive failures", ) circuit_breaker_state: CircuitBreakerState = Field( default=CircuitBreakerState.CLOSED, description="Circuit breaker state", ) tools_count: int = Field( default=0, description="Number of available tools", ) error_message: str | None = Field( default=None, description="Last error message if any", )