NoLang MCP Server

import json from pathlib import Path from typing import Any, Dict, List, Optional, Union from uuid import UUID import httpx from pydantic import FilePath, TypeAdapter, ValidationError from nolang_mcp.config import nolang_mcp_config from nolang_mcp.models import ( FilesPayload, MultipartFile, TemplateRecommendationResponse, VideoGenerationResponse, VideoListResponse, VideoModeEnum, VideoSettingsResponse, VideoStatusResponse, ) class NoLangAPI: def __init__(self, api_key: str) -> None: self.base_url = nolang_mcp_config.nolang_api_base_url self.headers = {"Authorization": f"Bearer {api_key}"} @staticmethod def _validate_file_path(file_path: str) -> Path: """Validate that the given file path exists and return Path instance.""" try: TypeAdapter(FilePath).validate_python(file_path) return Path(file_path) except ValidationError: raise FileNotFoundError(f"File not found: {file_path}") async def _post( self, path: str, data: Optional[Dict[str, Any]] = None, files: Optional[FilesPayload] = None, ) -> Dict[str, Any]: files_to_send = {k: v.to_tuple() for k, v in files.items()} if files else None async with httpx.AsyncClient(timeout=httpx.Timeout(60.0)) as client: url = f"{self.base_url}{path}" if data and "setting" in data and isinstance(data["setting"], dict): data = {**data, "setting": json.dumps(data["setting"], ensure_ascii=False)} r = await client.post(url, headers=self.headers, data=data, files=files_to_send) r.raise_for_status() return r.json() async def _get(self, path: str, params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: async with httpx.AsyncClient() as client: url = f"{self.base_url}{path}" r = await client.get(url, headers=self.headers, params=params) r.raise_for_status() try: return r.json() except Exception: raise def _prepare_setting_data(self, setting: Union[UUID, str, Dict[str, Any]]) -> tuple[str, Dict[str, Any]]: """Internal utility: returns API URL and data base from setting""" if isinstance(setting, dict): return "/unstable/videos/generate/", {"setting": setting} else: return "/videos/generate/", {"video_setting_id": str(setting)} async def _generate_video_with_file( self, setting: Union[UUID, str, Dict[str, Any]], file_path: str, file_field_name: str, mime_type: str, extra_data: Optional[Dict[str, Any]] = None, ) -> VideoGenerationResponse: path = self._validate_file_path(file_path) api_url, data = self._prepare_setting_data(setting) if extra_data: data.update(extra_data) with open(path, "rb") as f: files: FilesPayload = { file_field_name: MultipartFile(filename=path.name, content=f.read(), mime_type=mime_type) } response_data = await self._post(api_url, data=data, files=files) return VideoGenerationResponse(**response_data) async def generate_video_with_text( self, setting: Union[UUID, str, Dict[str, Any]], text: str, image_files: Optional[List[str]] = None, ) -> VideoGenerationResponse: api_url, data = self._prepare_setting_data(setting) data["text"] = text files: FilesPayload = {} if image_files: for image_path in image_files: try: path = self._validate_file_path(image_path) except FileNotFoundError: # Skip non-existent image paths gracefully continue with open(path, "rb") as f: files["image_files"] = MultipartFile( filename=path.name, content=f.read(), mime_type="image/jpeg", ) if files: response_data = await self._post(api_url, data=data, files=files) else: response_data = await self._post(api_url, data=data) return VideoGenerationResponse(**response_data) async def generate_video_with_pdf( self, setting: Union[UUID, str, Dict[str, Any]], pdf_path: str, ) -> VideoGenerationResponse: """Generate video by uploading PDF""" return await self._generate_video_with_file( setting, pdf_path, "pdf_file", "application/pdf", ) async def generate_video_with_pptx( self, setting: Union[UUID, str, Dict[str, Any]], pptx_path: str, ) -> VideoGenerationResponse: """Generate video by uploading PPTX""" return await self._generate_video_with_file( setting, pptx_path, "pptx_file", "application/vnd.openxmlformats-officedocument.presentationml.presentation", ) async def generate_video_with_audio( self, setting: Union[UUID, str, Dict[str, Any]], audio_path: str, ) -> VideoGenerationResponse: """Generate video by uploading audio file""" path = self._validate_file_path(audio_path) mime_types = { ".mp3": "audio/mpeg", ".wav": "audio/wav", ".m4a": "audio/mp4", ".aac": "audio/aac", } mime_type = mime_types.get(path.suffix.lower(), "audio/mpeg") return await self._generate_video_with_file( setting, audio_path, "audio_file", mime_type, ) async def generate_video_with_video( self, setting: Union[UUID, str, Dict[str, Any]], video_path: str, ) -> VideoGenerationResponse: """Generate video by uploading video file (encoding only)""" return await self._generate_video_with_file( setting, video_path, "video_file", "video/mp4", ) async def generate_video_with_pdf_and_text( self, setting: Union[UUID, str, Dict[str, Any]], pdf_path: str, text: str, ) -> VideoGenerationResponse: """Generate video with combination of PDF and text""" return await self._generate_video_with_file( setting, pdf_path, "pdf_file", "application/pdf", extra_data={"text": text}, ) async def get_video_status(self, video_id: UUID) -> VideoStatusResponse: response_data = await self._get(f"/videos/{video_id}/") return VideoStatusResponse(**response_data) async def list_videos(self, page: int = 1) -> VideoListResponse: response_data = await self._get("/videos/", params={"page": page}) return VideoListResponse(**response_data) async def list_video_settings(self, page: int = 1) -> VideoSettingsResponse: response_data = await self._get("/unstable/video-settings/", params={"page": page}) return VideoSettingsResponse(**response_data) async def get_video_setting_from_video_id(self, video_id: UUID) -> Dict[str, Any]: return await self._get( f"/unstable/video-settings/{video_id}/", ) async def recommend_template( self, video_mode: VideoModeEnum, query: Optional[str] = None, is_mobile_format: Optional[bool] = None, ) -> TemplateRecommendationResponse: """Template recommendation endpoint (unstable)""" params: Dict[str, Any] = {"video_mode": video_mode.value} if query is not None: params["query"] = query if is_mobile_format is not None: params["is_mobile_format"] = is_mobile_format try: response_data = await self._get( "/unstable/template/recommend/", params=params, ) return TemplateRecommendationResponse(**response_data) except Exception: raise def format_http_error(e: httpx.HTTPStatusError) -> str: try: data = e.response.json() return ( f"API Error {e.response.status_code}\n" f"Code : {data.get('code', '—')}\n" f"Message: {data.get('error', 'Unknown error')}" ) except Exception: return f"API Error {e.response.status_code}\nBody: {e.response.text}" nolang_api = NoLangAPI(nolang_mcp_config.nolang_api_key)