From ff1cc6a1b48e93830305daf8004ca27f780357b6 Mon Sep 17 00:00:00 2001 From: Jon B Date: Tue, 3 Mar 2026 22:17:27 -0600 Subject: [PATCH 1/2] chore(meshy): bump version to 0.1.2 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Patch release fixing the animation task responsePath bug (id → result). Co-Authored-By: Claude Opus 4.6 --- packages/meshy-content-generator/package.json | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/meshy-content-generator/package.json b/packages/meshy-content-generator/package.json index bb00563d..9e853b39 100644 --- a/packages/meshy-content-generator/package.json +++ b/packages/meshy-content-generator/package.json @@ -1,6 +1,6 @@ { "name": "@jbcom/agentic-meshy", - "version": "0.1.1", + "version": "0.1.2", "description": "Declarative Meshy content generation pipelines with a CLI and API.", "license": "MIT", "type": "module", From 45d86c9d0b9f1c0037db8873051617e32f2c0fba Mon Sep 17 00:00:00 2001 From: Jon B Date: Sat, 7 Mar 2026 00:30:04 -0600 Subject: [PATCH 2/2] feat(game-asset-mcp): add open-source MCP server for 3D game asset libraries MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit New package `game-asset-mcp` — a fully open-source MCP server for local 3D game asset libraries. Integrates directly into Claude Code, Cursor, and Windsurf via the Model Context Protocol. ## Core Features - SQLite catalog with FTS5 full-text search across all GLBs - Idempotent ingest: O(1) skip check via in-memory hash, stale detection - `browse_taxonomy` tool for macro→meso→micro→pack hierarchy navigation - Pure-Python GLB stats reader (vertices, faces, materials, animations, textures) - PolyHaven CC0 integration: search, preview, download with auto-taxonomy placement - Headless Blender preview generation (auto-detected binary, no bpy dep) ## Configuration - pydantic-settings: layered config (TOML → env vars → CLI flags) - TOML config at ~/.config/game-asset-mcp/config.toml - Fully configurable taxonomy: style_map, source_hints, skip_dirs - Interactive setup wizard: `game-asset-init` (or `--yes` for non-interactive) - pydantic-cli for IngestOptions: auto --help, JSON config file ingestion - Python 3.10–3.14 support (tomli fallback for 3.10) ## Testing - 140 unit/integration tests — all passing - conftest.py with GLB binary builder, tmp DB, taxonomy fixtures - Full coverage: catalog, ingest, glb_reader, polyhaven, server tools, wizard - E2E test suite that runs against real ASSETS_ROOT when available ## Monorepo Integration - Added to [tool.uv.workspace] members - project.json with Nx targets (test, test:e2e, lint, typecheck, ingest) - Root README updated with package listing and directory tree Co-Authored-By: Claude Sonnet 4.6 --- README.md | 4 +- packages/game-asset-mcp/CHANGELOG.md | 16 + packages/game-asset-mcp/README.md | 188 +++++++ packages/game-asset-mcp/project.json | 26 + packages/game-asset-mcp/pyproject.toml | 104 ++++ .../src/game_asset_mcp/__init__.py | 5 + .../src/game_asset_mcp/catalog.py | 204 ++++++++ .../src/game_asset_mcp/config.py | 157 ++++++ .../src/game_asset_mcp/glb_reader.py | 102 ++++ .../src/game_asset_mcp/ingest.py | 265 ++++++++++ .../src/game_asset_mcp/polyhaven.py | 223 ++++++++ .../src/game_asset_mcp/render_preview.py | 129 +++++ .../src/game_asset_mcp/search.py | 222 ++++++++ .../src/game_asset_mcp/server.py | 486 ++++++++++++++++++ .../src/game_asset_mcp/wizard.py | 180 +++++++ packages/game-asset-mcp/tests/__init__.py | 0 packages/game-asset-mcp/tests/conftest.py | 161 ++++++ packages/game-asset-mcp/tests/e2e/__init__.py | 0 packages/game-asset-mcp/tests/e2e/conftest.py | 38 ++ .../tests/e2e/test_ingest_e2e.py | 51 ++ packages/game-asset-mcp/tests/test_catalog.py | 276 ++++++++++ packages/game-asset-mcp/tests/test_config.py | 53 ++ .../game-asset-mcp/tests/test_glb_reader.py | 172 +++++++ packages/game-asset-mcp/tests/test_ingest.py | 231 +++++++++ .../game-asset-mcp/tests/test_polyhaven.py | 275 ++++++++++ .../game-asset-mcp/tests/test_server_tools.py | 356 +++++++++++++ packages/game-asset-mcp/tests/test_wizard.py | 157 ++++++ pyproject.toml | 4 +- 28 files changed, 4082 insertions(+), 3 deletions(-) create mode 100644 packages/game-asset-mcp/CHANGELOG.md create mode 100644 packages/game-asset-mcp/README.md create mode 100644 packages/game-asset-mcp/project.json create mode 100644 packages/game-asset-mcp/pyproject.toml create mode 100755 packages/game-asset-mcp/src/game_asset_mcp/__init__.py create mode 100755 packages/game-asset-mcp/src/game_asset_mcp/catalog.py create mode 100644 packages/game-asset-mcp/src/game_asset_mcp/config.py create mode 100755 packages/game-asset-mcp/src/game_asset_mcp/glb_reader.py create mode 100755 packages/game-asset-mcp/src/game_asset_mcp/ingest.py create mode 100644 packages/game-asset-mcp/src/game_asset_mcp/polyhaven.py create mode 100755 packages/game-asset-mcp/src/game_asset_mcp/render_preview.py create mode 100755 packages/game-asset-mcp/src/game_asset_mcp/search.py create mode 100755 packages/game-asset-mcp/src/game_asset_mcp/server.py create mode 100644 packages/game-asset-mcp/src/game_asset_mcp/wizard.py create mode 100644 packages/game-asset-mcp/tests/__init__.py create mode 100644 packages/game-asset-mcp/tests/conftest.py create mode 100644 packages/game-asset-mcp/tests/e2e/__init__.py create mode 100644 packages/game-asset-mcp/tests/e2e/conftest.py create mode 100644 packages/game-asset-mcp/tests/e2e/test_ingest_e2e.py create mode 100644 packages/game-asset-mcp/tests/test_catalog.py create mode 100644 packages/game-asset-mcp/tests/test_config.py create mode 100644 packages/game-asset-mcp/tests/test_glb_reader.py create mode 100644 packages/game-asset-mcp/tests/test_ingest.py create mode 100644 packages/game-asset-mcp/tests/test_polyhaven.py create mode 100644 packages/game-asset-mcp/tests/test_server_tools.py create mode 100644 packages/game-asset-mcp/tests/test_wizard.py diff --git a/README.md b/README.md index 8de1265e..4fd64c0c 100644 --- a/README.md +++ b/README.md @@ -28,6 +28,7 @@ | [`@jbcom/agentic-meshy`](packages/meshy-content-generator) | TypeScript | Declarative Meshy 3D asset generation pipelines | [Docs](https://agentic.coach/packages/meshy-content-generator/) | | [`@jbcom/agentic-providers`](packages/providers) | TypeScript | LLM provider implementations (Ollama, Jules, Cursor) | [Docs](https://agentic.coach/packages/control/) | | [`game-generator`](packages/game-generator) | Rust | Visual-first vintage game generator with AI assistance | [Docs](https://agentic.coach/packages/game-generator/) | +| [`game-asset-mcp`](packages/game-asset-mcp) | Python | MCP server for local 3D game asset libraries — search, browse, PolyHaven integration | [README](packages/game-asset-mcp/README.md) | ### Testing Plugins @@ -123,7 +124,8 @@ agentic/ │ ├── providers/ # @jbcom/agentic-providers (TypeScript) │ ├── game-generator/ # game-generator (Rust) │ ├── vitest-agentic-control/ # @jbcom/vitest-agentic (TypeScript) -│ └── pytest-agentic-crew/ # pytest-agentic-crew (Python) +│ ├── pytest-agentic-crew/ # pytest-agentic-crew (Python) +│ └── game-asset-mcp/ # game-asset-mcp (Python) ├── actions/ # GitHub Marketplace actions ├── docs/ # Documentation site (Astro + Starlight) └── scripts/ # Ecosystem automation diff --git a/packages/game-asset-mcp/CHANGELOG.md b/packages/game-asset-mcp/CHANGELOG.md new file mode 100644 index 00000000..bf7ccb14 --- /dev/null +++ b/packages/game-asset-mcp/CHANGELOG.md @@ -0,0 +1,16 @@ +# Changelog + +## [0.1.0] - 2026-03-07 + +### Added +- Initial release +- SQLite catalog with FTS5 full-text search +- Idempotent ingest (O(1) skip check, stale removal) +- `browse_taxonomy` tool for macro/meso/micro/pack hierarchy navigation +- `search_assets` hybrid keyword + FTS search +- `copy_asset`, `get_preview`, `generate_preview` tools +- PolyHaven CC0 integration (`search_polyhaven`, `download_polyhaven_asset`) +- Pydantic-settings configuration with TOML file support +- Interactive setup wizard (`game-asset-init`) +- Optional `bpy` extra for in-process Blender rendering +- Pure-Python GLB stats reader (no Blender required for ingest) diff --git a/packages/game-asset-mcp/README.md b/packages/game-asset-mcp/README.md new file mode 100644 index 00000000..41f112cc --- /dev/null +++ b/packages/game-asset-mcp/README.md @@ -0,0 +1,188 @@ +# game-asset-mcp + +**MCP server for local 3D game asset libraries** — search, browse, catalog, and download GLB/GLTF assets from your file system and PolyHaven. + +Built with [FastMCP](https://github.com/jlowin/fastmcp) · Works with Claude Code, Cursor, Windsurf, and any MCP-compatible client · CC0 PolyHaven integration included + +--- + +## Quick Start + +### Install + +```bash +pip install "game-asset-mcp[server,polyhaven]" +``` + +### Configure + +Set your asset library root: + +```bash +export ASSETS_ROOT=/path/to/your/3d-assets +export CATALOG_DB=~/.local/share/game-asset-mcp/catalog.db # optional, this is the default +``` + +### Ingest your library + +```bash +game-asset-ingest +``` + +This scans all `.glb` files in `ASSETS_ROOT`, extracts mesh stats (vertices, faces, materials, animations), and builds a searchable SQLite catalog. **Idempotent** — only re-processes files that have changed size. + +--- + +## Add to Claude Code + +```bash +claude mcp add game-asset-library \ + -e ASSETS_ROOT=/path/to/your/3d-assets \ + -- game-asset-mcp +``` + +Or add to `~/.claude.json` manually: + +```json +{ + "mcpServers": { + "game-asset-library": { + "type": "stdio", + "command": "game-asset-mcp", + "env": { + "ASSETS_ROOT": "/path/to/your/3d-assets" + } + } + } +} +``` + +## Add to Cursor + +Add to `.cursor/mcp.json` in your project (or `~/.cursor/mcp.json` globally): + +```json +{ + "mcpServers": { + "game-asset-library": { + "command": "game-asset-mcp", + "env": { + "ASSETS_ROOT": "/path/to/your/3d-assets" + } + } + } +} +``` + +## Add to Windsurf + +Add to `~/.codeium/windsurf/mcp_config.json`: + +```json +{ + "mcpServers": { + "game-asset-library": { + "command": "game-asset-mcp", + "env": { + "ASSETS_ROOT": "/path/to/your/3d-assets" + } + } + } +} +``` + +--- + +## Available Tools + +| Tool | Description | +|------|-------------| +| `search_assets` | Hybrid keyword + FTS search across all GLBs | +| `browse_taxonomy` | Navigate your directory taxonomy (macro → meso → micro → pack) | +| `list_categories` | List all categories with GLB counts | +| `get_asset_info` | Full metadata for one asset (mesh stats, preview path, etc.) | +| `copy_asset` | Copy a GLB into your game project directory | +| `get_preview` | Return path to an existing PNG thumbnail | +| `generate_preview` | Render a 512×512 thumbnail via headless Blender | +| `run_ingest` | Re-scan library and update catalog (idempotent) | +| `get_catalog_stats` | Summary statistics (total assets, with textures, with armatures) | +| `search_polyhaven` | Search [PolyHaven](https://polyhaven.com) for free CC0 models/HDRIs/textures | +| `download_polyhaven_asset` | Download a PolyHaven asset and auto-add to your catalog | + +--- + +## Taxonomy Convention + +`game-asset-mcp` works best with assets organized as: + +``` +ASSETS_ROOT/ +├── 3DLowPoly/ +│ ├── Characters/ +│ │ ├── Animated/ +│ │ │ └── / ← GLBs here +│ │ └── Animals/ +│ ├── Props/ +│ │ └── Weapons/ +│ └── Environment/ +│ └── Nature/ +├── 3DPSX/ +│ └── ... +└── 2DPhotorealistic/ + ├── HDRIs/ + └── Textures/ +``` + +The `browse_taxonomy` tool navigates this as **style → category → sub-category → pack**. Flat libraries work too — `search_assets` does full-text search on filenames and directory names regardless of structure. + +--- + +## PolyHaven Integration + +Search and download free CC0 assets from [polyhaven.com](https://polyhaven.com): + +``` +search_polyhaven("oak tree", asset_type="models") +download_polyhaven_asset("oak_tree", asset_type="models", resolution="1k") +``` + +Downloads are automatically placed in the correct taxonomy directory and added to the catalog. + +| PolyHaven Type | Local Path | +|----------------|------------| +| `models` | `ASSETS_ROOT/3DLowPoly//polyhaven//` | +| `hdris` | `ASSETS_ROOT/2DPhotorealistic/HDRIs/polyhaven//` | +| `textures` | `ASSETS_ROOT/2DPhotorealistic/Textures/polyhaven//` | + +--- + +## Environment Variables + +| Variable | Default | Description | +|----------|---------|-------------| +| `ASSETS_ROOT` | `~/assets` | Root directory of your 3D asset library | +| `CATALOG_DB` | `~/.local/share/game-asset-mcp/catalog.db` | SQLite catalog path | +| `BLENDER` | `/opt/homebrew/bin/blender` | Blender binary (only needed for `generate_preview`) | + +--- + +## Development + +```bash +# From the monorepo root +uv sync +uv run --package game-asset-mcp game-asset-mcp +``` + +Or directly: + +```bash +cd packages/game-asset-mcp +uv run game-asset-mcp +``` + +--- + +## License + +MIT © Jon Bogaty — PolyHaven assets are [CC0](https://polyhaven.com/license) diff --git a/packages/game-asset-mcp/project.json b/packages/game-asset-mcp/project.json new file mode 100644 index 00000000..e583dca0 --- /dev/null +++ b/packages/game-asset-mcp/project.json @@ -0,0 +1,26 @@ +{ + "name": "game-asset-mcp", + "tags": ["lang:py", "scope:mcp", "scope:game-assets"], + "targets": { + "test": { + "command": "uv run pytest tests/ -v --tb=short --ignore=tests/e2e/", + "inputs": ["{projectRoot}/src/**/*.py", "{projectRoot}/tests/**/*.py", "{projectRoot}/pyproject.toml"] + }, + "test:e2e": { + "command": "uv run pytest tests/e2e/ -v --tb=short -m e2e", + "inputs": ["{projectRoot}/src/**/*.py", "{projectRoot}/tests/**/*.py"] + }, + "lint": { + "command": "uvx ruff check {projectRoot}", + "inputs": ["{projectRoot}/src/**/*.py"] + }, + "typecheck": { + "command": "uvx mypy {projectRoot}/src/", + "inputs": ["{projectRoot}/src/**/*.py"] + }, + "ingest": { + "command": "uv run game-asset-ingest", + "inputs": [] + } + } +} diff --git a/packages/game-asset-mcp/pyproject.toml b/packages/game-asset-mcp/pyproject.toml new file mode 100644 index 00000000..1b8cfba7 --- /dev/null +++ b/packages/game-asset-mcp/pyproject.toml @@ -0,0 +1,104 @@ +[build-system] +requires = ["hatchling"] +build-backend = "hatchling.build" + +[project] +name = "game-asset-mcp" +version = "0.1.0" +description = "MCP server and catalog library for local 3D game asset libraries — search, browse, and manage GLB/GLTF assets" +requires-python = ">=3.10" +license = { text = "MIT" } +readme = "README.md" +authors = [{ name = "Jon Bogaty", email = "jon@jonbogaty.com" }] +keywords = [ + "mcp", "model-context-protocol", "3d-assets", "game-development", + "glb", "gltf", "polyhaven", "asset-management", "claude", "cursor", +] +classifiers = [ + "Development Status :: 3 - Alpha", + "Intended Audience :: Developers", + "License :: OSI Approved :: MIT License", + "Programming Language :: Python :: 3.10", + "Programming Language :: Python :: 3.11", + "Programming Language :: Python :: 3.12", + "Programming Language :: Python :: 3.13", + "Programming Language :: Python :: 3.14", + "Topic :: Games/Entertainment", + "Topic :: Multimedia :: Graphics :: 3D Modeling", + "Topic :: Software Development :: Libraries :: Python Modules", +] +# Core dependencies +dependencies = [ + "fastmcp>=3.0", + "tomli>=2.0; python_version < \"3.11\"", # tomllib backport + "pygltflib>=1.16.0", + "pydantic>=2.0", + "pydantic-settings>=2.0", + "pydantic-cli>=4.0", +] + +[project.optional-dependencies] +# PolyHaven integration (download free CC0 assets) +polyhaven = [ + "httpx>=0.27", +] +# Semantic search (vector embeddings) +semantic = [ + "sqlite-vec>=0.1", + "sentence-transformers>=3.0", + "langchain-community>=0.3", + "langchain>=0.3", +] +# Everything +all = [ + "fastmcp>=3.0", + "httpx>=0.27", + "sqlite-vec>=0.1", + "sentence-transformers>=3.0", + "langchain-community>=0.3", + "langchain>=0.3", +] +tests = [ + "pytest>=8.0", + "pytest-mock>=3.14", +] + +[project.scripts] +game-asset-mcp = "game_asset_mcp.server:main" +game-asset-ingest = "game_asset_mcp.ingest:main" +game-asset-init = "game_asset_mcp.wizard:main" + +[project.urls] +Homepage = "https://github.com/jbcom/agentic" +Repository = "https://github.com/jbcom/agentic" +Documentation = "https://github.com/jbcom/agentic/tree/main/packages/game-asset-mcp#readme" +Issues = "https://github.com/jbcom/agentic/issues" + +[tool.hatch.build.targets.wheel] +packages = ["src/game_asset_mcp"] + +[tool.semantic_release] +version_toml = ["pyproject.toml:project.version"] +build_command = "uv build" +commit_parser = "angular" +commit_author = "github-actions[bot] " +tag_format = "game-asset-mcp-v{version}" +changelog_file = "CHANGELOG.md" +upload_to_vcs_release = true + +[tool.semantic_release.commit_parser_options] +allowed_tags = ["feat", "fix", "docs", "style", "refactor", "perf", "test", "chore", "ci", "build"] +minor_tags = ["feat"] +patch_tags = ["fix", "perf"] + +[tool.semantic_release.branches.main] +match = "main" +prerelease = false + +[tool.semantic_release.remote] +type = "github" +token = { env = "GH_TOKEN" } + +[tool.semantic_release.publish] +dist_glob_patterns = ["dist/*"] +upload_to_vcs_release = true diff --git a/packages/game-asset-mcp/src/game_asset_mcp/__init__.py b/packages/game-asset-mcp/src/game_asset_mcp/__init__.py new file mode 100755 index 00000000..ba215d05 --- /dev/null +++ b/packages/game-asset-mcp/src/game_asset_mcp/__init__.py @@ -0,0 +1,5 @@ +"""game-asset-mcp — MCP server and catalog library for local 3D game asset libraries.""" +from __future__ import annotations + + +__version__ = "0.1.0" diff --git a/packages/game-asset-mcp/src/game_asset_mcp/catalog.py b/packages/game-asset-mcp/src/game_asset_mcp/catalog.py new file mode 100755 index 00000000..4674808b --- /dev/null +++ b/packages/game-asset-mcp/src/game_asset_mcp/catalog.py @@ -0,0 +1,204 @@ +""" +catalog.py — SQLite catalog database for the asset library. + +Database location: ~/.local/share/assets-mcp/catalog.db (local, fast) +Overridable via CATALOG_DB env var. +""" +from __future__ import annotations + + +import sqlite3 +import os +from pathlib import Path +from typing import Optional + +_DEFAULT_DB = Path.home() / ".local" / "share" / "game-asset-mcp" / "catalog.db" +DB_PATH = Path(os.environ.get("CATALOG_DB", str(_DEFAULT_DB))) + + +def get_connection(db_path: Path = DB_PATH) -> sqlite3.Connection: + conn = sqlite3.connect(db_path) + conn.row_factory = sqlite3.Row + conn.execute("PRAGMA journal_mode=WAL") + return conn + + +def init_db(db_path: Path = DB_PATH) -> None: + """Create tables if they don't exist.""" + conn = get_connection(db_path) + conn.executescript(""" + CREATE TABLE IF NOT EXISTS assets ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + path TEXT UNIQUE NOT NULL, + name TEXT NOT NULL, + style TEXT, -- '3DLowPoly' | '3DPSX' + category TEXT, -- e.g. 'Characters/Animated' + pack TEXT, -- pack directory name + source TEXT, -- 'Kenney' | 'Quaternius' | 'KayKit' | 'Custom' + meshes INTEGER, + vertices INTEGER, + faces INTEGER, + materials INTEGER, + textures INTEGER, + has_embedded_textures INTEGER DEFAULT 0, + has_armature INTEGER DEFAULT 0, + animations INTEGER DEFAULT 0, + extensions TEXT, -- JSON array + file_size_kb INTEGER, + preview_path TEXT, -- absolute path to PNG thumbnail + tags TEXT, -- space-separated searchable tags + ingested_at REAL -- unix timestamp + ); + + CREATE TABLE IF NOT EXISTS ingest_log ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + run_at REAL, + total INTEGER, + added INTEGER, + updated INTEGER, + skipped INTEGER + ); + """) + + # FTS5 virtual table for full-text search + conn.execute(""" + CREATE VIRTUAL TABLE IF NOT EXISTS assets_fts USING fts5( + name, + category, + pack, + source, + tags, + style, + content=assets, + content_rowid=id + ) + """) + conn.commit() + conn.close() + + +def upsert_asset(conn: sqlite3.Connection, asset: dict) -> int: + """Insert or replace an asset record. Returns the row id.""" + import json, time + + conn.execute(""" + INSERT INTO assets ( + path, name, style, category, pack, source, + meshes, vertices, faces, materials, textures, + has_embedded_textures, has_armature, animations, extensions, + file_size_kb, preview_path, tags, ingested_at + ) VALUES ( + :path, :name, :style, :category, :pack, :source, + :meshes, :vertices, :faces, :materials, :textures, + :has_embedded_textures, :has_armature, :animations, :extensions, + :file_size_kb, :preview_path, :tags, :ingested_at + ) ON CONFLICT(path) DO UPDATE SET + meshes=excluded.meshes, + vertices=excluded.vertices, + faces=excluded.faces, + materials=excluded.materials, + textures=excluded.textures, + has_embedded_textures=excluded.has_embedded_textures, + has_armature=excluded.has_armature, + animations=excluded.animations, + extensions=excluded.extensions, + file_size_kb=excluded.file_size_kb, + preview_path=excluded.preview_path, + tags=excluded.tags, + ingested_at=excluded.ingested_at + """, { + **asset, + "extensions": json.dumps(asset.get("extensions", [])), + "ingested_at": asset.get("ingested_at", time.time()), + }) + row = conn.execute("SELECT id FROM assets WHERE path=?", (asset["path"],)).fetchone() + return row["id"] + + +def rebuild_fts(conn: sqlite3.Connection) -> None: + """Rebuild the FTS5 index from the assets table.""" + conn.execute("INSERT INTO assets_fts(assets_fts) VALUES('rebuild')") + + +def search( + conn: sqlite3.Connection, + query: str, + style: Optional[str] = None, + category: Optional[str] = None, + has_armature: Optional[bool] = None, + has_textures: Optional[bool] = None, + max_results: int = 20, +) -> list[dict]: + """Full-text search over the asset catalog.""" + # Build FTS query + fts_query = query.strip() + if not fts_query: + # No text query — fall back to plain SELECT with filters + sql = "SELECT a.* FROM assets a WHERE 1=1" + params: list = [] + if style: + sql += " AND a.style = ?" + params.append(style) + if category: + sql += " AND a.category LIKE ?" + params.append(f"%{category}%") + if has_armature is not None: + sql += " AND a.has_armature = ?" + params.append(1 if has_armature else 0) + if has_textures is not None: + sql += " AND a.has_embedded_textures = ?" + params.append(1 if has_textures else 0) + sql += f" LIMIT {max_results}" + rows = conn.execute(sql, params).fetchall() + else: + sql = """ + SELECT a.* + FROM assets_fts fts + JOIN assets a ON a.id = fts.rowid + WHERE assets_fts MATCH ? + """ + params = [fts_query] + if style: + sql += " AND a.style = ?" + params.append(style) + if category: + sql += " AND a.category LIKE ?" + params.append(f"%{category}%") + if has_armature is not None: + sql += " AND a.has_armature = ?" + params.append(1 if has_armature else 0) + if has_textures is not None: + sql += " AND a.has_embedded_textures = ?" + params.append(1 if has_textures else 0) + sql += " ORDER BY rank" + sql += f" LIMIT {max_results}" + rows = conn.execute(sql, params).fetchall() + + return [dict(r) for r in rows] + + +def get_asset(conn: sqlite3.Connection, path: str) -> Optional[dict]: + row = conn.execute("SELECT * FROM assets WHERE path = ?", (path,)).fetchone() + return dict(row) if row else None + + +def list_categories(conn: sqlite3.Connection) -> list[dict]: + rows = conn.execute(""" + SELECT style, category, COUNT(*) as count + FROM assets + GROUP BY style, category + ORDER BY style, category + """).fetchall() + return [dict(r) for r in rows] + + +def get_stats(conn: sqlite3.Connection) -> dict: + row = conn.execute(""" + SELECT + COUNT(*) as total, + SUM(CASE WHEN has_embedded_textures=1 THEN 1 ELSE 0 END) as with_textures, + SUM(CASE WHEN has_armature=1 THEN 1 ELSE 0 END) as with_armature, + SUM(CASE WHEN preview_path IS NOT NULL THEN 1 ELSE 0 END) as with_previews + FROM assets + """).fetchone() + return dict(row) diff --git a/packages/game-asset-mcp/src/game_asset_mcp/config.py b/packages/game-asset-mcp/src/game_asset_mcp/config.py new file mode 100644 index 00000000..8969f91f --- /dev/null +++ b/packages/game-asset-mcp/src/game_asset_mcp/config.py @@ -0,0 +1,157 @@ +""" +config.py — Configuration for game-asset-mcp. + +Settings are loaded in priority order (lowest → highest): + 1. Built-in defaults + 2. TOML config file (~/.config/game-asset-mcp/config.toml or .game-asset-mcp.toml) + 3. Environment variables (GAME_ASSET_*) + 4. CLI flags (when invoked via pydantic-settings CLI source) + +Example TOML config: + [library] + assets_root = "/Volumes/home/assets" + catalog_db = "~/.local/share/game-asset-mcp/catalog.db" + + [taxonomy.style_map] + "3DLowPoly" = "3DLowPoly" + "3DPSX" = "3DPSX" + + [taxonomy.source_hints] + kenney = "Kenney" + quaternius = "Quaternius" + kaykit = "KayKit" +""" +from __future__ import annotations + +import os +try: + import tomllib +except ImportError: # Python 3.10 + import tomli as tomllib # type: ignore[no-redef] +from functools import lru_cache +from pathlib import Path +from typing import Optional + +from pydantic import Field, field_validator +from pydantic_settings import BaseSettings, SettingsConfigDict + +_DEFAULT_CONFIG_DIR = Path.home() / ".config" / "game-asset-mcp" +_DEFAULT_CONFIG_FILE = _DEFAULT_CONFIG_DIR / "config.toml" +_DEFAULT_LOCAL_CONFIG = Path(".game-asset-mcp.toml") # project-local override + + +class TaxonomyConfig: + """ + Taxonomy configuration (not a pydantic model — parsed from TOML manually + so we can support arbitrary style_map and source_hints keys). + """ + + def __init__( + self, + style_map: dict[str, str] | None = None, + source_hints: dict[str, str] | None = None, + skip_dirs: list[str] | None = None, + ): + self.style_map: dict[str, str] = style_map or { + "3DLowPoly": "3DLowPoly", + "3DPSX": "3DPSX", + } + self.source_hints: dict[str, str] = source_hints or { + "kenney": "Kenney", + "quaternius": "Quaternius", + "kaykit": "KayKit", + "kits": "KayKit", + "custom": "Custom", + "zappypixel": "Zappypixel", + } + self.skip_dirs: set[str] = set(skip_dirs or [ + "_Archive", "__pycache__", ".git", "node_modules", "Textures", "textures", + ]) + + @classmethod + def from_dict(cls, data: dict) -> "TaxonomyConfig": + return cls( + style_map=data.get("style_map"), + source_hints=data.get("source_hints"), + skip_dirs=data.get("skip_dirs"), + ) + + +class Settings(BaseSettings): + """ + Global settings for game-asset-mcp. + + Environment variable prefix: GAME_ASSET_ + Examples: + GAME_ASSET_ASSETS_ROOT=/mnt/assets + GAME_ASSET_CATALOG_DB=/tmp/catalog.db + """ + + model_config = SettingsConfigDict( + env_prefix="GAME_ASSET_", + env_file=".env", + env_file_encoding="utf-8", + extra="ignore", + ) + + assets_root: Path = Field( + default_factory=lambda: Path( + os.environ.get("ASSETS_ROOT", str(Path.home() / "assets")) + ), + description="Root directory of your 3D asset library", + ) + catalog_db: Path = Field( + default_factory=lambda: Path( + os.environ.get("CATALOG_DB", str(Path.home() / ".local" / "share" / "game-asset-mcp" / "catalog.db")) + ), + description="Path to the SQLite catalog database", + ) + blender: Path = Field( + default_factory=lambda: Path(os.environ.get("BLENDER", "/opt/homebrew/bin/blender")), + description="Path to Blender binary (used for generate_preview)", + ) + config_file: Optional[Path] = Field( + default=None, + description="Path to TOML config file (auto-detected if not set)", + ) + + @field_validator("assets_root", "catalog_db", "blender", mode="before") + @classmethod + def expand_path(cls, v: object) -> Path: + if isinstance(v, str): + return Path(v).expanduser() + if isinstance(v, Path): + return v.expanduser() + return v # type: ignore[return-value] + + def get_taxonomy(self) -> TaxonomyConfig: + """Load taxonomy config from TOML file if present, else return defaults.""" + config_path = self._find_config_file() + if config_path is None: + return TaxonomyConfig() + try: + with open(config_path, "rb") as f: + data = tomllib.load(f) + return TaxonomyConfig.from_dict(data.get("taxonomy", {})) + except Exception: + return TaxonomyConfig() + + def _find_config_file(self) -> Optional[Path]: + if self.config_file and self.config_file.exists(): + return self.config_file + if _DEFAULT_LOCAL_CONFIG.exists(): + return _DEFAULT_LOCAL_CONFIG + if _DEFAULT_CONFIG_FILE.exists(): + return _DEFAULT_CONFIG_FILE + return None + + +@lru_cache(maxsize=1) +def get_settings() -> Settings: + """Return the singleton Settings instance.""" + return Settings() + + +def reset_settings() -> None: + """Clear the cached settings (useful in tests or after config changes).""" + get_settings.cache_clear() diff --git a/packages/game-asset-mcp/src/game_asset_mcp/glb_reader.py b/packages/game-asset-mcp/src/game_asset_mcp/glb_reader.py new file mode 100755 index 00000000..0c5df4ff --- /dev/null +++ b/packages/game-asset-mcp/src/game_asset_mcp/glb_reader.py @@ -0,0 +1,102 @@ +""" +glb_reader.py — Pure-Python GLB/GLTF stats extraction. + +Reads GLB binary format directly (no Blender, no bpy, no external deps). +GLB format: 12-byte header + JSON chunk + optional BIN chunk. +""" +from __future__ import annotations + + +import struct +import json +import os + + +def read_glb_stats(path: str) -> dict | None: + """ + Extract mesh statistics from a GLB file without Blender. + + Returns a dict with: + meshes, primitives, vertices, faces, materials, + textures, has_embedded_textures, has_armature, + animations, file_size_kb + + Returns None if file is not a valid GLB. + """ + try: + file_size = os.path.getsize(path) + with open(path, "rb") as f: + header = f.read(12) + if len(header) < 12: + return None + magic, version, length = struct.unpack(" 0, + "animations": len(animations), + "extensions": extensions, + "file_size_kb": file_size // 1024, + } + + +def is_valid_glb(path: str) -> bool: + """Quick check — does this file start with the GLB magic bytes?""" + try: + with open(path, "rb") as f: + return f.read(4) == b"glTF" + except Exception: + return False diff --git a/packages/game-asset-mcp/src/game_asset_mcp/ingest.py b/packages/game-asset-mcp/src/game_asset_mcp/ingest.py new file mode 100755 index 00000000..3c08cc2e --- /dev/null +++ b/packages/game-asset-mcp/src/game_asset_mcp/ingest.py @@ -0,0 +1,265 @@ +""" +ingest.py — Scan the asset library and populate the SQLite catalog. + +Usage: + python -m game_asset_mcp.ingest [--root /Volumes/home/assets] [--force] [--dry-run] + +Does NOT require Blender. Uses pure-Python GLB reader for all mesh stats. +""" +from __future__ import annotations + + +import os +import re +import time +import json +from pathlib import Path + +from pydantic import BaseModel, Field + +from .glb_reader import read_glb_stats +from .catalog import DB_PATH, get_connection, init_db, upsert_asset, rebuild_fts + +from .config import get_settings + +ASSETS_ROOT = get_settings().assets_root + + +def _taxonomy(): + return get_settings().get_taxonomy() + +# Tags to derive from filename words +def derive_tags(name: str, category: str, pack: str) -> str: + """Build a space-separated tag string for FTS from path components.""" + parts = re.split(r"[_\-\s\.]+", name.lower()) + cat_parts = re.split(r"[/\\]+", category.lower()) + pack_parts = re.split(r"[_\-\s\.]+", pack.lower()) + all_parts = parts + cat_parts + pack_parts + # Deduplicate while preserving order + seen = set() + tags = [] + for p in all_parts: + if p and p not in seen and len(p) > 1: + seen.add(p) + tags.append(p) + return " ".join(tags) + + +def detect_source(path_parts: list[str]) -> str: + path_str = " ".join(p.lower() for p in path_parts) + for hint, source in _taxonomy().source_hints.items(): + if hint in path_str: + return source + return "Unknown" + + +def detect_style(rel_path: str) -> str: + for prefix, style in _taxonomy().style_map.items(): + if rel_path.startswith(prefix): + return style + return "Unknown" + + +def detect_category(rel_path: str, style: str) -> str: + """Extract the category portion (strip style prefix and pack name).""" + parts = Path(rel_path).parts + # parts[0] = style (3DLowPoly / 3DPSX) + # parts[1..n-2] = category path + # parts[-1] = filename.glb + if len(parts) >= 3: + return "/".join(parts[1:-1]) + return "/" + + +def detect_pack(rel_path: str) -> str: + """The immediate parent directory of a GLB is the pack.""" + return Path(rel_path).parent.name + + +def scan_glbs(root: Path) -> list[Path]: + """Walk the asset root and find all .glb files (skipping _Archive etc.).""" + result = [] + for dirpath, dirs, files in os.walk(root): + dirs[:] = sorted([ + d for d in dirs + if d not in _taxonomy().skip_dirs and not d.startswith("_") + ]) + for f in files: + if f.lower().endswith(".glb"): + result.append(Path(dirpath) / f) + return result + + +def ingest( + root: Path = ASSETS_ROOT, + db_path: Path = DB_PATH, + force: bool = False, + dry_run: bool = False, + verbose: bool = False, +) -> dict: + """Scan root for GLBs and populate the catalog DB.""" + init_db(db_path) + conn = get_connection(db_path) + + # Load all existing records into memory once — avoids N individual DB queries + known: dict[str, int] = {} + if not force: + for row in conn.execute("SELECT path, file_size_kb FROM assets"): + known[row["path"]] = row["file_size_kb"] + + glbs = scan_glbs(root) + total = len(glbs) + added = updated = skipped = errors = 0 + run_start = time.time() + + for glb_path in glbs: + rel = glb_path.relative_to(root) + rel_str = str(rel) + + # Fast in-memory skip check — no DB round-trip per file + if not force: + current_size_kb = glb_path.stat().st_size // 1024 + if known.get(str(glb_path)) == current_size_kb: + skipped += 1 + continue + + stats = read_glb_stats(str(glb_path)) + if stats is None: + if verbose: + print(f" SKIP (invalid GLB): {rel_str}") + errors += 1 + continue + + name = glb_path.stem + style = detect_style(rel_str) + category = detect_category(rel_str, style) + pack = detect_pack(rel_str) + source = detect_source(list(rel.parts)) + tags = derive_tags(name, category, pack) + + # Check for matching preview PNG + preview_path = None + for ext in (".png", ".jpg"): + candidate = glb_path.with_suffix(ext) + if candidate.exists(): + preview_path = str(candidate) + break + # Also check previews/ sibling directory + previews_dir = glb_path.parent / "previews" + candidate = previews_dir / (name + ".png") + if candidate.exists(): + preview_path = str(candidate) + + asset_record = { + "path": str(glb_path), + "name": name, + "style": style, + "category": category, + "pack": pack, + "source": source, + "meshes": stats["meshes"], + "vertices": stats["vertices"], + "faces": stats["faces"], + "materials": stats["materials"], + "textures": stats["textures"], + "has_embedded_textures": int(stats["has_embedded_textures"]), + "has_armature": int(stats["has_armature"]), + "animations": stats["animations"], + "extensions": stats["extensions"], + "file_size_kb": stats["file_size_kb"], + "preview_path": preview_path, + "tags": tags, + "ingested_at": time.time(), + } + + is_update = str(glb_path) in known + + if verbose: + status = "UPDATE" if is_update else "ADD" + print(f" [{status}] {rel_str}: {stats['faces']}f {stats['vertices']}v") + + if not dry_run: + upsert_asset(conn, asset_record) + if is_update: + updated += 1 + else: + added += 1 + else: + added += 1 # count as would-be-added in dry run + + # Remove stale entries — files in DB that no longer exist on disk + found_paths = {str(p) for p in glbs} + stale = [p for p in known if p not in found_paths] + removed = 0 + if stale and not dry_run: + for path in stale: + conn.execute("DELETE FROM assets WHERE path=?", (path,)) + removed = len(stale) + if verbose and stale: + for p in stale: + print(f" [REMOVE] {p}") + + if not dry_run: + rebuild_fts(conn) + elapsed = time.time() - run_start + conn.execute( + "INSERT INTO ingest_log (run_at, total, added, updated, skipped) VALUES (?,?,?,?,?)", + (run_start, total, added, updated, skipped) + ) + conn.commit() + + conn.close() + return { + "total_scanned": total, + "added": added, + "updated": updated, + "skipped": skipped, + "removed": removed, + "errors": errors, + } + + +class IngestOptions(BaseModel): + """Scan the asset library and populate the SQLite catalog.""" + + root: Path = Field( + default_factory=lambda: get_settings().assets_root, + description="Asset library root directory", + ) + db: Path = Field( + default_factory=lambda: get_settings().catalog_db, + description="SQLite catalog DB path", + ) + force: bool = Field(False, description="Re-ingest all files (ignore size cache)") + dry_run: bool = Field(False, description="Scan only — no DB writes") + verbose: bool = Field(False, description="Print per-file status") + + +def _run_ingest(opts: IngestOptions) -> int: + print(f"Ingesting from {opts.root} into {opts.db}") + if opts.dry_run: + print(" (dry run — no writes)") + + result = ingest( + root=opts.root, + db_path=opts.db, + force=opts.force, + dry_run=opts.dry_run, + verbose=opts.verbose, + ) + + print( + f"\nDone: {result['added']} added, {result['updated']} updated, " + f"{result['skipped']} skipped, {result['removed']} removed, {result['errors']} errors" + ) + print(f"Total GLBs scanned: {result['total_scanned']}") + return 0 + + +def main() -> None: + from pydantic_cli import run_and_exit + run_and_exit(IngestOptions, _run_ingest, description=__doc__ or "game-asset-ingest") + + +if __name__ == "__main__": + main() diff --git a/packages/game-asset-mcp/src/game_asset_mcp/polyhaven.py b/packages/game-asset-mcp/src/game_asset_mcp/polyhaven.py new file mode 100644 index 00000000..5102481c --- /dev/null +++ b/packages/game-asset-mcp/src/game_asset_mcp/polyhaven.py @@ -0,0 +1,223 @@ +""" +polyhaven.py — PolyHaven API integration. +Free 3D assets, HDRIs, and textures from polyhaven.com (CC0). +""" +from __future__ import annotations + +import os +import httpx +from pathlib import Path +from typing import Optional + +ASSETS_ROOT = Path(os.environ.get("ASSETS_ROOT", str(Path.home() / "assets"))) +PH_API = "https://api.polyhaven.com" + +MODEL_CATEGORY_MAP = { + "nature": "3DLowPoly/Environment/Nature/polyhaven", + "architecture": "3DLowPoly/Environment/City/polyhaven", + "furniture": "3DLowPoly/Props/Furniture/polyhaven", + "food": "3DLowPoly/Props/Food/polyhaven", + "vehicles": "3DLowPoly/Vehicles/polyhaven", + "electronics": "3DLowPoly/Props/Electronics/polyhaven", + "animals": "3DLowPoly/Characters/Animals/polyhaven", +} +DEFAULT_MODEL_PATH = "3DLowPoly/Props/Misc/polyhaven" + +# Maps used by texture download: keys are PolyHaven map keys → local filename suffixes +TEXTURE_MAP_KEYS = ["diffuse", "rough", "metal", "nor_gl", "ao", "disp", "arm"] + + +def ph_get(endpoint: str, params: dict = None) -> dict: + """GET from PolyHaven API with 30s timeout.""" + url = f"{PH_API}/{endpoint.lstrip('/')}" + response = httpx.get(url, params=params, timeout=30.0) + response.raise_for_status() + return response.json() + + +def search_ph( + query: str, + asset_type: str, + category: Optional[str] = None, + limit: int = 20, +) -> list[dict]: + """ + Search PolyHaven. query matches id/name/tags. + Returns list sorted by download_count descending. + """ + params: dict = {"type": asset_type} + if category: + params["categories"] = category + + assets = ph_get("assets", params=params) + + query_lower = query.lower() + query_words = query_lower.split() + + results = [] + for asset_id, info in assets.items(): + name = info.get("name", "") + tags = info.get("tags", []) + categories = info.get("categories", []) + + # Build searchable text from id, name, tags, categories + searchable = " ".join( + [asset_id, name.lower()] + [t.lower() for t in tags] + [c.lower() for c in categories] + ) + + if all(word in searchable for word in query_words): + results.append({ + "id": asset_id, + "name": name, + "type": info.get("type", asset_type), + "categories": categories, + "tags": tags, + "download_count": info.get("download_count", 0), + }) + + results.sort(key=lambda x: x["download_count"], reverse=True) + return results[:limit] + + +def get_ph_info(asset_id: str) -> dict: + """Get full info for one asset.""" + return ph_get(f"info/{asset_id}") + + +def get_taxonomy_path(asset_id: str, asset_type: str, categories: list[str]) -> Path: + """Map asset to local filesystem path.""" + if asset_type == "hdris": + return ASSETS_ROOT / "2DPhotorealistic" / "HDRIs" / "polyhaven" / asset_id + if asset_type == "textures": + return ASSETS_ROOT / "2DPhotorealistic" / "Textures" / "polyhaven" / asset_id + + # Models: match first category against the map + for cat in categories: + cat_lower = cat.lower() + if cat_lower in MODEL_CATEGORY_MAP: + return ASSETS_ROOT / MODEL_CATEGORY_MAP[cat_lower] / asset_id + + return ASSETS_ROOT / DEFAULT_MODEL_PATH / asset_id + + +def _download_bytes(url: str) -> bytes: + """Download raw bytes from a URL with 120s timeout.""" + response = httpx.get(url, timeout=120.0, follow_redirects=True) + response.raise_for_status() + return response.content + + +def download_ph_asset( + asset_id: str, + asset_type: str, + resolution: str = "1k", +) -> dict: + """ + Download asset from PolyHaven and save to taxonomy path. + + For models: downloads the GLB at the requested resolution. + For HDRIs: downloads the .hdr file at the requested resolution. + For textures: downloads all available map files (diffuse, rough, metal, + normal, ao, disp, arm) at the requested resolution. + + Returns: + {dest_dir: str, files: [str], asset_type: str, categories: list} + or {error: str} on failure. + """ + # Fetch asset info to get categories + try: + info = get_ph_info(asset_id) + except httpx.HTTPError as exc: + return {"error": f"Failed to fetch asset info: {exc}"} + + categories = info.get("categories", []) + + # Fetch file manifest + try: + files_data = ph_get(f"files/{asset_id}") + except httpx.HTTPError as exc: + return {"error": f"Failed to fetch file manifest: {exc}"} + + dest_dir = get_taxonomy_path(asset_id, asset_type, categories) + dest_dir.mkdir(parents=True, exist_ok=True) + + downloaded_files: list[str] = [] + + if asset_type == "models": + # Structure: {"gltf": {"1k": {"glb": {"url": "...", "size": N}, ...}, ...}, ...} + gltf_section = files_data.get("gltf", {}) + res_section = gltf_section.get(resolution) or next(iter(gltf_section.values()), None) + if not res_section: + return {"error": f"No gltf files found for {asset_id}"} + + glb_info = res_section.get("glb") + if not glb_info: + return {"error": f"No GLB variant found for {asset_id} at resolution {resolution}"} + + url = glb_info["url"] + dest_file = dest_dir / f"{asset_id}.glb" + try: + data = _download_bytes(url) + except httpx.HTTPError as exc: + return {"error": f"Download failed: {exc}"} + dest_file.write_bytes(data) + downloaded_files.append(str(dest_file)) + + elif asset_type == "hdris": + # Structure: {"hdri": {"1k": {"hdr": {"url": "...", "size": N}, "exr": {...}}, ...}} + hdri_section = files_data.get("hdri", {}) + res_section = hdri_section.get(resolution) or next(iter(hdri_section.values()), None) + if not res_section: + return {"error": f"No HDRI files found for {asset_id}"} + + # Prefer .hdr; fall back to .exr + for fmt in ("hdr", "exr"): + fmt_info = res_section.get(fmt) + if fmt_info: + url = fmt_info["url"] + dest_file = dest_dir / f"{asset_id}_{resolution}.{fmt}" + try: + data = _download_bytes(url) + except httpx.HTTPError as exc: + return {"error": f"Download failed: {exc}"} + dest_file.write_bytes(data) + downloaded_files.append(str(dest_file)) + break + + if not downloaded_files: + return {"error": f"No hdr/exr variant found for {asset_id} at resolution {resolution}"} + + elif asset_type == "textures": + # Structure: {"1k": {"diffuse": {"url": "...", ...}, "rough": {...}, ...}, ...} + res_section = files_data.get(resolution) or next(iter(files_data.values()), None) + if not res_section: + return {"error": f"No texture files found for {asset_id}"} + + for map_key, map_info in res_section.items(): + if not isinstance(map_info, dict) or "url" not in map_info: + continue + url = map_info["url"] + # Derive extension from URL + url_path = url.split("?")[0] + ext = url_path.rsplit(".", 1)[-1] if "." in url_path else "png" + dest_file = dest_dir / f"{asset_id}_{map_key}.{ext}" + try: + data = _download_bytes(url) + except httpx.HTTPError as exc: + # Don't abort entire texture set if one map fails + continue + dest_file.write_bytes(data) + downloaded_files.append(str(dest_file)) + + if not downloaded_files: + return {"error": f"No texture maps downloaded for {asset_id} at resolution {resolution}"} + + else: + return {"error": f"Unknown asset_type '{asset_type}'. Use 'models', 'hdris', or 'textures'."} + + return { + "dest_dir": str(dest_dir), + "files": downloaded_files, + "asset_type": asset_type, + "categories": categories, + } diff --git a/packages/game-asset-mcp/src/game_asset_mcp/render_preview.py b/packages/game-asset-mcp/src/game_asset_mcp/render_preview.py new file mode 100755 index 00000000..6920df9e --- /dev/null +++ b/packages/game-asset-mcp/src/game_asset_mcp/render_preview.py @@ -0,0 +1,129 @@ +""" +render_preview.py — Headless Blender script to render a GLB as a 512x512 PNG. + +Usage (called as subprocess from server.py): + blender --background --python render_preview.py -- --input model.glb --output preview.png + +Camera: isometric-ish 3/4 view. Lighting: 3-point studio. Output: 512x512 RGBA PNG. +""" +from __future__ import annotations + + +import sys +import os +import argparse +import math + +try: + import bpy # noqa: F401 + HAS_BPY = True +except ImportError: + HAS_BPY = False + + +def parse_args(): + if "--" in sys.argv: + args = sys.argv[sys.argv.index("--") + 1:] + else: + args = [] + p = argparse.ArgumentParser() + p.add_argument("--input", required=True) + p.add_argument("--output", required=True) + p.add_argument("--size", type=int, default=512) + return p.parse_args(args) + + +def main(): + import bpy + import mathutils + + args = parse_args() + + # Clear scene + bpy.ops.wm.read_factory_settings(use_empty=True) + + # Import GLB + bpy.ops.import_scene.gltf(filepath=os.path.abspath(args.input)) + + # Find imported mesh objects and compute bounding box + meshes = [o for o in bpy.context.scene.objects if o.type == "MESH"] + if not meshes: + print(f"[preview] ERROR: no mesh in {args.input}") + sys.exit(1) + + # Compute world bounding box center and radius + all_corners = [] + for obj in meshes: + for corner in obj.bound_box: + all_corners.append(obj.matrix_world @ mathutils.Vector(corner)) + + xs = [v.x for v in all_corners] + ys = [v.y for v in all_corners] + zs = [v.z for v in all_corners] + center = mathutils.Vector(( + (min(xs) + max(xs)) / 2, + (min(ys) + max(ys)) / 2, + (min(zs) + max(zs)) / 2, + )) + radius = max( + max(xs) - min(xs), + max(ys) - min(ys), + max(zs) - min(zs), + ) / 2 or 1.0 + + # Camera: 3/4 isometric view + cam_data = bpy.data.cameras.new("PreviewCam") + cam_data.type = "PERSP" + cam_data.lens = 50 + cam_obj = bpy.data.objects.new("PreviewCam", cam_data) + bpy.context.scene.collection.objects.link(cam_obj) + bpy.context.scene.camera = cam_obj + + distance = radius * 3.5 + angle_h = math.radians(45) + angle_v = math.radians(30) + cam_obj.location = center + mathutils.Vector(( + distance * math.cos(angle_v) * math.cos(angle_h), + distance * math.cos(angle_v) * math.sin(angle_h), + distance * math.sin(angle_v), + )) + direction = center - cam_obj.location + rot_quat = direction.to_track_quat("-Z", "Y") + cam_obj.rotation_euler = rot_quat.to_euler() + + # 3-point lighting + def add_light(name, light_type, energy, loc): + ld = bpy.data.lights.new(name, light_type) + ld.energy = energy + lo = bpy.data.objects.new(name, ld) + bpy.context.scene.collection.objects.link(lo) + lo.location = center + mathutils.Vector(loc) * radius * 3 + + add_light("Key", "SUN", 3.0, (1, -1, 2)) + add_light("Fill", "SUN", 1.0, (-2, 1, 1)) + add_light("Back", "SUN", 0.5, (0, 2, -0.5)) + + # World background: neutral grey + bpy.context.scene.world = bpy.data.worlds.new("World") + bpy.context.scene.world.use_nodes = True + bg = bpy.context.scene.world.node_tree.nodes["Background"] + bg.inputs["Color"].default_value = (0.15, 0.15, 0.15, 1.0) + bg.inputs["Strength"].default_value = 0.3 + + # Render settings + scene = bpy.context.scene + scene.render.engine = "CYCLES" + scene.cycles.samples = 32 + scene.cycles.use_denoising = True + scene.render.resolution_x = args.size + scene.render.resolution_y = args.size + scene.render.film_transparent = False + scene.render.image_settings.file_format = "PNG" + scene.render.filepath = os.path.abspath(args.output) + + bpy.ops.render.render(write_still=True) + print(f"[preview] Rendered: {args.output}") + + +if __name__ == "__main__": + main() diff --git a/packages/game-asset-mcp/src/game_asset_mcp/search.py b/packages/game-asset-mcp/src/game_asset_mcp/search.py new file mode 100755 index 00000000..488b409c --- /dev/null +++ b/packages/game-asset-mcp/src/game_asset_mcp/search.py @@ -0,0 +1,222 @@ +""" +search.py — Hybrid keyword + vector search over the asset catalog. + +Keyword search: SQLite FTS5 (fast, exact, no dependencies) +Semantic search: sqlite-vec + sentence-transformers (local, no API) + +The two results are merged and ranked: semantic hits boosted by keyword score. +""" +from __future__ import annotations + + +import os +import json +import sqlite3 +from pathlib import Path +from typing import Optional + +from .catalog import DB_PATH, get_connection + +# Lazy-import heavy deps so startup is fast even if they're not yet installed +_embedder = None +_vec_conn = None + + +def _get_embedder(): + global _embedder + if _embedder is None: + try: + from sentence_transformers import SentenceTransformer + _embedder = SentenceTransformer("all-MiniLM-L6-v2") + except ImportError: + _embedder = None + return _embedder + + +def embed_text(text: str) -> list[float] | None: + """Embed text using the local sentence-transformers model.""" + embedder = _get_embedder() + if embedder is None: + return None + vec = embedder.encode(text, convert_to_numpy=True) + return vec.tolist() + + +def _load_sqlite_vec(conn: sqlite3.Connection) -> bool: + """Load the sqlite-vec extension into the connection.""" + try: + import sqlite_vec + conn.enable_load_extension(True) + sqlite_vec.load(conn) + conn.enable_load_extension(False) + return True + except Exception: + return False + + +def ensure_vec_table(db_path: Path = DB_PATH) -> bool: + """Create the vec0 virtual table for vector search if it doesn't exist.""" + conn = get_connection(db_path) + if not _load_sqlite_vec(conn): + conn.close() + return False + try: + conn.execute(""" + CREATE VIRTUAL TABLE IF NOT EXISTS assets_vec USING vec0( + asset_id INTEGER PRIMARY KEY, + embedding FLOAT[384] + ) + """) + conn.commit() + return True + except Exception: + return False + finally: + conn.close() + + +def upsert_embedding(conn: sqlite3.Connection, asset_id: int, embedding: list[float]) -> None: + """Insert or replace an embedding in the vec0 table.""" + import struct + vec_bytes = struct.pack(f"{len(embedding)}f", *embedding) + conn.execute( + "INSERT OR REPLACE INTO assets_vec(asset_id, embedding) VALUES (?, ?)", + (asset_id, vec_bytes) + ) + + +def build_asset_text(asset: dict) -> str: + """Create a searchable text representation of an asset for embedding.""" + parts = [ + asset.get("name", ""), + asset.get("category", ""), + asset.get("pack", ""), + asset.get("source", ""), + asset.get("style", ""), + asset.get("tags", ""), + ] + return " ".join(p for p in parts if p) + + +def fts_search( + conn: sqlite3.Connection, + query: str, + style: Optional[str] = None, + category: Optional[str] = None, + has_armature: Optional[bool] = None, + has_textures: Optional[bool] = None, + limit: int = 20, +) -> list[dict]: + """FTS5 keyword search.""" + sql = """ + SELECT a.*, fts.rank as fts_rank + FROM assets_fts fts + JOIN assets a ON a.id = fts.rowid + WHERE assets_fts MATCH ? + """ + params: list = [query] + if style: + sql += " AND a.style = ?" + params.append(style) + if category: + sql += " AND a.category LIKE ?" + params.append(f"%{category}%") + if has_armature is not None: + sql += " AND a.has_armature = ?" + params.append(1 if has_armature else 0) + if has_textures is not None: + sql += " AND a.has_embedded_textures = ?" + params.append(1 if has_textures else 0) + sql += " ORDER BY rank LIMIT ?" + params.append(limit) + try: + rows = conn.execute(sql, params).fetchall() + return [dict(r) for r in rows] + except Exception: + return [] + + +def vec_search( + conn: sqlite3.Connection, + query: str, + style: Optional[str] = None, + category: Optional[str] = None, + limit: int = 20, +) -> list[dict]: + """Semantic vector search using sqlite-vec.""" + if not _load_sqlite_vec(conn): + return [] + + embedding = embed_text(query) + if embedding is None: + return [] + + import struct + vec_bytes = struct.pack(f"{len(embedding)}f", *embedding) + + sql = """ + SELECT a.*, v.distance as vec_distance + FROM assets_vec v + JOIN assets a ON a.id = v.asset_id + WHERE v.embedding MATCH ? + AND k = ? + """ + params: list = [vec_bytes, limit * 2] # fetch extra for post-filtering + if style: + sql += " AND a.style = ?" + params.append(style) + if category: + sql += " AND a.category LIKE ?" + params.append(f"%{category}%") + sql += " ORDER BY v.distance" + + try: + rows = conn.execute(sql, params).fetchall() + return [dict(r) for r in rows[:limit]] + except Exception: + return [] + + +def hybrid_search( + query: str, + style: Optional[str] = None, + category: Optional[str] = None, + has_armature: Optional[bool] = None, + has_textures: Optional[bool] = None, + max_results: int = 20, + db_path: Path = DB_PATH, +) -> list[dict]: + """ + Hybrid keyword + semantic search. + + Strategy: + 1. FTS5 keyword search (fast, exact) + 2. Vector semantic search (if embedder available) + 3. Merge, deduplicate, rank by combined score + 4. Fall back to FTS5-only if vector search unavailable + """ + conn = get_connection(db_path) + + kw_results = fts_search(conn, query, style, category, has_armature, has_textures, limit=max_results * 2) + vec_results = vec_search(conn, query, style, category, limit=max_results * 2) + + conn.close() + + # Merge by path (dedup) + seen_paths = {} + for r in kw_results: + seen_paths[r["path"]] = r + r["_score"] = 1.0 / (1.0 + abs(r.get("fts_rank", 0))) # FTS rank is negative + + for r in vec_results: + if r["path"] not in seen_paths: + seen_paths[r["path"]] = r + r["_score"] = 1.0 / (1.0 + r.get("vec_distance", 1.0)) + else: + # Boost combined hit + existing = seen_paths[r["path"]] + vec_score = 1.0 / (1.0 + r.get("vec_distance", 1.0)) + existing["_score"] = existing.get("_score", 0) + vec_score + + results = sorted(seen_paths.values(), key=lambda r: -r.get("_score", 0)) + return results[:max_results] diff --git a/packages/game-asset-mcp/src/game_asset_mcp/server.py b/packages/game-asset-mcp/src/game_asset_mcp/server.py new file mode 100755 index 00000000..cc5d3f95 --- /dev/null +++ b/packages/game-asset-mcp/src/game_asset_mcp/server.py @@ -0,0 +1,486 @@ +""" +server.py — FastMCP 3.0 stdio MCP server for local 3D game asset libraries. + +Install: + pip install "game-asset-mcp[server]" + +Add to Claude Code: + claude mcp add game-asset-library -- game-asset-mcp + +Configure via env vars: + ASSETS_ROOT=/path/to/assets CATALOG_DB=~/.local/share/game-asset-mcp/catalog.db +""" +from __future__ import annotations + + +import os +import shutil +import subprocess +from pathlib import Path +from typing import Optional + +from fastmcp import FastMCP + +from .catalog import get_connection, get_asset, list_categories, get_stats, DB_PATH +from .search import hybrid_search, fts_search + +from .config import get_settings + +_settings = get_settings() +ASSETS_ROOT = _settings.assets_root +BLENDER = _settings.blender +PREVIEW_SCRIPT = Path(__file__).parent / "render_preview.py" + +try: + import bpy as _bpy # noqa: F401 + HAS_BPY = True +except ImportError: + HAS_BPY = False + +HAS_BLENDER = HAS_BPY or BLENDER.exists() + +mcp = FastMCP( + "game-asset-library", + instructions=( + "Search, browse, and manage a local 3D game asset library. " + "Indexes GLB models by style, category, and pack with mesh stats. " + "Supports PolyHaven CC0 asset search and download." + ), +) + + +# ─── Tools ──────────────────────────────────────────────────────────────────── + + +@mcp.tool() +def search_assets( + query: str, + style: Optional[str] = None, + category: Optional[str] = None, + has_armature: Optional[bool] = None, + has_textures: Optional[bool] = None, + max_results: int = 20, +) -> list[dict]: + """ + Search the 3D asset catalog by keyword or natural language query. + + Args: + query: Search term — e.g. "tree", "PSX sword", "animated character walking" + style: Filter by '3DLowPoly' or '3DPSX' + category: Filter by category path — e.g. 'Characters/Animated', 'Environment/Nature' + has_armature: True to find rigged/animated models only + has_textures: True to find models with embedded textures only + max_results: Max number of results (default 20) + + Returns: + List of asset dicts with: path, name, style, category, pack, faces, vertices, + materials, has_armature, has_embedded_textures, file_size_kb, preview_path + """ + if not query.strip(): + conn = get_connection() + rows = conn.execute( + "SELECT * FROM assets ORDER BY ingested_at DESC LIMIT ?", (max_results,) + ).fetchall() + conn.close() + return [_format_asset(dict(r)) for r in rows] + + results = hybrid_search( + query=query, + style=style, + category=category, + has_armature=has_armature, + has_textures=has_textures, + max_results=max_results, + ) + return [_format_asset(r) for r in results] + + +@mcp.tool() +def list_categories(style: Optional[str] = None) -> list[dict]: + """ + List all asset categories with GLB counts. + + Args: + style: Filter by '3DLowPoly' or '3DPSX' (omit for all) + + Returns: + List of {style, category, count} dicts sorted by style/category. + """ + conn = get_connection() + if style: + rows = conn.execute( + "SELECT style, category, COUNT(*) as count FROM assets WHERE style=? " + "GROUP BY style, category ORDER BY category", + (style,) + ).fetchall() + else: + rows = conn.execute( + "SELECT style, category, COUNT(*) as count FROM assets " + "GROUP BY style, category ORDER BY style, category" + ).fetchall() + conn.close() + return [dict(r) for r in rows] + + +@mcp.tool() +def get_asset_info(path: str) -> dict: + """ + Get full metadata for a specific asset by its absolute path. + + Args: + path: Absolute path to the .glb file + + Returns: + Full asset record including mesh stats, category, source, preview path. + Returns {'error': '...'} if not found. + """ + conn = get_connection() + asset = get_asset(conn, path) + conn.close() + if asset is None: + # Try live read if not in catalog + from .glb_reader import read_glb_stats + if Path(path).exists(): + stats = read_glb_stats(path) + return {"path": path, "name": Path(path).stem, "note": "not in catalog", **(stats or {})} + return {"error": f"Asset not found: {path}"} + return _format_asset(asset) + + +@mcp.tool() +def copy_asset(src_path: str, dest_dir: str, rename: Optional[str] = None) -> dict: + """ + Copy a GLB asset to a destination directory (e.g. into your game project). + + Args: + src_path: Absolute path to the source .glb file + dest_dir: Destination directory (will be created if needed) + rename: Optional new filename without extension (default: keep original name) + + Returns: + {'dest': '/path/to/copied.glb', 'size_kb': 42} + """ + src = Path(src_path) + if not src.exists(): + return {"error": f"Source not found: {src_path}"} + dest = Path(dest_dir) + dest.mkdir(parents=True, exist_ok=True) + out_name = (rename or src.stem) + ".glb" + out_path = dest / out_name + shutil.copy2(src, out_path) + return {"dest": str(out_path), "size_kb": out_path.stat().st_size // 1024} + + +@mcp.tool() +def generate_preview(glb_path: str, output_dir: Optional[str] = None) -> dict: + """ + Render a 512×512 PNG thumbnail of a GLB model using headless Blender. + + The preview is saved alongside the GLB as .png (or in output_dir). + Uses a standardized 3/4-view camera with 3-point studio lighting. + + Args: + glb_path: Absolute path to the .glb file + output_dir: Optional output directory (default: same dir as GLB) + + Returns: + {'preview_path': '/path/to/preview.png'} or {'error': '...'} + """ + src = Path(glb_path) + if not src.exists(): + return {"error": f"GLB not found: {glb_path}"} + + if not HAS_BLENDER: + return {"error": "Blender not available. Install bpy (pip install 'game-asset-mcp[blender]') or set BLENDER env var."} + + if not PREVIEW_SCRIPT.exists(): + return {"error": f"Preview script not found: {PREVIEW_SCRIPT}"} + + out_dir = Path(output_dir) if output_dir else src.parent + out_dir.mkdir(parents=True, exist_ok=True) + out_png = out_dir / (src.stem + ".png") + + result = subprocess.run( + [ + str(BLENDER), "--background", "--python", str(PREVIEW_SCRIPT), + "--", "--input", str(src), "--output", str(out_png), + ], + capture_output=True, text=True, timeout=60 + ) + + if out_png.exists(): + # Update catalog with preview path + conn = get_connection() + conn.execute( + "UPDATE assets SET preview_path=? WHERE path=?", + (str(out_png), str(src)) + ) + conn.commit() + conn.close() + return {"preview_path": str(out_png)} + else: + return {"error": "Blender render failed", "stderr": result.stderr[-500:]} + + +@mcp.tool() +def get_preview(glb_path: str) -> dict: + """ + Return the path to an existing PNG preview for a GLB, if available. + + Does NOT generate a new preview — use generate_preview() for that. + + Args: + glb_path: Absolute path to the .glb file + + Returns: + {'preview_path': '/path/to/preview.png'} or {'preview_path': None, 'note': '...'} + """ + src = Path(glb_path) + # Check common preview locations + for candidate in [ + src.with_suffix(".png"), + src.parent / "previews" / (src.stem + ".png"), + ]: + if candidate.exists(): + return {"preview_path": str(candidate)} + + # Check catalog + conn = get_connection() + row = conn.execute( + "SELECT preview_path FROM assets WHERE path=?", (str(src),) + ).fetchone() + conn.close() + if row and row["preview_path"] and Path(row["preview_path"]).exists(): + return {"preview_path": row["preview_path"]} + + return {"preview_path": None, "note": "No preview. Run generate_preview() to create one."} + + +@mcp.tool() +def get_catalog_stats() -> dict: + """ + Return summary statistics about the asset catalog. + + Returns: + total GLB count, with_textures, with_armature, with_previews, by_style counts. + """ + conn = get_connection() + stats = get_stats(conn) + by_style = conn.execute( + "SELECT style, COUNT(*) as count FROM assets GROUP BY style" + ).fetchall() + conn.close() + return { + **stats, + "by_style": {r["style"]: r["count"] for r in by_style}, + } + + +@mcp.tool() +def browse_taxonomy( + style: Optional[str] = None, + meso: Optional[str] = None, + micro: Optional[str] = None, +) -> list[dict]: + """ + Navigate the asset library's nested category taxonomy at three levels. + + Macro (style): '3DLowPoly' or '3DPSX' + Meso (level 1): top category — 'Characters', 'Props', 'Environment', 'Vehicles', etc. + Micro (level 2): sub-category — 'Weapons', 'Animated', 'Nature', 'Buildings', etc. + Pack (level 3): individual pack directory name + + Call with no args → all meso categories with counts. + Pass style → filter to one library. + Pass meso → drill into sub-categories (micro level). + Pass meso + micro → list individual packs. + + Examples: + browse_taxonomy() + browse_taxonomy(style='3DPSX') + browse_taxonomy(style='3DPSX', meso='Props') + browse_taxonomy(meso='Props', micro='Weapons') + + Returns: + List of {style, meso, micro, pack, count, level} dicts. + """ + conn = get_connection() + + where = ["1=1"] + params: list = [] + if style: + where.append("style = ?") + params.append(style) + + where_sql = " AND ".join(where) + + if meso is None: + # Meso level: first path segment of category + rows = conn.execute(f""" + SELECT style, + CASE WHEN instr(category, '/') > 0 + THEN substr(category, 1, instr(category, '/') - 1) + ELSE category END AS meso, + COUNT(*) as count + FROM assets + WHERE {where_sql} + GROUP BY style, meso + ORDER BY style, meso + """, params).fetchall() + conn.close() + return [{"style": r["style"], "meso": r["meso"], "micro": None, "pack": None, + "count": r["count"], "level": "meso"} for r in rows] + + # Filter to assets under this meso category + where.append("(category = ? OR category LIKE ?)") + params += [meso, meso + "/%"] + where_sql = " AND ".join(where) + + if micro is None: + # Micro level: second path segment of category + rows = conn.execute(f""" + SELECT style, + CASE WHEN instr(category, '/') > 0 + THEN substr(category, 1, instr(category, '/') - 1) + ELSE category END AS meso, + CASE WHEN instr(category, '/') > 0 + THEN (SELECT CASE WHEN instr(rest, '/') > 0 + THEN substr(rest, 1, instr(rest, '/') - 1) + ELSE rest END + FROM (SELECT substr(category, instr(category, '/') + 1) as rest)) + ELSE NULL END AS micro, + COUNT(*) as count + FROM assets + WHERE {where_sql} + GROUP BY style, meso, micro + ORDER BY style, micro + """, params).fetchall() + conn.close() + return [{"style": r["style"], "meso": r["meso"], "micro": r["micro"], "pack": None, + "count": r["count"], "level": "micro"} for r in rows] + + # Filter to micro sub-category + where.append("(category = ? OR category LIKE ?)") + params += [meso + "/" + micro, meso + "/" + micro + "/%"] + where_sql = " AND ".join(where) + + # Pack level: individual pack directories + rows = conn.execute(f""" + SELECT style, pack, COUNT(*) as count + FROM assets + WHERE {where_sql} + GROUP BY style, pack + ORDER BY style, pack + """, params).fetchall() + conn.close() + return [{"style": r["style"], "meso": meso, "micro": micro, "pack": r["pack"], + "count": r["count"], "level": "pack"} for r in rows] + + +@mcp.tool() +def search_polyhaven( + query: str, + asset_type: str = "models", + category: Optional[str] = None, + limit: int = 20, +) -> list[dict]: + """ + Search polyhaven.com for free CC0 assets. + + Args: + query: Keyword to match against asset name/id/tags + asset_type: 'models', 'hdris', or 'textures' + category: Optional filter e.g. 'nature', 'furniture', 'architecture' + limit: Max results (default 20) + + Returns: + List of {id, name, type, categories, tags, download_count} dicts + """ + from .polyhaven import search_ph + return search_ph(query=query, asset_type=asset_type, category=category, limit=limit) + + +@mcp.tool() +def download_polyhaven_asset( + asset_id: str, + asset_type: str = "models", + resolution: str = "1k", +) -> dict: + """ + Download a PolyHaven asset and add it to the local library. + + Automatically places the asset in the correct taxonomy directory + and runs ingest to add it to the catalog. + + Args: + asset_id: PolyHaven asset ID (from search_polyhaven results) + asset_type: 'models', 'hdris', or 'textures' + resolution: '1k', '2k', '4k' (default '1k') + + Returns: + {dest_dir, files, asset_type, categories, ingested: bool} + """ + from .polyhaven import download_ph_asset + from .ingest import ingest + from pathlib import Path + + result = download_ph_asset(asset_id=asset_id, asset_type=asset_type, resolution=resolution) + if "error" in result: + return result + + # Auto-ingest if it's a model (GLB) + ingested = False + if asset_type == "models": + ingest_result = ingest() + ingested = ingest_result.get("added", 0) > 0 + + return {**result, "ingested": ingested} + + +@mcp.tool() +def run_ingest(force: bool = False) -> dict: + """ + Re-scan the asset library and update the catalog database. + + Only re-processes files whose size has changed (unless force=True). + + Args: + force: Re-ingest all files even if unchanged + + Returns: + {'added': N, 'updated': N, 'skipped': N, 'total_scanned': N} + """ + from .ingest import ingest + result = ingest(force=force) + return result + + +# ─── Helpers ────────────────────────────────────────────────────────────────── + + +def _format_asset(asset: dict) -> dict: + """Return a clean asset dict with only the most useful fields for agents.""" + return { + "path": asset.get("path"), + "name": asset.get("name"), + "style": asset.get("style"), + "category": asset.get("category"), + "pack": asset.get("pack"), + "source": asset.get("source"), + "faces": asset.get("faces"), + "vertices": asset.get("vertices"), + "materials": asset.get("materials"), + "has_armature": bool(asset.get("has_armature")), + "has_embedded_textures": bool(asset.get("has_embedded_textures")), + "animations": asset.get("animations", 0), + "file_size_kb": asset.get("file_size_kb"), + "preview_path": asset.get("preview_path"), + } + + +def main(): + mcp.run(transport="stdio") + + +if __name__ == "__main__": + main() diff --git a/packages/game-asset-mcp/src/game_asset_mcp/wizard.py b/packages/game-asset-mcp/src/game_asset_mcp/wizard.py new file mode 100644 index 00000000..014bf24a --- /dev/null +++ b/packages/game-asset-mcp/src/game_asset_mcp/wizard.py @@ -0,0 +1,180 @@ +""" +wizard.py — Interactive setup wizard for game-asset-mcp. + +Run: game-asset-init + +Asks questions, generates a TOML config, and optionally runs the initial ingest. +""" +from __future__ import annotations + +import argparse +import subprocess +import sys +import textwrap +from pathlib import Path + + +_CONFIG_DIR = Path.home() / ".config" / "game-asset-mcp" +_CONFIG_FILE = _CONFIG_DIR / "config.toml" + + +def _ask(prompt: str, default: str) -> str: + try: + value = input(f"{prompt} [{default}]: ").strip() + return value or default + except (EOFError, KeyboardInterrupt): + print() + sys.exit(0) + + +def _ask_bool(prompt: str, default: bool = True) -> bool: + default_str = "Y/n" if default else "y/N" + try: + value = input(f"{prompt} [{default_str}]: ").strip().lower() + if not value: + return default + return value.startswith("y") + except (EOFError, KeyboardInterrupt): + print() + sys.exit(0) + + +def _detect_styles(root: Path) -> list[tuple[str, str]]: + """Detect potential style directories in the asset root.""" + if not root.exists(): + return [] + styles = [] + for d in sorted(root.iterdir()): + if d.is_dir() and not d.name.startswith((".", "_")): + # Only suggest dirs that likely contain GLBs + glb_count = sum(1 for _ in d.rglob("*.glb") if True) if d.exists() else 0 + if glb_count > 0: + styles.append((d.name, d.name)) + return styles[:8] # cap at 8 suggestions + + +def run_wizard(non_interactive: bool = False, assets_root: str | None = None) -> None: + print("\n🎮 game-asset-mcp setup wizard\n" + "=" * 40) + + # Assets root + default_root = assets_root or str(Path.home() / "assets") + if non_interactive: + root_str = default_root + else: + root_str = _ask("Path to your 3D asset library", default_root) + root = Path(root_str).expanduser() + + # Catalog DB + default_db = str(Path.home() / ".local" / "share" / "game-asset-mcp" / "catalog.db") + if non_interactive: + db_str = default_db + else: + db_str = _ask("SQLite catalog DB path", default_db) + + # Detect styles + print(f"\nScanning {root} for style directories...") + detected = _detect_styles(root) + style_map: dict[str, str] = {} + + if detected: + print(f"Found {len(detected)} directories with GLBs:") + for name, _ in detected: + print(f" - {name}") + if non_interactive or _ask_bool("Use these as taxonomy styles?"): + style_map = {name: name for name, _ in detected} + + if not style_map: + # Fallback defaults + style_map = {"3DLowPoly": "3DLowPoly", "3DPSX": "3DPSX"} + + # Source hints + print("\nConfiguring asset source detection...") + default_hints = { + "kenney": "Kenney", + "quaternius": "Quaternius", + "kaykit": "KayKit", + "kits": "KayKit", + "custom": "Custom", + "zappypixel": "Zappypixel", + "polyhaven": "PolyHaven", + } + + # Write config + _CONFIG_DIR.mkdir(parents=True, exist_ok=True) + + style_map_toml = "\n".join(f'"{k}" = "{v}"' for k, v in style_map.items()) + source_hints_toml = "\n".join(f'{k} = "{v}"' for k, v in default_hints.items()) + + config_content = textwrap.dedent(f"""\ + # game-asset-mcp configuration + # Edit to customize your asset library setup. + # Full docs: https://github.com/jbcom/agentic/tree/main/packages/game-asset-mcp + + [library] + assets_root = "{root}" + catalog_db = "{db_str}" + + [taxonomy.style_map] + # Maps top-level directory prefix → style label + {style_map_toml} + + [taxonomy.source_hints] + # Maps lowercase path keywords → source name + {source_hints_toml} + + [taxonomy.skip_dirs] + # Directory names to skip during scan + dirs = ["_Archive", "__pycache__", ".git", "node_modules", "Textures", "textures"] + """) + + _CONFIG_FILE.write_text(config_content) + print(f"\n✓ Config written to {_CONFIG_FILE}") + print(f" assets_root = {root}") + print(f" catalog_db = {db_str}") + print(f" styles = {list(style_map.keys())}") + + # Offer to run ingest + if not non_interactive: + do_ingest = _ask_bool("\nRun initial ingest now? (scans all GLBs — may take a minute)", True) + else: + do_ingest = True + + if do_ingest: + print("\nRunning ingest...") + try: + subprocess.run( + [sys.executable, "-m", "game_asset_mcp.ingest"], + env={**__import__("os").environ, "ASSETS_ROOT": str(root), "CATALOG_DB": db_str}, + check=True, + ) + except subprocess.CalledProcessError as exc: + print(f"\n⚠ Ingest failed (exit {exc.returncode}). Run manually: game-asset-ingest") + else: + print("\nSkipped. Run later: game-asset-ingest") + + print("\n✓ Setup complete! Add to Claude Code:") + print(f' claude mcp add game-asset-library -e ASSETS_ROOT="{root}" -- game-asset-mcp\n') + + +def main() -> None: + parser = argparse.ArgumentParser( + description="Interactive setup wizard for game-asset-mcp", + formatter_class=argparse.RawDescriptionHelpFormatter, + epilog=textwrap.dedent("""\ + Examples: + game-asset-init # interactive + game-asset-init --root /mnt/assets # pre-fill root, interactive for rest + game-asset-init --root /mnt/assets --yes # non-interactive, accept defaults + """), + ) + parser.add_argument("--root", help="Asset library root directory") + parser.add_argument( + "--yes", "-y", action="store_true", + help="Non-interactive: accept all defaults (good for CI/Docker)", + ) + args = parser.parse_args() + run_wizard(non_interactive=args.yes, assets_root=args.root) + + +if __name__ == "__main__": + main() diff --git a/packages/game-asset-mcp/tests/__init__.py b/packages/game-asset-mcp/tests/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/packages/game-asset-mcp/tests/conftest.py b/packages/game-asset-mcp/tests/conftest.py new file mode 100644 index 00000000..b36aeaf6 --- /dev/null +++ b/packages/game-asset-mcp/tests/conftest.py @@ -0,0 +1,161 @@ +"""Shared fixtures for game-asset-mcp tests.""" +from __future__ import annotations + +import struct +import json +from pathlib import Path + +import pytest + +from game_asset_mcp.catalog import init_db, get_connection, upsert_asset, rebuild_fts +from game_asset_mcp.config import reset_settings + + +def _make_glb_bytes(gltf_dict: dict | None = None) -> bytes: + """Build a minimal valid GLB binary from a GLTF dict.""" + if gltf_dict is None: + gltf_dict = {"asset": {"version": "2.0"}} + + raw_json = json.dumps(gltf_dict).encode("utf-8") + # GLB JSON chunk must be 4-byte aligned, padded with spaces + pad = (4 - len(raw_json) % 4) % 4 + raw_json += b" " * pad + + # Header: magic + version + total_length + json_chunk_len = len(raw_json) + total_length = 12 + 8 + json_chunk_len # header + chunk_header + chunk_data + header = struct.pack(" bytes: + """Return a minimal valid GLB binary with no meshes/materials.""" + return _make_glb_bytes({"asset": {"version": "2.0"}}) + + +@pytest.fixture +def rich_glb_bytes() -> bytes: + """Return a GLB binary with mesh, material, and skin data.""" + gltf = { + "asset": {"version": "2.0"}, + "meshes": [ + { + "name": "Cube", + "primitives": [ + { + "attributes": {"POSITION": 0}, + "indices": 1, + } + ], + } + ], + "accessors": [ + {"count": 8, "componentType": 5126, "type": "VEC3"}, # POSITION: 8 verts + {"count": 36, "componentType": 5123, "type": "SCALAR"}, # indices: 36 → 12 tris + ], + "materials": [{"name": "Mat"}], + "images": [{"bufferView": 0}], # embedded texture + "skins": [{"name": "Armature", "joints": [0]}], + "animations": [{"name": "Walk", "channels": [], "samplers": []}], + } + return _make_glb_bytes(gltf) + + +@pytest.fixture +def tmp_db(tmp_path: Path) -> Path: + """Create and initialise a fresh SQLite catalog DB in a temp dir.""" + db = tmp_path / "test_catalog.db" + init_db(db) + return db + + +@pytest.fixture +def tmp_assets_root(tmp_path: Path) -> Path: + """Create a minimal taxonomy directory tree with fake GLB files.""" + root = tmp_path / "assets" + + # 3DLowPoly / Characters / Animated / kenney_pack / character.glb + char_dir = root / "3DLowPoly" / "Characters" / "Animated" / "kenney_pack" + char_dir.mkdir(parents=True) + glb_bytes = _make_glb_bytes({"asset": {"version": "2.0"}}) + (char_dir / "character.glb").write_bytes(glb_bytes) + + # 3DPSX / Props / Tools / blender_pack / tool.glb + psx_dir = root / "3DPSX" / "Props" / "Tools" / "blender_pack" + psx_dir.mkdir(parents=True) + (psx_dir / "tool.glb").write_bytes(glb_bytes) + + # 3DLowPoly / _Archive / archived.glb (should be skipped) + archive_dir = root / "3DLowPoly" / "_Archive" + archive_dir.mkdir(parents=True) + (archive_dir / "archived.glb").write_bytes(glb_bytes) + + return root + + +@pytest.fixture +def populated_db(tmp_db: Path) -> Path: + """Populate the DB with two sample asset records for search tests.""" + conn = get_connection(tmp_db) + + _sample_asset = dict( + path="/fake/3DLowPoly/Characters/Animated/kenney_pack/character.glb", + name="character", + style="3DLowPoly", + category="Characters/Animated/kenney_pack", + pack="kenney_pack", + source="Kenney", + meshes=1, + vertices=8, + faces=12, + materials=1, + textures=0, + has_embedded_textures=0, + has_armature=1, + animations=2, + extensions=[], + file_size_kb=4, + preview_path=None, + tags="character animated kenney pack", + ) + upsert_asset(conn, _sample_asset) + + _sample_asset2 = dict( + path="/fake/3DPSX/Props/Tools/blender_pack/tool.glb", + name="tool", + style="3DPSX", + category="Props/Tools/blender_pack", + pack="blender_pack", + source="Custom", + meshes=1, + vertices=24, + faces=12, + materials=1, + textures=1, + has_embedded_textures=1, + has_armature=0, + animations=0, + extensions=[], + file_size_kb=8, + preview_path=None, + tags="tool psx blender pack", + ) + upsert_asset(conn, _sample_asset2) + + rebuild_fts(conn) + conn.commit() + conn.close() + return tmp_db + + +@pytest.fixture(autouse=True) +def reset_settings_cache(): + """Reset the settings LRU cache before and after every test.""" + reset_settings() + yield + reset_settings() diff --git a/packages/game-asset-mcp/tests/e2e/__init__.py b/packages/game-asset-mcp/tests/e2e/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/packages/game-asset-mcp/tests/e2e/conftest.py b/packages/game-asset-mcp/tests/e2e/conftest.py new file mode 100644 index 00000000..a366dbbc --- /dev/null +++ b/packages/game-asset-mcp/tests/e2e/conftest.py @@ -0,0 +1,38 @@ +"""E2E test fixtures — require a real ASSETS_ROOT to be set.""" +from __future__ import annotations + +import os +from pathlib import Path + +import pytest + + +def pytest_configure(config): + """Register the e2e marker.""" + config.addinivalue_line( + "markers", + "e2e: end-to-end tests that require a real asset library (ASSETS_ROOT must be set)", + ) + + +@pytest.fixture(scope="session") +def real_assets_root() -> Path: + """ + Return the real asset library root from ASSETS_ROOT env var. + + Skips the test session if ASSETS_ROOT is not set or doesn't exist. + """ + root_str = os.environ.get("ASSETS_ROOT") or os.environ.get("GAME_ASSET_ASSETS_ROOT") + if not root_str: + pytest.skip("ASSETS_ROOT not set — skipping e2e tests") + root = Path(root_str) + if not root.exists(): + pytest.skip(f"ASSETS_ROOT does not exist: {root}") + return root + + +@pytest.fixture(scope="session") +def e2e_db(tmp_path_factory: pytest.TempPathFactory) -> Path: + """Provide a session-scoped temporary DB for e2e ingest tests.""" + db = tmp_path_factory.mktemp("e2e_db") / "e2e_catalog.db" + return db diff --git a/packages/game-asset-mcp/tests/e2e/test_ingest_e2e.py b/packages/game-asset-mcp/tests/e2e/test_ingest_e2e.py new file mode 100644 index 00000000..658f0148 --- /dev/null +++ b/packages/game-asset-mcp/tests/e2e/test_ingest_e2e.py @@ -0,0 +1,51 @@ +"""E2E ingest tests against the real ASSETS_ROOT (skipped if not set).""" +from __future__ import annotations + +from pathlib import Path + +import pytest + +from game_asset_mcp.catalog import get_connection, get_stats, list_categories +from game_asset_mcp.ingest import ingest, scan_glbs + + +@pytest.mark.e2e +class TestRealIngest: + def test_scan_finds_glbs(self, real_assets_root: Path) -> None: + """scan_glbs on the real asset root should find at least one GLB.""" + glbs = scan_glbs(real_assets_root) + assert len(glbs) > 0, "No GLBs found under ASSETS_ROOT" + + def test_ingest_dry_run_reports_assets( + self, real_assets_root: Path, e2e_db: Path + ) -> None: + """Dry-run ingest should count assets without writing to the DB.""" + from game_asset_mcp.catalog import init_db + init_db(e2e_db) + result = ingest(root=real_assets_root, db_path=e2e_db, dry_run=True) + assert result["total_scanned"] > 0 + # No records written + conn = get_connection(e2e_db) + stats = get_stats(conn) + conn.close() + assert stats["total"] == 0 + + def test_ingest_writes_records( + self, real_assets_root: Path, e2e_db: Path + ) -> None: + """Full ingest should populate the DB with asset records.""" + result = ingest(root=real_assets_root, db_path=e2e_db) + assert result["added"] > 0 or result["updated"] > 0 + conn = get_connection(e2e_db) + stats = get_stats(conn) + cats = list_categories(conn) + conn.close() + assert stats["total"] > 0 + assert len(cats) > 0 + + def test_second_ingest_skips_unchanged( + self, real_assets_root: Path, e2e_db: Path + ) -> None: + """Second ingest with no file changes should skip all previously indexed assets.""" + result = ingest(root=real_assets_root, db_path=e2e_db) + assert result["skipped"] >= result["total_scanned"] - result["errors"] diff --git a/packages/game-asset-mcp/tests/test_catalog.py b/packages/game-asset-mcp/tests/test_catalog.py new file mode 100644 index 00000000..9a3747a6 --- /dev/null +++ b/packages/game-asset-mcp/tests/test_catalog.py @@ -0,0 +1,276 @@ +"""Full catalog CRUD and FTS search tests.""" +from __future__ import annotations + +from pathlib import Path + +import pytest + +from game_asset_mcp.catalog import ( + get_connection, + get_stats, + init_db, + list_categories, + rebuild_fts, + search, + upsert_asset, + get_asset, +) + + +def _make_asset(**overrides) -> dict: + """Build a minimal asset dict, with optional field overrides.""" + base = dict( + path="/tmp/test/3DLowPoly/Characters/hero.glb", + name="hero", + style="3DLowPoly", + category="Characters", + pack="hero_pack", + source="Kenney", + meshes=1, + vertices=8, + faces=4, + materials=1, + textures=0, + has_embedded_textures=0, + has_armature=0, + animations=0, + extensions=[], + file_size_kb=2, + preview_path=None, + tags="hero character", + ) + base.update(overrides) + return base + + +class TestInitDb: + def test_creates_tables(self, tmp_db: Path) -> None: + """init_db should create assets, ingest_log and assets_fts tables.""" + conn = get_connection(tmp_db) + tables = { + row[0] + for row in conn.execute( + "SELECT name FROM sqlite_master WHERE type IN ('table', 'shadow')" + ).fetchall() + } + conn.close() + assert "assets" in tables + assert "ingest_log" in tables + + def test_idempotent(self, tmp_db: Path) -> None: + """Calling init_db twice should not raise.""" + init_db(tmp_db) # second call + conn = get_connection(tmp_db) + count = conn.execute("SELECT COUNT(*) FROM assets").fetchone()[0] + conn.close() + assert count == 0 + + +class TestUpsertAsset: + def test_insert_returns_row_id(self, tmp_db: Path) -> None: + """upsert_asset should return a positive integer row ID.""" + conn = get_connection(tmp_db) + row_id = upsert_asset(conn, _make_asset()) + conn.commit() + conn.close() + assert isinstance(row_id, int) + assert row_id >= 1 + + def test_duplicate_path_updates(self, tmp_db: Path) -> None: + """Inserting the same path twice should update, not duplicate.""" + conn = get_connection(tmp_db) + asset = _make_asset() + upsert_asset(conn, asset) + conn.commit() + + updated = _make_asset(vertices=999, faces=333) + upsert_asset(conn, updated) + conn.commit() + + count = conn.execute("SELECT COUNT(*) FROM assets").fetchone()[0] + row = conn.execute("SELECT vertices, faces FROM assets").fetchone() + conn.close() + assert count == 1 + assert row["vertices"] == 999 + assert row["faces"] == 333 + + def test_extensions_serialised_as_json(self, tmp_db: Path) -> None: + """extensions list should be stored as a JSON string.""" + conn = get_connection(tmp_db) + upsert_asset(conn, _make_asset(extensions=["KHR_draco_mesh_compression"])) + conn.commit() + row = conn.execute("SELECT extensions FROM assets").fetchone() + conn.close() + import json + exts = json.loads(row["extensions"]) + assert "KHR_draco_mesh_compression" in exts + + def test_ingested_at_auto_set(self, tmp_db: Path) -> None: + """ingested_at should be set automatically if not provided.""" + asset = _make_asset() + asset.pop("ingested_at", None) # ensure it's absent + conn = get_connection(tmp_db) + upsert_asset(conn, asset) + conn.commit() + row = conn.execute("SELECT ingested_at FROM assets").fetchone() + conn.close() + assert row["ingested_at"] is not None + assert row["ingested_at"] > 0 + + +class TestGetAsset: + def test_returns_dict_for_known_path(self, populated_db: Path) -> None: + """get_asset should return a dict for a path that exists in the DB.""" + conn = get_connection(populated_db) + asset = get_asset(conn, "/fake/3DLowPoly/Characters/Animated/kenney_pack/character.glb") + conn.close() + assert asset is not None + assert asset["name"] == "character" + + def test_returns_none_for_unknown_path(self, populated_db: Path) -> None: + """get_asset should return None for a path not in the DB.""" + conn = get_connection(populated_db) + result = get_asset(conn, "/nonexistent/path.glb") + conn.close() + assert result is None + + +class TestSearch: + def test_empty_query_returns_all(self, populated_db: Path) -> None: + """Empty query should return all assets (up to max_results).""" + conn = get_connection(populated_db) + results = search(conn, query="", max_results=10) + conn.close() + assert len(results) == 2 + + def test_fts_query_matches_name(self, populated_db: Path) -> None: + """FTS query for 'character' should return the character asset.""" + conn = get_connection(populated_db) + results = search(conn, query="character") + conn.close() + names = [r["name"] for r in results] + assert "character" in names + + def test_fts_query_matches_tags(self, populated_db: Path) -> None: + """FTS query on a tag token should match the relevant asset.""" + conn = get_connection(populated_db) + results = search(conn, query="kenney") + conn.close() + assert len(results) >= 1 + assert results[0]["name"] == "character" + + def test_style_filter(self, populated_db: Path) -> None: + """Filtering by style='3DPSX' should return only PSX assets.""" + conn = get_connection(populated_db) + results = search(conn, query="", style="3DPSX", max_results=10) + conn.close() + assert all(r["style"] == "3DPSX" for r in results) + assert len(results) == 1 + + def test_has_armature_filter(self, populated_db: Path) -> None: + """has_armature=True filter should return only rigged assets.""" + conn = get_connection(populated_db) + results = search(conn, query="", has_armature=True, max_results=10) + conn.close() + assert all(r["has_armature"] == 1 for r in results) + + def test_has_textures_filter(self, populated_db: Path) -> None: + """has_textures=True filter should return only textured assets.""" + conn = get_connection(populated_db) + results = search(conn, query="", has_textures=True, max_results=10) + conn.close() + assert all(r["has_embedded_textures"] == 1 for r in results) + + def test_category_filter(self, populated_db: Path) -> None: + """Category LIKE filter should narrow results.""" + conn = get_connection(populated_db) + results = search(conn, query="", category="Props", max_results=10) + conn.close() + assert len(results) == 1 + assert results[0]["name"] == "tool" + + def test_max_results_respected(self, populated_db: Path) -> None: + """max_results=1 should return at most 1 record.""" + conn = get_connection(populated_db) + results = search(conn, query="", max_results=1) + conn.close() + assert len(results) <= 1 + + def test_no_match_returns_empty(self, populated_db: Path) -> None: + """FTS query with no matching token should return empty list.""" + conn = get_connection(populated_db) + results = search(conn, query="zzznomatch") + conn.close() + assert results == [] + + +class TestListCategories: + def test_returns_list_of_dicts(self, populated_db: Path) -> None: + """list_categories should return a non-empty list of dicts.""" + conn = get_connection(populated_db) + cats = list_categories(conn) + conn.close() + assert isinstance(cats, list) + assert len(cats) >= 1 + + def test_dict_has_required_keys(self, populated_db: Path) -> None: + """Each category dict must contain style, category, and count keys.""" + conn = get_connection(populated_db) + cats = list_categories(conn) + conn.close() + for cat in cats: + assert "style" in cat + assert "category" in cat + assert "count" in cat + + def test_count_is_positive(self, populated_db: Path) -> None: + """Every category count should be a positive integer.""" + conn = get_connection(populated_db) + cats = list_categories(conn) + conn.close() + assert all(cat["count"] > 0 for cat in cats) + + +class TestGetStats: + def test_returns_required_keys(self, populated_db: Path) -> None: + """get_stats should return total, with_textures, with_armature, with_previews.""" + conn = get_connection(populated_db) + stats = get_stats(conn) + conn.close() + assert "total" in stats + assert "with_textures" in stats + assert "with_armature" in stats + assert "with_previews" in stats + + def test_total_matches_inserted(self, populated_db: Path) -> None: + """Total count should equal the number of inserted records.""" + conn = get_connection(populated_db) + stats = get_stats(conn) + conn.close() + assert stats["total"] == 2 + + def test_empty_db_stats(self, tmp_db: Path) -> None: + """Stats on an empty DB should return zeros.""" + conn = get_connection(tmp_db) + stats = get_stats(conn) + conn.close() + assert stats["total"] == 0 + + +class TestRebuildFts: + def test_rebuild_does_not_raise(self, populated_db: Path) -> None: + """rebuild_fts should succeed without raising exceptions.""" + conn = get_connection(populated_db) + # Should not raise + rebuild_fts(conn) + conn.commit() + conn.close() + + def test_rebuild_preserves_searchability(self, populated_db: Path) -> None: + """After a rebuild, FTS search should still find indexed records.""" + conn = get_connection(populated_db) + rebuild_fts(conn) + conn.commit() + results = search(conn, query="character") + conn.close() + assert len(results) >= 1 diff --git a/packages/game-asset-mcp/tests/test_config.py b/packages/game-asset-mcp/tests/test_config.py new file mode 100644 index 00000000..f7a6e850 --- /dev/null +++ b/packages/game-asset-mcp/tests/test_config.py @@ -0,0 +1,53 @@ +"""Basic smoke tests for game-asset-mcp config and catalog.""" + +import pytest +from pathlib import Path +from unittest.mock import patch + +from game_asset_mcp.config import Settings, TaxonomyConfig, reset_settings + + +def test_settings_defaults(): + s = Settings() + assert s.assets_root == Path.home() / "assets" + assert "game-asset-mcp" in str(s.catalog_db) + + +def test_settings_env_override(monkeypatch): + monkeypatch.setenv("GAME_ASSET_ASSETS_ROOT", "/tmp/test-assets") + reset_settings() + s = Settings() + assert s.assets_root == Path("/tmp/test-assets") + reset_settings() + + +def test_taxonomy_defaults(): + t = TaxonomyConfig() + assert "3DLowPoly" in t.style_map + assert "3DPSX" in t.style_map + assert "kenney" in t.source_hints + assert "_Archive" in t.skip_dirs + + +def test_taxonomy_custom(): + t = TaxonomyConfig( + style_map={"MyStyle": "MyStyle"}, + source_hints={"mycreator": "MyCreator"}, + ) + assert "MyStyle" in t.style_map + assert "3DLowPoly" not in t.style_map + assert "mycreator" in t.source_hints + + +def test_server_imports(): + """Verify server module imports without error.""" + from game_asset_mcp.server import mcp + assert mcp is not None + + +def test_ingest_options_defaults(): + from game_asset_mcp.ingest import IngestOptions + opts = IngestOptions() + assert opts.force is False + assert opts.dry_run is False + assert opts.verbose is False diff --git a/packages/game-asset-mcp/tests/test_glb_reader.py b/packages/game-asset-mcp/tests/test_glb_reader.py new file mode 100644 index 00000000..8e1de9d8 --- /dev/null +++ b/packages/game-asset-mcp/tests/test_glb_reader.py @@ -0,0 +1,172 @@ +"""GLB binary parsing tests using synthetic GLB bytes.""" +from __future__ import annotations + +import struct +import json +from pathlib import Path + +import pytest + +from game_asset_mcp.glb_reader import read_glb_stats, is_valid_glb + + +def _write_glb(path: Path, gltf_dict: dict) -> Path: + """Write a minimal GLB file to disk and return the path.""" + raw_json = json.dumps(gltf_dict).encode("utf-8") + # Pad to 4-byte alignment + pad = (4 - len(raw_json) % 4) % 4 + raw_json += b" " * pad + + json_chunk_len = len(raw_json) + total_length = 12 + 8 + json_chunk_len + header = struct.pack(" None: + """read_glb_stats on a minimal valid GLB should return a stats dict.""" + glb_file = tmp_path / "minimal.glb" + glb_file.write_bytes(minimal_glb_bytes) + result = read_glb_stats(str(glb_file)) + assert result is not None + assert isinstance(result, dict) + + def test_minimal_glb_has_zero_meshes(self, tmp_path: Path, minimal_glb_bytes: bytes) -> None: + """Minimal GLB with no meshes should report meshes=0.""" + glb_file = tmp_path / "minimal.glb" + glb_file.write_bytes(minimal_glb_bytes) + result = read_glb_stats(str(glb_file)) + assert result is not None + assert result["meshes"] == 0 + assert result["vertices"] == 0 + assert result["faces"] == 0 + + def test_rich_glb_mesh_stats(self, tmp_path: Path, rich_glb_bytes: bytes) -> None: + """GLB with one mesh (8 verts, 36 indices) should report correct stats.""" + glb_file = tmp_path / "rich.glb" + glb_file.write_bytes(rich_glb_bytes) + result = read_glb_stats(str(glb_file)) + assert result is not None + assert result["meshes"] == 1 + assert result["vertices"] == 8 + assert result["faces"] == 12 # 36 indices / 3 + + def test_rich_glb_has_armature(self, tmp_path: Path, rich_glb_bytes: bytes) -> None: + """GLB with skins should report has_armature=True.""" + glb_file = tmp_path / "rich.glb" + glb_file.write_bytes(rich_glb_bytes) + result = read_glb_stats(str(glb_file)) + assert result is not None + assert result["has_armature"] is True + + def test_rich_glb_has_embedded_textures(self, tmp_path: Path, rich_glb_bytes: bytes) -> None: + """GLB with images having bufferView should report has_embedded_textures=True.""" + glb_file = tmp_path / "rich.glb" + glb_file.write_bytes(rich_glb_bytes) + result = read_glb_stats(str(glb_file)) + assert result is not None + assert result["has_embedded_textures"] is True + + def test_rich_glb_animations_count(self, tmp_path: Path, rich_glb_bytes: bytes) -> None: + """GLB with one animation block should report animations=1.""" + glb_file = tmp_path / "rich.glb" + glb_file.write_bytes(rich_glb_bytes) + result = read_glb_stats(str(glb_file)) + assert result is not None + assert result["animations"] == 1 + + def test_materials_count(self, tmp_path: Path) -> None: + """GLB with 2 materials should report materials=2.""" + gltf = { + "asset": {"version": "2.0"}, + "materials": [{"name": "A"}, {"name": "B"}], + } + glb_file = _write_glb(tmp_path / "two_mats.glb", gltf) + result = read_glb_stats(str(glb_file)) + assert result is not None + assert result["materials"] == 2 + + def test_file_size_kb_is_int(self, tmp_path: Path, minimal_glb_bytes: bytes) -> None: + """file_size_kb should be a non-negative integer.""" + glb_file = tmp_path / "minimal.glb" + glb_file.write_bytes(minimal_glb_bytes) + result = read_glb_stats(str(glb_file)) + assert result is not None + assert isinstance(result["file_size_kb"], int) + assert result["file_size_kb"] >= 0 + + def test_extensions_are_listed(self, tmp_path: Path) -> None: + """extensionsUsed in GLTF JSON should appear in the result.""" + gltf = { + "asset": {"version": "2.0"}, + "extensionsUsed": ["KHR_draco_mesh_compression"], + } + glb_file = _write_glb(tmp_path / "ext.glb", gltf) + result = read_glb_stats(str(glb_file)) + assert result is not None + assert "KHR_draco_mesh_compression" in result["extensions"] + + def test_invalid_magic_returns_none(self, tmp_path: Path) -> None: + """File with wrong magic bytes should return None.""" + bad = tmp_path / "bad.glb" + bad.write_bytes(b"\x00\x00\x00\x00" * 10) + result = read_glb_stats(str(bad)) + assert result is None + + def test_truncated_file_returns_none(self, tmp_path: Path) -> None: + """File that is too short to be a valid GLB should return None.""" + short = tmp_path / "short.glb" + short.write_bytes(b"glTF\x02\x00") # only 6 bytes + result = read_glb_stats(str(short)) + assert result is None + + def test_nonexistent_file_returns_none(self, tmp_path: Path) -> None: + """Non-existent file path should return None.""" + result = read_glb_stats(str(tmp_path / "no_such_file.glb")) + assert result is None + + def test_no_embedded_textures_when_no_buffer_view(self, tmp_path: Path) -> None: + """Image with URI (no bufferView) should not set has_embedded_textures.""" + gltf = { + "asset": {"version": "2.0"}, + "images": [{"uri": "texture.png"}], + } + glb_file = _write_glb(tmp_path / "ext_tex.glb", gltf) + result = read_glb_stats(str(glb_file)) + assert result is not None + assert result["has_embedded_textures"] is False + + def test_result_has_all_expected_keys(self, tmp_path: Path, minimal_glb_bytes: bytes) -> None: + """Result dict should contain all documented keys.""" + glb_file = tmp_path / "minimal.glb" + glb_file.write_bytes(minimal_glb_bytes) + result = read_glb_stats(str(glb_file)) + assert result is not None + for key in ( + "meshes", "primitives", "vertices", "faces", "materials", + "textures", "has_embedded_textures", "has_armature", "animations", + "extensions", "file_size_kb", + ): + assert key in result, f"Missing key: {key}" + + +class TestIsValidGlb: + def test_valid_glb_returns_true(self, tmp_path: Path, minimal_glb_bytes: bytes) -> None: + """is_valid_glb should return True for a file starting with glTF magic.""" + glb_file = tmp_path / "valid.glb" + glb_file.write_bytes(minimal_glb_bytes) + assert is_valid_glb(str(glb_file)) is True + + def test_invalid_file_returns_false(self, tmp_path: Path) -> None: + """is_valid_glb should return False for a non-GLB file.""" + bad = tmp_path / "bad.bin" + bad.write_bytes(b"\xFF\xFE\xFD\xFC" + b"\x00" * 8) + assert is_valid_glb(str(bad)) is False + + def test_nonexistent_returns_false(self, tmp_path: Path) -> None: + """is_valid_glb should return False for a missing file.""" + assert is_valid_glb(str(tmp_path / "ghost.glb")) is False diff --git a/packages/game-asset-mcp/tests/test_ingest.py b/packages/game-asset-mcp/tests/test_ingest.py new file mode 100644 index 00000000..92951fdf --- /dev/null +++ b/packages/game-asset-mcp/tests/test_ingest.py @@ -0,0 +1,231 @@ +"""Taxonomy detection and ingest() integration tests.""" +from __future__ import annotations + +from pathlib import Path + +import pytest + +from game_asset_mcp.catalog import get_connection, get_stats +from game_asset_mcp.ingest import ( + IngestOptions, + derive_tags, + detect_category, + detect_pack, + detect_source, + detect_style, + ingest, + scan_glbs, +) + + +class TestDeriveTags: + def test_basic_split(self) -> None: + """Name tokens should appear in tags.""" + tags = derive_tags("hero_warrior", "Characters/Animated", "kenney_pack") + assert "hero" in tags + assert "warrior" in tags + + def test_category_parts_included(self) -> None: + """Category path parts should be included in tags.""" + tags = derive_tags("sword", "Props/Weapons", "custom_pack") + assert "props" in tags + assert "weapons" in tags + + def test_pack_parts_included(self) -> None: + """Pack name parts should be included in tags.""" + tags = derive_tags("tree", "Environment/Nature", "quaternius_nature_pack") + assert "quaternius" in tags + assert "nature" in tags + + def test_deduplication(self) -> None: + """Repeated tokens should appear only once.""" + tags = derive_tags("nature", "Environment/Nature", "nature_kit") + parts = tags.split() + assert parts.count("nature") == 1 + + def test_short_tokens_excluded(self) -> None: + """Single-character tokens should be excluded from tags.""" + tags = derive_tags("a_b_c", "X/Y", "z_q") + parts = tags.split() + for p in parts: + assert len(p) > 1 + + def test_returns_string(self) -> None: + """derive_tags should return a str.""" + result = derive_tags("something", "A/B", "packname") + assert isinstance(result, str) + + +class TestDetectStyle: + def test_3dlowpoly_prefix(self) -> None: + """Path starting with '3DLowPoly' should return '3DLowPoly'.""" + assert detect_style("3DLowPoly/Characters/hero.glb") == "3DLowPoly" + + def test_3dpsx_prefix(self) -> None: + """Path starting with '3DPSX' should return '3DPSX'.""" + assert detect_style("3DPSX/Props/Tools/tool.glb") == "3DPSX" + + def test_unknown_prefix(self) -> None: + """Path not matching any known style should return 'Unknown'.""" + assert detect_style("Audio/music.mp3") == "Unknown" + + +class TestDetectCategory: + def test_extracts_middle_parts(self) -> None: + """Category should be everything between style dir and filename.""" + cat = detect_category("3DLowPoly/Characters/Animated/kenney_pack/hero.glb", "3DLowPoly") + assert cat == "Characters/Animated/kenney_pack" + + def test_shallow_path_returns_root(self) -> None: + """Very shallow paths should fall back to '/'.""" + cat = detect_category("3DLowPoly/hero.glb", "3DLowPoly") + assert cat == "/" + + def test_psx_path(self) -> None: + """PSX-style paths should also extract correctly.""" + cat = detect_category("3DPSX/Props/Tools/blender_pack/tool.glb", "3DPSX") + assert cat == "Props/Tools/blender_pack" + + +class TestDetectPack: + def test_returns_parent_dir(self) -> None: + """Pack should be the immediate parent directory of the GLB.""" + assert detect_pack("3DLowPoly/Characters/kenney_pack/hero.glb") == "kenney_pack" + + def test_psx_pack(self) -> None: + """PSX paths should also extract the parent dir correctly.""" + assert detect_pack("3DPSX/Props/blender_pack/tool.glb") == "blender_pack" + + +class TestDetectSource: + def test_kenney_hint(self) -> None: + """Path parts containing 'kenney' should return 'Kenney'.""" + assert detect_source(["3DLowPoly", "Characters", "kenney_pack"]) == "Kenney" + + def test_quaternius_hint(self) -> None: + """Path parts containing 'quaternius' should return 'Quaternius'.""" + assert detect_source(["3DLowPoly", "Props", "Quaternius Nature Pack"]) == "Quaternius" + + def test_kaykit_hint(self) -> None: + """Path parts containing 'kaykit' should return 'KayKit'.""" + assert detect_source(["3DLowPoly", "Vehicles", "kaykit_pack"]) == "KayKit" + + def test_unknown_source(self) -> None: + """Path parts with no matching hint should return 'Unknown'.""" + assert detect_source(["3DLowPoly", "Misc", "random_pack"]) == "Unknown" + + +class TestScanGlbs: + def test_finds_glb_files(self, tmp_assets_root: Path) -> None: + """scan_glbs should find .glb files in the asset tree.""" + glbs = scan_glbs(tmp_assets_root) + names = [g.name for g in glbs] + assert "character.glb" in names + assert "tool.glb" in names + + def test_skips_archive_dirs(self, tmp_assets_root: Path) -> None: + """scan_glbs should skip _Archive and other skip_dirs.""" + glbs = scan_glbs(tmp_assets_root) + for glb in glbs: + assert "_Archive" not in glb.parts + + def test_returns_path_objects(self, tmp_assets_root: Path) -> None: + """scan_glbs should return a list of Path objects.""" + glbs = scan_glbs(tmp_assets_root) + assert all(isinstance(g, Path) for g in glbs) + + def test_empty_root_returns_empty(self, tmp_path: Path) -> None: + """scan_glbs on an empty directory should return an empty list.""" + empty = tmp_path / "empty" + empty.mkdir() + assert scan_glbs(empty) == [] + + +class TestIngest: + def test_basic_ingest_adds_assets(self, tmp_assets_root: Path, tmp_db: Path) -> None: + """ingest() should add GLB files to the catalog.""" + result = ingest(root=tmp_assets_root, db_path=tmp_db) + assert result["added"] >= 2 + assert result["errors"] == 0 + + def test_ingest_result_keys(self, tmp_assets_root: Path, tmp_db: Path) -> None: + """ingest() should return all expected summary keys.""" + result = ingest(root=tmp_assets_root, db_path=tmp_db) + for key in ("total_scanned", "added", "updated", "skipped", "removed", "errors"): + assert key in result + + def test_ingest_skips_unchanged_on_second_run(self, tmp_assets_root: Path, tmp_db: Path) -> None: + """Second ingest with unchanged files should skip all previously added.""" + ingest(root=tmp_assets_root, db_path=tmp_db) + result2 = ingest(root=tmp_assets_root, db_path=tmp_db) + # On second run with force=False, all files are skipped + assert result2["skipped"] >= 2 + assert result2["added"] == 0 + + def test_force_flag_reingest_all(self, tmp_assets_root: Path, tmp_db: Path) -> None: + """force=True should re-ingest all files even if unchanged.""" + ingest(root=tmp_assets_root, db_path=tmp_db) + result2 = ingest(root=tmp_assets_root, db_path=tmp_db, force=True) + # With force=True, no files are skipped + assert result2["skipped"] == 0 + assert result2["added"] + result2["updated"] >= 2 + + def test_dry_run_does_not_write(self, tmp_assets_root: Path, tmp_db: Path) -> None: + """dry_run=True should not persist any records to the DB.""" + result = ingest(root=tmp_assets_root, db_path=tmp_db, dry_run=True) + assert result["added"] >= 2 # counted as would-be-added + conn = get_connection(tmp_db) + stats = get_stats(conn) + conn.close() + assert stats["total"] == 0 # nothing actually written + + def test_ingest_detects_styles(self, tmp_assets_root: Path, tmp_db: Path) -> None: + """Ingested assets should have correct styles detected.""" + ingest(root=tmp_assets_root, db_path=tmp_db) + conn = get_connection(tmp_db) + rows = conn.execute("SELECT DISTINCT style FROM assets ORDER BY style").fetchall() + styles = {r["style"] for r in rows} + conn.close() + assert "3DLowPoly" in styles + assert "3DPSX" in styles + + def test_ingest_does_not_count_archive(self, tmp_assets_root: Path, tmp_db: Path) -> None: + """Files under _Archive should never appear in the catalog.""" + ingest(root=tmp_assets_root, db_path=tmp_db) + conn = get_connection(tmp_db) + rows = conn.execute("SELECT path FROM assets").fetchall() + conn.close() + for row in rows: + assert "_Archive" not in row["path"] + + def test_stale_records_removed(self, tmp_assets_root: Path, tmp_db: Path) -> None: + """Records whose files no longer exist should be pruned on re-ingest. + + Stale detection compares the `known` dict (populated from the DB when + force=False) against the set of GLBs found on disk. So the second + ingest must use force=False to populate `known`. + """ + ingest(root=tmp_assets_root, db_path=tmp_db) + # Delete one GLB from disk + char = tmp_assets_root / "3DLowPoly" / "Characters" / "Animated" / "kenney_pack" / "character.glb" + char.unlink() + # force=False so `known` is populated and stale entries are detected + result2 = ingest(root=tmp_assets_root, db_path=tmp_db, force=False) + assert result2["removed"] >= 1 + + +class TestIngestOptions: + def test_defaults(self) -> None: + """IngestOptions should have sensible defaults.""" + opts = IngestOptions() + assert opts.force is False + assert opts.dry_run is False + assert opts.verbose is False + + def test_custom_values(self, tmp_path: Path, tmp_db: Path) -> None: + """IngestOptions should accept custom root and db paths.""" + opts = IngestOptions(root=tmp_path, db=tmp_db, force=True, dry_run=True) + assert opts.root == tmp_path + assert opts.db == tmp_db + assert opts.force is True + assert opts.dry_run is True diff --git a/packages/game-asset-mcp/tests/test_polyhaven.py b/packages/game-asset-mcp/tests/test_polyhaven.py new file mode 100644 index 00000000..e8548950 --- /dev/null +++ b/packages/game-asset-mcp/tests/test_polyhaven.py @@ -0,0 +1,275 @@ +"""Mocked API tests for polyhaven.py.""" +from __future__ import annotations + +from pathlib import Path +from unittest.mock import MagicMock, patch + +import pytest + +from game_asset_mcp.polyhaven import ( + download_ph_asset, + get_taxonomy_path, + search_ph, + get_ph_info, +) + + +# ─── Helpers ────────────────────────────────────────────────────────────────── + +def _mock_response(data: dict, status_code: int = 200) -> MagicMock: + """Build a mock httpx.Response that returns `data` from .json().""" + resp = MagicMock() + resp.status_code = status_code + resp.json.return_value = data + resp.raise_for_status = MagicMock() + return resp + + +FAKE_ASSETS_RESPONSE = { + "oak_tree": { + "name": "Oak Tree", + "type": 0, + "categories": ["nature"], + "tags": ["tree", "oak", "plant"], + "download_count": 5000, + }, + "pine_tree": { + "name": "Pine Tree", + "type": 0, + "categories": ["nature"], + "tags": ["tree", "pine", "conifer"], + "download_count": 3000, + }, + "wooden_chair": { + "name": "Wooden Chair", + "type": 0, + "categories": ["furniture"], + "tags": ["chair", "wooden", "seat"], + "download_count": 1000, + }, +} + +FAKE_FILES_RESPONSE = { + "gltf": { + "1k": { + "glb": { + "url": "https://dl.polyhaven.com/oak_tree_1k.glb", + "size": 204800, + } + } + } +} + +FAKE_INFO_RESPONSE = { + "name": "Oak Tree", + "categories": ["nature"], + "tags": ["tree", "oak"], +} + + +# ─── Tests ──────────────────────────────────────────────────────────────────── + + +class TestSearchPh: + def test_returns_matching_assets(self) -> None: + """search_ph should return assets whose name/id/tags match the query.""" + with patch("game_asset_mcp.polyhaven.ph_get", return_value=FAKE_ASSETS_RESPONSE): + results = search_ph("tree", asset_type="models") + ids = [r["id"] for r in results] + assert "oak_tree" in ids + assert "pine_tree" in ids + + def test_filters_by_query_word(self) -> None: + """search_ph with 'chair' should not return tree assets.""" + with patch("game_asset_mcp.polyhaven.ph_get", return_value=FAKE_ASSETS_RESPONSE): + results = search_ph("chair", asset_type="models") + ids = [r["id"] for r in results] + assert "wooden_chair" in ids + assert "oak_tree" not in ids + + def test_sorted_by_download_count(self) -> None: + """Results should be sorted by download_count descending.""" + with patch("game_asset_mcp.polyhaven.ph_get", return_value=FAKE_ASSETS_RESPONSE): + results = search_ph("tree", asset_type="models") + counts = [r["download_count"] for r in results] + assert counts == sorted(counts, reverse=True) + + def test_limit_respected(self) -> None: + """limit parameter should cap results.""" + with patch("game_asset_mcp.polyhaven.ph_get", return_value=FAKE_ASSETS_RESPONSE): + results = search_ph("tree", asset_type="models", limit=1) + assert len(results) <= 1 + + def test_no_match_returns_empty(self) -> None: + """Query that matches nothing should return an empty list.""" + with patch("game_asset_mcp.polyhaven.ph_get", return_value=FAKE_ASSETS_RESPONSE): + results = search_ph("zzznomatch", asset_type="models") + assert results == [] + + def test_result_dict_has_required_keys(self) -> None: + """Each result dict must have id, name, type, categories, tags, download_count.""" + with patch("game_asset_mcp.polyhaven.ph_get", return_value=FAKE_ASSETS_RESPONSE): + results = search_ph("tree", asset_type="models") + for r in results: + for key in ("id", "name", "type", "categories", "tags", "download_count"): + assert key in r, f"Missing key '{key}' in result: {r}" + + def test_multi_word_query(self) -> None: + """Multi-word query should require all words to match.""" + with patch("game_asset_mcp.polyhaven.ph_get", return_value=FAKE_ASSETS_RESPONSE): + results = search_ph("oak tree", asset_type="models") + ids = [r["id"] for r in results] + assert "oak_tree" in ids + assert "pine_tree" not in ids + + +class TestGetTaxonomyPath: + def test_hdri_type_maps_to_2d_photorealistic(self, tmp_path: Path, monkeypatch) -> None: + """HDRI assets should be placed under 2DPhotorealistic/HDRIs/polyhaven.""" + monkeypatch.setattr("game_asset_mcp.polyhaven.ASSETS_ROOT", tmp_path) + path = get_taxonomy_path("sky_01", "hdris", []) + assert "2DPhotorealistic" in path.parts + assert "HDRIs" in path.parts + + def test_texture_type_maps_to_textures(self, tmp_path: Path, monkeypatch) -> None: + """Texture assets should be placed under 2DPhotorealistic/Textures/polyhaven.""" + monkeypatch.setattr("game_asset_mcp.polyhaven.ASSETS_ROOT", tmp_path) + path = get_taxonomy_path("wood_01", "textures", []) + assert "2DPhotorealistic" in path.parts + assert "Textures" in path.parts + + def test_nature_model_maps_correctly(self, tmp_path: Path, monkeypatch) -> None: + """Models with 'nature' category should go to 3DLowPoly/Environment/Nature/polyhaven.""" + monkeypatch.setattr("game_asset_mcp.polyhaven.ASSETS_ROOT", tmp_path) + path = get_taxonomy_path("oak_tree", "models", ["nature"]) + assert "Environment" in path.parts + assert "Nature" in path.parts + + def test_furniture_model_maps_correctly(self, tmp_path: Path, monkeypatch) -> None: + """Models with 'furniture' category should go to 3DLowPoly/Props/Furniture/polyhaven.""" + monkeypatch.setattr("game_asset_mcp.polyhaven.ASSETS_ROOT", tmp_path) + path = get_taxonomy_path("chair_01", "models", ["furniture"]) + assert "Furniture" in path.parts + + def test_unknown_category_uses_default(self, tmp_path: Path, monkeypatch) -> None: + """Models with unrecognised category should use the default path.""" + monkeypatch.setattr("game_asset_mcp.polyhaven.ASSETS_ROOT", tmp_path) + path = get_taxonomy_path("thing_01", "models", ["unknown_category"]) + assert "Misc" in path.parts + + def test_asset_id_is_leaf(self, tmp_path: Path, monkeypatch) -> None: + """The asset_id should be the leaf directory of the returned path.""" + monkeypatch.setattr("game_asset_mcp.polyhaven.ASSETS_ROOT", tmp_path) + path = get_taxonomy_path("oak_tree", "models", ["nature"]) + assert path.name == "oak_tree" + + +class TestDownloadPhAsset: + def test_model_download_success(self, tmp_path: Path, monkeypatch) -> None: + """Successful model download should return dest_dir, files, and categories.""" + monkeypatch.setattr("game_asset_mcp.polyhaven.ASSETS_ROOT", tmp_path) + + with ( + patch("game_asset_mcp.polyhaven.get_ph_info", return_value=FAKE_INFO_RESPONSE), + patch("game_asset_mcp.polyhaven.ph_get", return_value=FAKE_FILES_RESPONSE), + patch("game_asset_mcp.polyhaven._download_bytes", return_value=b"fake-glb-data"), + ): + result = download_ph_asset("oak_tree", asset_type="models", resolution="1k") + + assert "error" not in result + assert "dest_dir" in result + assert "files" in result + assert len(result["files"]) == 1 + assert result["files"][0].endswith(".glb") + + def test_model_download_writes_file(self, tmp_path: Path, monkeypatch) -> None: + """Downloaded GLB should actually exist on disk.""" + monkeypatch.setattr("game_asset_mcp.polyhaven.ASSETS_ROOT", tmp_path) + + with ( + patch("game_asset_mcp.polyhaven.get_ph_info", return_value=FAKE_INFO_RESPONSE), + patch("game_asset_mcp.polyhaven.ph_get", return_value=FAKE_FILES_RESPONSE), + patch("game_asset_mcp.polyhaven._download_bytes", return_value=b"fake-glb-data"), + ): + result = download_ph_asset("oak_tree", asset_type="models", resolution="1k") + + assert Path(result["files"][0]).exists() + + def test_info_fetch_failure_returns_error(self, tmp_path: Path, monkeypatch) -> None: + """HTTP error fetching asset info should return an error dict.""" + import httpx + + monkeypatch.setattr("game_asset_mcp.polyhaven.ASSETS_ROOT", tmp_path) + + with patch("game_asset_mcp.polyhaven.get_ph_info", side_effect=httpx.HTTPError("not found")): + result = download_ph_asset("bad_id", asset_type="models") + + assert "error" in result + + def test_no_gltf_section_returns_error(self, tmp_path: Path, monkeypatch) -> None: + """Files response with no gltf section should return an error dict.""" + monkeypatch.setattr("game_asset_mcp.polyhaven.ASSETS_ROOT", tmp_path) + + with ( + patch("game_asset_mcp.polyhaven.get_ph_info", return_value=FAKE_INFO_RESPONSE), + patch("game_asset_mcp.polyhaven.ph_get", return_value={}), + ): + result = download_ph_asset("oak_tree", asset_type="models") + + assert "error" in result + + def test_unknown_asset_type_returns_error(self, tmp_path: Path, monkeypatch) -> None: + """Unsupported asset_type should return an error dict.""" + monkeypatch.setattr("game_asset_mcp.polyhaven.ASSETS_ROOT", tmp_path) + + with ( + patch("game_asset_mcp.polyhaven.get_ph_info", return_value=FAKE_INFO_RESPONSE), + patch("game_asset_mcp.polyhaven.ph_get", return_value=FAKE_FILES_RESPONSE), + ): + result = download_ph_asset("oak_tree", asset_type="unknown_type") + + assert "error" in result + assert "unknown_type" in result["error"] + + def test_hdri_download(self, tmp_path: Path, monkeypatch) -> None: + """HDRI download should save an .hdr file.""" + monkeypatch.setattr("game_asset_mcp.polyhaven.ASSETS_ROOT", tmp_path) + fake_hdri_info = {"name": "Sky", "categories": []} + fake_hdri_files = { + "hdri": { + "1k": { + "hdr": {"url": "https://example.com/sky_1k.hdr", "size": 1024} + } + } + } + + with ( + patch("game_asset_mcp.polyhaven.get_ph_info", return_value=fake_hdri_info), + patch("game_asset_mcp.polyhaven.ph_get", return_value=fake_hdri_files), + patch("game_asset_mcp.polyhaven._download_bytes", return_value=b"hdr-data"), + ): + result = download_ph_asset("sky_01", asset_type="hdris", resolution="1k") + + assert "error" not in result + assert any(f.endswith(".hdr") for f in result["files"]) + + def test_texture_download(self, tmp_path: Path, monkeypatch) -> None: + """Texture download should save multiple map files.""" + monkeypatch.setattr("game_asset_mcp.polyhaven.ASSETS_ROOT", tmp_path) + fake_tex_info = {"name": "Brick Wall", "categories": []} + fake_tex_files = { + "1k": { + "diffuse": {"url": "https://example.com/brick_diff_1k.png"}, + "rough": {"url": "https://example.com/brick_rough_1k.png"}, + } + } + + with ( + patch("game_asset_mcp.polyhaven.get_ph_info", return_value=fake_tex_info), + patch("game_asset_mcp.polyhaven.ph_get", return_value=fake_tex_files), + patch("game_asset_mcp.polyhaven._download_bytes", return_value=b"img-data"), + ): + result = download_ph_asset("brick_wall", asset_type="textures", resolution="1k") + + assert "error" not in result + assert len(result["files"]) == 2 diff --git a/packages/game-asset-mcp/tests/test_server_tools.py b/packages/game-asset-mcp/tests/test_server_tools.py new file mode 100644 index 00000000..79cd33d5 --- /dev/null +++ b/packages/game-asset-mcp/tests/test_server_tools.py @@ -0,0 +1,356 @@ +"""Smoke tests for each MCP tool function in server.py. + +These tests call the tool functions directly (without the MCP runtime). +The catalog DB is routed to a tmp_path DB via monkeypatching. +""" +from __future__ import annotations + +import os +from pathlib import Path +from unittest.mock import MagicMock, patch + +import pytest + +import game_asset_mcp.server as server_module +from game_asset_mcp.catalog import ( + get_connection, + init_db, + rebuild_fts, + upsert_asset, +) +from game_asset_mcp.server import ( + browse_taxonomy, + copy_asset, + download_polyhaven_asset, + generate_preview, + get_asset_info, + get_catalog_stats, + get_preview, + list_categories, + run_ingest, + search_assets, + search_polyhaven, + _format_asset, +) + + +# ─── Helpers ────────────────────────────────────────────────────────────────── + + +def _patch_db(monkeypatch, db_path: Path) -> None: + """Redirect all server catalog calls to use db_path instead of the default.""" + from game_asset_mcp import catalog as cat_module + + def _get_conn(dp=None): + return get_connection(db_path) + + monkeypatch.setattr(server_module, "get_connection", _get_conn) + # Also patch catalog-level calls that accept no arguments from server.py + monkeypatch.setattr(cat_module, "DB_PATH", db_path) + + +def _insert_sample(db_path: Path, **overrides) -> None: + """Insert a single test asset record into db_path.""" + base = dict( + path="/fake/3DLowPoly/Characters/hero.glb", + name="hero", + style="3DLowPoly", + category="Characters/Animated", + pack="kenney_pack", + source="Kenney", + meshes=1, vertices=8, faces=4, + materials=1, textures=0, + has_embedded_textures=0, has_armature=1, animations=1, + extensions=[], file_size_kb=4, preview_path=None, + tags="hero character kenney", + ) + base.update(overrides) + conn = get_connection(db_path) + upsert_asset(conn, base) + rebuild_fts(conn) + conn.commit() + conn.close() + + +# ─── Tests ──────────────────────────────────────────────────────────────────── + + +class TestFormatAsset: + def test_contains_expected_keys(self) -> None: + """_format_asset should include path, name, style, faces, etc.""" + asset = dict( + path="/a/b.glb", name="b", style="3DLowPoly", category="X", + pack="p", source="S", faces=4, vertices=8, materials=1, + has_armature=1, has_embedded_textures=0, animations=0, + file_size_kb=2, preview_path=None, + ) + result = _format_asset(asset) + for key in ("path", "name", "style", "faces", "vertices", "has_armature", + "has_embedded_textures", "preview_path", "file_size_kb"): + assert key in result + + def test_has_armature_is_bool(self) -> None: + """_format_asset should convert has_armature to a Python bool.""" + asset = dict( + path="/a/b.glb", name="b", style="3DLowPoly", category="X", + pack="p", source="S", faces=4, vertices=8, materials=1, + has_armature=1, has_embedded_textures=0, animations=0, + file_size_kb=2, preview_path=None, + ) + result = _format_asset(asset) + assert isinstance(result["has_armature"], bool) + assert result["has_armature"] is True + + +class TestSearchAssets: + def test_empty_query_returns_list(self, populated_db: Path, monkeypatch) -> None: + """search_assets with empty query should return a list.""" + _patch_db(monkeypatch, populated_db) + results = search_assets(query="") + assert isinstance(results, list) + + def test_fts_query_returns_list(self, populated_db: Path, monkeypatch) -> None: + """search_assets with FTS query should return a list. + + hybrid_search has DB_PATH as a default arg (captured at import time), + so we patch the name in server.py's namespace directly. + """ + fake_results = [ + { + "path": "/fake/char.glb", "name": "character", "style": "3DLowPoly", + "category": "Characters", "pack": "k", "source": "Kenney", + "faces": 4, "vertices": 8, "materials": 1, "has_armature": 0, + "has_embedded_textures": 0, "animations": 0, "file_size_kb": 2, + "preview_path": None, + } + ] + monkeypatch.setattr(server_module, "hybrid_search", lambda **kw: fake_results) + results = search_assets(query="character") + assert isinstance(results, list) + assert len(results) == 1 + + def test_each_result_is_formatted(self, populated_db: Path, monkeypatch) -> None: + """Each result should have 'name' and 'path' from _format_asset.""" + _patch_db(monkeypatch, populated_db) + results = search_assets(query="") + for r in results: + assert "name" in r + assert "path" in r + + +class TestListCategories: + def test_returns_list(self, populated_db: Path, monkeypatch) -> None: + """list_categories should return a list of category dicts.""" + _patch_db(monkeypatch, populated_db) + result = list_categories() + assert isinstance(result, list) + + def test_style_filter(self, populated_db: Path, monkeypatch) -> None: + """list_categories(style='3DLowPoly') should only return 3DLowPoly entries.""" + _patch_db(monkeypatch, populated_db) + result = list_categories(style="3DLowPoly") + assert all(r["style"] == "3DLowPoly" for r in result) + + +class TestGetAssetInfo: + def test_existing_asset(self, populated_db: Path, monkeypatch) -> None: + """get_asset_info should return asset dict for a known path.""" + _patch_db(monkeypatch, populated_db) + info = get_asset_info("/fake/3DLowPoly/Characters/Animated/kenney_pack/character.glb") + assert "name" in info + assert info["name"] == "character" + + def test_unknown_path_not_on_disk(self, populated_db: Path, monkeypatch) -> None: + """get_asset_info should return error dict for missing path.""" + _patch_db(monkeypatch, populated_db) + result = get_asset_info("/totally/nonexistent/path.glb") + assert "error" in result + + def test_live_read_for_uncatalogued_glb( + self, tmp_path: Path, tmp_db: Path, minimal_glb_bytes: bytes, monkeypatch + ) -> None: + """get_asset_info on a real GLB not in catalog should return live stats.""" + glb_file = tmp_path / "uncatalogued.glb" + glb_file.write_bytes(minimal_glb_bytes) + _patch_db(monkeypatch, tmp_db) + result = get_asset_info(str(glb_file)) + assert "error" not in result + assert result.get("name") == "uncatalogued" + + +class TestCopyAsset: + def test_copies_file(self, tmp_path: Path, minimal_glb_bytes: bytes) -> None: + """copy_asset should copy the file to dest_dir.""" + src = tmp_path / "src" / "model.glb" + src.parent.mkdir() + src.write_bytes(minimal_glb_bytes) + dest_dir = tmp_path / "dest" + + result = copy_asset(str(src), str(dest_dir)) + assert "error" not in result + assert Path(result["dest"]).exists() + + def test_rename_option(self, tmp_path: Path, minimal_glb_bytes: bytes) -> None: + """copy_asset with rename should use the new filename.""" + src = tmp_path / "src" / "model.glb" + src.parent.mkdir() + src.write_bytes(minimal_glb_bytes) + dest_dir = tmp_path / "dest" + + result = copy_asset(str(src), str(dest_dir), rename="renamed_model") + assert Path(result["dest"]).name == "renamed_model.glb" + + def test_missing_source_returns_error(self, tmp_path: Path) -> None: + """copy_asset with non-existent source should return error dict.""" + result = copy_asset("/no/such/file.glb", str(tmp_path)) + assert "error" in result + + def test_creates_dest_dir(self, tmp_path: Path, minimal_glb_bytes: bytes) -> None: + """copy_asset should create the destination directory if it doesn't exist.""" + src = tmp_path / "model.glb" + src.write_bytes(minimal_glb_bytes) + deep_dest = tmp_path / "a" / "b" / "c" + + result = copy_asset(str(src), str(deep_dest)) + assert "error" not in result + assert deep_dest.exists() + + +class TestGetPreview: + def test_no_preview_returns_none(self, tmp_path: Path, tmp_db: Path, monkeypatch) -> None: + """get_preview for a GLB with no preview should return preview_path=None.""" + glb = tmp_path / "model.glb" + glb.write_bytes(b"fake") + _patch_db(monkeypatch, tmp_db) + result = get_preview(str(glb)) + assert result["preview_path"] is None + + def test_existing_preview_sibling(self, tmp_path: Path) -> None: + """get_preview should find a .png next to the GLB.""" + glb = tmp_path / "model.glb" + glb.write_bytes(b"fake") + png = tmp_path / "model.png" + png.write_bytes(b"PNG") + result = get_preview(str(glb)) + assert result["preview_path"] == str(png) + + +class TestGeneratePreview: + def test_missing_glb_returns_error(self, tmp_path: Path) -> None: + """generate_preview on a non-existent GLB should return error dict.""" + result = generate_preview("/no/such/file.glb") + assert "error" in result + + def test_no_blender_returns_error( + self, tmp_path: Path, minimal_glb_bytes: bytes, monkeypatch + ) -> None: + """generate_preview without Blender available should return error dict.""" + glb = tmp_path / "model.glb" + glb.write_bytes(minimal_glb_bytes) + monkeypatch.setattr(server_module, "HAS_BLENDER", False) + result = generate_preview(str(glb)) + assert "error" in result + + +class TestGetCatalogStats: + def test_returns_dict(self, populated_db: Path, monkeypatch) -> None: + """get_catalog_stats should return a dict with total key.""" + _patch_db(monkeypatch, populated_db) + stats = get_catalog_stats() + assert isinstance(stats, dict) + assert "total" in stats + + def test_by_style_present(self, populated_db: Path, monkeypatch) -> None: + """get_catalog_stats should include by_style breakdown.""" + _patch_db(monkeypatch, populated_db) + stats = get_catalog_stats() + assert "by_style" in stats + assert isinstance(stats["by_style"], dict) + + def test_total_is_two(self, populated_db: Path, monkeypatch) -> None: + """populated_db has 2 assets, so total should be 2.""" + _patch_db(monkeypatch, populated_db) + stats = get_catalog_stats() + assert stats["total"] == 2 + + +class TestBrowseTaxonomy: + def test_no_args_returns_meso_level(self, populated_db: Path, monkeypatch) -> None: + """browse_taxonomy() with no args should return meso-level entries.""" + _patch_db(monkeypatch, populated_db) + result = browse_taxonomy() + assert isinstance(result, list) + assert all(r["level"] == "meso" for r in result) + + def test_style_filter(self, populated_db: Path, monkeypatch) -> None: + """browse_taxonomy(style='3DLowPoly') should only return 3DLowPoly rows.""" + _patch_db(monkeypatch, populated_db) + result = browse_taxonomy(style="3DLowPoly") + assert all(r["style"] == "3DLowPoly" for r in result) + + def test_meso_filter_returns_micro_level(self, populated_db: Path, monkeypatch) -> None: + """browse_taxonomy(meso='Characters') should return micro-level entries.""" + _patch_db(monkeypatch, populated_db) + result = browse_taxonomy(meso="Characters") + assert isinstance(result, list) + if result: + assert result[0]["level"] == "micro" + + +class TestSearchPolyhaven: + def test_calls_search_ph(self) -> None: + """search_polyhaven should return what search_ph returns. + + server.py does `from .polyhaven import search_ph` inside the function body, + so we patch at the source module (game_asset_mcp.polyhaven.search_ph). + """ + fake_results = [{"id": "oak_tree", "name": "Oak Tree"}] + with patch("game_asset_mcp.polyhaven.search_ph", return_value=fake_results): + result = search_polyhaven("tree") + assert result == fake_results + + +class TestDownloadPolyhavenAsset: + def test_delegates_to_download_ph_asset(self) -> None: + """download_polyhaven_asset should call download_ph_asset and return result. + + server.py imports download_ph_asset and ingest inside the function body, + so we patch at the source modules. + """ + fake_dl_result = { + "dest_dir": "/tmp/fake", + "files": ["/tmp/fake/x.glb"], + "asset_type": "models", + "categories": [], + } + with ( + patch("game_asset_mcp.polyhaven.download_ph_asset", return_value=fake_dl_result), + patch("game_asset_mcp.ingest.ingest", return_value={"added": 1}), + ): + result = download_polyhaven_asset("oak_tree", asset_type="models") + assert "error" not in result + assert result["ingested"] is True + + def test_error_propagates(self) -> None: + """If download_ph_asset returns error, it should propagate unchanged.""" + with patch("game_asset_mcp.polyhaven.download_ph_asset", return_value={"error": "not found"}): + result = download_polyhaven_asset("bad_id") + assert "error" in result + + +class TestRunIngest: + def test_returns_summary_dict(self) -> None: + """run_ingest should return a dict with added/updated/skipped keys. + + server.py imports ingest inside the function body, so patch at the + source module (game_asset_mcp.ingest.ingest). + """ + fake_result = { + "total_scanned": 2, "added": 2, "updated": 0, + "skipped": 0, "removed": 0, "errors": 0, + } + with patch("game_asset_mcp.ingest.ingest", return_value=fake_result) as mock: + result = run_ingest(force=False) + assert "added" in result + assert result["added"] == 2 + mock.assert_called_once_with(force=False) diff --git a/packages/game-asset-mcp/tests/test_wizard.py b/packages/game-asset-mcp/tests/test_wizard.py new file mode 100644 index 00000000..d6081053 --- /dev/null +++ b/packages/game-asset-mcp/tests/test_wizard.py @@ -0,0 +1,157 @@ +"""Wizard smoke tests — non-interactive mode only.""" +from __future__ import annotations + +from pathlib import Path +from unittest.mock import patch + +import pytest + +from game_asset_mcp.wizard import _detect_styles, run_wizard + + +class TestDetectStyles: + def test_empty_root_returns_empty(self, tmp_path: Path) -> None: + """_detect_styles on an empty directory should return [].""" + result = _detect_styles(tmp_path) + assert result == [] + + def test_nonexistent_root_returns_empty(self, tmp_path: Path) -> None: + """_detect_styles on a non-existent directory should return [].""" + result = _detect_styles(tmp_path / "no_such_dir") + assert result == [] + + def test_detects_dir_with_glbs(self, tmp_path: Path, minimal_glb_bytes: bytes) -> None: + """_detect_styles should return dirs that contain at least one GLB.""" + style_dir = tmp_path / "3DLowPoly" + style_dir.mkdir() + (style_dir / "hero.glb").write_bytes(minimal_glb_bytes) + result = _detect_styles(tmp_path) + names = [name for name, _ in result] + assert "3DLowPoly" in names + + def test_skips_dirs_without_glbs(self, tmp_path: Path) -> None: + """_detect_styles should skip directories with no GLBs.""" + (tmp_path / "EmptyDir").mkdir() + result = _detect_styles(tmp_path) + names = [name for name, _ in result] + assert "EmptyDir" not in names + + def test_skips_hidden_dirs(self, tmp_path: Path, minimal_glb_bytes: bytes) -> None: + """_detect_styles should skip directories starting with '.' or '_'.""" + hidden = tmp_path / ".hidden" + hidden.mkdir() + (hidden / "asset.glb").write_bytes(minimal_glb_bytes) + private = tmp_path / "_Private" + private.mkdir() + (private / "asset.glb").write_bytes(minimal_glb_bytes) + result = _detect_styles(tmp_path) + names = [name for name, _ in result] + assert ".hidden" not in names + assert "_Private" not in names + + def test_caps_at_eight_results(self, tmp_path: Path, minimal_glb_bytes: bytes) -> None: + """_detect_styles should return at most 8 directories.""" + for i in range(12): + d = tmp_path / f"Style{i}" + d.mkdir() + (d / "asset.glb").write_bytes(minimal_glb_bytes) + result = _detect_styles(tmp_path) + assert len(result) <= 8 + + +class TestRunWizardNonInteractive: + def test_writes_config_file(self, tmp_path: Path) -> None: + """run_wizard(non_interactive=True) should write a config.toml.""" + config_dir = tmp_path / "config" / "game-asset-mcp" + config_file = config_dir / "config.toml" + + with ( + patch("game_asset_mcp.wizard._CONFIG_DIR", config_dir), + patch("game_asset_mcp.wizard._CONFIG_FILE", config_file), + patch("game_asset_mcp.wizard.subprocess.run"), # suppress ingest subprocess + ): + run_wizard(non_interactive=True, assets_root=str(tmp_path)) + + assert config_file.exists() + + def test_config_contains_assets_root(self, tmp_path: Path) -> None: + """Written config.toml should reference the provided assets_root path.""" + config_dir = tmp_path / "config" / "game-asset-mcp" + config_file = config_dir / "config.toml" + + with ( + patch("game_asset_mcp.wizard._CONFIG_DIR", config_dir), + patch("game_asset_mcp.wizard._CONFIG_FILE", config_file), + patch("game_asset_mcp.wizard.subprocess.run"), + ): + run_wizard(non_interactive=True, assets_root=str(tmp_path)) + + content = config_file.read_text() + assert str(tmp_path) in content + + def test_config_contains_style_map_section(self, tmp_path: Path) -> None: + """Written config.toml should include the [taxonomy.style_map] section.""" + config_dir = tmp_path / "config" / "game-asset-mcp" + config_file = config_dir / "config.toml" + + with ( + patch("game_asset_mcp.wizard._CONFIG_DIR", config_dir), + patch("game_asset_mcp.wizard._CONFIG_FILE", config_file), + patch("game_asset_mcp.wizard.subprocess.run"), + ): + run_wizard(non_interactive=True, assets_root=str(tmp_path)) + + content = config_file.read_text() + assert "[taxonomy.style_map]" in content + + def test_uses_detected_styles_in_config( + self, tmp_path: Path, minimal_glb_bytes: bytes + ) -> None: + """When GLB dirs are found, config should include them as styles.""" + style_dir = tmp_path / "MyStyle" + style_dir.mkdir() + (style_dir / "asset.glb").write_bytes(minimal_glb_bytes) + + config_dir = tmp_path / "config" / "game-asset-mcp" + config_file = config_dir / "config.toml" + + with ( + patch("game_asset_mcp.wizard._CONFIG_DIR", config_dir), + patch("game_asset_mcp.wizard._CONFIG_FILE", config_file), + patch("game_asset_mcp.wizard.subprocess.run"), + ): + run_wizard(non_interactive=True, assets_root=str(tmp_path)) + + content = config_file.read_text() + assert "MyStyle" in content + + def test_falls_back_to_default_styles(self, tmp_path: Path) -> None: + """If no GLBs found, config should use default 3DLowPoly/3DPSX styles.""" + config_dir = tmp_path / "config" / "game-asset-mcp" + config_file = config_dir / "config.toml" + empty_root = tmp_path / "empty" + empty_root.mkdir() + + with ( + patch("game_asset_mcp.wizard._CONFIG_DIR", config_dir), + patch("game_asset_mcp.wizard._CONFIG_FILE", config_file), + patch("game_asset_mcp.wizard.subprocess.run"), + ): + run_wizard(non_interactive=True, assets_root=str(empty_root)) + + content = config_file.read_text() + assert "3DLowPoly" in content + + def test_ingest_subprocess_called_in_non_interactive(self, tmp_path: Path) -> None: + """run_wizard(non_interactive=True) should invoke the ingest subprocess.""" + config_dir = tmp_path / "config" / "game-asset-mcp" + config_file = config_dir / "config.toml" + + with ( + patch("game_asset_mcp.wizard._CONFIG_DIR", config_dir), + patch("game_asset_mcp.wizard._CONFIG_FILE", config_file), + patch("game_asset_mcp.wizard.subprocess.run") as mock_run, + ): + run_wizard(non_interactive=True, assets_root=str(tmp_path)) + + mock_run.assert_called_once() diff --git a/pyproject.toml b/pyproject.toml index c5bdd371..4df40787 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -11,7 +11,7 @@ authors = [{ name = "Jon Bogaty" }] package = false [tool.uv.workspace] -members = ["packages/agentic-crew", "packages/pytest-agentic-crew"] +members = ["packages/agentic-crew", "packages/pytest-agentic-crew", "packages/game-asset-mcp"] [dependency-groups] dev = [ @@ -39,7 +39,7 @@ line-length = 120 select = ["E", "F", "I", "UP", "B", "C4", "SIM"] [tool.pytest.ini_options] -testpaths = ["packages/agentic-crew", "packages/pytest-agentic-crew"] +testpaths = ["packages/agentic-crew", "packages/pytest-agentic-crew", "packages/game-asset-mcp"] markers = [ "e2e: end-to-end tests (may require external services)", "slow: slow-running tests",