#!/usr/bin/env python3 """ GumYum NPC API Python SDK - Single File Distribution Cross-engine AI-powered NPC dialogue and quest system client Example usage: import asyncio from gumyum_npc_sdk import GumYumClient async def main(): client = GumYumClient("https://npc.gumyum.com") # Authenticate user = await client.auth.register("username", "password", "email@example.com") # Copy a public universe universe_id = await client.universes.copy_public("blade-runner", "My World") # Spawn an NPC npc = await client.npcs.spawn(universe_id, seed=12345) # Chat with the NPC response = await client.chat.completions( npc_id=npc.npc_id, messages=[{"role": "user", "content": "Hello!"}] ) print(response.choices[0].message.content) asyncio.run(main()) Version: 0.1.0 Author: GumYum NPC API Team Contact: timehexon@gumyum.com """ import asyncio import json from datetime import datetime from enum import Enum from typing import Dict, List, Optional, Any, Union, AsyncIterator, TYPE_CHECKING from urllib.parse import urljoin try: import httpx except ImportError: httpx = None # Import shared data models and exceptions # Import pydantic utilities try: from pydantic import Field except ImportError: Field = None from gumyum_models import ( # Exceptions GumYumError, GumYumAPIError, GumYumAuthError, GumYumNotFoundError, GumYumValidationError, GumYumServerError, GumYumNetworkError, GumYumTimeoutError, GumYumRateLimitError, # Models ChatRole, ChatMessage, ChatChoice, ChatCompletion, SpawnedData, NPCProfile, NPC as NPCBase, UniverseThemeInfo, UniverseListItem, UniverseData, AuthToken, UserProfile, APIUsage, NPCStats, PublicUniverse, ApiKey, ApiKeyPair, MoodData, MoodItem, MoodUploadResponse, ) # ============================================================================ # NPC CHAT PROXY # ============================================================================ class NPCChatProxy: """Proxy object that provides chat.completions() interface for NPC""" def __init__(self, npc: "NPC"): self.npc = npc async def completions( self, messages: List[Union[ChatMessage, dict]], model: str = "gumyum-npc", temperature: float = 0.8, max_tokens: Optional[int] = None, stream: bool = False, **kwargs, ) -> Union[ChatCompletion, AsyncIterator[dict]]: """ Generate AI chat completion with this NPC's context Args: messages: Chat message history model: AI model to use temperature: Response randomness (0.0-2.0) max_tokens: Maximum response length stream: Enable streaming responses **kwargs: Additional OpenAI-compatible parameters Returns: Chat completion response or async iterator for streaming Examples: # Basic chat response = await npc.chat.completions( messages=[{"role": "user", "content": "Hello!"}] ) print(response.choices[0].message.content) # With streaming async for chunk in await npc.chat.completions( messages=[{"role": "user", "content": "Tell me a story"}], stream=True ): print(chunk["choices"][0]["delta"]["content"], end="") """ if not self.npc.client: raise ValueError("NPC has no client reference. Was it spawned properly?") # Build npc_params with the NPC's context npc_params = { "universe_id": self.npc.universe_id, "world_seed": self.npc.world_seed, "npc_id": self.npc.npc_id, } # If not streaming, wrap response to track history if not stream: response = await self.npc.client.chat.completions( npc_params=npc_params, messages=messages, model=model, temperature=temperature, max_tokens=max_tokens, stream=stream, **kwargs, ) # Track history if response and hasattr(response, "choices") and response.choices: # Add user message to history if messages and messages[-1].get("role") == "user": self.npc.chat_history.append(messages[-1]) # Add assistant response to history assistant_msg = response.choices[0].message self.npc.chat_history.append( {"role": assistant_msg.role, "content": assistant_msg.content} ) return response else: # For streaming, return the stream directly return await self.npc.client.chat.completions( npc_params=npc_params, messages=messages, model=model, temperature=temperature, max_tokens=max_tokens, stream=stream, **kwargs, ) async def chat_with_history( self, message: str, model: Optional[str] = None, temperature: float = 0.8, max_tokens: int = 2000, stream: bool = False, **kwargs: Any, ) -> Union[ChatCompletion, AsyncIterator[Dict[str, Any]]]: """Chat with conversation history - includes all previous messages in this session. Args: message: The user's message model: Model to use (defaults to gumyum-npc) temperature: Sampling temperature max_tokens: Maximum tokens to generate stream: Whether to stream the response **kwargs: Additional API parameters Returns: ChatCompletion or async iterator for streaming """ # Build messages array with history plus new message messages_with_history = self.npc.chat_history.copy() messages_with_history.append({"role": "user", "content": message}) # Use regular completions which will also update history return await self.completions( messages=messages_with_history, model=model, temperature=temperature, max_tokens=max_tokens, stream=stream, **kwargs, ) # ============================================================================ # NPC EXTENSIONS # ============================================================================ class NPCAsync(NPCBase): """Async NPC with client integration and chat capabilities""" model_config = {"arbitrary_types_allowed": True} # Client reference - excluded from serialization client: Optional["GumYumClient"] = Field(default=None, exclude=True) # Chat proxy - created on demand _chat: Optional[NPCChatProxy] = None @property def chat(self) -> NPCChatProxy: """Get the chat proxy for this NPC""" if self._chat is None: self._chat = NPCChatProxy(self) return self._chat async def completions( self, messages: List[Union[ChatMessage, dict]], **kwargs ) -> Union[ChatCompletion, AsyncIterator[dict]]: """ Convenience alias for npc.chat.completions() Args: messages: Chat message history **kwargs: Additional parameters passed to chat.completions Returns: Chat completion response or async iterator for streaming Example: # Direct usage response = await npc.completions([ {"role": "user", "content": "Hello!"} ]) """ return await self.chat.completions(messages, **kwargs) async def chat_with_history( self, message: str, **kwargs ) -> Union[ChatCompletion, AsyncIterator[dict]]: """ Chat with conversation history - includes all previous messages in this session. Args: message: The user's message **kwargs: Additional parameters passed to chat.completions Returns: Chat completion response or async iterator for streaming Example: # Continue conversation with context response = await npc.chat_with_history("Tell me more about that") print(response.choices[0].message.content) """ return await self.chat.chat_with_history(message, **kwargs) def clear_history(self): """Clear the chat history for this NPC""" self.chat_history = [] def chat_history_to_json(self) -> str: """Convert chat history to JSON string""" return json.dumps(self.chat_history) def to_dict(self) -> dict: """Convert NPC to dictionary""" # Use pydantic's model_dump if available, otherwise dict if hasattr(self, "model_dump"): return self.model_dump(exclude={"client", "_chat"}) else: data = self.dict(exclude={"client", "_chat"}) return data def to_json(self) -> str: """Convert NPC to JSON string""" return json.dumps(self.to_dict()) @classmethod def from_dict( cls, data: dict, client: Optional["GumYumClient"] = None ) -> "NPCAsync": """Create NPC from dictionary""" npc = cls(**data) if client: npc.client = client return npc @classmethod def from_json( cls, json_str: str, client: Optional["GumYumClient"] = None ) -> "NPCAsync": """Create NPC from JSON string""" try: return cls.from_dict(json.loads(json_str), client) except json.JSONDecodeError as e: raise ValueError(f"Failed to parse NPC JSON: {e}") except Exception as e: raise ValueError(f"Failed to parse NPC JSON: {e}") # Use NPCAsync as NPC for compatibility NPC = NPCAsync # ============================================================================ # MANAGER CLASSES # ============================================================================ class AuthManager: """Handles authentication operations""" def __init__(self, client: "GumYumClient"): self.client = client async def register(self, username: str, password: str, email: str) -> AuthToken: """ Register a new user account Args: username: Unique username password: Password (minimum 8 characters) email: Valid email address Returns: Authentication token data """ data = {"username": username, "password": password, "email": email} response = await self.client.post("auth/register", json_data=data) token = AuthToken(**response) # Automatically set token for subsequent requests self.client.set_token(token.access_token, token.refresh_token, token.expires_in) return token async def login(self, username: str, password: str) -> AuthToken: """ Login with existing account Args: username: Username or email password: Password Returns: Authentication token data """ data = {"username": username, "password": password} response = await self.client.post("auth/login", json_data=data) token = AuthToken(**response) # Automatically set token for subsequent requests self.client.set_token(token.access_token, token.refresh_token, token.expires_in) return token async def refresh_token(self, refresh_token: str) -> AuthToken: """ Refresh access token using refresh token Args: refresh_token: Valid refresh token Returns: New authentication token data """ data = {"refresh_token": refresh_token} response = await self.client.post("auth/refresh", json_data=data) # Extract new tokens and update client new_access_token = response["access_token"] new_refresh_token = response.get( "refresh_token", refresh_token ) # Server now returns new refresh token expires_in = response.get("expires_in", 3600) self.client.set_token(new_access_token, new_refresh_token, expires_in) return AuthToken( access_token=new_access_token, refresh_token=new_refresh_token, # Use the new refresh token token_type=response.get("token_type", "Bearer"), expires_in=response.get("expires_in", 3600), user_id=response.get("user_id", ""), username=response.get("username", ""), ) def logout(self): """ Logout by clearing the authentication token """ self.client.set_token(None, None) async def get_profile(self) -> UserProfile: """ Get current user profile Returns: User profile data Raises: GumYumAuthError: If not authenticated """ response = await self.client.get("auth/profile") return UserProfile(**response) def is_authenticated(self) -> bool: """ Check if client has an authentication token Returns: True if authenticated, False otherwise """ return self.client._token is not None async def delete_account(self) -> Dict[str, Any]: """ Delete the current user's account (GDPR compliance) Returns: Deletion confirmation with cleanup statistics Raises: GumYumAuthError: If not authenticated """ response = await self.client.delete("auth/account") # Clear token after successful deletion self.client.set_token(None, None) return response # API Key Management Methods async def create_api_key( self, name: str, description: str = "", permissions: Optional[List[str]] = None, rate_limits: Optional[Dict[str, int]] = None, ) -> ApiKeyPair: """ Create a new API key pair for game integration Args: name: Descriptive name for the API key description: Optional description permissions: List of permissions (defaults to standard game permissions) rate_limits: Rate limiting configuration Returns: API key pair with both public and secret keys (shown only once) Raises: GumYumAuthError: If not authenticated """ data = {"name": name, "description": description} if permissions is not None: data["permissions"] = permissions if rate_limits is not None: data["rate_limits"] = rate_limits response = await self.client.post("auth/api-keys", json_data=data) return ApiKeyPair(**response) async def list_api_keys(self) -> List[ApiKey]: """ List all API keys for the current user Returns: List of API key metadata (secret keys never included) Raises: GumYumAuthError: If not authenticated """ response = await self.client.get("auth/api-keys") return [ApiKey(**key) for key in response["api_keys"]] async def get_api_key(self, key_id: str) -> ApiKey: """ Get details for a specific API key Args: key_id: API key ID Returns: API key metadata (secret key not included) Raises: GumYumAuthError: If not authenticated GumYumError: If key not found """ response = await self.client.get(f"auth/api-keys/{key_id}") return ApiKey(**response) async def update_api_key( self, key_id: str, name: Optional[str] = None, description: Optional[str] = None, permissions: Optional[List[str]] = None, rate_limits: Optional[Dict[str, int]] = None, status: Optional[str] = None, ) -> bool: """ Update API key metadata Args: key_id: API key ID name: New name (optional) description: New description (optional) permissions: New permissions (optional) rate_limits: New rate limits (optional) status: New status (optional) Returns: True if successful Raises: GumYumAuthError: If not authenticated GumYumError: If key not found """ data = {} if name is not None: data["name"] = name if description is not None: data["description"] = description if permissions is not None: data["permissions"] = permissions if rate_limits is not None: data["rate_limits"] = rate_limits if status is not None: data["status"] = status response = await self.client.put(f"auth/api-keys/{key_id}", json_data=data) return response.get("success", False) async def revoke_api_key(self, key_id: str) -> bool: """ Permanently revoke an API key Args: key_id: API key ID Returns: True if successful Raises: GumYumAuthError: If not authenticated GumYumError: If key not found """ response = await self.client.delete(f"auth/api-keys/{key_id}") return response.get("success", False) async def regenerate_api_key(self, key_id: str) -> Dict[str, str]: """ Generate new API key pair for existing key ID Args: key_id: API key ID Returns: New public and secret key pair (shown only once) Raises: GumYumAuthError: If not authenticated GumYumError: If key not found """ response = await self.client.post(f"auth/api-keys/{key_id}/regenerate") return { "public_key": response["public_key"], "secret_key": response["secret_key"], } class UniverseManager: """Handles universe operations""" def __init__(self, client: "GumYumClient"): self.client = client async def list_public(self) -> List[PublicUniverse]: """ List all available public universes Returns: List of public universe metadata """ response = await self.client.get("public/universes") return [PublicUniverse(**universe) for universe in response["public_universes"]] async def copy_public( self, public_universe_id: str, custom_name: Optional[str] = None, mood_id: Optional[str] = None, ) -> str: """ Copy a public universe to user's vault Args: public_universe_id: ID of the public universe to copy custom_name: Optional custom name for the copied universe mood_id: Optional mood UUID or name to link to universe Returns: New universe ID in user's vault """ data = {"public_universe_id": public_universe_id} if custom_name: data["custom_name"] = custom_name if mood_id: data["mood_id"] = mood_id response = await self.client.post("public/universes/copy", json_data=data) return response["universe_id"] async def list_user(self) -> List[UniverseListItem]: """ List user's private universes Returns: List of user's universe metadata Raises: GumYumAuthError: If not authenticated """ response = await self.client.get("universes") return [UniverseListItem(**universe) for universe in response["universes"]] async def create( self, universe_data: UniverseData, mood_id: Optional[str] = None ) -> str: """ Create a new universe in user's vault Args: universe_data: Universe configuration data mood_id: Optional mood UUID or name to link to universe Returns: New universe ID Raises: GumYumAuthError: If not authenticated GumYumValidationError: If universe data is invalid """ # Convert Pydantic model to dict if needed if hasattr(universe_data, "model_dump"): data = universe_data.model_dump(exclude_none=True) elif hasattr(universe_data, "dict"): data = universe_data.dict(exclude_none=True) else: data = universe_data # Add mood_id if provided if mood_id: data["mood_id"] = mood_id response = await self.client.post("universes", json_data=data) return response["universe_id"] async def get(self, universe_id: str) -> UniverseData: """ Get a specific universe by ID Args: universe_id: Universe identifier Returns: Universe data Raises: GumYumNotFoundError: If universe not found GumYumAuthError: If not owner of private universe """ response = await self.client.get(f"universes/{universe_id}") # The API returns the raw universe data, we need to wrap it return UniverseData( universe_id=universe_id, name=response.get("theme_info", {}).get("name", "Unknown"), hash="", # Not returned by API data=response, ) async def update( self, universe_id: str, universe_data: Union[UniverseData, dict] ) -> UniverseData: """ Update an existing universe Args: universe_id: Universe identifier universe_data: Updated universe data Returns: Updated universe data Raises: GumYumNotFoundError: If universe not found GumYumAuthError: If not owner of universe """ # Convert Pydantic model to dict if needed if hasattr(universe_data, "model_dump"): data = universe_data.model_dump(exclude_none=True) elif hasattr(universe_data, "dict"): data = universe_data.dict(exclude_none=True) else: data = universe_data response = await self.client.put(f"universes/{universe_id}", json_data=data) # Return the updated universe return await self.get(universe_id) async def fork(self, universe_id: str, name: Optional[str] = None) -> str: """ Fork (copy) an existing universe Args: universe_id: Source universe ID to fork name: Optional name for the forked universe Returns: New forked universe ID Raises: GumYumNotFoundError: If source universe not found """ data = {} if name: data["name"] = name response = await self.client.post( f"universes/{universe_id}/fork", json_data=data ) return response["universe_id"] async def delete(self, universe_id: str) -> bool: """ Delete a universe from user's vault Args: universe_id: Universe identifier to delete Returns: True if successfully deleted Raises: GumYumNotFoundError: If universe not found GumYumAuthError: If not owner of universe """ await self.client.delete(f"universes/{universe_id}") return True async def list(self) -> List[UniverseListItem]: """ List user's universes (alias for list_user) Returns: List of user's universe metadata Raises: GumYumAuthError: If not authenticated """ return await self.list_user() # Mood-related methods async def upload_mood( self, mood_name: str, mood_data: Union[MoodData, Dict[str, Any]] ) -> MoodUploadResponse: """ Upload a custom mood configuration Args: mood_name: Human-readable name for the mood mood_data: Mood categories data (MoodData object or dict with 'categories') Returns: Upload response with mood UUID Raises: GumYumAuthError: If not authenticated GumYumValidationError: If mood data is invalid """ # Convert to dict if needed if isinstance(mood_data, MoodData): mood_dict = mood_data.model_dump() elif hasattr(mood_data, "dict"): mood_dict = mood_data.dict() else: mood_dict = mood_data data = {"mood_id": mood_name, "mood_data": mood_dict} response = await self.client.post("moods/upload", json_data=data) return MoodUploadResponse(**response) async def list_moods(self) -> List[MoodItem]: """ List user's custom moods Returns: List of mood items Raises: GumYumAuthError: If not authenticated """ response = await self.client.get("moods") return [MoodItem(**mood) for mood in response["moods"]] async def delete_mood(self, mood_id: str) -> bool: """ Delete a custom mood Args: mood_id: Mood UUID or name to delete Returns: True if successfully deleted Raises: GumYumNotFoundError: If mood not found GumYumAuthError: If not authenticated """ await self.client.delete(f"moods/{mood_id}") return True class NPCManager: """Handles NPC operations""" def __init__(self, client: "GumYumClient"): self.client = client async def spawn( self, universe_id: str, seed: int, npc_id: Optional[int] = None, ) -> NPC: """ Spawn a deterministic NPC Args: universe_id: Universe to spawn NPC in seed: Deterministic seed for generation npc_id: Specific 64-bit NPC ID (optional, generates random from seed if None) location_filter: Filter NPCs by location types Returns: Spawned NPC data Raises: GumYumNotFoundError: If universe not found GumYumAuthError: If not authenticated """ # Ensure authentication BEFORE making any requests await self.client._ensure_authenticated() if not self.client._token or not self.client._token.strip(): raise GumYumAuthError("Authentication required for NPC spawning") if npc_id is not None: # Spawn with specific 9-digit NPC ID params = {"universe_id": universe_id, "seed": seed, "npc_id": npc_id} response = await self.client.get("npc/spawn", params=params) npc = NPC(**response) npc.client = self.client # Set world_seed if not in response (API may not return it) if npc.world_seed is None: npc.world_seed = seed return npc else: # Generate random NPC ID from seed and redirect params = {"universe_id": universe_id, "seed": seed} response = await self.client.get("npc/spawn", params=params) npc = NPC(**response) npc.client = self.client # Set world_seed if not in response (API may not return it) if npc.world_seed is None: npc.world_seed = seed return npc async def spawn_random(self, universe_id: str, seed: int) -> NPC: """ Spawn random NPC using deterministic seed Args: universe_id: Universe to spawn NPC in seed: Deterministic seed for generation Returns: Spawned NPC data """ return await self.spawn(universe_id, seed, None) async def spawn_filtered( self, universe_id: str, world_seed: int, filters: Dict[str, Any], start_npc_id: int = 1, max_attempts: int = 10000, ) -> Optional[NPC]: """ Spawn NPC matching specific filters using 64-bit crawling Args: universe_id: Universe to spawn NPC in world_seed: World seed for generation filters: Dictionary of filter criteria start_npc_id: Starting NPC ID for search max_attempts: Maximum search attempts Returns: NPC if found, None if no match within attempts Filter options: - profession: List[str] - Specific professions - personality_type: List[int] - Personality types (1-9) - wing: List[int] - Wing types (1-9) - stress_level: Dict[str, int] - {'min': 1, 'max': 9} - mood: List[str] - Specific moods - gender: List[str] - ['male', 'female', 'neutral'] - location: List[str] - Location names to filter by Example: filters = { "profession": ["warrior", "mage"], "personality_type": [1, 8], # Perfectionist or Challenger "stress_level": {"min": 1, "max": 3} # Low stress } """ data = { "universe_id": universe_id, "world_seed": world_seed, "filters": filters, "start_npc_id": start_npc_id, "max_attempts": max_attempts, } try: response = await self.client.post("npc/spawn/filtered", json_data=data) except GumYumNotFoundError as e: # If the error is specifically "No NPC found matching filters", return None if "No NPC found matching filters" in str(e): return None # Re-raise other NotFound errors raise # Check if NPC was found (response will have npc_id if found) if not response.get("npc_id"): return None # Convert response to NPC spawned_data = None if "spawned" in response and response["spawned"]: spawned_info = response["spawned"] spawned_data = SpawnedData( location=spawned_info.get("location"), mood=spawned_info.get("mood"), stress_level=spawned_info.get("stress_level"), ) npc = NPC( npc_id=response["npc_id"], name=response["name"], profession=response["profession"], personality_type=response["personality_type"], mood=( response.get("spawned", {}).get("mood") if "spawned" in response else None ), spawned=spawned_data, universe_id=universe_id, # Use the passed universe_id world_seed=world_seed, # Use the passed world_seed cached=False, cache_url=None, ) npc.client = self.client return npc async def get_profile(self, npc_id: int) -> NPCProfile: """ Get full NPC profile data Args: npc_id: NPC identifier Returns: Complete NPC profile Raises: GumYumNotFoundError: If NPC not found """ response = await self.client.get(f"npc/saved/{npc_id}") # API returns data wrapped in 'npc' field if "npc" in response: return NPCProfile(**response["npc"]) return NPCProfile(**response) async def save( self, universe_id: str, seed: int, npc_id: int, custom_name: Optional[str] = None, ) -> NPCProfile: """ Save an NPC to user's vault for persistence Args: universe_id: Universe the NPC belongs to seed: Seed used for generation npc_id: 64-bit NPC ID custom_name: Optional custom name override Returns: Saved NPC profile Raises: GumYumAuthError: If not authenticated """ data = { "universe_id": universe_id, "world_seed": seed, # API expects world_seed for save endpoint "npc_id": npc_id, # API expects npc_id for save endpoint } if custom_name: data["custom_name"] = custom_name response = await self.client.post("npc/save", json_data=data) return NPCProfile(**response) async def list_saved(self) -> List[NPC]: """ List user's saved NPCs Returns: List of saved NPCs with client attached, ready for chat Raises: GumYumAuthError: If not authenticated """ response = await self.client.get("npc/list") npcs = [] for npc_data in response["npcs"]: # Convert list response data to NPC npc = NPC( npc_id=npc_data["npc_id"], name=npc_data["name"], profession=npc_data["profession"], personality_type=npc_data.get( "personality_type", 5 ), # Default if missing spawned=SpawnedData(**npc_data["spawned"]), universe_id=npc_data["universe_id"], seed=npc_data.get("seed", 0), # Default seed if not provided cached=False, cache_url=None, ) # Attach client reference for chat functionality npc.client = self.client npcs.append(npc) return npcs async def update(self, npc_id: int, updates: dict) -> NPCProfile: """ Update a saved NPC's data Args: npc_id: NPC identifier updates: Fields to update Returns: Updated NPC profile Raises: GumYumNotFoundError: If NPC not found GumYumAuthError: If not owner of NPC """ response = await self.client.put(f"npc/saved/{npc_id}", json_data=updates) return NPCProfile(**response) async def delete(self, npc_id: int) -> bool: """ Delete a saved NPC Args: npc_id: NPC identifier to delete Returns: True if successfully deleted Raises: GumYumNotFoundError: If NPC not found GumYumAuthError: If not owner of NPC """ await self.client.delete(f"npc/saved/{npc_id}") return True async def list(self) -> List[NPC]: """ List user's saved NPCs Returns: List of saved NPCs with client attached, ready for chat """ return await self.list_saved() async def stats(self) -> NPCStats: """ Get user's NPC statistics Returns: NPC usage statistics """ return await self.get_stats() async def get_stats(self) -> NPCStats: """ Get user's NPC statistics Returns: NPC usage statistics Raises: GumYumAuthError: If not authenticated """ response = await self.client.get("npc/stats") return NPCStats(**response) class ChatManager: """Handles AI chat completions with NPCs""" def __init__(self, client: "GumYumClient"): self.client = client async def completions( self, npc_id: Optional[int] = None, messages: Optional[List[Union[ChatMessage, dict]]] = None, npc_params: Optional[dict] = None, model: str = "gumyum-npc", temperature: float = 0.8, max_tokens: Optional[int] = None, stream: bool = False, **kwargs, ) -> Union[ChatCompletion, AsyncIterator[dict]]: """ Generate AI chat completion with NPC context Args: npc_id: Saved NPC ID to chat with messages: Chat message history npc_params: Parameters to spawn temporary NPC (if npc_id not provided) Should contain: universe_id, world_seed, npc_id model: AI model to use temperature: Response randomness (0.0-2.0) max_tokens: Maximum response length stream: Enable streaming responses **kwargs: Additional OpenAI-compatible parameters Returns: Chat completion response or async iterator for streaming Examples: # Chat with saved NPC response = await client.chat.completions( npc_id="npc_123", messages=[{"role": "user", "content": "Hello!"}] ) # Chat with temporary NPC response = await client.chat.completions( npc_params={ "universe_id": "universe_123", "world_seed": 12345, "npc_id": 123456789 }, messages=[{"role": "user", "content": "Hello!"}] ) # Streaming chat async for chunk in client.chat.completions( npc_id="npc_123", messages=[{"role": "user", "content": "Tell me a story"}], stream=True ): print(chunk["choices"][0]["delta"]["content"], end="") """ # Build request data data = {"model": model, "temperature": temperature, "stream": stream, **kwargs} if max_tokens is not None: data["max_tokens"] = max_tokens # Add NPC context if npc_id is not None: data["npc_id"] = str(npc_id) # Convert to string as expected by Elixir API elif npc_params: data["npc_params"] = npc_params else: raise ValueError("Either npc_id or npc_params must be provided") # Add messages if messages: # Convert ChatMessage objects to dicts if needed formatted_messages = [] for msg in messages: if isinstance(msg, ChatMessage): if hasattr(msg, "model_dump"): formatted_messages.append(msg.model_dump()) elif hasattr(msg, "dict"): formatted_messages.append(msg.dict()) else: formatted_messages.append( {"role": msg.role, "content": msg.content} ) else: formatted_messages.append(msg) data["messages"] = formatted_messages else: data["messages"] = [] if stream: # For streaming, we need to return the async generator # Since this method is async, we can't return an async generator directly # We need to yield from it async def stream_wrapper(): async for chunk in self._stream_completions(data): yield chunk return stream_wrapper() else: response = await self.client.post("chat/completions", json_data=data) return ChatCompletion(**response) async def _stream_completions(self, data: dict) -> AsyncIterator[dict]: """ Handle streaming chat completions Args: data: Request data Yields: Streaming response chunks """ # Note: This is a simplified streaming implementation # In a real implementation, you'd handle Server-Sent Events (SSE) async with self.client._client.stream( "POST", f"{self.client.api_url}/chat/completions", json=data, headers=self.client.get_headers(), ) as response: if response.status_code >= 400: # For streaming responses, we need to read the content first content = await response.aread() error_text = content.decode("utf-8") if response.status_code == 401: raise GumYumAuthError( f"HTTP {response.status_code}: {error_text}", response.status_code, ) elif response.status_code == 404: raise GumYumNotFoundError( f"HTTP {response.status_code}: {error_text}", response.status_code, ) elif response.status_code in (400, 422): raise GumYumValidationError( f"HTTP {response.status_code}: {error_text}", response.status_code, ) else: raise GumYumServerError( f"HTTP {response.status_code}: {error_text}", response.status_code, ) # Check if we got a streaming response or a regular JSON response content_type = response.headers.get("content-type", "") if "event-stream" not in content_type: # API returned non-streaming response, convert to streaming format try: # Read the response content first content = await response.aread() json_response = json.loads(content.decode("utf-8")) # Emit the content as chunks if "choices" in json_response and json_response["choices"]: message_content = json_response["choices"][0]["message"][ "content" ] # Split content into words to simulate streaming words = message_content.split() import asyncio for i, word in enumerate(words): chunk = { "choices": [ { "delta": { "content": word + (" " if i < len(words) - 1 else "") } } ] } yield chunk # Add small delay to simulate streaming await asyncio.sleep(0.02) except Exception as e: # If we can't parse it, fall back to non-streaming pass else: # Handle actual streaming response async for line in response.aiter_lines(): if line.startswith("data: "): chunk_data = line[6:] # Remove "data: " prefix if chunk_data.strip() == "[DONE]": break try: chunk = json.loads(chunk_data) yield chunk except json.JSONDecodeError: continue # ============================================================================ class MoodManager: """Handles custom mood management operations""" def __init__(self, client: "GumYumClient"): self.client = client async def upload(self, mood_id: str, mood_data: dict) -> dict: """ Upload a custom mood JSON file Args: mood_id: Unique identifier for the mood (alphanumeric, underscore, hyphen only) mood_data: Mood data with 'categories' field containing mood definitions Returns: Upload response with success status and categories count Raises: GumYumAuthError: If not authenticated GumYumBadRequestError: If mood_id format is invalid or mood_data structure is wrong """ payload = {"mood_id": mood_id, "mood_data": mood_data} response = await self.client.post("moods/upload", json_data=payload) return response async def list(self) -> List[dict]: """ List all available custom mood files Returns: List of custom moods with mood_id, categories_count, and filename Raises: GumYumAuthError: If not authenticated """ response = await self.client.get("moods") return response.get("moods", []) async def delete(self, mood_id: str) -> dict: """ Delete a custom mood file Args: mood_id: The mood ID to delete Returns: Deletion response with success status Raises: GumYumAuthError: If not authenticated GumYumNotFoundError: If mood not found GumYumBadRequestError: If trying to delete default GumYum moods """ response = await self.client.delete(f"moods/{mood_id}") return response async def create_simple_mood( self, mood_id: str, category_name: str, moods: dict ) -> dict: """ Create a simple custom mood with a single category Args: mood_id: Unique identifier for the mood category_name: Name of the mood category (e.g., "ENERGETIC", "CALM") moods: Dictionary of mood_name -> description pairs Returns: Upload response Example: await client.moods.create_simple_mood( "my_energetic_moods", "ENERGETIC", { "pumped": "Full of energy and ready for action", "hyped": "Extremely excited and enthusiastic" } ) """ mood_data = { "categories": { category_name: { mood_name: {"description": description} for mood_name, description in moods.items() } } } return await self.upload(mood_id, mood_data) # MAIN CLIENT # ============================================================================ class GumYumClient: """ Main client for GumYum NPC API Example: client = GumYumClient("https://api.gumyum.com") user = await client.auth.register("username", "password", "email") npc = await client.npcs.spawn("universe_id", world_seed=12345) response = await client.chat.completions(npc.npc_id, "Hello!") """ def __init__( self, base_url: str = "http://localhost:6543", api_key: Optional[str] = None, timeout: float = 75.0, # Higher than server's 65s timeout max_retries: int = 3, **kwargs, ): """ Initialize GumYum client Args: base_url: API base URL api_key: Optional API key for authentication timeout: Request timeout in seconds max_retries: Maximum number of retries for failed requests **kwargs: Additional httpx client options """ if httpx is None: raise ImportError("httpx is required. Install with: pip install httpx") self.base_url = base_url.rstrip("/") self.api_url = f"{self.base_url}/v1" self.timeout = timeout self.max_retries = max_retries # HTTP client configuration client_kwargs = {"timeout": timeout, "follow_redirects": True, **kwargs} self._client = httpx.AsyncClient(**client_kwargs) self._token: Optional[str] = None self._refresh_token: Optional[str] = None self._token_expires_at: Optional[float] = None self._api_key: Optional[str] = api_key self._public_key: Optional[str] = None self._secret_key: Optional[str] = None self._jwt_obtained: bool = False # Track if we've exchanged API key for JWT # Parse API key if provided (format: "public_key:secret_key") if api_key and ":" in api_key: parts = api_key.split(":", 1) self._public_key = parts[0] self._secret_key = parts[1] # Initialize managers self.auth = AuthManager(self) self.universes = UniverseManager(self) self.npcs = NPCManager(self) self.chat = ChatManager(self) self.moods = MoodManager(self) async def __aenter__(self): """Async context manager entry""" return self async def __aexit__(self, exc_type, exc_val, exc_tb): """Async context manager exit""" await self.close() async def close(self): """Close the HTTP client""" await self._client.aclose() def set_token( self, token: Optional[str], refresh_token: Optional[str] = None, expires_in: Optional[int] = None, ): """Set authentication token""" self._token = token if refresh_token is not None: self._refresh_token = refresh_token if expires_in is not None: import time self._token_expires_at = time.time() + expires_in # Mark that user has explicitly authenticated if token: self._auth_attempted_by_user = True def set_api_key(self, public_key: str, secret_key: str): """Set API key credentials for authentication""" self._public_key = public_key self._secret_key = secret_key self._api_key = f"{public_key}:{secret_key}" # Mark that user has explicitly authenticated self._auth_attempted_by_user = True def get_headers(self) -> Dict[str, str]: """Get request headers with authentication""" headers = { "Content-Type": "application/json", "User-Agent": "GumYum-Python-SDK/0.1.0", } # For API key auth: use JWT if we have it, otherwise use API key if self._public_key and self._secret_key: if self._jwt_obtained and self._token and self._token.strip(): # We've already exchanged API key for JWT, use JWT headers["Authorization"] = f"Bearer {self._token}" else: # First request with API key - will trigger JWT exchange headers["Authorization"] = ( f"Bearer {self._public_key}:{self._secret_key}" ) elif self._token and self._token.strip(): # Direct JWT auth (from login) headers["Authorization"] = f"Bearer {self._token}" else: # Explicitly avoid sending empty/null Authorization headers # which can cause HTTP 500 errors on the server pass return headers async def _ensure_authenticated(self): """Ensure client is authenticated, auto-login if needed""" # Only auto-authenticate if we've never had a token (prevents overriding explicit auth) if not self._token or not self._token.strip(): if not hasattr(self, "_auth_attempted_by_user"): # Auto-authenticate with default credentials if no token (max 2 attempts) if not hasattr(self, "_auto_auth_attempts"): self._auto_auth_attempts = 0 if self._auto_auth_attempts < 2: self._auto_auth_attempts += 1 try: # Try to login with load test credentials import time timestamp = int(time.time()) username = f"auto_user_{timestamp}_{self._auto_auth_attempts}" await self.auth.register( username, "auto_password123", f"{username}@example.com" ) except Exception: # If registration fails, try login on second attempt if self._auto_auth_attempts == 2: try: await self.auth.login(username, "auto_password123") except Exception: # Both attempts failed - client is now permanently failed pass async def _exchange_api_key_for_jwt(self) -> bool: """Exchange API key for JWT token using the dedicated exchange endpoint""" if not self._public_key or not self._secret_key: return False try: # Use the dedicated exchange endpoint headers = { "Content-Type": "application/json", "User-Agent": "GumYum-Python-SDK/0.1.0", } data = {"public_key": self._public_key, "secret_key": self._secret_key} url = urljoin(f"{self.api_url}/", "auth/exchange") response = await self._client.post(url, json=data, headers=headers) if response.status_code == 200: result = response.json() # Store JWT tokens self._token = result.get("access_token") self._refresh_token = result.get("refresh_token") self._jwt_obtained = True # Calculate expiry time if "expires_in" in result: import time self._token_expires_at = time.time() + result["expires_in"] return True # If exchange fails, continue with API key auth return False except Exception: # If exchange fails, continue with API key auth return False async def _ensure_jwt_valid(self): """Ensure JWT is valid, refresh if needed""" # Check if token is expired for any auth type (not just API key auth) if self._token_expires_at: import time if ( time.time() >= self._token_expires_at - 60 ): # Refresh 1 minute before expiry await self._refresh_jwt() async def _refresh_jwt(self): """Refresh JWT using refresh token""" from urllib.parse import urljoin if not self._refresh_token: # No refresh token, try to re-exchange API key if self._public_key and self._secret_key: await self._exchange_api_key_for_jwt() return else: raise GumYumAuthError("No refresh token available") # Prevent recursive refresh attempts if getattr(self, "_refreshing", False): raise GumYumAuthError("Already attempting to refresh token") self._refreshing = True try: data = {"refresh_token": self._refresh_token} # Use _client directly to avoid going through _request() which could trigger another refresh url = urljoin(f"{self.api_url}/", "auth/refresh") headers = self.get_headers() response = await self._client.post(url, json=data, headers=headers) # Check if refresh was successful if response.status_code == 200: response_data = response.json() if "access_token" in response_data: self._token = response_data["access_token"] # Store the new refresh token (server rotates it) if "refresh_token" in response_data: self._refresh_token = response_data["refresh_token"] if "expires_in" in response_data: import time self._token_expires_at = ( time.time() + response_data["expires_in"] ) else: # Refresh failed - raise appropriate error error_data = {} try: error_data = response.json() except: pass raise GumYumAuthError( f"Token refresh failed: {error_data.get('error', 'Unknown error')}", response.status_code, ) except GumYumAuthError: # Re-raise auth errors raise except Exception as e: # For other exceptions, try to re-exchange API key if self._public_key and self._secret_key: self._jwt_obtained = False await self._exchange_api_key_for_jwt() else: raise GumYumAuthError(f"Token refresh failed: {str(e)}") finally: self._refreshing = False async def _request( self, method: str, endpoint: str, params: Optional[Dict[str, Any]] = None, json_data: Optional[Dict[str, Any]] = None, **kwargs, ) -> Dict[str, Any]: """ Make HTTP request with error handling and retries Args: method: HTTP method endpoint: API endpoint (relative to api_url) params: Query parameters json_data: JSON request body **kwargs: Additional request options Returns: Response JSON data Raises: GumYumError: On API errors """ # Ensure client is authenticated await self._ensure_authenticated() # Exchange API key for JWT on first request if using API key auth if self._public_key and self._secret_key and not self._jwt_obtained: await self._exchange_api_key_for_jwt() # Ensure JWT is still valid await self._ensure_jwt_valid() url = urljoin(f"{self.api_url}/", endpoint.lstrip("/")) # Save extra headers if provided extra_headers = kwargs.pop("headers", {}) retries = 0 while retries <= self.max_retries: # Build headers inside the loop so we get fresh auth token headers = self.get_headers() # Merge extra headers, but sanitize Authorization to prevent empty arrays for key, value in extra_headers.items(): if key.lower() == "authorization" and ( not value or value == [] or value == [""] ): # Skip malformed Authorization headers continue headers[key] = value try: # Final aggressive check: remove ANY malformed Authorization header if "Authorization" in headers: auth_value = headers["Authorization"] if ( auth_value == [] or auth_value == [""] or auth_value == "" or auth_value is None or (isinstance(auth_value, list) and len(auth_value) == 0) or (isinstance(auth_value, str) and not auth_value.strip()) ): # DEBUG: Log what we're removing import sys print( f"WARNING: Removing malformed Authorization header: {repr(auth_value)}", file=sys.stderr, ) del headers["Authorization"] # Make a clean copy of headers to avoid any mutation issues clean_headers = { k: v for k, v in headers.items() if not ( k == "Authorization" and ( v == [] or v == [""] or v == "" or v is None or (isinstance(v, list) and len(v) == 0) or (isinstance(v, str) and not v.strip()) ) ) } response = await self._client.request( method=method, url=url, params=params, json=json_data, headers=clean_headers, **kwargs, ) # Handle successful responses if 200 <= response.status_code < 300: if response.headers.get("content-type", "").startswith( "application/json" ): return response.json() else: return {"data": response.text} # Handle errors await self._handle_error_response(response) except GumYumAuthError as e: # Handle 401 errors intelligently if e.status_code == 401 and retries == 0: error_msg = str(e).lower() error_data = getattr(e, "response_data", {}) # Check if it's a token error that might benefit from refresh # The server returns "Invalid token" for expired tokens too error_str = str(error_data.get("error", "")).lower() message_str = str(error_data.get("message", "")).lower() is_token_error = ( "expired" in error_msg or "token expired" in error_str or "token has expired" in message_str or "invalid token" in error_str or "authentication token is invalid" in message_str ) # Don't refresh for API key errors or missing auth is_api_key_error = ( "api key" in error_msg or "api key" in error_str or "authentication required" in message_str ) if is_token_error and not is_api_key_error: # Try to refresh if we have a refresh token if self._refresh_token: try: await self._refresh_jwt() # Retry the request with the new token retries += 1 # Increment retry count # Don't clear token or raise - let the loop continue except GumYumAuthError as refresh_error: # Refresh failed - check if it's because refresh token is invalid (403) refresh_error_data = getattr( refresh_error, "response_data", {} ) refresh_error_msg = str(refresh_error).lower() # Check for 403 status (invalid/expired refresh token) if refresh_error.status_code == 403 or ( "expired" in refresh_error_msg or "invalid refresh token" in refresh_error_msg or "expired" in refresh_error_data.get("error", "").lower() or "expired" in refresh_error_data.get("message", "").lower() ): # Both tokens expired - try to re-authenticate if self._public_key and self._secret_key: # API key auth - exchange for new tokens try: self._jwt_obtained = False await self._exchange_api_key_for_jwt() retries += 1 # Retry with new tokens except Exception: # Final failure self._token = None self._refresh_token = None raise GumYumAuthError( "Authentication failed - all tokens expired", e.status_code, ) else: # Password auth - try to use refresh token if available if self._refresh_token: try: await self._refresh_jwt() # Retry the request with the new token retries += 1 continue except: # Refresh failed, clear tokens self._token = None self._refresh_token = None raise GumYumAuthError( "Session expired - please login again", e.status_code, ) else: # No refresh token, must re-login self._token = None self._refresh_token = None raise GumYumAuthError( "Session expired - please login again", e.status_code, ) else: # Other refresh failure self._token = None raise GumYumAuthError( "Authentication failed - token refresh failed", e.status_code, ) except Exception as refresh_error: # Unexpected error during refresh self._token = None raise GumYumAuthError( f"Authentication failed - token cleared: {str(refresh_error)}", getattr(refresh_error, "status_code", None), ) else: # No refresh token, clear token and raise self._token = None raise GumYumAuthError( "Authentication failed - token cleared", e.status_code ) else: # Not a token error or is API key error # Clear the invalid token self._token = None # Don't retry without proper auth raise GumYumAuthError( "Authentication failed - token cleared", e.status_code ) else: # Don't retry, bubble up the auth error raise except GumYumServerError as e: # Retry on 503 Service Unavailable if e.status_code == 503 and retries < self.max_retries: retries += 1 # Exponential backoff with jitter await asyncio.sleep(min(2**retries + 0.1 * retries, 10)) continue else: raise except httpx.TimeoutException as e: if retries >= self.max_retries: raise GumYumTimeoutError( f"Request timeout after {self.timeout}s" ) from e except httpx.NetworkError as e: if retries >= self.max_retries: raise GumYumNetworkError(f"Network error: {e}") from e retries += 1 if retries <= self.max_retries: # Exponential backoff await asyncio.sleep(2**retries) raise GumYumError("Max retries exceeded") async def _handle_error_response(self, response: httpx.Response): """Handle HTTP error responses""" try: error_data = response.json() message = error_data.get( "message", error_data.get("error", f"HTTP {response.status_code}") ) if not message or message == f"HTTP {response.status_code}": # Try to extract more details if "errors" in error_data: message = f"HTTP {response.status_code}: {error_data['errors']}" else: message = f"HTTP {response.status_code}: {response.text[:200]}" except: message = f"HTTP {response.status_code}: {response.text[:200]}" error_data = {} # Map status codes to exception types if response.status_code == 401: raise GumYumAuthError(message, response.status_code, error_data) elif response.status_code == 403: raise GumYumAuthError(message, response.status_code, error_data) elif response.status_code == 404: raise GumYumNotFoundError(message, response.status_code, error_data) elif response.status_code in (400, 422): validation_errors = error_data.get("errors", []) raise GumYumValidationError( message, response.status_code, error_data, validation_errors ) elif response.status_code == 429: retry_after = response.headers.get("Retry-After") raise GumYumRateLimitError( message, response.status_code, error_data, int(retry_after) if retry_after else None, ) elif 500 <= response.status_code < 600: raise GumYumServerError(message, response.status_code, error_data) else: raise GumYumAPIError(message, response.status_code, error_data) async def get(self, endpoint: str, **kwargs) -> Dict[str, Any]: """Make GET request""" return await self._request("GET", endpoint, **kwargs) async def post(self, endpoint: str, **kwargs) -> Dict[str, Any]: """Make POST request""" return await self._request("POST", endpoint, **kwargs) async def put(self, endpoint: str, **kwargs) -> Dict[str, Any]: """Make PUT request""" return await self._request("PUT", endpoint, **kwargs) async def delete(self, endpoint: str, **kwargs) -> Dict[str, Any]: """Make DELETE request""" return await self._request("DELETE", endpoint, **kwargs) async def health_check(self) -> Dict[str, Any]: """Check API health status""" return await self.get("health") # ============================================================================ # EXPORTS # ============================================================================ __version__ = "0.1.0" __author__ = "GumYum NPC API Team" __email__ = "timehexon@gumyum.com" __all__ = [ # Main client "GumYumClient", # Exceptions "GumYumError", "GumYumAPIError", "GumYumAuthError", "GumYumNotFoundError", "GumYumValidationError", "GumYumServerError", "GumYumNetworkError", "GumYumTimeoutError", "GumYumRateLimitError", # Models "NPCProfile", "UniverseData", "ChatCompletion", "ChatMessage", "AuthToken", "NPC", "NPCChatProxy", "UserProfile", "NPCStats", "PublicUniverse", "UniverseThemeInfo", "ChatChoice", "ChatRole", ] # ============================================================================ # EXAMPLE USAGE # ============================================================================ if __name__ == "__main__": async def demo(): """Example usage of the GumYum NPC SDK""" async with GumYumClient("http://localhost:6543") as client: try: # Register or login print("Authenticating...") try: user = await client.auth.login("demo_user", "demo_password123") except GumYumAuthError: user = await client.auth.register( "demo_user", "demo_password123", "demo@example.com" ) print(f"Logged in as: {user.username}") # List public universes print("\nFetching public universes...") universes = await client.universes.list_public() if universes: print(f"Found {len(universes)} public universes") print(f"Using: {universes[0].name}") # Copy a universe universe_id = await client.universes.copy_public( universes[0].id, "My Demo World" ) print(f"Copied universe: {universe_id}") # Spawn an NPC print("\nSpawning NPC...") npc = await client.npcs.spawn( universe_id, seed=12345, npc_id=123456789 ) print(f"Spawned: {npc.name} ({npc.profession})") # Chat with the NPC print(f"\nChatting with {npc.name}...") response = await client.chat.simple_chat( npc.npc_id, "Hello! What's your story?" ) print(f"{npc.name}: {response}") # Streaming chat example print(f"\n{npc.name} (streaming): ", end="") async for chunk in await client.chat.completions( npc_id=npc.npc_id, messages=[ {"role": "user", "content": "Tell me about your day"} ], stream=True, ): if "choices" in chunk and chunk["choices"]: delta = chunk["choices"][0].get("delta", {}) content = delta.get("content", "") if content: print(content, end="", flush=True) print("\n\nDemo completed!") except Exception as e: print(f"Error: {e}") # Run demo if executed directly asyncio.run(demo())