创建 MCP 服务器

MCP 服务器是我们的 Pull Request 代理的核心。它提供了我们的代理将用于与 Hugging Face Hub 交互的工具，特别是用于读取和更新模型仓库标签。在本节中，我们将使用 FastMCP 和 Hugging Face Hub Python SDK 构建服务器。

理解 MCP 服务器架构

我们的 MCP 服务器提供两个基本工具：

工具	描述
`get_current_tags`	从模型仓库检索现有标签
`add_new_tag`	通过 pull request 向仓库添加新标签

这些工具抽象了 Hub API 交互的复杂性，并为我们的代理提供了清晰的接口。

MCP Server Tools

完整的 MCP 服务器实现

让我们一步步创建 mcp_server.py 文件。我们将逐步构建，以便您理解每个组件以及它们如何协同工作。

1. 导入和配置

首先，让我们设置所有必要的导入和配置。

#!/usr/bin/env python3
"""
Simplified MCP Server for HuggingFace Hub Tagging Operations using FastMCP
"""

import os
import json
from fastmcp import FastMCP
from huggingface_hub import HfApi, model_info, ModelCard, ModelCardData
from huggingface_hub.utils import HfHubHTTPError
from dotenv import load_dotenv

load_dotenv()

以上导入为我们构建 MCP 服务器提供了所需的一切。FastMCP 提供了服务器框架，而 huggingface_hub 导入则提供了与模型仓库交互的工具。

load_dotenv() 调用会自动从 .env 文件加载环境变量，从而便于在开发过程中管理 API 令牌等机密信息。

如果您正在使用 uv，可以在项目根目录中创建一个 .env 文件，如果使用 uv run 运行服务器，则无需使用 load_dotenv()。

接下来，我们将使用必要的凭据配置服务器并创建 FastMCP 实例。

# Configuration
HF_TOKEN = os.getenv("HF_TOKEN")

# Initialize HF API client
hf_api = HfApi(token=HF_TOKEN) if HF_TOKEN else None

# Create the FastMCP server
mcp = FastMCP("hf-tagging-bot")

此配置块完成三项重要任务：

从环境变量中检索 Hugging Face 令牌
创建一个经过身份验证的 API 客户端（仅当令牌可用时）
使用描述性名称初始化我们的 FastMCP 服务器

hf_api 的条件创建确保了即使没有令牌，我们的服务器也能启动，这对于测试基本结构非常有用。

2. 获取当前标签工具

现在让我们实现第一个工具——get_current_tags。这个工具从模型仓库中检索现有标签。

@mcp.tool()
def get_current_tags(repo_id: str) -> str:
    """Get current tags from a HuggingFace model repository"""
    print(f"🔧 get_current_tags called with repo_id: {repo_id}")

    if not hf_api:
        error_result = {"error": "HF token not configured"}
        json_str = json.dumps(error_result)
        print(f"❌ No HF API token - returning: {json_str}")
        return json_str

该函数首先进行验证——检查我们是否有经过身份验证的 API 客户端。请注意我们如何返回 JSON 字符串而不是 Python 对象。这对于 MCP 通信至关重要。

所有 MCP 工具都必须返回字符串，而不是 Python 对象。这就是我们使用 json.dumps() 将结果转换为 JSON 字符串的原因。这确保了 MCP 服务器和客户端之间可靠的数据交换。

让我们继续 get_current_tags 函数的主要逻辑：

    try:
        print(f"📡 Fetching model info for: {repo_id}")
        info = model_info(repo_id=repo_id, token=HF_TOKEN)
        current_tags = info.tags if info.tags else []
        print(f"🏷️ Found {len(current_tags)} tags: {current_tags}")

        result = {
            "status": "success",
            "repo_id": repo_id,
            "current_tags": current_tags,
            "count": len(current_tags),
        }
        json_str = json.dumps(result)
        print(f"✅ get_current_tags returning: {json_str}")
        return json_str

    except Exception as e:
        print(f"❌ Error in get_current_tags: {str(e)}")
        error_result = {"status": "error", "repo_id": repo_id, "error": str(e)}
        json_str = json.dumps(error_result)
        print(f"❌ get_current_tags error returning: {json_str}")
        return json_str

