Hermes/ayon-mcp
1
0
Fork 0
Code Issues 5 Pull Requests Actions Packages Projects Releases Wiki Activity

v0.3.0: Duplicate detection, GraphQL read-back, create_folder, path normalisation

Browse Source 
- 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
This commit is contained in:
Hermes Agent
2026-06-28 11:48:53 +00:00
parent 76683d5e24
commit 049b3c0901
2 changed files with 298 additions and 126 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ayon_mcp/server.py
+157 -84
View File
@@ -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))