| """
|
| MCP Server for ManimStudio - Allows Claude Desktop to render Manim animations
|
| """
|
|
|
| import asyncio
|
| import httpx
|
| import os
|
| from typing import Any
|
| from mcp.server.models import InitializationOptions
|
| from mcp.server import NotificationOptions, Server
|
| from mcp.server.stdio import stdio_server
|
| from mcp.types import (
|
| Tool,
|
| TextContent,
|
| ImageContent,
|
| EmbeddedResource,
|
| )
|
|
|
|
|
| API_BASE_URL = os.getenv("MANIM_API_URL", "http://localhost:7860")
|
|
|
|
|
| mcp_server = Server("manim-studio")
|
|
|
|
|
| @mcp_server.list_tools()
|
| async def handle_list_tools() -> list[Tool]:
|
| """
|
| List available tools for rendering Manim animations
|
| """
|
| return [
|
| Tool(
|
| name="render_manim_animation",
|
| description=(
|
| "Render a mathematical animation using Manim. "
|
| "Provide Python code with a Scene class that defines the animation. "
|
| "Returns a URL to download the rendered video. "
|
| "Supports multiple quality presets (4k, 2k, 1080p, 720p, 480p) "
|
| "and output formats (mp4, gif, webm)."
|
| ),
|
| inputSchema={
|
| "type": "object",
|
| "properties": {
|
| "code": {
|
| "type": "string",
|
| "description": (
|
| "Python code containing Manim Scene class. "
|
| "Must include 'from manim import *' and define a Scene with construct() method."
|
| ),
|
| },
|
| "quality": {
|
| "type": "string",
|
| "description": "Quality preset: 4k, 2k, 1080p, 720p, or 480p",
|
| "enum": ["4k", "2k", "1080p", "720p", "480p"],
|
| "default": "720p",
|
| },
|
| "fps": {
|
| "type": "integer",
|
| "description": "Custom FPS (1-120). Overrides quality preset default.",
|
| "minimum": 1,
|
| "maximum": 120,
|
| },
|
| "output_format": {
|
| "type": "string",
|
| "description": "Output format: mp4, gif, webm, or all",
|
| "enum": ["mp4", "gif", "webm", "all"],
|
| "default": "mp4",
|
| },
|
| "scene_name": {
|
| "type": "string",
|
| "description": "Scene class name to render (auto-detected if only one scene exists)",
|
| },
|
| },
|
| "required": ["code"],
|
| },
|
| ),
|
| Tool(
|
| name="check_manim_health",
|
| description="Check if the Manim rendering service is healthy and available",
|
| inputSchema={
|
| "type": "object",
|
| "properties": {},
|
| },
|
| ),
|
| Tool(
|
| name="get_render_status",
|
| description="Check the status of a previously submitted render request",
|
| inputSchema={
|
| "type": "object",
|
| "properties": {
|
| "request_id": {
|
| "type": "string",
|
| "description": "The request ID returned from a previous render",
|
| },
|
| },
|
| "required": ["request_id"],
|
| },
|
| ),
|
| ]
|
|
|
|
|
| @mcp_server.call_tool()
|
| async def handle_call_tool(name: str, arguments: dict) -> list[TextContent]:
|
| """
|
| Handle tool execution requests
|
| """
|
|
|
| if name == "render_manim_animation":
|
| return await render_animation(arguments)
|
|
|
| elif name == "check_manim_health":
|
| return await check_health()
|
|
|
| elif name == "get_render_status":
|
| return await get_status(arguments)
|
|
|
| else:
|
| raise ValueError(f"Unknown tool: {name}")
|
|
|
|
|
| async def render_animation(args: dict) -> list[TextContent]:
|
| """
|
| Render a Manim animation by calling the API
|
| """
|
| async with httpx.AsyncClient(timeout=300.0) as client:
|
| try:
|
|
|
| response = await client.post(
|
| f"{API_BASE_URL}/render",
|
| json={
|
| "code": args["code"],
|
| "quality": args.get("quality", "720p"),
|
| "fps": args.get("fps"),
|
| "output_format": args.get("output_format", "mp4"),
|
| "scene_name": args.get("scene_name"),
|
| },
|
| )
|
|
|
| response.raise_for_status()
|
| result = response.json()
|
|
|
|
|
| if result["status"] == "success":
|
| urls = result.get("urls", {})
|
| url_list = "\n".join([f" - {fmt.upper()}: {url}" for fmt, url in urls.items()])
|
|
|
| message = f"""✅ Animation rendered successfully!
|
|
|
| Request ID: {result['request_id']}
|
| Scene: {result['message']}
|
|
|
| 📥 Download URLs:
|
| {url_list}
|
|
|
| ⏰ Files expire at: {result['expires_at']}
|
|
|
| The animation is ready to download. Files will be automatically deleted after 1 hour.
|
| """
|
| return [TextContent(type="text", text=message)]
|
|
|
| else:
|
| error_msg = f"""❌ Rendering failed
|
|
|
| Error: {result['message']}
|
|
|
| Details: {result.get('error_details', 'No additional details')}
|
|
|
| Please check your Manim code for syntax errors or invalid imports.
|
| """
|
| return [TextContent(type="text", text=error_msg)]
|
|
|
| except httpx.HTTPError as e:
|
| return [
|
| TextContent(
|
| type="text",
|
| text=f"❌ API request failed: {str(e)}\n\nMake sure the Manim API is running at {API_BASE_URL}"
|
| )
|
| ]
|
| except Exception as e:
|
| return [
|
| TextContent(
|
| type="text",
|
| text=f"❌ Unexpected error: {str(e)}"
|
| )
|
| ]
|
|
|
|
|
| async def check_health() -> list[TextContent]:
|
| """
|
| Check the health of the Manim API
|
| """
|
| async with httpx.AsyncClient(timeout=10.0) as client:
|
| try:
|
| response = await client.get(f"{API_BASE_URL}/health")
|
| response.raise_for_status()
|
| health = response.json()
|
|
|
| status_emoji = "✅" if health["status"] == "healthy" else "⚠️"
|
| ffmpeg_emoji = "✅" if health["ffmpeg_available"] else "❌"
|
|
|
| message = f"""{status_emoji} Manim Studio Health Check
|
|
|
| Status: {health['status'].upper()}
|
| Manim Version: {health['manim_version']}
|
| FFmpeg Available: {ffmpeg_emoji} {health['ffmpeg_available']}
|
| Disk Space: {health.get('disk_space_mb', 0):.2f} MB
|
| Active Renders: {health.get('active_renders', 0)}
|
|
|
| API URL: {API_BASE_URL}
|
| """
|
| return [TextContent(type="text", text=message)]
|
|
|
| except Exception as e:
|
| return [
|
| TextContent(
|
| type="text",
|
| text=f"❌ Health check failed: {str(e)}\n\nMake sure the API is running at {API_BASE_URL}"
|
| )
|
| ]
|
|
|
|
|
| async def get_status(args: dict) -> list[TextContent]:
|
| """
|
| Get the status of a render request
|
| """
|
| async with httpx.AsyncClient(timeout=10.0) as client:
|
| try:
|
| request_id = args["request_id"]
|
| response = await client.get(f"{API_BASE_URL}/status/{request_id}")
|
| response.raise_for_status()
|
| status = response.json()
|
|
|
| if status["status"] == "ready":
|
| urls = status.get("urls", {})
|
| url_list = "\n".join([f" - {fmt.upper()}: {url}" for fmt, url in urls.items()])
|
|
|
| message = f"""✅ Render Status: READY
|
|
|
| Request ID: {request_id}
|
| Available Formats: {', '.join(status.get('files_available', []))}
|
|
|
| 📥 Download URLs:
|
| {url_list}
|
|
|
| ⏰ Expires at: {status['expires_at']}
|
| """
|
| return [TextContent(type="text", text=message)]
|
|
|
| elif status["status"] == "expired":
|
| return [TextContent(type="text", text=f"⏰ Render has expired (request ID: {request_id})")]
|
|
|
| elif status["status"] == "not_found":
|
| return [TextContent(type="text", text=f"❌ Render not found (request ID: {request_id})")]
|
|
|
| else:
|
| return [TextContent(type="text", text=f"Status: {status['status']}")]
|
|
|
| except Exception as e:
|
| return [
|
| TextContent(
|
| type="text",
|
| text=f"❌ Failed to get status: {str(e)}"
|
| )
|
| ]
|
|
|
|
|
| async def main():
|
| """
|
| Main entry point for the MCP server
|
| """
|
| async with stdio_server() as (read_stream, write_stream):
|
| await mcp_server.run(
|
| read_stream,
|
| write_stream,
|
| InitializationOptions(
|
| server_name="manim-studio",
|
| server_version="1.0.0",
|
| capabilities=mcp_server.get_capabilities(
|
| notification_options=NotificationOptions(),
|
| experimental_capabilities={},
|
| ),
|
| ),
|
| )
|
|
|
|
|
| if __name__ == "__main__":
|
| asyncio.run(main())
|
|
|