此实现遵循清晰的模式：

使用 Hugging Face Hub API 获取数据
处理响应以提取标签信息
以一致的 JSON 格式构建结果
以详细的错误消息优雅地处理错误

大量的日志记录可能看起来有些多余，但在服务器运行时有助于调试和监控。请记住，您的应用程序将自主地响应来自 Hub 的事件，因此您无法实时查看日志。

3. 添加新标签工具

现在来看更复杂的工具——add_new_tag。该工具通过创建拉取请求向仓库添加新标签。让我们从初始设置和验证开始：

@mcp.tool()
def add_new_tag(repo_id: str, new_tag: str) -> str:
    """Add a new tag to a HuggingFace model repository via PR"""
    print(f"🔧 add_new_tag called with repo_id: {repo_id}, new_tag: {new_tag}")

    if not hf_api:
        error_result = {"error": "HF token not configured"}
        json_str = json.dumps(error_result)
        print(f"❌ No HF API token - returning: {json_str}")
        return json_str

与我们的第一个工具类似，我们从验证开始。现在让我们获取当前的仓库状态以检查标签是否已存在：

    try:
        # Get current model info and tags
        print(f"📡 Fetching current model info for: {repo_id}")
        info = model_info(repo_id=repo_id, token=HF_TOKEN)
        current_tags = info.tags if info.tags else []
        print(f"🏷️ Current tags: {current_tags}")

        # Check if tag already exists
        if new_tag in current_tags:
            print(f"⚠️ Tag '{new_tag}' already exists in {current_tags}")
            result = {
                "status": "already_exists",
                "repo_id": repo_id,
                "tag": new_tag,
                "message": f"Tag '{new_tag}' already exists",
            }
            json_str = json.dumps(result)
            print(f"🏷️ add_new_tag (already exists) returning: {json_str}")
            return json_str

本节演示了一个重要原则：**行动前验证**。我们检查标签是否已存在，以避免创建不必要的拉取请求。

在进行更改之前，请务必检查当前状态。这可以防止重复工作并提供更好的用户反馈。当创建拉取请求时，这一点尤为重要，因为重复的拉取请求可能会使仓库混乱。

接下来，我们将准备更新后的标签列表并处理模型卡。

        # Add the new tag to existing tags
        updated_tags = current_tags + [new_tag]
        print(f"🆕 Will update tags from {current_tags} to {updated_tags}")

        # Create model card content with updated tags
        try:
            # Load existing model card
            print(f"📄 Loading existing model card...")
            card = ModelCard.load(repo_id, token=HF_TOKEN)
            if not hasattr(card, "data") or card.data is None:
                card.data = ModelCardData()
        except HfHubHTTPError:
            # Create new model card if none exists
            print(f"📄 Creating new model card (none exists)")
            card = ModelCard("")
            card.data = ModelCardData()

        # Update tags - create new ModelCardData with updated tags
        card_dict = card.data.to_dict()
        card_dict["tags"] = updated_tags
        card.data = ModelCardData(**card_dict)

本节处理模型卡管理。我们首先尝试加载现有模型卡，如果不存在，则创建一个新的。这确保了我们的工具适用于任何仓库，即使它是空的。

模型卡（README.md）包含仓库元数据，包括标签。通过更新模型卡数据并创建拉取请求，我们遵循 Hugging Face 标准的工作流程来处理元数据更改。

