Initial commit: AYON MCP server — AI agents publish CGI assets
Tools: - ayon_list_projects / ayon_list_folders / ayon_list_products - ayon_get_product - ayon_publish_texture_set (multiple maps as one product) - ayon_publish_model (ABC, USD, FBX, ...) FastMCP-based, wraps AYON REST API with X-Api-Key auth. Compatible with Hermes native MCP client and any MCP-compatible agent.
This commit is contained in:
@@ -0,0 +1,3 @@
|
||||
"""AYON MCP Server — AI agents publish CGI assets to AYON."""
|
||||
|
||||
__version__ = "0.1.0"
|
||||
@@ -0,0 +1,138 @@
|
||||
"""AYON REST API client with API key auth."""
|
||||
|
||||
import os
|
||||
from typing import Optional
|
||||
|
||||
import httpx
|
||||
|
||||
|
||||
class AyonClient:
|
||||
"""Minimal wrapper around AYON's REST API.
|
||||
|
||||
Requires AYON_SERVER_URL and AYON_API_KEY environment variables.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
url: Optional[str] = None,
|
||||
api_key: Optional[str] = None,
|
||||
):
|
||||
self.url = (url or os.environ["AYON_SERVER_URL"]).rstrip("/")
|
||||
self.api_key = api_key or os.environ["AYON_API_KEY"]
|
||||
self._client = httpx.Client(
|
||||
base_url=self.url,
|
||||
headers={
|
||||
"X-Api-Key": self.api_key,
|
||||
"Content-Type": "application/json",
|
||||
},
|
||||
timeout=30,
|
||||
)
|
||||
|
||||
# ── Projects ──────────────────────────────────────────
|
||||
|
||||
def list_projects(self) -> list[dict]:
|
||||
r = self._client.get("/api/projects")
|
||||
r.raise_for_status()
|
||||
return r.json()["projects"]
|
||||
|
||||
# ── Folders ───────────────────────────────────────────
|
||||
|
||||
def list_folders(self, project: str) -> list[dict]:
|
||||
r = self._client.get(f"/api/projects/{project}/folders")
|
||||
r.raise_for_status()
|
||||
return r.json()["folders"]
|
||||
|
||||
# ── Products ──────────────────────────────────────────
|
||||
|
||||
def list_products(self, project: str, folder_id: str) -> list[dict]:
|
||||
r = self._client.get(
|
||||
f"/api/projects/{project}/products",
|
||||
params={"folderId": folder_id},
|
||||
)
|
||||
r.raise_for_status()
|
||||
return r.json()["products"]
|
||||
|
||||
def create_product(
|
||||
self,
|
||||
project: str,
|
||||
name: str,
|
||||
product_type: str,
|
||||
folder_id: str,
|
||||
) -> dict:
|
||||
r = self._client.post(
|
||||
f"/api/projects/{project}/products",
|
||||
json={
|
||||
"name": name,
|
||||
"productType": product_type,
|
||||
"folderId": folder_id,
|
||||
},
|
||||
)
|
||||
r.raise_for_status()
|
||||
return r.json()
|
||||
|
||||
# ── Versions ──────────────────────────────────────────
|
||||
|
||||
def create_version(
|
||||
self,
|
||||
project: str,
|
||||
product_id: str,
|
||||
version: int,
|
||||
task_id: Optional[str] = None,
|
||||
) -> dict:
|
||||
body: dict = {"version": version, "productId": product_id}
|
||||
if task_id:
|
||||
body["taskId"] = task_id
|
||||
r = self._client.post(
|
||||
f"/api/projects/{project}/versions",
|
||||
json=body,
|
||||
)
|
||||
r.raise_for_status()
|
||||
return r.json()
|
||||
|
||||
def get_last_version(self, project: str, product_id: str) -> Optional[dict]:
|
||||
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
|
||||
|
||||
# ── Representations ───────────────────────────────────
|
||||
|
||||
def create_representation(
|
||||
self,
|
||||
project: str,
|
||||
version_id: str,
|
||||
name: str,
|
||||
files: list[dict],
|
||||
) -> dict:
|
||||
"""files: [{"name": "BaseColor.png", "size": 12345}, ...]"""
|
||||
r = self._client.post(
|
||||
f"/api/projects/{project}/representations",
|
||||
json={
|
||||
"name": name,
|
||||
"versionId": version_id,
|
||||
"files": files,
|
||||
},
|
||||
)
|
||||
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")
|
||||
@@ -0,0 +1,267 @@
|
||||
"""AYON MCP Server — main entry point.
|
||||
|
||||
Usage:
|
||||
AYON_SERVER_URL=https://ayon.example.com \\
|
||||
AYON_API_KEY=*** \\
|
||||
ayon-mcp
|
||||
|
||||
Hermes config.yaml:
|
||||
mcp_servers:
|
||||
ayon:
|
||||
command: "ayon-mcp"
|
||||
env:
|
||||
AYON_SERVER_URL: "https://ayon.niklashmotion.art"
|
||||
AYON_API_KEY: "sk-..."
|
||||
"""
|
||||
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
from typing import Optional
|
||||
|
||||
from mcp.server.fastmcp import FastMCP
|
||||
|
||||
from .client import AyonClient
|
||||
|
||||
# ── Bootstrap ────────────────────────────────────────────
|
||||
|
||||
app = FastMCP("ayon-publish")
|
||||
client = AyonClient()
|
||||
|
||||
|
||||
def _ok(**kwargs) -> str:
|
||||
return json.dumps(dict(status="ok", **kwargs), indent=2, ensure_ascii=False)
|
||||
|
||||
|
||||
def _err(msg: str) -> str:
|
||||
return json.dumps({"error": msg}, indent=2)
|
||||
|
||||
|
||||
# ── Discovery Tools ──────────────────────────────────────
|
||||
|
||||
@app.tool()
|
||||
async def ayon_list_projects() -> str:
|
||||
"""List all available AYON projects."""
|
||||
try:
|
||||
projects = client.list_projects()
|
||||
summary = [{"name": p["name"], "code": p.get("code")} for p in projects]
|
||||
return json.dumps(summary, indent=2)
|
||||
except Exception as e:
|
||||
return _err(str(e))
|
||||
|
||||
|
||||
@app.tool()
|
||||
async def ayon_list_folders(project: str) -> str:
|
||||
"""List folders in an AYON project.
|
||||
|
||||
Args:
|
||||
project: Project name (e.g. 'MyProject')
|
||||
"""
|
||||
try:
|
||||
folders = client.list_folders(project)
|
||||
summary = [
|
||||
{"path": f.get("path"), "type": f.get("folderType"), "id": f.get("id")}
|
||||
for f in folders
|
||||
]
|
||||
return json.dumps(summary, indent=2)
|
||||
except Exception as e:
|
||||
return _err(str(e))
|
||||
|
||||
|
||||
@app.tool()
|
||||
async def ayon_list_products(project: str, folder_path: str) -> str:
|
||||
"""List published products in a folder.
|
||||
|
||||
Args:
|
||||
project: Project name
|
||||
folder_path: Folder path (e.g. '/assets/chair')
|
||||
"""
|
||||
try:
|
||||
folder = client.find_folder(project, folder_path)
|
||||
if not folder:
|
||||
return _err(f"Folder '{folder_path}' not found in '{project}'")
|
||||
products = client.list_products(project, folder["id"])
|
||||
summary = [
|
||||
{"name": p["name"], "type": p.get("productType"), "id": p["id"]}
|
||||
for p in products
|
||||
]
|
||||
return json.dumps(summary, indent=2)
|
||||
except Exception as e:
|
||||
return _err(str(e))
|
||||
|
||||
|
||||
@app.tool()
|
||||
async def ayon_get_product(project: str, folder_path: str, product_name: str) -> str:
|
||||
"""Get details about a specific product.
|
||||
|
||||
Args:
|
||||
project: Project name
|
||||
folder_path: Folder path
|
||||
product_name: Product name to find
|
||||
"""
|
||||
try:
|
||||
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}'")
|
||||
except Exception as e:
|
||||
return _err(str(e))
|
||||
|
||||
|
||||
# ── Publishing Tools ─────────────────────────────────────
|
||||
|
||||
@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.
|
||||
|
||||
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}, ...]
|
||||
"""
|
||||
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}'")
|
||||
|
||||
full_name = product_name
|
||||
if variant:
|
||||
full_name = f"{product_name}_{variant}"
|
||||
|
||||
product = client.create_product(
|
||||
project=project,
|
||||
name=full_name,
|
||||
product_type="texture",
|
||||
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="texture_set",
|
||||
files=textures_list,
|
||||
)
|
||||
|
||||
return _ok(
|
||||
product=full_name,
|
||||
type="texture",
|
||||
version=version_num,
|
||||
maps=[t["name"] for t in textures_list],
|
||||
folder=folder_path,
|
||||
product_id=product["id"],
|
||||
version_id=version["id"],
|
||||
)
|
||||
except Exception as e:
|
||||
return _err(str(e))
|
||||
|
||||
|
||||
@app.tool()
|
||||
async def ayon_publish_model(
|
||||
project: str,
|
||||
folder_path: str,
|
||||
product_name: str,
|
||||
model_type: str = "abc",
|
||||
variant: str = "",
|
||||
files: str = "[]",
|
||||
) -> str:
|
||||
"""Publish a 3D model (ABC, USD, FBX, etc.).
|
||||
|
||||
Args:
|
||||
project: Project name
|
||||
folder_path: Folder path (e.g. '/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}]
|
||||
"""
|
||||
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(
|
||||
project=project,
|
||||
name=full_name,
|
||||
product_type="model",
|
||||
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="model",
|
||||
format=model_type,
|
||||
version=version_num,
|
||||
files=[f["name"] for f in files_list],
|
||||
folder=folder_path,
|
||||
)
|
||||
except Exception as e:
|
||||
return _err(str(e))
|
||||
|
||||
|
||||
# ── Entry Point ──────────────────────────────────────────
|
||||
|
||||
def main():
|
||||
if not os.environ.get("AYON_SERVER_URL"):
|
||||
print("ERROR: AYON_SERVER_URL not set", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
if not os.environ.get("AYON_API_KEY"):
|
||||
print("ERROR: AYON_API_KEY not set", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
app.run()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user