""" Fractafrag MCP Server — AI agent interface to the shader platform. Enables Claude, GPT, and other MCP clients to: - Browse and search shaders - Get full shader details by ID - Submit new shaders - Update existing shaders (push revisions) - View version history - Browse the desire queue """ import os import json import httpx from mcp.server.fastmcp import FastMCP API_BASE = os.environ.get("API_BASE_URL", "http://api:8000") INTERNAL_AUTH = {"Authorization": "Bearer internal:mcp-service"} mcp = FastMCP( "Fractafrag", stateless_http=True, json_response=True, host="0.0.0.0", port=3200, ) async def api_get(path: str, params: dict | None = None): async with httpx.AsyncClient(base_url=API_BASE, timeout=15.0) as client: resp = await client.get(f"/api/v1{path}", params=params) resp.raise_for_status() return resp.json() async def api_post(path: str, data: dict): async with httpx.AsyncClient(base_url=API_BASE, timeout=15.0) as client: resp = await client.post(f"/api/v1{path}", json=data, headers=INTERNAL_AUTH) resp.raise_for_status() return resp.json() async def api_post_with_params(path: str, params: dict): """POST with query parameters (not JSON body). Used for endpoints like fulfill.""" async with httpx.AsyncClient(base_url=API_BASE, timeout=15.0) as client: resp = await client.post(f"/api/v1{path}", params=params, headers=INTERNAL_AUTH) resp.raise_for_status() return resp.json() async def api_put(path: str, data: dict): async with httpx.AsyncClient(base_url=API_BASE, timeout=15.0) as client: resp = await client.put(f"/api/v1{path}", json=data, headers=INTERNAL_AUTH) resp.raise_for_status() return resp.json() @mcp.tool() async def browse_shaders(query: str = "", tags: str = "", shader_type: str = "", sort: str = "trending", limit: int = 20) -> str: """Browse and search shaders on Fractafrag. Args: query: Search text (matches title) tags: Comma-separated tag filter (e.g. "fractal,colorful") shader_type: Filter by type: 2d, 3d, or audio-reactive sort: Sort order: trending, new, or top limit: Number of results (1-50) """ params: dict = {"sort": sort, "limit": min(limit, 50)} if query: params["q"] = query if tags: params["tags"] = [t.strip() for t in tags.split(",") if t.strip()] if shader_type: params["shader_type"] = shader_type shaders = await api_get("/shaders", params) results = [{"id": s["id"], "title": s["title"], "description": s.get("description", ""), "shader_type": s["shader_type"], "tags": s.get("tags", []), "score": s.get("score", 0), "view_count": s.get("view_count", 0), "is_system": s.get("is_system", False), "current_version": s.get("current_version", 1)} for s in shaders] return json.dumps({"count": len(results), "shaders": results}) @mcp.tool() async def get_shader(shader_id: str) -> str: """Get full details of a shader by its ID, including GLSL source code. Args: shader_id: UUID of the shader """ s = await api_get(f"/shaders/{shader_id}") return json.dumps({"id": s["id"], "title": s["title"], "description": s.get("description"), "glsl_code": s["glsl_code"], "shader_type": s["shader_type"], "tags": s.get("tags", []), "status": s.get("status"), "is_system": s.get("is_system", False), "style_metadata": s.get("style_metadata"), "current_version": s.get("current_version", 1), "score": s.get("score", 0), "view_count": s.get("view_count", 0), "forked_from": s.get("forked_from"), "created_at": s.get("created_at"), "updated_at": s.get("updated_at")}) @mcp.tool() async def get_shader_versions(shader_id: str) -> str: """Get the version history of a shader. Args: shader_id: UUID of the shader """ versions = await api_get(f"/shaders/{shader_id}/versions") return json.dumps({"shader_id": shader_id, "version_count": len(versions), "versions": [{"version_number": v["version_number"], "title": v["title"], "change_note": v.get("change_note"), "created_at": v["created_at"]} for v in versions]}) @mcp.tool() async def get_shader_version_code(shader_id: str, version_number: int) -> str: """Get the GLSL code of a specific version of a shader. Args: shader_id: UUID of the shader version_number: Version number to retrieve """ v = await api_get(f"/shaders/{shader_id}/versions/{version_number}") return json.dumps({"shader_id": shader_id, "version_number": v["version_number"], "title": v["title"], "glsl_code": v["glsl_code"], "tags": v.get("tags", []), "change_note": v.get("change_note")}) @mcp.tool() async def submit_shader(title: str, glsl_code: str, description: str = "", tags: str = "", shader_type: str = "2d", status: str = "published", fulfills_desire_id: str = "") -> str: """Submit a new GLSL shader to Fractafrag. Shader format: void mainImage(out vec4 fragColor, in vec2 fragCoord) Uniforms: iTime (float), iResolution (vec3), iMouse (vec4) Args: title: Shader title (max 120 chars) glsl_code: Complete GLSL fragment shader code description: Optional description tags: Comma-separated tags (e.g. "fractal,noise,colorful") shader_type: 2d, 3d, or audio-reactive status: "published" to go live, "draft" to save privately fulfills_desire_id: Optional UUID of a desire this shader fulfills """ tag_list = [t.strip() for t in tags.split(",") if t.strip()] if tags else [] payload = {"title": title, "glsl_code": glsl_code, "description": description, "tags": tag_list, "shader_type": shader_type, "status": status} if fulfills_desire_id: payload["fulfills_desire_id"] = fulfills_desire_id result = await api_post("/shaders", payload) return json.dumps({"id": result["id"], "title": result["title"], "status": result.get("status"), "current_version": result.get("current_version", 1), "message": f"Shader '{result['title']}' created.", "url": f"/shader/{result['id']}"}) @mcp.tool() async def update_shader(shader_id: str, glsl_code: str = "", title: str = "", description: str = "", tags: str = "", status: str = "", change_note: str = "") -> str: """Update an existing shader. Creates a new version in the history. Use this to iterate — change code, adjust metadata, or publish a draft. Every code change creates an immutable version snapshot. Args: shader_id: UUID of the shader to update glsl_code: New GLSL code (empty = keep current) title: New title (empty = keep current) description: New description (empty = keep current) tags: New comma-separated tags (empty = keep current) status: Change status: draft, published, or archived change_note: Brief note about what changed (e.g. "made it pinker") """ payload = {} if glsl_code: payload["glsl_code"] = glsl_code if title: payload["title"] = title if description: payload["description"] = description if tags: payload["tags"] = [t.strip() for t in tags.split(",") if t.strip()] if status: payload["status"] = status if change_note: payload["change_note"] = change_note if not payload: return json.dumps({"error": "No changes provided."}) result = await api_put(f"/shaders/{shader_id}", payload) return json.dumps({"id": result["id"], "title": result["title"], "status": result.get("status"), "current_version": result.get("current_version"), "message": f"Updated to v{result.get('current_version', '?')}.", "url": f"/shader/{result['id']}"}) @mcp.tool() async def get_trending(limit: int = 10) -> str: """Get currently trending shaders. Args: limit: Number of results (1-50) """ shaders = await api_get("/feed/trending", {"limit": min(limit, 50)}) return json.dumps({"count": len(shaders), "shaders": [{"id": s["id"], "title": s["title"], "shader_type": s["shader_type"], "tags": s.get("tags", []), "score": s.get("score", 0)} for s in shaders]}) @mcp.tool() async def get_similar_shaders(shader_id: str, limit: int = 10) -> str: """Find shaders visually similar to a given shader (by tag overlap). Args: shader_id: UUID of the reference shader limit: Number of results (1-30) """ shaders = await api_get(f"/feed/similar/{shader_id}", {"limit": min(limit, 30)}) return json.dumps({"reference": shader_id, "count": len(shaders), "similar": [{"id": s["id"], "title": s["title"], "shader_type": s["shader_type"], "tags": s.get("tags", []), "score": s.get("score", 0)} for s in shaders]}) @mcp.tool() async def get_desire_queue(min_heat: float = 0, limit: int = 10) -> str: """Get open shader desires/bounties with cluster context and style hints. Returns community requests ranked by heat. Use cluster_count to identify high-demand desires (many similar requests). Use style_hints to understand the visual direction requested. Args: min_heat: Minimum heat score (higher = more demand) limit: Number of results (1-20) """ desires = await api_get("/desires", {"min_heat": min_heat, "limit": min(limit, 20)}) return json.dumps({"count": len(desires), "desires": [{"id": d["id"], "prompt_text": d["prompt_text"], "heat_score": d.get("heat_score", 0), "cluster_count": d.get("cluster_count", 0), "style_hints": d.get("style_hints"), "tip_amount_cents": d.get("tip_amount_cents", 0), "status": d.get("status"), "fulfilled_by_shader": d.get("fulfilled_by_shader")} for d in desires]}) @mcp.tool() async def fulfill_desire(desire_id: str, shader_id: str) -> str: """Mark a desire as fulfilled by linking it to a published shader. The shader must be published. The desire must be open. Use get_desire_queue to find open desires, then submit_shader or use an existing shader ID to fulfill one. Args: desire_id: UUID of the desire to fulfill shader_id: UUID of the published shader that fulfills this desire """ try: result = await api_post_with_params( f"/desires/{desire_id}/fulfill", {"shader_id": shader_id} ) return json.dumps({"status": "fulfilled", "desire_id": desire_id, "shader_id": shader_id, "message": f"Desire {desire_id} fulfilled by shader {shader_id}."}) except httpx.HTTPStatusError as e: try: error_detail = e.response.json().get("detail", str(e)) except Exception: error_detail = str(e) return json.dumps({"error": error_detail, "status_code": e.response.status_code}) @mcp.resource("fractafrag://platform-info") def platform_info() -> str: """Platform overview and shader writing guidelines.""" return """# Fractafrag — GLSL Shader Platform ## Shader Format (Shadertoy-compatible, WebGL2 / GLSL ES 3.00) ```glsl void mainImage(out vec4 fragColor, in vec2 fragCoord) { vec2 uv = fragCoord / iResolution.xy; fragColor = vec4(color, 1.0); } ``` ## Uniforms: iTime (float), iResolution (vec3), iMouse (vec4) ## Workflow 1. browse_shaders to see what exists 2. get_shader to read code by ID 3. submit_shader to create new, or update_shader to revise existing 4. Every code update creates a versioned snapshot 5. get_shader_versions / get_shader_version_code for history """ if __name__ == "__main__": print(f"Fractafrag MCP server starting on :3200") print(f"API backend: {API_BASE}") mcp.run(transport="streamable-http")