#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ AICraft Video Generation MCP Server ==================================== 让 Codex++ / Codex CLI 能直接调用 AICraft 平台的视频生成模型(可灵/Seedance/Veo/Vidu)。 背景:Codex 是编码智能体,视频模型是生成式 API,不是聊天模型。 本 MCP server 把视频生成封装成 Codex 可调用的工具: list_video_models() → 可用模型清单 generate_video(prompt, model, res?) → 提交任务,返回 request_id + status_url check_video(status_url) → 轮询任务状态,完成返回视频链接 计费走用户平台余额:提交即扣全价,失败自动退费。 协议:Model Context Protocol (MCP) over stdio,newline-delimited JSON-RPC 2.0。 用法(Codex config.toml): [mcp_servers.video] command = "python" args = ["C:/Users/dell/.codex/mcp/video_mcp.py"] env = { "PYTHONUTF8" = "1" } """ import base64 import json import os import sys import time import urllib.request import urllib.error # ── 配置 ────────────────────────────────────────────────────────────── DEFAULT_API_URL = "https://aicraftapi.com/v1/videos/generations" LOG_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "video_mcp.log") # 可用视频模型(来自平台 MODEL_MAP) VIDEO_MODELS = [ {"id": "kling-v3", "name": "可灵 Kling V3", "type": "文生视频"}, {"id": "kling-video-o1", "name": "可灵 Kling O1", "type": "文生视频"}, {"id": "kling-v2-6", "name": "可灵 Kling V2.6 Pro", "type": "文生视频"}, {"id": "kling-v2-1", "name": "可灵 Kling V2.1 Pro", "type": "文生视频"}, {"id": "bytedance/doubao-seedance-2-0-260128", "name": "豆包 Seedance 2.0", "type": "文生视频"}, {"id": "bytedance/doubao-seedance-2-0-fast-260128", "name": "豆包 Seedance 2.0 Fast", "type": "文生视频"}, {"id": "bytedance/doubao-seedance-2-0-mini-260615", "name": "豆包 Seedance 2.0 Mini", "type": "文生视频"}, {"id": "veo-3.1-generate-001", "name": "Google Veo 3.1", "type": "文生视频"}, {"id": "veo-3.1-fast-generate-001", "name": "Google Veo 3.1 Fast", "type": "文生视频"}, {"id": "viduq2", "name": "Vidu Q2", "type": "文生视频"}, {"id": "viduq1", "name": "Vidu Q1", "type": "文生视频"}, {"id": "kling-image-o1", "name": "可灵 Kling 生图 O1", "type": "文生图"}, ] # 可用生图模型(8/15 实测:只有带命名空间的模型名在 /v1/images/generations 有效,裸名/kling-image-o1 报 503/400) IMAGE_MODELS = [ {"id": "openai/gpt-image-2", "name": "GPT Image 2", "type": "文生图"}, {"id": "google/gemini-3-pro-image", "name": "Gemini 3 Pro Image", "type": "文生图"}, {"id": "google/gemini-3.1-flash-image", "name": "Gemini 3.1 Flash Image", "type": "文生图"}, {"id": "google/gemini-3.1-flash-lite-image","name": "Gemini 3.1 Flash Lite Image","type": "文生图"}, ] # 生图模型一句话定位(CEO铁律:不限制顾客选模型,只教怎么用) _IMAGE_USAGE = { "openai/gpt-image-2": "GPT Image 2，通用全能，最稳", "google/gemini-3-pro-image": "Gemini 3 Pro Image，细节与文字排版强", "google/gemini-3.1-flash-image": "Gemini 3.1 Flash，速度快", "google/gemini-3.1-flash-lite-image": "Gemini 3.1 Flash Lite，最省", } # 模型一句话定位(教顾客怎么选。CEO铁律:不限制顾客选模型,只教怎么用;不设默认绑定/不替顾客换模型) _MODEL_USAGE = { "kling-v3": "可灵 V3，写实电影感，适合剧情、人物、动作大片", "kling-video-o1": "可灵 O1，长镜头与复杂运镜，适合纪录片式叙事", "kling-v2-6": "可灵 V2.6，老牌写实，性价比高", "kling-v2-1": "可灵 V2.1，老牌写实，基础款", "bytedance/doubao-seedance-2-0-260128": "豆包 Seedance 2.0，中文指令理解强，适合电商、口播、分镜", "bytedance/doubao-seedance-2-0-fast-260128": "豆包 Seedance 2.0 Fast，出片快，适合时效性内容", "bytedance/doubao-seedance-2-0-mini-260615": "豆包 Seedance 2.0 Mini，轻量快速，适合日常短视频", "veo-3.1-generate-001": "Google Veo 3.1，真实世界物理，画质天花板", "veo-3.1-fast-generate-001": "Google Veo 3.1 Fast，兼顾质量与速度", "viduq2": "Vidu Q2，风格多样，适合创意、广告", "viduq1": "Vidu Q1，创意风格基础款", "kling-image-o1": "可灵生图 O1（文生图）", } def _log(msg: str) -> None: try: with open(LOG_PATH, "a", encoding="utf-8") as f: f.write(f"{time.strftime('%Y-%m-%dT%H:%M:%S')} {msg}\n") except Exception: pass def _clean(s) -> str: """去掉代理字符(不合法 Unicode),防止编码异常。""" try: return str(s).encode("utf-8", errors="ignore").decode("utf-8") except Exception: return "" def _gbk_safe(s: str) -> str: """把文本过滤成 GBK 可编码字符。 Codex++(中文 Windows)的中继把 MCP 工具结果按 GBK 编码发给模型, 任何 GBK 没有的字符(如 ✓✅❌⏳→★)都会让中继解析失败 → 模型拿到 "PARSE ERR 'gbk' codec can't encode..." → 以为工具失败 → 死循环重试。 中文都在 GBK 内,只会被去掉这些符号,不影响内容。 """ try: return s.encode("gbk", errors="ignore").decode("gbk") except Exception: return s def _load_api_key() -> str: """解析 API key: 环境变量 > settings.json(顶层 relayApiKey > profile 级 > authContents)。""" for var in ("AICRAFT_MCP_KEY", "SEARCH_API_KEY", "VIDEO_API_KEY"): v = os.environ.get(var, "").strip() if v: return v try: sp = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "..", ".codex-session-delete", "settings.json") sp = os.path.abspath(sp) with open(sp, encoding="utf-8") as f: data = json.load(f) # 1) 顶层 relayApiKey(新版 Codex++ 存这里) k = str(data.get("relayApiKey") or "").strip() if k: return k # 2) profile 级 relayApiKey for prof in data.get("relayProfiles", []): k = str(prof.get("relayApiKey") or "").strip() if k: return k # 3) profile authContents 里的 OPENAI_API_KEY for prof in data.get("relayProfiles", []): ac = prof.get("authContents") or "" if isinstance(ac, str): try: ac = json.loads(ac) except Exception: ac = {} k = str((ac or {}).get("OPENAI_API_KEY") or "").strip() if k: return k except Exception: pass return "" def _headers(key: str) -> dict: h = {"Content-Type": "application/json"} if key: h["Authorization"] = "Bearer " + key return h def _req(url: str, key: str, method: str, payload: dict = None, timeout: int = 60) -> dict: data = None if payload is not None: data = json.dumps(payload, ensure_ascii=False).encode("utf-8", errors="ignore") req = urllib.request.Request(url, data=data, headers=_headers(key), method=method) try: with urllib.request.urlopen(req, timeout=timeout) as resp: body = resp.read().decode("utf-8", errors="replace") return {"ok": True, "status": resp.status, "data": json.loads(body)} except urllib.error.HTTPError as e: err = e.read().decode("utf-8", errors="replace") try: err = json.loads(err) except Exception: pass # 8/14 修复:平台对 FAILED 任务返回 502 + 完整任务对象(含 status 字段)。 # 这是业务失败而非传输错误,降级为 data 处理,否则 check_video 永远走不进 FAILED 分支。 if isinstance(err, dict) and err.get("status"): return {"ok": True, "status": e.code, "data": err} return {"ok": False, "status": e.code, "error": err} except Exception as e: return {"ok": False, "status": None, "error": f"{type(e).__name__}: {e}"} def _fmt_error(res: dict) -> str: err = res.get("error") if isinstance(err, dict): return err.get("message", json.dumps(err, ensure_ascii=False)) return _clean(err) # ── 工具实现 ───────────────────────────────────────────────────────── # 防重复提交护栏:同一 (model, prompt) 在短时间内不重复提交,直接返回已有任务。 # 2026-08-12 事故:Codex 死循环同 prompt 提交 39 次 kling-v3,每次真扣 ¥0.2。 import hashlib _recent_submits: dict = {} # key -> {"status_url": str, "ts": float} def _submit_key(model: str, prompt: str, resolution: str) -> str: h = hashlib.sha256((model + "|" + prompt + "|" + resolution).encode("utf-8", errors="ignore")) return h.hexdigest() def _list_video_models() -> str: lines = ["可用视频模型:"] for m in VIDEO_MODELS: lines.append(f"- {m['id']} ({m['name']} · {m['type']})") lines.append("\n不确定选哪个:用 create_video 并把 model 留空,会列出每个模型的适合场景,由你决定用哪个。") return "\n".join(lines) def _list_image_models() -> str: lines = ["可用生图模型(同步接口,提交后直接返回图片链接):"] for m in IMAGE_MODELS: usage = _IMAGE_USAGE.get(m["id"], "") lines.append(f"- {m['id']} ({m['name']})" + (f"，{usage}" if usage else "")) lines.append("\n不确定选哪个:把 model 留空,会列出适合场景由你决定,平台不替你默认选。") return "\n".join(lines) def _generate_video(prompt: str, model: str, resolution: str) -> str: prompt = _clean(prompt).strip() model = _clean(model).strip() resolution = _clean(resolution).strip() if not prompt: return "参数错误:prompt 不能为空" if not model: return "参数错误:model 不能为空(用 list_video_models 查看可用模型)" payload = {"model": model, "prompt": prompt} if resolution: payload["resolution"] = resolution key = _load_api_key() url = os.environ.get("VIDEO_API_URL", DEFAULT_API_URL) # 防重复提交护栏:同一 (model, prompt, resolution) 在 180s 内已提交 → 复用已有任务 now = time.time() skey = _submit_key(model, prompt, resolution) prev = _recent_submits.get(skey) if prev and (now - prev["ts"]) < 180: _log(f" ! 重复提交拦截:同 prompt 180s 内已提交,复用 {prev['status_url']}") return (f"检测到相同请求已在 180 秒内提交,为避免重复扣费已直接复用已有任务。\n" f"status_url: {prev['status_url']}\n" f"请用 check_video 工具传入此 status_url 查询结果。") _log(f"generate_video model={model} res={resolution or '-'} q={prompt[:40]}") res = _req(url, key, "POST", payload, timeout=120) _log(f" → HTTP {res.get('status')} ok={res.get('ok')} err={_fmt_error(res)[:120]}") if not res.get("ok"): if res.get("status") == 402: return ("余额不足:视频生成需预扣费用,请到 aicraftapi.com/dashboard.html 充值后重试。" f"详情:{_fmt_error(res)}") if res.get("status") == 401: return f"鉴权失败(API key 无效):{_fmt_error(res)}" if res.get("status") == 403: return f"API key 已过期:{_fmt_error(res)}" return f"提交失败(HTTP {res.get('status')}):{_fmt_error(res)}" data = res.get("data") or {} rid = data.get("request_id", "") status_url = data.get("status_url", "") lines = [f"任务已提交 request_id={rid}", f"状态查询地址(status_url): {status_url}"] lines.append("请用 check_video 工具传入上面的 status_url 轮询结果(生成通常需 1-5 分钟)。") return "\n".join(lines) def _create_video(prompt: str, model: str, resolution: str) -> str: """一站式生成视频。model 不填 → 列出全部可选模型与各自定位引导顾客指定, 绝不偷偷替顾客默认选/换模型(CEO铁律:不限制顾客选模型,只教怎么用)。""" prompt = _clean(prompt).strip() model = _clean(model).strip() resolution = _clean(resolution).strip() if not prompt: return "参数错误:prompt 不能为空" if not model: lines = ["未指定 model 参数。平台不替你默认选模型——请明说想用哪个,可选模型与适合场景如下:"] for m in VIDEO_MODELS: usage = _MODEL_USAGE.get(m["id"], "") lines.append(f"- {m['id']} {m['name']}" + (f"，{usage}" if usage else "")) lines.append("确定后,把模型 ID 填进 create_video 的 model 参数即可(也可直接说「用可灵 V3 生成」)。") return "\n".join(lines) return _generate_video(prompt, model, resolution) def _generate_image(prompt: str, model: str, size: str) -> str: """同步生图(8/15 加):调 /v1/images/generations,直接返回图片链接。 与视频不同,生图是同步接口,不产生长时任务。同样带 180s 防重复护栏防 Codex 死循环。""" prompt = _clean(prompt).strip() model = _clean(model).strip() size = _clean(size).strip() if not prompt: return "参数错误:prompt 不能为空" if not model: lines = ["未指定 model 参数。平台不替你默认选模型——请明说想用哪个,可选生图模型如下:"] for m in IMAGE_MODELS: usage = _IMAGE_USAGE.get(m["id"], "") lines.append(f"- {m['id']} {m['name']}" + (f"，{usage}" if usage else "")) lines.append("确定后,把模型 ID 填进 generate_image 的 model 参数即可(也可直接说「用可灵生图 O1 生成」)。") return "\n".join(lines) payload = {"model": model, "prompt": prompt} if size: payload["size"] = size key = _load_api_key() url = "https://aicraftapi.com/v1/images/generations" # 防重复提交护栏:同一 (model, prompt, size) 在 180s 内已生成 → 直接复用结果 now = time.time() h = hashlib.sha256((model + "|" + prompt + "|" + size).encode("utf-8", errors="ignore")).hexdigest() prev = _recent_submits.get(h) if prev and (now - prev["ts"]) < 180: _log(f" ! 重复生图拦截:同请求 180s 内已生成,复用 {prev['status_url'][:80]}") return (f"检测到相同生图请求已在 180 秒内提交,为避免重复扣费已直接复用上次结果。\n" f"图片地址: {prev['status_url']}") _log(f"generate_image model={model} size={size or '-'} q={prompt[:40]}") res = _req(url, key, "POST", payload, timeout=180) _log(f" → HTTP {res.get('status')} ok={res.get('ok')} err={_fmt_error(res)[:120]}") if not res.get("ok"): if res.get("status") == 402: return ("余额不足:生图需扣费,请到 aicraftapi.com/dashboard.html 充值后重试。" f"详情:{_fmt_error(res)}") if res.get("status") == 401: return f"鉴权失败(API key 无效):{_fmt_error(res)}" if res.get("status") == 403: return f"API key 已过期:{_fmt_error(res)}" return f"生成失败(HTTP {res.get('status')}):{_fmt_error(res)}" data = res.get("data") or {} items = data.get("data") if isinstance(data, dict) else None first = items[0] if isinstance(items, list) and items else {} if not isinstance(first, dict): return "未在返回中找到图片,原始返回: " + json.dumps(data, ensure_ascii=False)[:500] # 情况A: 直接返回图片直链 url img_url = first.get("url") or "" if img_url: _recent_submits[h] = {"status_url": img_url, "ts": now} return ("图片已生成!\n图片地址(浏览器直接打开即可看/右键保存):\n" + img_url) # 情况B: 返回 b64_json(实测 gpt-image-2/gemini 走这个) → 解码存本地,返回本地路径 b64 = first.get("b64_json") or "" if b64: try: img_bytes = base64.b64decode(b64) base_dir = os.path.join(os.path.expanduser("~"), "Downloads") os.makedirs(base_dir, exist_ok=True) fname = "aicraft_img_" + hashlib.md5(b64[:2048].encode("utf-8", errors="ignore")).hexdigest()[:8] + ".png" dest = os.path.join(base_dir, fname) with open(dest, "wb") as f: f.write(img_bytes) _recent_submits[h] = {"status_url": dest, "ts": now} return (f"图片已生成并保存到本地!\n本地路径: {dest}\n" f"大小: {len(img_bytes)/1024:.0f} KB\n请直接打开或双击查看。") except Exception as e: _log(f"generate_image b64 decode error: {type(e).__name__}: {e}") return f"图片已生成但本地保存失败: {type(e).__name__}: {e}" return "未在返回中找到图片,原始返回: " + json.dumps(data, ensure_ascii=False)[:500] def _check_video(status_url: str) -> str: status_url = _clean(status_url).strip() if not status_url.startswith("http"): return "参数错误:status_url 必须是 generate_video 返回的完整地址(以 http 开头)" key = _load_api_key() _log(f"check_video {status_url[:90]}") res = _req(status_url, key, "GET", timeout=60) if not res.get("ok"): return f"查询失败(HTTP {res.get('status')}):{_fmt_error(res)}" data = res.get("data") or {} status = str(data.get("status", "?")).upper() _log(f" status={status}") if status == "COMPLETED": # 平台结算后返回的视频地址可能在各字段,统一列出来方便 Codex 读取 out = ["视频生成完成!"] # 8/14:viduq1/veo 等模型响应结构是 task_result.videos[0].url,原遍历集合漏了它 tr = data.get("task_result") if isinstance(tr, dict): vids = tr.get("videos") if isinstance(vids, list) and vids and isinstance(vids[0], dict): u = vids[0].get("url") or vids[0].get("video_url") or "" if u: out.append("video_url: " + u) for k in ("video_url", "url", "output_url", "result", "video", "file_url"): v = data.get(k) if isinstance(v, dict): v = v.get("url") or v.get("video_url") or list(v.values())[0] if v else None if v: out.append(f"{k}: {v}") if len(out) == 1: out.append("未找到视频地址字段,原始返回: " + json.dumps(data, ensure_ascii=False)[:500]) return "\n".join(out) if status == "FAILED": er = data.get("error") or data.get("message") or "" if isinstance(er, dict): er = er.get("message") or er.get("code") or json.dumps(er, ensure_ascii=False) return "视频生成失败。" + (" 原因: " + _clean(er) if er else " 请检查提示词后重试。") # PROCESSING / QUEUED ... return f"生成中(status={status}),请稍后再次用本工具查询同一条 status_url。" def _download_video(url_or_status: str, save_path: str) -> str: """下载视频到本地（8/13 加，顾客用 Codex 拿视频的堵点根治）。 Codex++ 客户端无直接联网能力，一切网络靠 MCP 工具； check_video 只能查状态，缺"下载到本地"工具 → Codex 卡在"需要联网"死循环。 download_video 内网用 urllib 下载：传 status_url 自动查状态拿直链，或直接传 video.url。 输出过 _gbk_safe，Codex++(中文Windows GBK 中继)可正常解析。""" url_or_status = _clean(url_or_status).strip() if not url_or_status.startswith("http"): return "参数错误:download_video 需要 status_url 或视频直链 url(http 开头)" key = _load_api_key() video_url = url_or_status if "/requests/" in url_or_status and "/status" in url_or_status: # 是 status_url → 先查状态拿直链 res = _req(url_or_status, key, "GET", timeout=60) if not res.get("ok"): return f"状态查询失败(HTTP {res.get('status')}):{_fmt_error(res)}" data = res.get("data") or {} status = data.get("status", "?") if status != "COMPLETED": return f"任务尚未完成(status={status}),无法下载,请稍后再试。" try: video_url = data["result"]["video"]["url"] or "" except Exception: video_url = "" if not video_url: return "未在结果中找到视频直链,原始返回: " + json.dumps(data, ensure_ascii=False)[:300] # 直接下载直链 try: base = _clean(save_path).strip() or os.path.join(os.path.expanduser("~"), "Downloads") os.makedirs(base, exist_ok=True) fname = video_url.rstrip("/").split("/")[-1].split("?")[0] if not fname or "." not in fname: fname = "video_" + hashlib.md5(video_url.encode("utf-8", errors="ignore")).hexdigest()[:8] + ".mp4" dest = os.path.join(base, fname) req = urllib.request.Request(video_url, headers={"User-Agent": "Mozilla/5.0"}) total = 0 with urllib.request.urlopen(req, timeout=180) as r, open(dest, "wb") as f: while True: chunk = r.read(262144) if not chunk: break f.write(chunk) total += len(chunk) _log(f"download_video -> {dest} ({total} bytes)") return f"视频已下载到本地: {dest}\n大小: {total/1024/1024:.1f} MB,请直接打开或双击播放。" except Exception as e: _log(f"download_video error: {type(e).__name__}: {e}") return f"下载失败: {type(e).__name__}: {e}" # ── MCP 协议处理 ───────────────────────────────────────────────────── TOOLS = [ { "name": "list_video_models", "description": "列出平台可用的视频/图像生成模型ID与名称。调用 generate_video 前建议先查。", "inputSchema": {"type": "object", "properties": {}, "required": []}, }, { "name": "generate_video", "description": "提交文生视频任务。平台自动扣费、异步生成。返回 request_id 和 status_url,之后用 check_video 轮询。", "inputSchema": { "type": "object", "properties": { "prompt": {"type": "string", "description": "视频内容描述,越具体越好(主体/动作/镜头/画风/时长等)"}, "model": {"type": "string", "description": "模型ID,见 list_video_models,如 kling-v3 / bytedance/doubao-seedance-2-0-260128"}, "resolution": {"type": "string", "description": "可选:720p 或 1080p(部分模型支持,默认上游)"}, }, "required": ["prompt", "model"], }, }, { "name": "create_video", "description": "一站式生成视频。prompt 必填;model 可选——想用指定模型就填其 ID(如 kling-v3 / bytedance/doubao-seedance-2-0-260128 / veo-3.1-generate-001),不填则列出全部可选模型与各自适合场景由你决定,平台不会替你默认选或偷偷换模型。返回 request_id 和 status_url,之后用 check_video 轮询。", "inputSchema": { "type": "object", "properties": { "prompt": {"type": "string", "description": "视频内容描述,越具体越好(主体/动作/镜头/画风/时长等)"}, "model": {"type": "string", "description": "可选:模型ID。不填则返回全部可选模型引导你指定"}, "resolution": {"type": "string", "description": "可选:720p 或 1080p(部分模型支持,默认上游)"}, }, "required": ["prompt"], }, }, { "name": "check_video", "description": "查询视频生成任务状态。传入 generate_video 返回的 status_url。完成时返回视频下载地址。", "inputSchema": { "type": "object", "properties": { "status_url": {"type": "string", "description": "generate_video 返回的 status_url 完整地址"}, }, "required": ["status_url"], }, }, { "name": "download_video", "description": "把已生成的视频下载到本地文件。传入 check_video/generate_video 返回的 status_url,或直接传视频直链(video.url)。MCP 内部自动查状态并下载到指定目录(默认用户下载文件夹),成功后返回本地文件路径。", "inputSchema": { "type": "object", "properties": { "status_url": {"type": "string", "description": "status_url(推荐)或视频直链 url"}, "save_path": {"type": "string", "description": "可选:保存目录,默认系统下载文件夹"}, }, "required": ["status_url"], }, }, { "name": "list_image_models", "description": "列出平台可用的生图模型ID与名称。调用 generate_image 前建议先查。", "inputSchema": {"type": "object", "properties": {}, "required": []}, }, { "name": "generate_image", "description": "同步生成图片。调用 /v1/images/generations,提交后直接返回图片链接。model 可选——想用指定模型就填其 ID(如 kling-image-o1 / gpt-image-2 / gemini-3-pro-image),不填则列出全部可选生图模型与各自适合场景由你决定,平台不会替你默认选。同步返回图片地址,不用轮询。", "inputSchema": { "type": "object", "properties": { "prompt": {"type": "string", "description": "图片内容描述,越具体越好(主体/环境/画风/构图/光线等)"}, "model": {"type": "string", "description": "可选:模型ID,见 list_image_models,如 kling-image-o1 / gpt-image-2"}, "size": {"type": "string", "description": "可选:尺寸,如 1024x1024(部分模型支持,默认上游)"}, }, "required": ["prompt"], }, }, ] def _rpc(id_, result=None, error=None) -> str: msg = {"jsonrpc": "2.0", "id": id_} if error is not None: msg["error"] = error else: msg["result"] = result return json.dumps(msg, ensure_ascii=False) def _handle_call(name: str, args: dict) -> dict: if name == "list_video_models": return {"content": [{"type": "text", "text": _gbk_safe(_list_video_models())}]} if name == "generate_video": text = _generate_video( _clean(args.get("prompt", "")), _clean(args.get("model", "")), _clean(args.get("resolution", "")), ) return {"content": [{"type": "text", "text": _gbk_safe(text)}]} if name == "create_video": text = _create_video( _clean(args.get("prompt", "")), _clean(args.get("model", "")), _clean(args.get("resolution", "")), ) return {"content": [{"type": "text", "text": _gbk_safe(text)}]} if name == "check_video": return {"content": [{"type": "text", "text": _gbk_safe(_check_video(args.get("status_url", "")))}]} if name == "download_video": return {"content": [{"type": "text", "text": _gbk_safe(_download_video(_clean(args.get("status_url", "")), _clean(args.get("save_path", ""))))}]} if name == "list_image_models": return {"content": [{"type": "text", "text": _gbk_safe(_list_image_models())}]} if name == "generate_image": text = _generate_image( _clean(args.get("prompt", "")), _clean(args.get("model", "")), _clean(args.get("size", "")), ) return {"content": [{"type": "text", "text": _gbk_safe(text)}]} return {"content": [{"type": "text", "text": f"未知工具: {name}"}]} def main() -> None: if sys.platform == "win32": sys.stdout.reconfigure(encoding="utf-8") sys.stderr.reconfigure(encoding="utf-8") os.environ.setdefault("PYTHONUTF8", "1") _log("MCP server 启动") for raw in sys.stdin: raw = raw.strip() if not raw: continue try: msg = json.loads(raw) except json.JSONDecodeError: continue method = msg.get("method") mid = msg.get("id") params = msg.get("params") or {} if mid is None: continue if method == "initialize": print(_rpc(mid, { "protocolVersion": params.get("protocolVersion", "2024-11-05"), "capabilities": {"tools": {}}, "serverInfo": {"name": "aicraft-video", "version": "1.0.0"}, }), flush=True) elif method == "ping": print(_rpc(mid, {}), flush=True) elif method == "tools/list": print(_rpc(mid, {"tools": TOOLS}), flush=True) elif method == "tools/call": name = params.get("name", "") args = params.get("arguments") or {} print(_rpc(mid, _handle_call(name, args)), flush=True) else: print(_rpc(mid, error={"code": -32601, "message": f"未知方法: {method}"}), flush=True) if __name__ == "__main__": main()