2167 lines
74 KiB
Python
2167 lines
74 KiB
Python
#!/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())
|