Source code for tools.game_asset_upload

"""GameGirl Color -- Asset upload tool.

Allows the LLM to register game assets (images) uploaded by
players during a game session.
# 🎨💀 ASSET CORRUPTION PROTOCOL
"""

from __future__ import annotations

import json
import logging
from typing import TYPE_CHECKING

if TYPE_CHECKING:
    from tool_context import ToolContext

logger = logging.getLogger(__name__)

TOOL_NAME = "upload_game_asset"
TOOL_DESCRIPTION = (
    "Upload/register an image asset for the active GameGirl Color game session. "
    "Assets are categorized (title_screen, ui, enemy, character, item, "
    "background, special) and labeled with a name for reference during "
    "game narration. Requires an active game session."
)
TOOL_PARAMETERS = {
    "type": "object",
    "properties": {
        "asset_name": {
            "type": "string",
            "description": "Name/label for the asset (e.g. 'Dark Slime', 'Forest Background').",
        },
        "category": {
            "type": "string",
            "description": (
                "Asset category. One of: title_screen, ui, enemy, "
                "character, item, background, special."
            ),
            "enum": [
                "title_screen", "ui", "enemy", "character",
                "item", "background", "special",
            ],
        },
        "image_url": {
            "type": "string",
            "description": "URL of the image to register as an asset.",
        },
    },
    "required": ["asset_name", "category", "image_url"],
}


[docs] async def run( asset_name: str, category: str, image_url: str, ctx: ToolContext | None = None, ) -> str: """Register a game asset. Args: asset_name: Name/label for the asset. category: Asset category. image_url: URL of the image. ctx: Tool execution context. Returns: str: JSON result. """ if ctx is None: return json.dumps({"error": "No tool context available."}) from game_session import get_session channel_id = str(ctx.channel_id) session = get_session(channel_id) if session is None or not session.active: return json.dumps({ "error": "No active game session in this channel. " "Boot a game first with boot_game.", }) redis = getattr(ctx, "redis", None) user_id = str(getattr(ctx, "user_id", "")) from game_assets import upload_asset result = await upload_asset( game_id=session.game_id, name=asset_name, category=category, url=image_url, user_id=user_id, turn=session.turn_number, redis=redis, ) return json.dumps(result)