Coverage for integrations / service_tools / wan2gp_tool.py: 100.0%

1""" 

2Wan2GP tool wrapper — video generation from text prompts. 

3 

4Service: Wan2GP (https://github.com/deepbeepmeep/Wan2GP) 

5Port: Dynamic (assigned by RuntimeToolManager) 

6Async task pattern: submit → poll (same as ACE-Step) 

7""" 

8 

9from .registry import ServiceToolInfo, service_tool_registry 

10 

11 

12class Wan2GPTool: 

13 """Thin wrapper to register Wan2GP video generation with the ServiceToolRegistry.""" 

14 

15 REPO_URL = "https://github.com/deepbeepmeep/Wan2GP" 

16 

17 @classmethod 

18 def create_tool_info(cls, base_url: str) -> ServiceToolInfo: 

19 return ServiceToolInfo( 

20 name="wan2gp", 

21 description=( 

22 "AI video generation from text prompts. Creates short video clips " 

23 "from text descriptions. Supports text-to-video and image-to-video " 

24 "modes with various resolution and duration settings." 

25 ), 

26 base_url=base_url, 

27 endpoints={ 

28 "generate": { 

29 "path": "/generate", 

30 "method": "POST", 

31 "description": ( 

32 "Submit a video generation task. " 

33 "Input: JSON with 'prompt' (text description of video), " 

34 "'num_frames' (int, default 49), 'width' (int, default 512), " 

35 "'height' (int, default 320), 'num_inference_steps' (int, default 25). " 

36 "Returns task_id to check result." 

37 ), 

38 "params_schema": { 

39 "prompt": {"type": "string", "description": "Video description prompt"}, 

40 "num_frames": {"type": "integer", "description": "Number of frames", "default": 49}, 

41 "width": {"type": "integer", "description": "Video width", "default": 512}, 

42 "height": {"type": "integer", "description": "Video height", "default": 320}, 

43 "num_inference_steps": {"type": "integer", "description": "Inference steps", "default": 25}, 

44 }, 

45 }, 

46 "check_result": { 

47 "path": "/check_result", 

48 "method": "POST", 

49 "description": ( 

50 "Check status of a video generation task. " 

51 "Input: JSON with 'task_id' (string from generate). " 

52 "Returns status and video URL when complete." 

53 ), 

54 "params_schema": { 

55 "task_id": {"type": "string", "description": "Task ID from generate"}, 

56 }, 

57 }, 

58 }, 

59 health_endpoint="/health", 

60 tags=["video", "generation", "text-to-video", "ai"], 

61 timeout=300, 

62 ) 

63 

64 @classmethod 

65 def register(cls, base_url: str) -> bool: 

66 """Register Wan2GP with the global service_tool_registry.""" 

67 tool_info = cls.create_tool_info(base_url) 

68 return service_tool_registry.register_tool(tool_info)