1.前言
中药信息的发展是一个跨越数千年的演进过程，它不仅反映了中医药学自身的理论体系构建，也体现了人类记录、传播和利用知识方式的变革。从原始社会的口耳相传，到甲骨金石的刻画，再到纸张印刷的普及，直至现代的数字化技术，每一种信息载体的革新都推动了中医药知识的积累与传播。
我这里也有一个中药信息发展时间
然而据广州调查显示，近90%小学生对中医药"一问三不知"，仅知"针灸""凉茶"等概念，无法辨识决明子等常用药材。
北京中医药大学调查指出，青少年普遍缺乏中药基础知识。
之前也给大家介绍过一个使用dify来实现的一个中医药的工作流。之前的工作流主要是实现了中药的药理介绍，中药的图片展示。本期我们给大家制作一个短视频的中药介绍MCP。那么我们看一下生成的中药的视频介绍。
我们看一下cherry studio生成的效果
那么这样的MCP 是如何制作的呢？下面给大家介绍一下。
2.MCP-Server
API 代码编写
这里我们用到火山引擎的deepseek-v3模型已经文生视频、文生图模型。我们可以在火山引擎上开通这些模型
接下来我们找到火山引擎的api 文档 https://www.volcengine.com/docs/82379/1587798 找到最新的模型使用文档
找到对应的API 文件。
文字转语音的我这里就没有使用火山引擎提供的文字转语音。我们用了免费的edgetts,这个免费的TTS 我在Cloudflare 部署了一个edge-tts-openai-cf-worker 代理从而实现免费访问微软的TTS。关于这块大家可以看我之前的文章。【技术分享】Edge-TTS与Cloudflare Worker结合，免费TTS服务轻松搭建！
编写的MCP 代码如下
zhongyao_mcp_server.py
# zhongyao\_mcp\_server.py
import time
import json
import requests
import configparser
import os
import datetime
import random
import tempfile
import platform
import asyncio # 新增
import httpx # 新增: 用于异步HTTP请求
from typing importAny, Dict
from openai import OpenAI
from mcp.server.fastmcp import FastMCP
from qcloud\_cos import CosConfig, CosS3Client
# 导入视频处理库
try:
import moviepy.editor as mpe
from moviepy.config import change\_settings;
except ImportError:
print("错误: moviepy 库未安装。请运行 'pip install moviepy' 安装。")
print("注意: moviepy 依赖 ffmpeg，请确保您的系统中已安装 ffmpeg。")
mpe = None
# 创建MCP服务器实例
mcp = FastMCP("Zhongyao AI Generation Server")
# --- 全局配置 (将从 config.ini 加载) ---
API\_KEY = None
BASE\_URL = None
DEFAULT\_CHAT\_MODEL = None
DEFAULT\_IMAGE\_MODEL = None
DEFAULT\_VIDEO\_MODEL = None
TTS\_API\_KEY = None
TTS\_BASE\_URL = None
COS\_REGION = None
COS\_SECRET\_ID = None
COS\_SECRET\_KEY = None
COS\_BUCKET = None
IMAGEMAGICK\_BINARY = None
FONT\_PATH = None
VIDEO\_GENERATION\_TIMEOUT = None
# --- 加载配置 ---
defload\_config():
"""从 config.ini 文件加载配置"""
global API\_KEY, BASE\_URL, DEFAULT\_CHAT\_MODEL, DEFAULT\_IMAGE\_MODEL, DEFAULT\_VIDEO\_MODEL
global TTS\_API\_KEY, TTS\_BASE\_URL, COS\_REGION, COS\_SECRET\_ID, COS\_SECRET\_KEY, COS\_BUCKET
global IMAGEMAGICK\_BINARY, FONT\_PATH
global VIDEO\_GENERATION\_TIMEOUT
config = configparser.ConfigParser()
config\_file = 'config.ini'
ifnot os.path.exists(config\_file):
print(f"错误: 配置文件 '{config\_file}' 未找到。请根据文档创建。")
return
config.read(config\_file)
try:
VIDEO\_GENERATION\_TIMEOUT = config.getint('Models', 'video\_generation\_timeout', fallback=480)
API\_KEY = config.get('API', 'api\_key', fallback=None)
BASE\_URL = config.get('API', 'base\_url', fallback='https://ark.cn-beijing.volces.com/api/v3')
DEFAULT\_CHAT\_MODEL = config.get('Models', 'chat\_model', fallback='deepseek-V3')
DEFAULT\_IMAGE\_MODEL = config.get('Models', 'image\_model', fallback='doubao-seedream-3-0-t2i-250415')
DEFAULT\_VIDEO\_MODEL = config.get('Models', 'video\_model', fallback='doubao-seedance-1-0-lite-t2v-250428')
TTS\_API\_KEY = config.get('edgetts', 'tts\_api\_key', fallback=None)
TTS\_BASE\_URL = config.get('edgetts', 'tts\_base\_url', fallback=None)
COS\_REGION = config.get('common', 'cos\_region', fallback=None)
COS\_SECRET\_ID = config.get('common', 'cos\_secret\_id', fallback=None)
COS\_SECRET\_KEY = config.get('common', 'cos\_secret\_key', fallback=None)
COS\_BUCKET = config.get('common', 'cos\_bucket', fallback=None)
IMAGEMAGICK\_BINARY = config.get('common', 'imagemagick\_binary', fallback=None)
FONT\_PATH = config.get('common', 'font\_path', fallback=None)
print("配置已从 config.ini 成功加载。")
print(f"视频生成任务超时设置为: {VIDEO\_GENERATION\_TIMEOUT} 秒")
ifnot API\_KEY or'YOUR\_API\_KEY'in API\_KEY:
print("警告: 'config.ini' 中的 [API] api\_key 未设置或仍为占位符。")
API\_KEY = None
ifnot COS\_SECRET\_ID or'YOUR\_COS\_SECRET\_ID'in COS\_SECRET\_ID:
print("警告: 'config.ini' 中的 [common] COS配置 未正确设置。TTS和视频合成功能将不可用。")
except (configparser.NoSectionError, configparser.NoOptionError) as e:
print(f"读取配置文件时出错: {e}")
# --- 程序执行流程 ---
# 1. 首先加载配置
load\_config()
# 2. 根据加载的配置设置 MoviePy
if mpe:
print(f"当前操作系统: {platform.system()}")
if IMAGEMAGICK\_BINARY:
IMAGEMAGICK\_BINARY = os.path.normpath(IMAGEMAGICK\_BINARY)
change\_settings({"IMAGEMAGICK\_BINARY": IMAGEMAGICK\_BINARY})
print(f"MoviePy ImageMagick 已配置: {IMAGEMAGICK\_BINARY}")
else:
print("警告: 'imagemagick\_binary' 未在 config.ini 中配置。字幕功能可能失败。")
if FONT\_PATH:
print(f"MoviePy 字体已配置: {FONT\_PATH}")
if platform.system() != "Windows"andnot os.path.exists(FONT\_PATH):
print(f"警告: 字体文件 '{FONT\_PATH}' 不存在！请检查 config.ini 中的 'font\_path' 配置。")
else:
print("警告: 'font\_path' 未在 config.ini 中配置。字幕可能使用默认字体或失败。")
# --- 提示词模板 (无变化) ---
PROMPT\_TEMPLATES = {
"info": { "system": "...", "user": "..." }, # 内容省略
"summary": { "system": "...", "user": "..." }, # 内容省略
"image": { "prompt": "..." }, # 内容省略
"video": { "prompt": "..." } # 内容省略
}
# 为了简洁，这里省略了模板的具体内容，实际代码中它们是存在的。
PROMPT\_TEMPLATES = {
"info": {
"system": "你是一个专业的中医药专家，请提供准确、详细且格式正确的中药材信息。",
"user": """请以JSON格式返回关于中药材"{herb\_name}"的详细信息，必须包含以下字段：
1. "name": 药材名称 (string)
2. "property": 药性, 例如: '寒', '热', '温', '凉' (string)
3. "taste": 药味, 例如: '酸', '苦', '甘', '辛', '咸' (list of strings)
4. "meridian": 归经, 例如: '肝经', '心经' (list of strings)
5. "function": 功效主治 (string)
6. "usage\_and\_dosage": 用法用量 (string)
7. "contraindications": 使用禁忌 (string)
8. "description": 简要描述，介绍药材来源和形态特征 (string)
请确保返回的是一个结构完整的、合法的JSON对象，不要在JSON前后添加任何多余的文字或解释。
"""
},
"summary": {
"system": "你是一位短视频文案专家，擅长将复杂信息提炼成简洁、精炼、引人入胜的口播稿，严格控制时长。",
"user": """请根据以下关于中药'{herb\_name}'的JSON信息，撰写一段**长度严格控制在20到30字之间**的口播文案，以适配一个10秒左右的短视频。文案需要流畅、易于听懂，并突出该药材最核心的功效。最终输出纯文本，不要包含任何标题或额外说明。
中药信息如下：
{herb\_info}
"""
},
"image": {
"prompt": "一张关于中药'{herb\_name}'的高清摄影照片，展示其作为药材的真实形态、颜色和纹理细节。背景干净纯白，光线明亮均匀，突出药材本身，具有百科全书式的专业质感。"
},
"video": {
"prompt": "一段关于中药'{herb\_name}'的短视频。视频风格：纪录片、特写镜头。画面内容：首先是{herb\_name}药材的特写镜头，缓慢旋转展示细节；然后展示其生长的自然环境；最后是它被用于传统中医的场景，比如煎药或者入药。整个视频节奏舒缓，配乐为典雅的中国古典音乐。"
}
}
# --- 辅助函数 ---
definitialize\_client():
ifnot API\_KEY:
raise ValueError("豆包 API key (api\_key) is required. Please set it in config.ini.")
return OpenAI(api\_key=API\_KEY, base\_url=BASE\_URL)
def\_generate\_timestamp\_filename(extension='mp3'):
timestamp = datetime.datetime.now().strftime("%Y%m%d%H%M%S")
random\_number = random.randint(1000, 9999)
filename = f"{timestamp}\_{random\_number}.{extension}"
return filename
# 注意: COS 上传仍然是同步阻塞操作，对于大型文件可能也需要优化，但暂时保持原样
def\_upload\_to\_cos\_from\_memory(file\_content: bytes, file\_name: str) -> Dict[str, Any]:
try:
config = CosConfig(Region=COS\_REGION, SecretId=COS\_SECRET\_ID, SecretKey=COS\_SECRET\_KEY)
client = CosS3Client(config)
response = client.put\_object(
Bucket=COS\_BUCKET,
Body=file\_content,
Key=file\_name,
EnableMD5=False
)
if response and response.get('ETag'):
url = f"https://{COS\_BUCKET}.cos.{COS\_REGION}.myqcloud.com/{file\_name}"
return {"success": True, "url": url, "etag": response['ETag']}
else:
return {"success": False, "error": f"Upload to COS failed. Response: {response}"}
except Exception as e:
return {"success": False, "error": f"An error occurred during COS upload: {str(e)}"}
# --- MODIFIED: 将视频合成改为异步函数 ---
asyncdef\_combine\_video\_audio\_text(video\_url: str, audio\_url: str, subtitle\_text: str, herb\_name: str) -> Dict[str, Any]:
"""
将视频、音频和文字（字幕）合成为一个新的视频文件，并上传到COS。
这是 CPU 和 IO 密集型操作，我们将它放在 asyncio 的 executor 中运行以避免阻塞事件循环。
"""
ifnot mpe:
return {"success": False, "error": "MoviePy library is not available. Cannot combine video."}
ifnotall([COS\_REGION, COS\_SECRET\_ID, COS\_SECRET\_KEY, COS\_BUCKET]):
return {"success": False, "error": "COS configuration is missing. Cannot upload final video."}
ifnot IMAGEMAGICK\_BINARY:
return {"success": False, "error": "ImageMagick binary path is not configured in config.ini. Cannot generate subtitles."}
ifnot FONT\_PATH:
return {"success": False, "error": "Font path is not configured in config.ini. Cannot generate subtitles."}
loop = asyncio.get\_running\_loop()
# 将所有阻塞操作封装在一个函数中，以便在 executor 中运行
defblocking\_operations():
video\_clip = audio\_clip = final\_clip = None
try:
with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as temp\_video\_file, \
tempfile.NamedTemporaryFile(suffix=".mp3", delete=False) as temp\_audio\_file:
print("Downloading video and audio for processing...")
# 使用同步的 requests
video\_content = requests.get(video\_url, stream=True).content
temp\_video\_file.write(video\_content)
temp\_video\_file.flush()
audio\_content = requests.get(audio\_url, stream=True).content
temp\_audio\_file.write(audio\_content)
temp\_audio\_file.flush()
print("Combining video, audio, and subtitles with MoviePy...")
video\_clip = mpe.VideoFileClip(temp\_video\_file.name)
audio\_clip = mpe.AudioFileClip(temp\_audio\_file.name)
final\_clip = video\_clip.set\_audio(audio\_clip)
if final\_clip.duration > audio\_clip.duration:
final\_clip = final\_clip.subclip(0, audio\_clip.duration)
txt\_clip = mpe.TextClip(
subtitle\_text, fontsize=40, color='yellow', font=FONT\_PATH,
bg\_color='rgba(0, 0, 0, 0.5)', size=(final\_clip.w * 0.9, None), method='caption'
)
txt\_clip = txt\_clip.set\_position(('center', 'bottom')).set\_duration(final\_clip.duration)
video\_with\_subs = mpe.CompositeVideoClip([final\_clip, txt\_clip])
print("Exporting final video...")
final\_filename = f"final\_{\_generate\_timestamp\_filename('mp4')}"
final\_filepath\_temp = os.path.join(tempfile.gettempdir(), final\_filename)
video\_with\_subs.write\_videofile(final\_filepath\_temp, codec="libx264", audio\_codec="aac")
print(f"Uploading final video '{final\_filename}' to COS...")
withopen(final\_filepath\_temp, 'rb') as f\_final:
final\_video\_content = f\_final.read()
# COS 上传也是阻塞的
upload\_result = \_upload\_to\_cos\_from\_memory(final\_video\_content, final\_filename)
try:
os.remove(final\_filepath\_temp)
except OSError as e:
print(f"Error removing temporary file {final\_filepath\_temp}: {e}")
return upload\_result
except Exception as e:
import traceback
traceback.print\_exc()
return {"success": False, "error": f"Failed during video combination process: {str(e)}"}
finally:
if video\_clip: video\_clip.close()
if audio\_clip: audio\_clip.close()
if final\_clip: final\_clip.close()
# 在默认的 executor (线程池) 中运行阻塞函数
result = await loop.run\_in\_executor(None, blocking\_operations)
return result
# --- 核心高层工具 ---
# 注意：工具函数本身不需要是 async，mcp 会处理。
# 但它们调用的底层函数如果是 IO 密集型，最好是 async。
@mcp.tool()
defget\_chinese\_herb\_info(herb\_name: str, model: str = None) -> Dict[str, Any]:
model\_to\_use = model or DEFAULT\_CHAT\_MODEL
try:
system\_prompt = PROMPT\_TEMPLATES["info"]["system"]
user\_prompt = PROMPT\_TEMPLATES["info"]["user"].format(herb\_name=herb\_name)
# \_chat\_completion 是同步的，对于快速的API调用可以接受
response = \_chat\_completion(prompt=user\_prompt, system\_prompt=system\_prompt, model=model\_to\_use)
ifnot response.get("success"): return response
raw\_content = response.get("content", "")
if"```json"in raw\_content:
raw\_content = raw\_content.split("```json")[1].split("```")[0].strip()
try:
return {"success": True, "data": json.loads(raw\_content)}
except json.JSONDecodeError as e:
return {"success": False, "error": f"Failed to parse model response as JSON: {e}", "raw\_content": raw\_content}
except Exception as e:
return {"success": False, "error": f"An unexpected error occurred while getting herb info: {str(e)}"}
@mcp.tool()
defget\_chinese\_herb\_image(herb\_name: str, size: str = "1024x1024", model: str = None) -> Dict[str, Any]:
model\_to\_use = model or DEFAULT\_IMAGE\_MODEL
try:
prompt = PROMPT\_TEMPLATES["image"]["prompt"].format(herb\_name=herb\_name)
result = \_text\_to\_image(prompt=prompt, size=size, model=model\_to\_use)
if result.get("success"):
return {"success": True, "herb\_name": herb\_name, "image\_url": result.get("image\_url")}
else:
return result
except Exception as e:
return {"success": False, "error": f"An unexpected error occurred while generating herb image: {str(e)}"}
# --- MODIFIED: 将此工具改为 async ---
@mcp.tool()
asyncdefget\_chinese\_herb\_video(herb\_name: str, duration: str = "8", ratio: str = "16:9", model: str = None) -> Dict[str, Any]:
model\_to\_use = model or DEFAULT\_VIDEO\_MODEL
try:
prompt = PROMPT\_TEMPLATES["video"]["prompt"].format(herb\_name=herb\_name)
# 调用异步版本的 \_text\_to\_video
result = await \_text\_to\_video(prompt=prompt, duration=duration, ratio=ratio, model=model\_to\_use)
if result.get("success"):
return {"success": True, "herb\_name": herb\_name, "video\_url": result.get("video\_url"), "task\_id": result.get("task\_id")}
else:
return result
except Exception as e:
return {"success": False, "error": f"An unexpected error occurred while generating herb video: {str(e)}"}
@mcp.tool()
defgenerate\_audio\_from\_text(text: str, voice: str = "zh-CN-XiaoxiaoNeural", speed: float = 1.0) -> Dict[str, Any]:
ifnotall([TTS\_BASE\_URL, COS\_REGION, COS\_SECRET\_ID, COS\_SECRET\_KEY, COS\_BUCKET]):
return {"success": False, "error": "TTS or COS configuration is missing."}
try:
tts\_client = OpenAI(api\_key=TTS\_API\_KEY, base\_url=TTS\_BASE\_URL)
response = tts\_client.audio.speech.create(model="tts-1", input=text, voice=voice, response\_format="mp3", speed=speed)
upload\_result = \_upload\_to\_cos\_from\_memory(response.content, \_generate\_timestamp\_filename('mp3'))
if upload\_result.get("success"):
return {"success": True, "audio\_url": upload\_result.get("url")}
else:
return upload\_result
except Exception as e:
return {"success": False, "error": f"An unexpected error occurred during TTS generation or upload: {str(e)}"}
# --- MODIFIED: 关键修改，将主工具函数改为 async def ---
@mcp.tool()
asyncdefgenerate\_herb\_short\_video(herb\_name: str) -> Dict[str, Any]:
print(f"--- 开始为 '{herb\_name}' 生成完整短视频 ---")
try:
print(f"[1/5] 正在获取 '{herb\_name}' 的详细信息...")
# 这个调用是同步的，但通常很快
info\_result = get\_chinese\_herb\_info(herb\_name)
ifnot info\_result.get("success"):
return {"success": False, "error": f"步骤1失败: {info\_result.get('error')}"}
herb\_info\_data = info\_result["data"]
print(f"成功获取信息。")
print(f"[2/5] 正在为 '{herb\_name}' 生成口播文案...")
summary\_prompt = PROMPT\_TEMPLATES["summary"]["user"].format(herb\_name=herb\_name, herb\_info=json.dumps(herb\_info\_data, ensure\_ascii=False, indent=2))
# 同步调用
summary\_result = \_chat\_completion(prompt=summary\_prompt, system\_prompt=PROMPT\_TEMPLATES["summary"]["system"], model=DEFAULT\_CHAT\_MODEL)
ifnot summary\_result.get("success"):
return {"success": False, "error": f"步骤2失败: {summary\_result.get('error')}"}
summary\_text = summary\_result["content"].strip()
print(f"成功生成文案: {summary\_text[:50]}...")
print(f"[3/5] 正在为文案生成语音...")
# 同步调用
audio\_result = generate\_audio\_from\_text(text=summary\_text)
ifnot audio\_result.get("success"):
return {"success": False, "error": f"步骤3失败: {audio\_result.get('error')}"}
audio\_url = audio\_result["audio\_url"]
print(f"成功生成语音，URL: {audio\_url}")
print(f"[4/5] 正在生成 '{herb\_name}' 的背景视频...")
# --- MODIFIED: 使用 await 调用异步函数 ---
video\_result = await get\_chinese\_herb\_video(herb\_name, duration="8")
ifnot video\_result.get("success"):
return {"success": False, "error": f"步骤4失败: {video\_result.get('error')}"}
video\_url = video\_result["video\_url"]
print(f"成功生成视频，URL: {video\_url}")
print(f"[5/5] 正在合成最终视频...")
# --- MODIFIED: 使用 await 调用异步函数 ---
final\_video\_result = await \_combine\_video\_audio\_text(video\_url, audio\_url, summary\_text, herb\_name)
if final\_video\_result.get("success"):
print(f"--- 成功为 '{herb\_name}' 生成完整短视频 ---")
return {"success": True, "message": f"Successfully generated a complete short video for {herb\_name}.", "final\_video\_url": final\_video\_result.get("url")}
else:
return {"success": False, "error": f"步骤5失败: {final\_video\_result.get('error')}"}
except Exception as e:
import traceback
traceback.print\_exc()
return {"success": False, "error": f"生成短视频过程中发生意外错误: {str(e)}"}
# --- 底层API工具 ---
@mcp.tool()
defset\_api\_key(api\_key: str) -> str:
global API\_KEY
API\_KEY = api\_key
return"API key set successfully for this session."
# 同步版本，用于快速调用
def\_chat\_completion(prompt: str, model: str, system\_prompt: str) -> Dict[str, Any]:
try:
response = initialize\_client().chat.completions.create(model=model, messages=[{"role": "system", "content": system\_prompt}, {"role": "user", "content": prompt}])
return {"success": True, "content": response.choices[0].message.content}
except Exception as e:
return {"success": False, "error": f"Text generation failed: {str(e)}"}
# 同步版本
def\_text\_to\_image(prompt: str, size: str, model: str) -> Dict[str, Any]:
try:
response = initialize\_client().images.generate(model=model, prompt=prompt, size=size, response\_format="url", n=1)
return {"success": True, "image\_url": response.data[0].url} if response.data else {"success": False, "error": "No image data returned."}
except Exception as e:
return {"success": False, "error": f"Image generation failed: {str(e)}"}
# --- MODIFIED: 视频生成函数改为 async，并使用 httpx ---
asyncdef\_text\_to\_video(prompt: str, duration: str, ratio: str, model: str) -> Dict[str, Any]:
try:
if ratio and"--ratio"notin prompt: prompt += f" --ratio {ratio}"
if duration and"--duration"notin prompt and"--dur"notin prompt: prompt += f" --duration {duration}"
headers = {"Content-Type": "application/json", "Authorization": f"Bearer {API\_KEY}"}
request\_data = {"model": model, "content": [{"type": "text", "text": prompt}]}
# 使用 httpx 进行异步请求
asyncwith httpx.AsyncClient() as client:
create\_resp = await client.post(f"{BASE\_URL}/contents/generations/tasks", headers=headers, json=request\_data, timeout=30.0)
if create\_resp.status\_code != 200:
return {"success": False, "error": f"Failed to create video task. Status: {create\_resp.status\_code}, Info: {create\_resp.text}"}
task\_id = create\_resp.json().get("id")
ifnot task\_id: return {"success": False, "error": "Could not get task ID."}
polling\_interval = 5
max\_retries = VIDEO\_GENERATION\_TIMEOUT // polling\_interval
for i inrange(max\_retries):
await asyncio.sleep(polling\_interval)
print(f"Checking video task status... Attempt {i+1}/{max\_retries}")
task\_resp = await client.get(f"{BASE\_URL}/contents/generations/tasks/{task\_id}", headers=headers, timeout=30.0)
if task\_resp.status\_code != 200: continue
task\_data = task\_resp.json()
status = task\_data.get("status")
if status == "succeeded":
return {"success": True, "video\_url": task\_data.get("content", {}).get("video\_url"), "task\_id": task\_id}
elif status in ("failed", "canceled"):
return {"success": False, "error": f"Video task status: {status}, Info: {task\_data.get('error')}"}
return {"success": False, "error": f"Video generation timed out after {VIDEO\_GENERATION\_TIMEOUT} seconds."}
except Exception as e:
return {"success": False, "error": f"Video generation failed: {str(e)}"}
# --- 服务器入口 ---
defmain():
"""主函数入口点"""
print("Zhongyao AI Generation Server is running.")
mcp.settings.host = "0.0.0.0"
mcp.settings.port = 8003
# 不再需要错误的超时设置
mcp.run(transport="sse")
if \_\_name\_\_ == "\_\_main\_\_":
# 在运行前，确保已安装 httpx
# pip install httpx
main()
这代码配置文件我们使用config.ini 内容如下
# config.ini
[API]
api\_key = YOUR\_API\_KEY\_HERE
base\_url = https://ark.cn-beijing.volces.com/api/v3
[Models]
# 语言模型
chat\_model = deepseek-v3-250324
# 文生图模型
image\_model = doubao-seedream-3-0-t2i-250415
# 文生视频模型
video\_model = doubao-seedance-1-0-lite-t2v-250428
[edgetts]
tts\_api\_key =zhouhuizhou
tts\_base\_url=https://edgettsapi.duckcloud.fun/v1
[common]
cos\_region = ap-nanjing # 腾讯云OSS存储Region
cos\_secret\_id = AKID003XXXXXXgO9qPl # 腾讯云OSS存储SecretId
cos\_secret\_key = IZhavCXXXXXXXXXXX6i9NXUFqGTUOFvS # 腾讯云OSS存储SecretKey
cos\_bucket =tts-1258720957 # 腾讯云OSS存储bucket
# font\_path = /usr/share/fonts/truetype/wqy/wqy-zenhei.ttc
# [Windows 用户配置]
imagemagick\_binary = D:\develop\ImageMagick-7.1.1-Q16\magick.exe
font\_path = SimHei
# [Linux 用户配置]
# imagemagick\_binary = /home/ImageMagick/magick
# font\_path = /usr/share/fonts/truetype/wqy/wqy-zenhei.ttc
这个MCP 依赖包我们使用uv 管理所以pyproject.toml
dependencies = [
"mcp[cli]>=1.9.4", # 添加requests依赖
"requests>=2.31.0",
"openai>=1.86.0",
"cos-python-sdk-v5==1.9.33",
"moviepy==1.0.3",
"httpx",
]
ImageMagick
这个程序用到moviepy包，moviepy 是一个用于视频编辑的 Python 库，可处理视频剪辑、拼接、添加音频、特效等操作。它支持多种视频格式，提供了直观的 API，适合自动化视频处理任务。以下是其主要功能：
视频剪辑 ：切割、合并视频片段
音频处理 ：添加背景音乐、调整音量
特效添加 ：文字叠加、滤镜、转场效果
格式转换 ：支持 MP4、AVI、GIF 等格式
批处理 ：自动化处理多个视频文件
此外我们还用到ImageMagick 软件，这个需要我们在本地电脑上安装
在 windows 安装
下载地址 https://imagemagick.org/script/download.php#windows
下载本地电脑上安装即可，我的电脑安装到D 盘
安装后我们需要设置或者检查一下环境变量是否设置成功
在linux 安装
下载地址
https://imagemagick.org/script/download.php
复制magick 文件到linux 服务器上。
然后给它授权
chmod 755 magick
启动
python3 zhongyao\_mcp\_server.py
启动运行
我们使用开发工具 trae 或者Visual Studio Code 都是可以 启动程序
3.验证及测试
我们这里使用trae 启动，所以我们用Cherry Studio客户端做验证测试。
Cherry Studio 配置
配置完成后我们可以在工具栏查看有6个工具可以调用的
Cherry Studio验证测试
我们在 Cherry Studio聊天窗口中输入 下面问题
请帮我生成一个 薄荷 中药带文字，语音，和视频的短视频，其中 api key ='719f1aec-xxxx-1fc26a95df73'
我们一步到位让它把API key设置好，后面让它自动完成中药短视频生成。
我们点击cherrry studio 生成的视频下载
下载后打开视频
程序控制台也打印出视频合成的地址
以上我们就完成了中药短视频的MCP 服务了，当然你也可以把这个打包改成studio的方式。关于studio 打包可以看我之前的文章
mcp-server案例分享-用豆包大模型 1.6 手搓文生图视频 MCP-server发布到PyPI官网
4.总结
今天主要带大家了解并实现了使用火山引擎相关模型制作中药短视频 MCP（模型上下文协议）服务的全流程。借助 MCP Server，我们解决了在中药信息展示方面数据分散、展示形式单一等问题，为用户提供了一种集文字、语音和视频于一体的直观、生动的中药科普方式。
我们详细介绍了如何开通火山引擎的 deepseek - v3 等模型，找到对应的 API 文档和文件，编写 MCP 代码，配置 config.ini 文件和 pyproject.toml 文件。同时，还说明了 moviepy 包和 ImageMagick 软件的作用及安装方法，包括在 Windows 和 Linux 系统上的安装步骤。另外，我们也展示了如何使用开发工具启动程序，以及如何利用 Cherry Studio 客户端进行验证测试。
通过本文的方案，开发者可以轻松搭建自己的中药短视频 MCP 服务，为中医药科普和推广添加强大的 AI 能力。感兴趣的小伙伴可以按照本文步骤去尝试制作自己的 MCP 服务，也可以将其打包改成 studio 的方式。今天的分享就到这里结束了，我们下一篇文章见。
dify案例分享-用 Dify 一键生成 长安的荔枝金句 HTML 页面，三步搞定！
谷歌 Gemini CLI 重磅发布！手把手教你用命令行玩转 AI 多模态开发（附保姆级教程）
实测！豆包 AI 播客没下载按钮？F12 抓包 + 剪映处理全教程