Synphony MCP

Synphony MCP（可译为“Synphony MCP系统”或根据具体上下文调整，若“Synphony”为特定品牌或产品名，则保持原样）

一个集成Hugging Face Hub的FastMCP服务器，用于管理视频数据集。

快速入门

安装

1. 克隆仓库：

git clone https://github.com/rukasuamarike/synphony-mcp.git
cd synphony-mcp

1. 安装依赖项：

pip install -r requirements.txt

1. 复制并配置环境变量：

cp .env.example .env
# Edit .env with your settings

1. 测试服务器：

python server.py

Claude 桌面版安装

1. 添加到您的Claude桌面MCP配置文件中（~/Library/Application Support/Claude/claude_desktop_config.json （在 macOS 上）：

{
  "mcpServers": {
    "synphony-mcp": {
      "command": "python",
      "args": ["/absolute/path/to/synphony-mcp/server.py"],
      "env": {
        "SYNPHONY_ROOT_DIR": "/path/to/your/video/directory"
      }
    }
  }
}

1. 重启Claude桌面版

可用工具

• list_videos - 列出目录中的视频文件
• get_server_info - 获取服务器配置和状态
• validate_setup - 验证服务器设置和配置

______________________________________________________________________

实施细节

好极了——让我们添加一个专用的FastMCP工具，该工具使用Hub的Python库将本地视频上传到Hugging Face数据集存储库。这利用HfApi进行可靠上传（相比HfFileSystem，性能更优，因此推荐使用），同时仍然支持fsspec路径，以防你之后需要。

计划

1. 添加Hugging Face Hub配置

• 环境变量：HF_TOKEN，HF_DATASET_REPO_ID。 • 验证令牌和仓库是否存在，如果不存在则创建数据集仓库。

1. 选择文件

• 验证SYNPHONY_ROOT_DIR下的相对路径。 • 确保仅使用视频扩展功能。

1. 上传到中心（或上传至集线器）

• 使用 HfApi.upload_file 方法，为每个文件指定目标路径布局为 datasets/{repo_id}/videos/… • 返回每个文件的状态，并捕获其中的错误。

1. 安全

• 不要在代码中硬编码HF令牌；请使用环境变量。 • 防止超出ROOT_DIR的路径遍历。

以下是您可以添加到现有Synphony MCP服务器中的工具。它增加了一个名为upload_to_hf_datasets的新上传工具。# synphony_mcp_server.py（添加了Hugging Face上传功能）

要求：

使用 pip 安装 fastmcp、huggingface_hub 和 python-dotenv

环境：

SYNPHONY_ROOT_DIR=/本地数据路径

HF_TOKEN=hf\_... # 您具有写入权限的个人或组织令牌

HF_DATASET_REPO_ID=username/my-dataset # 目标数据集仓库（必须已存在或将被创建）

import os（注：这行代码本身在中文语境下无需翻译，但若要解释其含义，可译为“导入os模块”，不过通常在编程语境中直接保留原样） 从 pathlib 模块导入 Path 从 typing 模块导入 List、Dict 和 Optional

从fastmcp导入FastMCP from fastmcp.types import ToolError 翻译为中文是：来自 fastmcp.types 的 ToolError（工具错误） 从 dotenv 模块导入 load_dotenv 函数 从huggingface_hub导入HfApi，HfHubHTTPError

\load_dotenv()\ 翻译成中文是：“加载环境变量文件”。

mcp = FastMCP("Synphony MCP 🚀") 翻译为中文是：mcp = 快速MCP("Synphony MCP 🚀")

ROOT_DIR = os.environ.get("SYNPHONY_ROOT_DIR", os.getcwd()) 翻译为中文是：ROOT_DIR = 获取环境变量 "SYNPHONY_ROOT_DIR" 的值，如果未设置则使用当前工作目录 HF_TOKEN = os.environ.get("HF_TOKEN") 翻译为中文是：HF_TOKEN = 从环境变量中获取 "HF_TOKEN" 的值 \HF_DATASET_REPO_ID = os.environ.get("HF_DATASET_REPO_ID")\ # 例如，"synphony/videos" 或 "username/my-dataset"

