#!/usr/bin/env python3 """ GumYum NPC API Python SDK - Synchronous Version Cross-engine AI-powered NPC dialogue and quest system client Example usage: from gumyum_npc_sync import GumYumClient def main(): client = GumYumClient("https://npc.gumyum.com") # Authenticate user = client.auth.register("username", "password", "email@example.com") # Copy a public universe universe_id = client.universes.copy_public("blade-runner", "My World") # Spawn an NPC npc = client.npcs.spawn(universe_id, seed=12345) # Chat with the NPC response = client.chat.completions( npc_id=npc.npc_id, messages=[{"role": "user", "content": "Hello!"}] ) print(response.choices[0].message.content) main() Version: 0.1.0 Author: GumYum NPC API Team Contact: timehexon@gumyum.com """ import json import time import threading import re import base64 import hashlib import logging from datetime import datetime, timedelta, timezone from enum import Enum from typing import Dict, List, Optional, Any, Union, Iterator, Callable, Tuple from urllib.parse import urljoin from functools import wraps try: import requests except ImportError: requests = None try: from pydantic import BaseModel, Field except ImportError: BaseModel = None Field = None # Import shared data models and exceptions from gumyum_models import ( # Exceptions GumYumError, GumYumAPIError, GumYumAuthError, GumYumNotFoundError, GumYumValidationError, GumYumServerError, GumYumNetworkError, GumYumTimeoutError, GumYumRateLimitError, # Models ChatRole, ChatMessage, ChatChoice, ChatCompletion, SpawnedData, NPCProfile, NPC as AsyncNPC, # Rename to avoid conflict UniverseThemeInfo, UniverseListItem, UniverseData, AuthToken, UserProfile, APIUsage, NPCStats, PublicUniverse, ApiKey, ApiKeyPair, MoodData, MoodItem, MoodUploadResponse, ) # ============================================================================ # SYNC-SPECIFIC DATA MODELS # ============================================================================ class NPCChatProxy: """Proxy object that provides chat.completions() interface for NPC""" def __init__(self, npc: "NPC"): self.npc = npc 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, Iterator[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 iterator for streaming Examples: # Basic chat response = npc.chat.completions( messages=[{"role": "user", "content": "Hello!"}] ) print(response.choices[0].message.content) # With streaming for chunk in 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 = 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 self.npc.client.chat.completions( npc_params=npc_params, messages=messages, model=model, temperature=temperature, max_tokens=max_tokens, stream=stream, **kwargs, ) def chat_with_history( self, message: str, model: Optional[str] = None, temperature: float = 0.8, max_tokens: int = 2000, stream: bool = False, **kwargs, ) -> Union[ChatCompletion, Iterator[dict]]: """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 response or 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 self.completions( messages=messages_with_history, model=model, temperature=temperature, max_tokens=max_tokens, stream=stream, **kwargs, ) class NPC(AsyncNPC): """Synchronous version of NPC""" 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 def completions( self, messages: List[Union[ChatMessage, dict]], **kwargs ) -> Union[ChatCompletion, Iterator[dict]]: """ Convenience alias for npc.chat.completions() Args: messages: Chat message history **kwargs: Additional parameters passed to chat.completions Returns: Chat completion response or iterator for streaming Example: # Direct usage response = npc.completions([ {"role": "user", "content": "Hello!"} ]) """ return self.chat.completions(messages, **kwargs) def chat_with_history( self, message: str, **kwargs ) -> Union[ChatCompletion, Iterator[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 iterator for streaming Example: # Continue conversation with context response = npc.chat_with_history("Tell me more about that") print(response.choices[0].message.content) """ return 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) -> "NPC": """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) -> "NPC": """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}") # ============================================================================ # STREAMING SUPPORT # ============================================================================ class StreamingResponse: """Handle streaming Server-Sent Events responses""" def __init__(self, response: requests.Response): self.response = response self._buffer = "" def __iter__(self) -> Iterator[dict]: """Iterate over streaming chunks""" try: for line in self.response.iter_lines(decode_unicode=True): 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 finally: self.response.close() # ============================================================================ # MANAGER CLASSES # ============================================================================ class MoodManager: """Handles custom mood management operations""" def __init__(self, client: "GumYumClient"): self.client = client 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 GumYumValidationError: If mood_id format is invalid or mood_data structure is wrong """ payload = {"mood_id": mood_id, "mood_data": mood_data} response = self.client.post("moods/upload", json=payload) return response 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 = self.client.get("moods") return response.get("moods", []) 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 GumYumValidationError: If trying to delete default GumYum moods """ response = self.client.delete(f"moods/{mood_id}") return response 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: 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 self.upload(mood_id, mood_data) class AuthManager: """Handles authentication operations""" def __init__(self, client: "GumYumClient"): self.client = client def register(self, username: str, password: str, email: str) -> AuthToken: """Register a new user account""" data = {"username": username, "password": password, "email": email} response = self.client.post("auth/register", json=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 def login(self, username: str, password: str) -> AuthToken: """Login with existing account""" data = {"username": username, "password": password} response = self.client.post("auth/login", json=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 def refresh_token(self, refresh_token: str) -> AuthToken: """Refresh access token using refresh token""" data = {"refresh_token": refresh_token} # Use a special flag to avoid auto-auth during refresh self.client._is_refreshing = True try: response = self.client.post("auth/refresh", json=data) finally: self.client._is_refreshing = False # Extract new tokens and update client new_access_token = response["access_token"] new_refresh_token = response.get( "refresh_token", refresh_token ) # Server may rotate refresh token expires_in = response.get("expires_in", 3600) # Debug: Check if server provided a new refresh token # print(f"[DEBUG] Refresh response has refresh_token: {'refresh_token' in response}") 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 if provided token_type=response.get("token_type", "Bearer"), expires_in=expires_in, 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) def get_profile(self) -> UserProfile: """Get current user profile""" response = self.client.get("auth/profile") return UserProfile(**response) def is_authenticated(self) -> bool: """Check if client has an authentication token""" return self.client._token is not None def create_api_key(self, name: str, scopes: List[str]) -> dict: """ Create a new API key Args: name: Human-readable name for the API key scopes: List of permission scopes (e.g., ['npc.spawn', 'chat.completions']) Returns: API key details including public_key and secret_key Raises: GumYumAuthError: If not authenticated with JWT """ data = {"name": name, "scopes": scopes} response = self.client.post("auth/api-keys", json=data) return response def list_api_keys(self) -> List[dict]: """ List all API keys for the current user Returns: List of API key details (without secret keys) Raises: GumYumAuthError: If not authenticated """ response = self.client.get("auth/api-keys") return response.get("api_keys", []) def get_api_key(self, api_key_id: str) -> dict: """ Get details of a specific API key Args: api_key_id: The API key ID Returns: API key details (without secret key) Raises: GumYumAuthError: If not authenticated GumYumNotFoundError: If API key not found """ response = self.client.get(f"auth/api-keys/{api_key_id}") return response def update_api_key( self, api_key_id: str, name: Optional[str] = None, scopes: Optional[List[str]] = None, ) -> dict: """ Update API key metadata Args: api_key_id: The API key ID name: New name for the API key scopes: New scopes for the API key Returns: Updated API key details Raises: GumYumAuthError: If not authenticated GumYumNotFoundError: If API key not found """ data = {} if name is not None: data["name"] = name if scopes is not None: data["scopes"] = scopes response = self.client.put(f"auth/api-keys/{api_key_id}", json=data) return response def revoke_api_key(self, api_key_id: str) -> dict: """ Revoke (delete) an API key Args: api_key_id: The API key ID to revoke Returns: Deletion confirmation Raises: GumYumAuthError: If not authenticated GumYumNotFoundError: If API key not found """ response = self.client.delete(f"auth/api-keys/{api_key_id}") return response def regenerate_api_key(self, api_key_id: str) -> dict: """ Regenerate an API key (creates new public/secret pair) Args: api_key_id: The API key ID to regenerate Returns: New API key details including new public_key and secret_key Raises: GumYumAuthError: If not authenticated GumYumNotFoundError: If API key not found """ response = self.client.post(f"auth/api-keys/{api_key_id}/regenerate") return response def delete_account(self) -> dict: """ Delete the current user account (GDPR compliance) Returns: Deletion confirmation Raises: GumYumAuthError: If not authenticated """ response = self.client.delete("auth/account") return response class UniverseManager: """Handles universe operations""" def __init__(self, client: "GumYumClient"): self.client = client def list_public(self) -> List[PublicUniverse]: """List all available public universes""" response = self.client.get("public/universes") return [PublicUniverse(**universe) for universe in response["public_universes"]] 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""" data = {"public_universe_id": public_universe_id} if custom_name: data["custom_name"] = custom_name if mood_id: data["mood_id"] = mood_id response = self.client.post("public/universes/copy", json=data) return response["universe_id"] def list(self) -> List[dict]: """List user's private universes""" response = self.client.get("universes") return response["universes"] # Alias for backwards compatibility def list_user(self) -> List[dict]: """List user's private universes (alias for list())""" return self.list() def create(self, universe_data: dict, mood_id: Optional[str] = None) -> str: """Create a new universe in user's vault""" if mood_id: universe_data["mood_id"] = mood_id response = self.client.post("universes", json=universe_data) return response["universe_id"] def get(self, universe_id: str) -> UniverseData: """Get a specific universe by ID""" response = 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, ) def update(self, universe_id: str, universe_data: dict) -> UniverseData: """Update an existing universe""" response = self.client.put(f"universes/{universe_id}", json=universe_data) # Return the updated universe return self.get(universe_id) def fork(self, universe_id: str, name: Optional[str] = None) -> str: """Fork (copy) an existing universe""" data = {} if name: data["name"] = name response = self.client.post(f"universes/{universe_id}/fork", json=data) return response["universe_id"] def delete(self, universe_id: str) -> bool: """Delete a universe from user's vault""" self.client.delete(f"universes/{universe_id}") return True class NPCManager: """Handles NPC operations""" def __init__(self, client: "GumYumClient"): self.client = client def spawn( self, universe_id: str, seed: int, npc_id: Optional[int] = None, ) -> NPC: """Spawn a deterministic NPC""" if npc_id is not None: # Spawn with specific 64-bit NPC ID params = {"universe_id": universe_id, "seed": seed, "npc_id": npc_id} response = 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 = 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 def spawn_random(self, universe_id: str, seed: int) -> NPC: """Spawn random NPC using deterministic seed""" return self.spawn(universe_id, seed, None) 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 """ data = { "universe_id": universe_id, "world_seed": world_seed, "filters": filters, "start_npc_id": start_npc_id, "max_attempts": max_attempts, } try: response = self.client.post("npc/spawn/filtered", json=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 if not response or "npc_id" not in response: # If we get an error response or no NPC data if response and response.get("error"): if "No NPC found" in response["error"]: return None raise GumYumNotFoundError(response["error"]) 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 def get_profile(self, npc_id: Union[int, str]) -> NPCProfile: """Get full NPC profile data""" response = 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) 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""" data = { "universe_id": universe_id, "world_seed": seed, # API expects world_seed for save endpoint "npc_id": npc_id, } if custom_name: data["custom_name"] = custom_name response = self.client.post("npc/save", json=data) return NPCProfile(**response) def list_saved(self) -> List[NPC]: """ List user's saved NPCs Returns: List of saved NPCs with client attached, ready for chat """ response = 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 def update(self, npc_id: Union[int, str], updates: dict) -> NPCProfile: """Update a saved NPC's data""" response = self.client.put(f"npc/saved/{npc_id}", json=updates) return NPCProfile(**response) def delete(self, npc_id: Union[int, str]) -> bool: """Delete a saved NPC""" self.client.delete(f"npc/saved/{npc_id}") return True def get_stats(self) -> NPCStats: """Get user's NPC statistics""" response = self.client.get("npc/stats") return NPCStats(**response) class ChatManager: """Handles AI chat completions with NPCs""" def __init__(self, client: "GumYumClient"): self.client = client def completions( self, npc_id: Optional[str] = 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, StreamingResponse]: """ 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 """ # 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: data["npc_id"] = npc_id 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): formatted_messages.append(msg.to_dict()) else: formatted_messages.append(msg) data["messages"] = formatted_messages else: data["messages"] = [] if stream: return self._stream_completions(data) else: response = self.client.post("chat/completions", json=data) # Convert response to ChatCompletion object choices = [] for choice_data in response.get("choices", []): message_data = choice_data.get("message", {}) message = ChatMessage( role=message_data.get("role", "assistant"), content=message_data.get("content", ""), ) choice = ChatChoice( index=choice_data.get("index", 0), message=message, finish_reason=choice_data.get("finish_reason"), ) choices.append(choice) return ChatCompletion( id=response.get("id", ""), created=response.get("created", int(time.time())), model=response.get("model", model), choices=choices, usage=response.get("usage"), npc_context=response.get("npc_context"), mood_transition=response.get("mood_transition"), ) def _stream_completions(self, data: dict) -> StreamingResponse: """Handle streaming chat completions""" url = urljoin(f"{self.client.api_url}/", "chat/completions") headers = self.client.get_headers() response = self.client._session.post( url, json=data, headers=headers, stream=True, timeout=self.client.timeout ) if response.status_code >= 400: self.client._handle_error_response(response) # 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: json_response = response.json() # Create a fake streaming response from the JSON class FakeStreamingResponse: def __init__(self, json_data): self.json_data = json_data self.response = response def __iter__(self): # Emit the content as chunks if "choices" in self.json_data and self.json_data["choices"]: content = self.json_data["choices"][0]["message"]["content"] # Split content into words to simulate streaming words = content.split() 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 time.sleep(0.02) return FakeStreamingResponse(json_response) except: # If we can't parse it, return the original streaming response pass return StreamingResponse(response) # ============================================================================ # MAIN CLIENT # ============================================================================ class GumYumClient: """ Synchronous client for GumYum NPC API Example: client = GumYumClient("https://npc.gumyum.com") user = client.auth.register("username", "password", "email") npc = client.npcs.spawn("universe_id", seed=12345) response = 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, username: Optional[str] = None, password: Optional[str] = None, public_key: Optional[str] = None, secret_key: Optional[str] = None, **kwargs, ): """ Initialize GumYum client Args: base_url: API base URL api_key: Optional API key for authentication (deprecated, use public_key/secret_key) timeout: Request timeout in seconds max_retries: Maximum number of retries for failed requests username: Username for login password: Password for login public_key: Public API key secret_key: Secret API key **kwargs: Additional requests session options """ if requests is None: raise ImportError( "requests is required. Install with: pip install requests" ) self.base_url = base_url.rstrip("/") self.api_url = f"{self.base_url}/v1" self.timeout = timeout self.max_retries = max_retries # HTTP session configuration self._session = requests.Session() self._session.timeout = timeout # Apply additional session options for key, value in kwargs.items(): setattr(self._session, key, value) 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] = public_key self._secret_key: Optional[str] = secret_key self._jwt_obtained: bool = False # Track if we've exchanged API key for JWT self._is_refreshing: bool = False # Track if we're in a refresh operation # Initialize managers self.auth = AuthManager(self) self.moods = MoodManager(self) self.universes = UniverseManager(self) self.npcs = NPCManager(self) self.chat = ChatManager(self) # Auto-authenticate if credentials provided if username and password: self.login(username, password) elif public_key and secret_key: self.set_api_key(public_key, secret_key) elif api_key: # Legacy support parts = api_key.split(":") if len(parts) == 2: self.set_api_key(parts[0], parts[1]) def __enter__(self): """Context manager entry""" return self def __exit__(self, exc_type, exc_val, exc_tb): """Context manager exit""" self.close() def close(self): """Close the HTTP session""" self._session.close() 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: 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) -> None: """ Set API key credentials for authentication Args: public_key: The public API key secret_key: The secret API key """ 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 # Exchange API key for JWT immediately self._exchange_api_key_for_jwt() def _exchange_api_key_for_jwt(self) -> None: """Exchange API key for JWT token""" if not self._public_key or not self._secret_key: raise GumYumAuthError("API keys not set") url = f"{self.api_url}/auth/exchange" headers = { "X-API-Key-Public": self._public_key, "X-API-Key-Secret": self._secret_key, "Content-Type": "application/json", "User-Agent": "GumYum-Python-SDK-Sync/0.2.0", } try: response = self._session.post(url, headers=headers, timeout=self.timeout) response.raise_for_status() data = response.json() self._token = data["jwt_token"] self._refresh_token = data["refresh_token"] self._token_expires_at = time.time() + data.get("expires_in", 3600) self._jwt_obtained = True logger.debug("Successfully exchanged API key for JWT") except requests.HTTPError as e: if e.response.status_code == 401: raise GumYumAuthError("Invalid API key credentials") else: raise GumYumAPIError(f"API key exchange failed: {e}") except Exception as e: raise GumYumAPIError(f"API key exchange failed: {e}") def get_headers(self) -> Dict[str, str]: """Get request headers with authentication""" headers = { "Content-Type": "application/json", "User-Agent": "GumYum-Python-SDK-Sync/0.1.0", } if self._token and self._token.strip(): headers["Authorization"] = f"Bearer {self._token}" return headers def _ensure_authenticated(self): """Ensure client is authenticated, auto-login if needed""" # Check if token needs refresh if self._token and self._token_expires_at: if ( time.time() >= self._token_expires_at - 60 ): # Refresh 1 minute before expiry if self._refresh_token and not self._is_refreshing: try: self.auth.refresh_token(self._refresh_token) except Exception: # If refresh fails, try to re-authenticate with API keys if self._public_key and self._secret_key: try: self._exchange_api_key_for_jwt() except Exception: pass # Only auto-authenticate if we've never had a token (prevents overriding explicit auth) if not self._token or not self._token.strip(): # Try API key auth first if self._public_key and self._secret_key and not self._jwt_obtained: try: self._exchange_api_key_for_jwt() return except Exception: pass 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 timestamp = int(time.time()) username = f"auto_user_{timestamp}_{self._auto_auth_attempts}" 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: self.auth.login(username, "auto_password123") except Exception: # Both attempts failed - client is now permanently failed pass def _request( self, method: str, endpoint: str, params: Optional[Dict[str, Any]] = None, json: 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: JSON request body **kwargs: Additional request options Returns: Response JSON data Raises: GumYumError: On API errors """ # Check if token needs refresh before making request if self._token and self._token_expires_at: if ( time.time() >= self._token_expires_at - 60 ): # Refresh 1 minute before expiry # Skip refresh for auth endpoints that don't need it (to prevent recursion) auth_endpoints_to_skip = [ "auth/refresh", "auth/login", "auth/register", "auth/exchange", "health", "public/universes", ] if self._refresh_token and not any( endpoint.startswith(ep) for ep in auth_endpoints_to_skip ): try: self.auth.refresh_token(self._refresh_token) except Exception: # If refresh fails, continue anyway and let 401 handling take over pass # Auto-authenticate for protected endpoints protected_endpoints = ["npc", "chat", "universes", "npcs"] if any(endpoint.startswith(ep) for ep in protected_endpoints): self._ensure_authenticated() if not self._token or not self._token.strip(): raise GumYumAuthError("Authentication required for protected endpoints") url = urljoin(f"{self.api_url}/", endpoint.lstrip("/")) headers = self.get_headers() # Merge headers, but sanitize Authorization to prevent empty arrays if "headers" in kwargs: extra_headers = kwargs.pop("headers") 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 retries = 0 while retries <= self.max_retries: 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()) ): 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 = self._session.request( method=method, url=url, params=params, json=json, headers=clean_headers, timeout=self.timeout, **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 self._handle_error_response(response) except GumYumAuthError as e: # Handle 401 authentication errors if e.status_code == 401 and retries == 0: # Try to refresh token if we have a refresh token auth_endpoints_to_skip = [ "auth/refresh", "auth/login", "auth/register", "auth/exchange", "health", "public/universes", ] if self._refresh_token and not any( endpoint.startswith(ep) for ep in auth_endpoints_to_skip ): try: # Call refresh endpoint refresh_response = self.auth.refresh_token( self._refresh_token ) # Token is already updated by refresh_token method # Retry the original request retries += 1 continue except GumYumAuthError as refresh_error: # If refresh returns 403, the refresh token is invalid if refresh_error.status_code == 403: # Clear tokens - refresh token is invalid self._token = None self._refresh_token = None raise GumYumAuthError( "Refresh token is invalid or expired - please login again", 403, ) else: # Other refresh errors self._token = None self._refresh_token = None raise GumYumAuthError( "Authentication failed - token refresh failed", e.status_code, ) except Exception: # Refresh failed, clear tokens self._token = None self._refresh_token = None raise GumYumAuthError( "Authentication failed - token refresh failed", e.status_code, ) # No refresh token or refresh failed self._token = None self._refresh_token = None 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 time.sleep(min(2**retries + 0.1 * retries, 10)) continue else: raise except requests.exceptions.Timeout as e: if retries >= self.max_retries: raise GumYumTimeoutError( f"Request timeout after {self.timeout}s" ) from e except requests.exceptions.RequestException as e: if retries >= self.max_retries: raise GumYumNetworkError(f"Network error: {e}") from e retries += 1 if retries <= self.max_retries: # Exponential backoff time.sleep(2**retries) raise GumYumError("Max retries exceeded") def _handle_error_response(self, response: requests.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) def get(self, endpoint: str, **kwargs) -> Dict[str, Any]: """Make GET request""" return self._request("GET", endpoint, **kwargs) def post(self, endpoint: str, **kwargs) -> Dict[str, Any]: """Make POST request""" return self._request("POST", endpoint, **kwargs) def put(self, endpoint: str, **kwargs) -> Dict[str, Any]: """Make PUT request""" return self._request("PUT", endpoint, **kwargs) def delete(self, endpoint: str, **kwargs) -> Dict[str, Any]: """Make DELETE request""" return self._request("DELETE", endpoint, **kwargs) def health_check(self) -> Dict[str, Any]: """Check API health status""" return self.get("health") def login(self, username: str, password: str) -> AuthToken: """ Login with username and password Args: username: User's username password: User's password Returns: AuthToken with access and refresh tokens """ return self.auth.login(username, password) def register(self, username: str, password: str, email: str) -> AuthToken: """ Register a new account Args: username: Desired username password: User's password email: User's email Returns: AuthToken with access and refresh tokens """ return self.auth.register(username, password, email) # ============================================================================ # EXPORTS # ============================================================================ __version__ = "0.2.0" __author__ = "GumYum NPC API Team" __email__ = "timehexon@gumyum.com" logger = logging.getLogger(__name__) __all__ = [ # Main client "GumYumClient", # Exceptions "GumYumError", "GumYumAPIError", "GumYumAuthError", "GumYumNotFoundError", "GumYumValidationError", "GumYumServerError", "GumYumNetworkError", "GumYumTimeoutError", "GumYumRateLimitError", # Models "NPCProfile", "ChatCompletion", "ChatMessage", "AuthToken", "NPC", "PublicUniverse", "ChatChoice", "ChatRole", "StreamingResponse", ] # ============================================================================ # EXAMPLE USAGE # ============================================================================ if __name__ == "__main__": def demo(): """Example usage of the synchronous GumYum NPC SDK""" with GumYumClient("http://localhost:6543") as client: try: # Register or login print("Authenticating...") try: user = client.auth.login("demo_user", "demo_password123") except GumYumAuthError: user = 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 = client.universes.list_public() if universes: print(f"Found {len(universes)} public universes") print(f"Using: {universes[0].name}") # Copy a universe universe_id = client.universes.copy_public( universes[0].id, "My Demo World" ) print(f"Copied universe: {universe_id}") # Spawn an NPC print("\nSpawning NPC...") npc = 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 = 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="") stream = client.chat.completions( npc_id=npc.npc_id, messages=[ {"role": "user", "content": "Tell me about your day"} ], stream=True, ) for chunk in stream: 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 demo()