现在开始创建拉取请求——这是我们工具的主要部分：

        # Create a pull request with the updated model card
        pr_title = f"Add '{new_tag}' tag"
        pr_description = f"""
## Add tag: {new_tag}

This PR adds the `{new_tag}` tag to the model repository.

**Changes:**
- Added `{new_tag}` to model tags
- Updated from {len(current_tags)} to {len(updated_tags)} tags

**Current tags:** {", ".join(current_tags) if current_tags else "None"}
**New tags:** {", ".join(updated_tags)}

🤖 This is a pull request created by the Hugging Face Hub Tagging Bot.
"""

        print(f"🚀 Creating PR with title: {pr_title}")

我们创建了一个详细的拉取请求描述，解释了正在更改的内容以及原因。这种透明度对于将审查拉取请求的仓库维护者至关重要。

清晰、详细的 PR 描述对于自动化拉取请求至关重要。它们有助于仓库维护者了解正在发生的事情，并就是否合并更改做出明智的决定。

此外，清楚地说明 PR 是由自动化工具创建的也是一个好习惯。这有助于仓库维护者了解如何处理 PR。

最后，我们创建提交和拉取请求。

        # Create commit with updated model card using CommitOperationAdd
        from huggingface_hub import CommitOperationAdd

        commit_info = hf_api.create_commit(
            repo_id=repo_id,
            operations=[
                CommitOperationAdd(
                    path_in_repo="README.md", path_or_fileobj=str(card).encode("utf-8")
                )
            ],
            commit_message=pr_title,
            commit_description=pr_description,
            token=HF_TOKEN,
            create_pr=True,
        )

        # Extract PR URL from commit info
        pr_url_attr = commit_info.pr_url
        pr_url = pr_url_attr if hasattr(commit_info, "pr_url") else str(commit_info)

        print(f"✅ PR created successfully! URL: {pr_url}")

        result = {
            "status": "success",
            "repo_id": repo_id,
            "tag": new_tag,
            "pr_url": pr_url,
            "previous_tags": current_tags,
            "new_tags": updated_tags,
            "message": f"Created PR to add tag '{new_tag}'",
        }
        json_str = json.dumps(result)
        print(f"✅ add_new_tag success returning: {json_str}")
        return json_str

带有 create_pr=True 的 create_commit 函数是我们自动化的关键。它会使用更新后的 README.md 文件创建一个提交，并自动打开一个拉取请求以供审查。

不要忘记此复杂操作的错误处理。

    except Exception as e:
        print(f"❌ Error in add_new_tag: {str(e)}")
        print(f"❌ Error type: {type(e)}")
        import traceback
        print(f"❌ Traceback: {traceback.format_exc()}")

        error_result = {
            "status": "error",
            "repo_id": repo_id,
            "tag": new_tag,
            "error": str(e),
        }
        json_str = json.dumps(error_result)
        print(f"❌ add_new_tag error returning: {json_str}")
        return json_str

全面的错误处理包括完整的堆栈跟踪，这对于在出现问题时进行调试非常有价值。

日志消息中的表情符号可能看起来很傻，但它们可以大大加快日志扫描速度。🔧 表示函数调用，📡 表示 API 请求，✅ 表示成功，❌ 表示错误，这些视觉模式可以帮助您快速找到所需内容。

在构建此应用程序时，很容易意外地创建无限循环的 PR。这是因为带有 create_pr=True 的 create_commit 函数将为每个提交创建 PR。如果 PR 未合并，create_commit 函数将一次又一次地被调用……

我们已经添加了检查以防止这种情况，但这是您需要注意的事情。

下一步

现在我们已经实现了带有强大标签工具的 MCP 服务器，我们需要：

创建 MCP 客户端 - 构建代理和 MCP 服务器之间的接口
实现 Webhook 处理 - 监听 Hub 讨论事件
集成代理逻辑 - 将 Webhook 与 MCP 工具调用连接起来
测试整个系统 - 验证端到端功能

在下一节中，我们将创建 MCP 客户端，它将允许我们的 webhook 处理程序智能地与这些工具交互。

MCP 服务器作为一个独立于主应用程序的进程运行。这种隔离提供了更好的错误处理，并允许服务器被多个客户端或应用程序重用。

< > 在 GitHub 上更新