VIDEO_EXTS = { “.mp4”，“.mov”，“.mkv”，“.avi”，“.wmv”，“.flv”，“.webm”，“.m4v”，“.mpeg”，“.mpg”，“.3gp”，“.ts” } 最大上传批次数 = 50

def \_normalize_and_validate_path(candidate: str) -> Path: base = Path(ROOT_DIR).resolve() 翻译成中文是：基础路径 = 解析出 ROOT_DIR 的实际路径 \p = (base / candidate).resolve()\ 翻译成中文是：“p 等于 base 除以 candidate 后的解析结果”。不过，这里的“resolve”具体含义可能根据上下文有所不同，它可能表示路径解析、问题解决或其他类型的解析操作。在没有具体上下文的情况下，上述翻译是一个较为通用的表述 如果 \base\ 不在 \p\ 的父节点列表中且 \p\ 不等于 \base\： 抛出 ToolError 异常，信息为：“路径 '{candidate}' 不在 ROOT_DIR 之内” 返回 p

def \_is_video_file(path: Path) -> bool: （翻译为中文）定义一个函数 \_is_video_file，接收一个路径参数 path（类型为 Path），返回一个布尔值 返回路径后缀的小写形式是否在视频扩展名列表中

def \_ensure_hf_dataset_repo(api: HfApi, repo_id: str, token: str) -> None: """ 确保数据集仓库存在；如果不存在，请创建它。 """ 尝试： \# 如果未找到则会抛出异常 \api.repo_info(repo_id=repo_id, repo_type="dataset", token=token)\ 翻译成中文是：调用API获取仓库信息，其中仓库ID为\repo_id\，仓库类型为“数据集”，使用\token\作为身份验证令牌 除了捕获 HfHubHTTPError 异常 e： \# 如果仓库不存在（404），则创建它 如果 e.response 不为 None 且 e.response.status_code 等于 404： \api.create_repo(repo_id=repo_id, repo_type="dataset", token=token, exist_ok=True, private=True)\ 翻译成中文是：使用指定的令牌创建一个仓库，仓库ID为\repo_id\，仓库类型为“数据集”，允许仓库已存在（\exist_ok=True\），并设置为私有仓库（\private=True\） 否则： 提高；举起；饲养；抚养

