From 049b3c09012fa1ed8e9d1e2fd1ca1c6c2471f117 Mon Sep 17 00:00:00 2001 From: Hermes Agent Date: Sun, 28 Jun 2026 11:48:53 +0000 Subject: [PATCH] v0.3.0: Duplicate detection, GraphQL read-back, create_folder, path normalisation - client.py: GraphQL-backed list_products() and get_last_version() (AYON >=1.15 compat, replaces broken REST endpoints) - client.py: find_folder() normalises leading slash ('/assets/chair' and 'assets/chair' both work) - client.py: create_folder() + find_or_create_folder() with automatic parent creation - client.py: find_product_by_name() for duplicate detection - server.py: ayon_create_folder tool (creates folder tree) - server.py: publish tools now detect existing products and auto-version instead of creating duplicates - server.py: response includes 'action' field ('created_new' vs 'versioned_existing') - server.py: model_type and folder_type enum validation - server.py: ayon_get_product uses GraphQL for read-back --- src/ayon_mcp/client.py | 183 ++++++++++++++++++++++++------- src/ayon_mcp/server.py | 241 +++++++++++++++++++++++++++-------------- 2 files changed, 298 insertions(+), 126 deletions(-) diff --git a/src/ayon_mcp/client.py b/src/ayon_mcp/client.py index 7ca3a53..6ca2ffd 100644 --- a/src/ayon_mcp/client.py +++ b/src/ayon_mcp/client.py @@ -1,5 +1,6 @@ -"""AYON REST API client with API key auth.""" +"""AYON REST + GraphQL API client with API key auth.""" +import json import os from typing import Optional @@ -7,7 +8,7 @@ import httpx class AyonClient: - """Minimal wrapper around AYON's REST API. + """Wrapper around AYON's REST and GraphQL APIs. Requires AYON_URL and AYON_KEY environment variables. AYON_KEY is used instead of AYON_API_KEY to avoid Hermes @@ -31,6 +32,17 @@ class AyonClient: timeout=30, ) + # ── GraphQL helper ──────────────────────────────────── + + def _graphql(self, query: str) -> dict: + """Execute a GraphQL query. Returns data dict or raises.""" + r = self._client.post("/graphql", json={"query": query}) + r.raise_for_status() + body = r.json() + if body.get("errors"): + raise RuntimeError(body["errors"][0]["message"]) + return body.get("data", {}) + # ── Projects ────────────────────────────────────────── def list_projects(self) -> list[dict]: @@ -45,21 +57,119 @@ class AyonClient: r.raise_for_status() return r.json()["folders"] - # ── Products ────────────────────────────────────────── + def find_folder(self, project: str, path: str) -> Optional[dict]: + """Find a folder by its path. + + Normalises leading slash — both '/assets/chair' and 'assets/chair' work. + """ + clean = path.lstrip("/") + folders = self.list_folders(project) + for f in folders: + fp = f.get("path", "").lstrip("/") + if fp == clean: + return f + return None + + def create_folder( + self, + project: str, + name: str, + parent_id: str, + folder_type: str = "Asset", + ) -> dict: + """Create a new folder under a parent.""" + r = self._client.post( + f"/api/projects/{project}/folders", + json={ + "name": name, + "parentId": parent_id, + "folderType": folder_type, + }, + ) + r.raise_for_status() + return r.json() + + def find_or_create_folder( + self, project: str, path: str, folder_type: str = "Asset" + ) -> dict: + """Find existing folder or create it (including parents).""" + f = self.find_folder(project, path) + if f: + return f + + parts = path.strip("/").split("/") + current_parent = None + + # Find root parent + folders = self.list_folders(project) + for part in parts: + found = None + for f in folders: + fp = f.get("path", "").lstrip("/") + pid = f.get("parentId") or "" + if current_parent is None and not pid and fp == part: + found = f + break + if current_parent and f.get("parentId") == current_parent["id"] and fp.endswith(part): + found = f + break + + if found: + current_parent = found + else: + parent_id = current_parent["id"] if current_parent else self._find_root_folder_id(project) + current_parent = self.create_folder( + project, part, parent_id, folder_type + ) + # Refresh folder list for next iteration + folders = self.list_folders(project) + + if not current_parent: + raise RuntimeError(f"Could not find or create folder '{path}'") + return current_parent + + def _find_root_folder_id(self, project: str) -> str: + from urllib.parse import urlparse + path_parts = urlparse(self.url).path.strip("/").split("/") + # Default to 'assets' parent — find it + folders = self.list_folders(project) + for f in folders: + if f.get("name") == "assets" and not f.get("parentId"): + return f["id"] + # Fallback: first folder without parent + for f in folders: + if not f.get("parentId"): + return f["id"] + raise RuntimeError(f"No root folder found in project '{project}'") + + # ── Products (via GraphQL, AYON >=1.15) ─────────────── def list_products(self, project: str, folder_id: str) -> list[dict]: - # AYON >=1.15 removed GET /products?folderId= — returns 404. + """List products in a folder via GraphQL (REST endpoint removed in >=1.15).""" + query = ( + '{project(name:"' + project + '")' + '{folder(id:"' + folder_id + '")' + "{products{edges{node{id name productType createdAt updatedAt}}}}}}" + ) try: - r = self._client.get( - f"/api/projects/{project}/products", - params={"folderId": folder_id}, - ) - r.raise_for_status() - return r.json().get("products", []) - except httpx.HTTPStatusError as e: - if e.response.status_code == 404: + data = self._graphql(query) + folder = data.get("project", {}).get("folder") + if not folder: return [] - raise + edges = folder.get("products", {}).get("edges", []) + return [e["node"] for e in edges] + except Exception: + return [] + + def find_product_by_name( + self, project: str, folder_id: str, product_name: str + ) -> Optional[dict]: + """Find a product by name in a folder. Returns None if not found.""" + products = self.list_products(project, folder_id) + for p in products: + if p["name"] == product_name: + return p + return None def create_product( self, @@ -99,19 +209,26 @@ class AyonClient: return r.json() def get_last_version(self, project: str, product_id: str) -> Optional[dict]: - # AYON >=1.15 removed GET /versions?productId= — returns 404. + """Get latest version via GraphQL (REST endpoint removed in >=1.15).""" + query = ( + '{project(name:"' + project + '")' + '{product(id:"' + product_id + '")' + "{versions{edges{node{id version author createdAt}}}}}}" + ) try: - r = self._client.get( - f"/api/projects/{project}/versions", - params={"productId": product_id, "latest": "true"}, - ) - r.raise_for_status() - versions = r.json().get("versions", []) - return versions[0] if versions else None - except httpx.HTTPStatusError as e: - if e.response.status_code == 404: + data = self._graphql(query) + product = data.get("project", {}).get("product") + if not product: return None - raise + edges = product.get("versions", {}).get("edges", []) + if not edges: + return None + # Sort by version desc + versions = [e["node"] for e in edges] + versions.sort(key=lambda v: v["version"], reverse=True) + return versions[0] + except Exception: + return None # ── Representations ─────────────────────────────────── @@ -133,21 +250,3 @@ class AyonClient: ) r.raise_for_status() return r.json() - - # ── Helpers ─────────────────────────────────────────── - - def find_folder(self, project: str, path: str) -> Optional[dict]: - """Find a folder by its path (e.g. '/assets/chair').""" - folders = self.list_folders(project) - for f in folders: - if f.get("path") == path: - return f - return None - - def get_or_create_folder(self, project: str, path: str, folder_type: str = "Folder") -> dict: - """Get existing folder or create it.""" - f = self.find_folder(project, path) - if f: - return f - # For create we'd need the parent — but let's keep it simple for now - raise NotImplementedError("Folder creation via REST not yet implemented — create manually or via pyblish") diff --git a/src/ayon_mcp/server.py b/src/ayon_mcp/server.py index 2258121..f6231b6 100644 --- a/src/ayon_mcp/server.py +++ b/src/ayon_mcp/server.py @@ -1,17 +1,12 @@ """AYON MCP Server — main entry point. Usage: - AYON_URL=https://ayon.example.com \\ + AYON_URL=http://172.18.0.2:5000 \\ AYON_KEY=sk-... \\ ayon-mcp -Hermes config.yaml: - mcp_servers: - ayon: - command: "ayon-mcp" - env: - AYON_URL: "https://ayon.niklashmotion.art" - AYON_KEY: "sk-..." +Supports stdio (Claude Desktop, Cursor) and HTTP (Docker, Hermes) transport. +Set AYON_TRANSPORT=http for HTTP mode. """ import json @@ -78,9 +73,12 @@ async def ayon_list_folders(project: str) -> str: async def ayon_list_products(project: str, folder_path: str) -> str: """List published products in a folder. + Uses GraphQL for AYON >=1.15 compatibility. + Accepts paths with or without leading slash. + Args: project: Project name - folder_path: Folder path (e.g. '/assets/chair') + folder_path: Folder path (e.g. '/assets/chair' or 'assets/chair') """ try: folder = client.find_folder(project, folder_path) @@ -98,7 +96,9 @@ async def ayon_list_products(project: str, folder_path: str) -> str: @app.tool() async def ayon_get_product(project: str, folder_path: str, product_name: str) -> str: - """Get details about a specific product. + """Get details about a specific product, including latest version. + + Uses GraphQL for AYON >=1.15 compatibility. Args: project: Project name @@ -109,56 +109,115 @@ async def ayon_get_product(project: str, folder_path: str, product_name: str) -> folder = client.find_folder(project, folder_path) if not folder: return _err(f"Folder '{folder_path}' not found") - products = client.list_products(project, folder["id"]) - for p in products: - if p["name"] == product_name: - last = client.get_last_version(project, p["id"]) - return json.dumps({ - "name": p["name"], - "type": p.get("productType"), - "id": p["id"], - "latest_version": last["version"] if last else None, - }, indent=2) - return _err(f"Product '{product_name}' not found in '{folder_path}'") + product = client.find_product_by_name(project, folder["id"], product_name) + if not product: + return _err(f"Product '{product_name}' not found in '{folder_path}'") + last = client.get_last_version(project, product["id"]) + return json.dumps({ + "name": product["name"], + "type": product.get("productType"), + "id": product["id"], + "latest_version": last["version"] if last else None, + }, indent=2) + except Exception as e: + return _err(str(e)) + + +# ── Folder Management ──────────────────────────────────── + +@app.tool() +async def ayon_create_folder( + project: str, + path: str, + folder_type: str = "Asset", +) -> str: + """Create a new folder in an AYON project. Creates parent folders as needed. + + Args: + project: Project name (e.g. 'lumenfjord') + path: Folder path, slash-separated (e.g. 'assets/test/my_folder') + folder_type: AYON folder type: 'Asset', 'Folder', 'Shot', 'Sequence', + 'Episode', 'Library' (default: 'Asset') + """ + VALID_TYPES = {"Folder", "Library", "Asset", "Episode", "Sequence", "Shot"} + if folder_type not in VALID_TYPES: + return _err(f"Invalid folder type '{folder_type}'. Must be one of: {', '.join(sorted(VALID_TYPES))}") + + try: + folder = client.find_or_create_folder(project, path, folder_type) + return json.dumps({ + "name": folder["name"], + "path": folder.get("path"), + "id": folder["id"], + "created": True, + }, indent=2) except Exception as e: return _err(str(e)) # ── Publishing Tools ───────────────────────────────────── -@app.tool() -async def ayon_publish_texture_set( +def _publish_product( project: str, folder_path: str, product_name: str, + product_type: str, + representation_name: str, + files_list: list[dict], variant: str = "", - textures: str = "[]", ) -> str: - """Publish a texture set (multiple maps) as a single product. + """Core publish logic — creates product or auto-versions existing one. - Args: - project: Project name (e.g. 'DemoProject') - folder_path: Folder path (e.g. '/assets/chair') - product_name: Base product name (e.g. 'texture_chair') - variant: Optional variant (e.g. 'brown_leather') - textures: JSON array of texture info: - [{"name": "BaseColor.png", "size": 1048576}, ...] + Returns a JSON string with status and details. + The caller decides if this is modelMain or imageMain. """ - try: - textures_list = json.loads(textures) + folder = client.find_folder(project, folder_path) + if not folder: + return _err(f"Folder '{folder_path}' not found in '{project}'") - folder = client.find_folder(project, folder_path) - if not folder: - return _err(f"Folder '{folder_path}' not found in '{project}'") + full_name = product_name + if variant: + full_name = f"{product_name}_{variant}" - full_name = product_name - if variant: - full_name = f"{product_name}_{variant}" + # Check if product already exists + existing = client.find_product_by_name(project, folder["id"], full_name) + if existing: + # Product exists — auto-version + last = client.get_last_version(project, existing["id"]) + version_num = (last["version"] + 1) if last else 1 + + version = client.create_version( + project=project, + product_id=existing["id"], + version=version_num, + ) + + rep = client.create_representation( + project=project, + version_id=version["id"], + name=representation_name, + files=files_list, + ) + + return _ok( + action="versioned_existing", + product=full_name, + type=product_type, + version=version_num, + existing_product=True, + product_id=existing["id"], + version_id=version["id"], + folder=folder_path, + files=[f["name"] for f in files_list], + note=f"Product '{full_name}' already existed. Created version {version_num}.", + ) + else: + # New product product = client.create_product( project=project, name=full_name, - product_type="imageMain", + product_type=product_type, folder_id=folder["id"], ) @@ -174,18 +233,55 @@ async def ayon_publish_texture_set( rep = client.create_representation( project=project, version_id=version["id"], - name="texture_set", - files=textures_list, + name=representation_name, + files=files_list, ) return _ok( + action="created_new", product=full_name, - type="imageMain", + type=product_type, version=version_num, - maps=[t["name"] for t in textures_list], - folder=folder_path, + existing_product=False, product_id=product["id"], version_id=version["id"], + folder=folder_path, + files=[f["name"] for f in files_list], + note=f"New product '{full_name}' created with version {version_num}.", + ) + + +@app.tool() +async def ayon_publish_texture_set( + project: str, + folder_path: str, + product_name: str, + variant: str = "", + textures: str = "[]", +) -> str: + """Publish a texture set (multiple maps) as a single product. + + If a product with the same name already exists, a new VERSION is created + instead of a duplicate product. The response indicates which happened. + + Args: + project: Project name (e.g. 'DemoProject') + folder_path: Folder path (e.g. '/assets/chair' or 'assets/chair') + product_name: Base product name (e.g. 'texture_chair') + variant: Optional variant (e.g. 'brown_leather') + textures: JSON array of texture info: + [{"name": "BaseColor.png", "size": 1048576}, ...] + """ + try: + textures_list = json.loads(textures) + return _publish_product( + project=project, + folder_path=folder_path, + product_name=product_name, + product_type="imageMain", + representation_name="texture_set", + files_list=textures_list, + variant=variant, ) except Exception as e: return _err(str(e)) @@ -202,56 +298,33 @@ async def ayon_publish_model( ) -> str: """Publish a 3D model (ABC, USD, FBX, etc.). + If a product with the same name already exists, a new VERSION is created + instead of a duplicate product. The response indicates which happened + via the 'action' field ('created_new' or 'versioned_existing'). + Args: project: Project name - folder_path: Folder path (e.g. '/assets/chair') + folder_path: Folder path (e.g. '/assets/chair' or 'assets/chair') product_name: Product name (e.g. 'modelMain') model_type: File format: 'abc', 'usd', 'fbx', 'obj', 'blend' variant: Optional variant (e.g. 'highpoly') files: JSON array of file info: [{"name": "chair.abc", "size": 2048000}] """ + VALID_FORMATS = {"abc", "usd", "fbx", "obj", "blend"} + if model_type not in VALID_FORMATS: + return _err(f"Invalid model type '{model_type}'. Must be one of: {', '.join(sorted(VALID_FORMATS))}") + try: files_list = json.loads(files) - - folder = client.find_folder(project, folder_path) - if not folder: - return _err(f"Folder '{folder_path}' not found") - - full_name = product_name - if variant: - full_name = f"{product_name}_{variant}" - - product = client.create_product( + return _publish_product( project=project, - name=full_name, + folder_path=folder_path, + product_name=product_name, product_type="modelMain", - folder_id=folder["id"], - ) - - last = client.get_last_version(project, product["id"]) - version_num = (last["version"] + 1) if last else 1 - - version = client.create_version( - project=project, - product_id=product["id"], - version=version_num, - ) - - rep = client.create_representation( - project=project, - version_id=version["id"], - name=model_type, - files=files_list, - ) - - return _ok( - product=full_name, - type="modelMain", - format=model_type, - version=version_num, - files=[f["name"] for f in files_list], - folder=folder_path, + representation_name=model_type, + files_list=files_list, + variant=variant, ) except Exception as e: return _err(str(e))