#!/usr/bin/env python3 """ Quorvid MCP server -- lets an AI assistant (Claude Desktop, Claude Code, or any MCP-compatible app) make videos, shorts, audio, images and thumbnails with your Quorvid account. Setup (see https://quorvid.com/connect-claude/): 1. pip install mcp httpx 2. In Quorvid: Connect to Claude -> Create key. Copy the key (it is shown once). 3. Add this file to your assistant's MCP settings with the key in the QUORVID_API_KEY environment variable. Environment: QUORVID_API_KEY required. Your personal key, starts with qv_ QUORVID_API_URL optional. Default https://app.quorvid.com Everything runs against your own account and credits. The key has its own daily spending limit, set in Quorvid, so an assistant can never spend more than that in a day. Run `python quorvid_mcp.py --check` to test your setup. """ from __future__ import annotations import asyncio import json import os import sys from typing import Any, Optional import httpx try: # mcp >= 2 from mcp.server.mcpserver import MCPServer as _Server from mcp.server.mcpserver.exceptions import ToolError except ImportError: # mcp 1.x from mcp.server.fastmcp import FastMCP as _Server from mcp.server.fastmcp.exceptions import ToolError from mcp.types import ToolAnnotations DEFAULT_URL = "https://app.quorvid.com" MAX_WAIT_SECONDS = 50 # one tool call stays well under typical client time-outs; call again to keep waiting _TRANSPORT: Optional[httpx.AsyncBaseTransport] = None # tests inject an in-process transport here INSTRUCTIONS = """Quorvid turns a topic or script into a finished, narrated, captioned video (plus shorts, audio, images and thumbnails). It uses the user's own credits. How to use it well: - Spending credits is real. Before an expensive request, call quote_video, or pass max_credits to create_video so it refuses anything pricier. Tell the user the price when it is more than a few credits. - Standard voices are free-ish (2 credits for a video, and the account gets free daily credits that can cover it). Premium voices cost about 12 credits per minute of narration, AI images 2 each. Prefer voice="standard" unless the user asks for premium quality. - Videos take a few minutes to render. create_video returns a job_id at once; call get_video(job_id, wait_seconds=45) repeatedly until status is "completed", then give the user the download_url. - Never invent a download link; only use the ones returned by the tools.""" mcp = _Server("quorvid", instructions=INSTRUCTIONS) _READ = ToolAnnotations(readOnlyHint=True, openWorldHint=False) _MAKE = ToolAnnotations(readOnlyHint=False, destructiveHint=False, openWorldHint=True) class QuorvidError(ToolError): """A problem the assistant (and the user) should be told about, as-is. Subclassing ToolError makes the MCP library return it as a normal error result instead of treating it as a crash.""" # ---------------------------------------------------------------------------------------------------------- HTTP def _base_url() -> str: return (os.environ.get("QUORVID_API_URL") or DEFAULT_URL).rstrip("/") def _abs(url: Optional[str]) -> Optional[str]: if not url: return None return url if url.startswith("http") else _base_url() + url def _detail(response: httpx.Response) -> str: try: body = response.json() except ValueError: return response.text[:300] or response.reason_phrase detail = body.get("detail", body) if isinstance(body, dict) else body if isinstance(detail, list): # FastAPI validation errors return "; ".join(f"{'.'.join(str(p) for p in e.get('loc', [])[1:])}: {e.get('msg')}" for e in detail if isinstance(e, dict)) return str(detail) async def _call(method: str, path: str, *, json_body: Any = None, params: Optional[dict] = None) -> Any: key = (os.environ.get("QUORVID_API_KEY") or "").strip() if not key: raise QuorvidError("QUORVID_API_KEY isn't set. Create a key in Quorvid (Connect to Claude) and add it to this server's settings.") try: async with httpx.AsyncClient( base_url=_base_url(), headers={"Authorization": f"Bearer {key}"}, timeout=60, transport=_TRANSPORT, ) as client: response = await client.request(method, path, json=json_body, params=params) except httpx.HTTPError as exc: raise QuorvidError(f"Couldn't reach Quorvid at {_base_url()} ({exc}). Check QUORVID_API_URL and your connection.") from exc if response.status_code >= 400: hint = { 401: "The API key was rejected: it is wrong, revoked or missing. Create a new one in Quorvid under Connect to Claude.", 402: "Not enough credits. The user can buy more in Quorvid.", 403: "API keys can't do that. It has to be done in the Quorvid app.", 404: "Not found (check the id).", 429: "Limit reached.", }.get(response.status_code, "Request failed.") raise QuorvidError(f"{hint} Details: {_detail(response)}") return response.json() if response.content else {} def _job(record: dict) -> dict: """The parts of a job the assistant needs, with absolute download links.""" request = record.get("request") or {} out = { "job_id": record["id"], "status": record.get("status"), "progress_percent": record.get("progress"), "message": record.get("message"), "title": request.get("title") or request.get("prompt"), "download_url": _abs(record.get("output_url")), "thumbnail_url": _abs(record.get("thumbnail_url")), "duration_seconds": record.get("duration_seconds"), "error": record.get("error"), } if record.get("subtitle_srt_url"): out["subtitles_srt_url"] = _abs(record["subtitle_srt_url"]) if record.get("script") and record.get("status") == "completed": out["script"] = record["script"][:4000] return {k: v for k, v in out.items() if v not in (None, "")} def _voice(value: str) -> str: v = (value or "standard").strip() if v.lower() in ("standard", "free"): return "standard" if v.lower() in ("premium", "auto", "best"): return "auto" return v # an exact voice name from list_voices, or "elevenlabs:" async def _wait(path: str, wait_seconds: int, finished=("completed", "failed")) -> dict: record = await _call("GET", path) deadline = max(0, min(int(wait_seconds or 0), MAX_WAIT_SECONDS)) waited = 0 while record.get("status") not in finished and waited < deadline: await asyncio.sleep(3) waited += 3 record = await _call("GET", path) return record def _video_body(topic, script, voice, length_seconds, aspect_ratio, video_type, captions, music, visual_source, narration_quality) -> dict: if not (topic or script): raise QuorvidError("Give either a topic or a script.") body: dict = { "mode": "script" if script else "prompt", "video_type": video_type, "aspect_ratio": aspect_ratio, "target_duration_seconds": int(length_seconds), "voice": _voice(voice), "captions": captions, "music": music, "visual_source": visual_source, "narration_quality": narration_quality, } body["script" if script else "prompt"] = script or topic return body # ---------------------------------------------------------------------------------------------------------- tools @mcp.tool(annotations=_READ) async def check_account() -> dict: """Show the connected Quorvid account: email, credit balance and how many free daily credits are left today.""" me = await _call("GET", "/api/auth/me") costs = await _call("GET", "/api/billing/costs") result = {"email": me.get("email"), "credits": me.get("credit_balance")} if costs.get("charged") and costs.get("free"): result["free_daily_credits_left"] = costs["free"]["left"] result["free_daily_credits_per_day"] = costs["free"]["per_day"] return result @mcp.tool(annotations=_READ) async def price_list() -> dict: """List what each Quorvid action costs in credits (some are charged per minute or per scene).""" costs = await _call("GET", "/api/billing/costs") if not costs.get("charged"): return {"note": "This account isn't charged for usage."} return {"prices": {c["label"]: c["cost"] for c in costs["items"]}} @mcp.tool(annotations=_READ) async def list_voices(language: str = "en", limit: int = 30) -> list[dict]: """List standard (free) voices for a language code such as en, ur, hi, ar, es, fr. Use the short_name as the voice when creating a video. Standard voices also work with voice="standard", which picks one automatically.""" picked = await _call("GET", "/api/voices", params={"language": (language or "").strip()} if language else None) return [ {"voice": v["short_name"], "name": v["display_name"], "gender": v["gender"], "locale": v["locale"]} for v in picked[: max(1, min(int(limit), 100))] ] @mcp.tool(annotations=_READ) async def quote_video( length_seconds: int = 45, voice: str = "standard", visual_source: str = "auto", video_type: str = "explainer", narration_quality: str = "best", script: Optional[str] = None, ) -> dict: """Price a video in credits WITHOUT making it. voice is "standard" (cheap, covered by free daily credits when short), "premium" (about 12 credits per minute) or an exact voice name. visual_source: auto, stock, ai (2 credits per image) or ai_video (paid clips). Pass script if you will supply your own text, so its length is used.""" body = {"kind": "video", "mode": "script" if script else "prompt", "script": script, "target_duration_seconds": int(length_seconds), "voice": _voice(voice), "visual_source": visual_source, "video_type": video_type, "narration_quality": narration_quality} quote = await _call("POST", "/api/billing/quote", json_body=body) if not quote.get("charged"): return {"credits": 0, "note": "This account isn't charged for usage."} return {"credits_to_pay": quote["credits"], "covered_by_free_daily_credits": quote.get("covered", 0), "breakdown": quote.get("lines", []), "free_daily_credits_left": (quote.get("free") or {}).get("left")} @mcp.tool(annotations=_MAKE) async def create_video( topic: Optional[str] = None, script: Optional[str] = None, voice: str = "standard", length_seconds: int = 45, aspect_ratio: str = "9:16", video_type: str = "explainer", captions: bool = True, music: bool = True, visual_source: str = "auto", narration_quality: str = "best", max_credits: Optional[int] = None, ) -> dict: """Make a narrated, captioned video from a topic (Quorvid writes the script) or your own script. Uses credits. Returns a job_id immediately; the render takes a few minutes, so call get_video(job_id, wait_seconds=45) until it is completed. aspect_ratio: 9:16 (Shorts/Reels/TikTok), 16:9 (YouTube) or 1:1. video_type: explainer, history, story, motivational, listicle, news, cartoon, movie_explainer, game_lore or custom. voice: "standard" (default, cheapest), "premium", or an exact name from list_voices. Set max_credits to refuse anything that costs more.""" body = _video_body(topic, script, voice, length_seconds, aspect_ratio, video_type, captions, music, visual_source, narration_quality) quote = await _call("POST", "/api/billing/quote", json_body={ "kind": "video", "mode": body["mode"], "script": body.get("script"), "target_duration_seconds": body["target_duration_seconds"], "voice": body["voice"], "visual_source": visual_source, "video_type": video_type, "narration_quality": narration_quality, }) price = quote.get("credits", 0) if quote.get("charged") else 0 if max_credits is not None and price > int(max_credits): return {"created": False, "reason": f"This would cost {price} credits, more than your limit of {max_credits}. Nothing was charged.", "credits_to_pay": price} record = await _call("POST", "/api/videos", json_body=body) return {"created": True, "job_id": record["id"], "credits_to_pay": price, "next_step": "Rendering takes a few minutes. Call get_video(job_id, wait_seconds=45) until status is completed."} @mcp.tool(annotations=_READ) async def get_video(job_id: str, wait_seconds: int = 0) -> dict: """Check a video, short, or dubbed video by job_id. With wait_seconds (up to 50) it waits for the render to finish. When status is "completed" the result has download_url (the MP4).""" return _job(await _wait(f"/api/videos/{job_id}", wait_seconds)) @mcp.tool(annotations=_READ) async def list_videos(limit: int = 10) -> list[dict]: """List the account's most recent videos and their status.""" records = await _call("GET", "/api/videos") return [_job(r) for r in records[: max(1, min(int(limit), 50))]] @mcp.tool(annotations=_MAKE) async def create_audio(script: str, voice: str = "standard", max_credits: Optional[int] = None) -> dict: """Turn text into narration (MP3). Standard voice is about 1 credit per minute, premium about 12. Returns a job_id; call get_audio(job_id, wait_seconds=45) for the download_url.""" quote = await _call("POST", "/api/billing/quote", json_body={"kind": "audio", "script": script, "voice": _voice(voice)}) price = quote.get("credits", 0) if quote.get("charged") else 0 if max_credits is not None and price > int(max_credits): return {"created": False, "reason": f"This would cost {price} credits, more than your limit of {max_credits}.", "credits_to_pay": price} record = await _call("POST", "/api/audio", json_body={"script": script, "voice": _voice(voice)}) return {"created": True, "job_id": record["id"], "credits_to_pay": price} @mcp.tool(annotations=_READ) async def get_audio(job_id: str, wait_seconds: int = 0) -> dict: """Check a narration job; waits up to 50 seconds if asked. download_url is the MP3 when completed.""" return _job(await _wait(f"/api/audio/{job_id}", wait_seconds)) @mcp.tool(annotations=_MAKE) async def generate_image(prompt: str, aspect_ratio: str = "1:1", style: str = "photorealistic") -> dict: """Generate an AI image (about 3 credits). style: photorealistic, cinematic, cartoon, anime, fantasy_art, digital_painting, watercolor, three_d_render, sketch or minimalist. aspect_ratio e.g. 1:1, 16:9, 9:16, 4:3, 3:2, 21:9. Returns a job_id; call get_image(job_id) for the download_url.""" record = await _call("POST", "/api/images", json_body={"prompt": prompt, "aspect_ratio": aspect_ratio, "style": style}) return {"created": True, "job_id": record["id"]} @mcp.tool(annotations=_READ) async def get_image(job_id: str, wait_seconds: int = 0) -> dict: """Check an image job; waits up to 50 seconds if asked. download_url is the image when completed.""" return _job(await _wait(f"/api/images/{job_id}", wait_seconds)) @mcp.tool(annotations=_MAKE) async def make_shorts(video_job_id: str, count: int = 3, min_seconds: int = 20, max_seconds: int = 60) -> list[dict]: """Cut a finished Quorvid video into 1-10 vertical Shorts (5 credits each, covered by free daily credits when available). Returns one job per short; check each with get_video.""" body = {"source_job_id": video_job_id, "num_shorts": int(count), "min_duration": int(min_seconds), "max_duration": int(max_seconds), "aspect_ratio": "9:16", "captions": True, "music": False} records = await _call("POST", "/api/repurpose", json_body=body) return [{"job_id": r["id"], "title": (r.get("request") or {}).get("title")} for r in records] @mcp.tool(annotations=_MAKE) async def translate_video(video_job_id: str, language_code: str, language_name: str, voice: str = "standard") -> dict: """Dub a finished Quorvid video into another language, e.g. language_code "ur", language_name "Urdu". Priced per minute of video (standard voice about 4 credits a minute, premium 12). Returns a job_id; check it with get_video.""" record = await _call("POST", "/api/translate", json_body={ "source_job_id": video_job_id, "target_language_code": language_code, "target_language_name": language_name, "voice": _voice(voice)}) return {"created": True, "job_id": record["id"]} @mcp.tool(annotations=_MAKE) async def make_thumbnail(video_job_id: str, headline: Optional[str] = None, ai_background: bool = False) -> dict: """Make a thumbnail for a finished video. By default it uses the video's best frame (free); ai_background=True makes a new AI background (4 credits). headline is optional text (max 60 characters); one is written if you leave it out.""" result = await _call("POST", f"/api/videos/{video_job_id}/thumbnail", json_body={ "text": headline, "source": "ai" if ai_background else "frame"}) return {"thumbnail_url": _abs(result.get("url")), "headline": result.get("text"), "credits_charged": result.get("charged", 0), "note": result.get("note")} # ---------------------------------------------------------------------------------------------------------- run def main() -> None: if "--check" in sys.argv: try: print(json.dumps(asyncio.run(check_account()), indent=2)) except QuorvidError as exc: print(f"Setup problem: {exc}") sys.exit(1) return mcp.run() # stdio, which is what Claude Desktop and Claude Code expect if __name__ == "__main__": main()