@mcp.tool（注：这个翻译保持了原样的形式，因为“@mcp.tool”可能是一个特定的用户名、工具名或项目名，在中文中通常不直接翻译，而是保留原样。） def upload_to_hf_datasets( 路径列表：List\[str\]， target_dir: 可选的字符串类型，默认值为 "videos"， 提交信息（commit message）: 可选字符串（str）= "从Synphony MCP上传视频" 字典 """ 将选定的本地视频文件上传到 Hugging Face 数据集仓库。 \- paths：SYNPHONY_ROOT_DIR 内的相对路径列表 \- target_dir: 数据集仓库内的子目录，用于存放文件（默认：'videos'） \- commit_message：此批量上传的提交信息

Returns per-file status with repository and destination paths.
"""
if not HF_TOKEN:
    raise ToolError("HF_TOKEN not configured. Set HF_TOKEN in environment.")
if not HF_DATASET_REPO_ID:
    raise ToolError("HF_DATASET_REPO_ID not configured. Set HF_DATASET_REPO_ID in environment.")
if not isinstance(paths, list) or len(paths) == 0:
    raise ToolError("Provide a non-empty list of relative file paths.")
if len(paths) > MAX_UPLOAD_BATCH:
    raise ToolError(f"Too many files in one call. Max {MAX_UPLOAD_BATCH}.")

base = Path(ROOT_DIR).resolve()

# Validate files
valid_files: List[Path] = []
validations: List[Dict] = []
for p in paths:
    try:
        fp = _normalize_and_validate_path(p)
        if not fp.exists():
            validations.append({"path": p, "valid": False, "reason": "not found"})
            continue
        if not fp.is_file():
            validations.append({"path": p, "valid": False, "reason": "not a file"})
            continue
        if not _is_video_file(fp):
            validations.append({"path": p, "valid": False, "reason": "not a recognized video extension"})
            continue
        validations.append({"path": p, "valid": True})
        valid_files.append(fp)
    except ToolError as te:
        validations.append({"path": p, "valid": False, "reason": str(te)})
    except Exception as e:
        validations.append({"path": p, "valid": False, "reason": str(e)})

if not valid_files:
    return {
        "repo": HF_DATASET_REPO_ID,
        "root_dir": str(base),
        "validated": validations,
        "uploads": [],
        "summary": "No valid files to upload.",
    }

api = HfApi(token=HF_TOKEN)

# Ensure repo exists (create if missing, private by default)
try:
    _ensure_hf_dataset_repo(api, HF_DATASET_REPO_ID, HF_TOKEN)
except Exception as e:
    raise ToolError(f"Failed to ensure datasets repo '{HF_DATASET_REPO_ID}': {e}")

uploads: List[Dict] = []
# Upload files one-by-one (HfApi handles efficient chunking and retry)
for fp in valid_files:
    repo_path = fp.name if not target_dir else f"{target_dir.rstrip('/')}/{fp.name}"
    try:
        # Using upload_file for reliability (preferred over HfFileSystem put_file for performance)
        api.upload_file(
            path_or_fileobj=str(fp),
            path_in_repo=repo_path,
            repo_id=HF_DATASET_REPO_ID,
            repo_type="dataset",
            token=HF_TOKEN,
            commit_message=commit_message,
        )
        uploads.append({
            "source": str(fp),
            "dest": f"datasets/{HF_DATASET_REPO_ID}/{repo_path}",
            "status": "uploaded",
        })
    except Exception as e:
        uploads.append({
            "source": str(fp),
            "dest": f"datasets/{HF_DATASET_REPO_ID}/{repo_path}",
            "status": "error",
            "error": str(e),
        })

return {
    "repo": HF_DATASET_REPO_ID,
    "root_dir": str(base),
    "validated": validations,
    "uploads": uploads,
    "notes": [
        "Uses HfApi.upload_file for reliability and performance.",
        "For bulk commits, consider HfApi.upload_large_files or batch operations to reduce commit overhead.",
        "Never hardcode HF_TOKEN; store in environment or a secret manager.",
    ],
}

如何跑步 • 安装依赖：使用 pip 安装 fastmcp、huggingface_hub 和 python-dotenv • 设置环境： ▫ SYNPHONY_ROOT_DIR 指向您本地包含视频的文件夹 ▫ HF_TOKEN 拥有对您的账户/组织的写入访问权限 ▫ HF_DATASET_REPO_ID 类似于 username/my-dataset • 启动服务器：python synphony_mcp_server.py • 从MCP客户端调用工具并传递参数：

{ "name": "上传到HF数据集"， "arguments": { "路径": \["clips/robot_arm_001.mp4", "captures/test.mov"\]， "目标目录": "videos/robotics"， "commit_message": "初步的机器人视频片段" } }

注释 • 对于非常大的批次，您可能更倾向于一次性提交多个文件；我们可以切换到使用create_commit调用add/remove操作来进行分阶段提交。 • 如果您需要fsspec风格的用法，我们可以使用HfFileSystem.put_file添加一个替代路径，但文档建议使用HfApi以提高性能和可靠性：通过文件系统API（huggingface.co/93）与Hub进行交互。

希望我实现按日期/项目自动归档存储库文件夹，并在数据集中添加校验和元数据文件吗？我可以扩展这个工具，使其每次上传时都生成一个清单JSON文件。