建立 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 上更新