Dockerized the solution

.gitignore

@@ -0,0 +1,178 @@
+# Byte-compiled / optimized / DLL files
+__pycache__/
+*.py[cod]
+*$py.class
+
+# C extensions
+*.so
+
+# Distribution / packaging
+.Python
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+wheels/
+share/python-wheels/
+*.egg-info/
+.installed.cfg
+*.egg
+MANIFEST
+
+# PyInstaller
+#  Usually these files are written by a python script from a template
+#  before PyInstaller builds the exe, so as to inject date/other infos into it.
+*.manifest
+*.spec
+
+# Installer logs
+pip-log.txt
+pip-delete-this-directory.txt
+
+# Unit test / coverage reports
+htmlcov/
+.tox/
+.nox/
+.coverage
+.coverage.*
+.cache
+nosetests.xml
+coverage.xml
+*.cover
+*.py,cover
+.hypothesis/
+.pytest_cache/
+cover/
+
+# Translations
+*.mo
+*.pot
+
+# Django stuff:
+*.log
+local_settings.py
+db.sqlite3
+db.sqlite3-journal
+
+# Flask stuff:
+instance/
+.webassets-cache
+
+# Scrapy stuff:
+.scrapy
+
+# Sphinx documentation
+docs/_build/
+
+# PyBuilder
+.pybuilder/
+target/
+
+# Jupyter Notebook
+.ipynb_checkpoints
+
+# IPython
+profile_default/
+ipython_config.py
+
+# pyenv
+#   For a library or package, you might want to ignore these files since the code is
+#   intended to run in multiple environments; otherwise, check them in:
+# .python-version
+
+# pipenv
+#   According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
+#   However, in case of collaboration, if having platform-specific dependencies or dependencies
+#   having no cross-platform support, pipenv may install dependencies that don't work, or not
+#   install all needed dependencies.
+#Pipfile.lock
+
+# UV
+#   Similar to Pipfile.lock, it is generally recommended to include uv.lock in version control.
+#   This is especially recommended for binary packages to ensure reproducibility, and is more
+#   commonly ignored for libraries.
+#uv.lock
+
+# poetry
+#   Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control.
+#   This is especially recommended for binary packages to ensure reproducibility, and is more
+#   commonly ignored for libraries.
+#   https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control
+#poetry.lock
+
+# pdm
+#   Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control.
+#pdm.lock
+#   pdm stores project-wide configurations in .pdm.toml, but it is recommended to not include it
+#   in version control.
+#   https://pdm.fming.dev/latest/usage/project/#working-with-version-control
+.pdm.toml
+.pdm-python
+.pdm-build/
+
+# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm
+__pypackages__/
+
+# Celery stuff
+celerybeat-schedule
+celerybeat.pid
+
+# SageMath parsed files
+*.sage.py
+
+# Environments
+.env
+.venv
+env/
+venv/
+ENV/
+env.bak/
+venv.bak/
+
+# Spyder project settings
+.spyderproject
+.spyproject
+
+# Rope project settings
+.ropeproject
+
+# mkdocs documentation
+/site
+
+# mypy
+.mypy_cache/
+.dmypy.json
+dmypy.json
+
+# Pyre type checker
+.pyre/
+
+# pytype static type analyzer
+.pytype/
+
+# Cython debug symbols
+cython_debug/
+
+# PyCharm
+#  JetBrains specific template is maintained in a separate JetBrains.gitignore that can
+#  be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
+#  and can be added to the global gitignore or merged into this file.  For a more nuclear
+#  option (not recommended) you can uncomment the following to ignore the entire idea folder.
+#.idea/
+
+# Ruff stuff:
+.ruff_cache/
+
+# PyPI configuration file
+.pypirc
+
+.DS_Store
+
+videos/

.python-version

@@ -0,0 +1 @@
+3.13

Dockerfile

@@ -0,0 +1,31 @@
+FROM python:3.13-slim
+
+# Create a non-root user
+RUN useradd -m -u 1000 appuser
+USER appuser
+
+# Environment
+ENV PATH="/home/appuser/.local/bin:$PATH"
+
+# Set working directory
+WORKDIR /app
+RUN chown -R appuser:1000 /app
+
+# Install pip and uv
+RUN pip install --upgrade pip && pip install uv
+
+# Copy project files
+COPY --chown=appuser:1000 pyproject.toml .
+COPY --chown=appuser:1000 app.py .
+COPY --chown=appuser:1000 Dockerfile .
+COPY --chown=appuser:1000 README.md .
+COPY --chown=appuser:1000 uv.lock .
+
+# Install dependencies
+RUN uv sync --frozen
+
+# Expose port
+EXPOSE 7860
+
+# Run the application
+CMD ["uv", "run", "app.py", "--host", "0.0.0.0", "--port", "7860"]

README.md

@@ -1,14 +1,27 @@
 ---
 title: Unit3 HF Tagging Bot
 emoji: ⚡
-colorFrom: red
+colorFrom: purple
 colorTo: yellow
 sdk: gradio
 sdk_version: 5.34.1
 app_file: app.py
-pinned: false
+pinned: true
+base_path: /gradio
 license: apache-2.0
-short_description: This is a bot that tags HuggingFace models when they are men
+short_description: Bot that tags HuggingFace models when they are mentioned
+tags:
+    - mcp-course
+    - huggingface
+    - pr-agent
 ---
 
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+# HF Tagging Bot
+
+This is a bot that tags HuggingFace models when they are mentioned in discussions.
+
+## How it works
+
+1. The bot listens to discussions on the HuggingFace Hub
+2. When a discussion is created, the bot checks for tag mentions in the comment
+3. If a tag is mentioned, the bot adds the tag to the model repository via a PR

app.py

@@ -0,0 +1,558 @@
+import os
+import re
+import json
+from datetime import datetime
+from typing import List, Dict, Any, Optional, Literal
+
+from fastapi import FastAPI, Request, BackgroundTasks
+from fastapi.middleware.cors import CORSMiddleware
+import gradio as gr
+import uvicorn
+from pydantic import BaseModel
+from huggingface_hub.inference._mcp.agent import Agent
+from dotenv import load_dotenv
+
+load_dotenv()
+
+# Configuration
+WEBHOOK_SECRET = os.getenv("WEBHOOK_SECRET")
+HF_TOKEN = os.getenv("HF_TOKEN")
+HF_MODEL = os.getenv("HF_MODEL", "microsoft/DialoGPT-medium")
+# Use a valid provider literal from the documentation
+DEFAULT_PROVIDER: Literal["hf-inference"] = "hf-inference"
+HF_PROVIDER = os.getenv("HF_PROVIDER", DEFAULT_PROVIDER)
+
+# Simple storage for processed tag operations
+tag_operations_store: List[Dict[str, Any]] = []
+
+# Agent instance
+agent_instance: Optional[Agent] = None
+
+# Common ML tags that we recognize for auto-tagging
+RECOGNIZED_TAGS = {
+    "pytorch",
+    "tensorflow",
+    "jax",
+    "transformers",
+    "diffusers",
+    "text-generation",
+    "text-classification",
+    "question-answering",
+    "text-to-image",
+    "image-classification",
+    "object-detection",
+    "   ",
+    "fill-mask",
+    "token-classification",
+    "translation",
+    "summarization",
+    "feature-extraction",
+    "sentence-similarity",
+    "zero-shot-classification",
+    "image-to-text",
+    "automatic-speech-recognition",
+    "audio-classification",
+    "voice-activity-detection",
+    "depth-estimation",
+    "image-segmentation",
+    "video-classification",
+    "reinforcement-learning",
+    "tabular-classification",
+    "tabular-regression",
+    "time-series-forecasting",
+    "graph-ml",
+    "robotics",
+    "computer-vision",
+    "nlp",
+    "cv",
+    "multimodal",
+}
+
+
+class WebhookEvent(BaseModel):
+    event: Dict[str, str]
+    comment: Dict[str, Any]
+    discussion: Dict[str, Any]
+    repo: Dict[str, str]
+
+
+app = FastAPI(title="HF Tagging Bot")
+app.add_middleware(CORSMiddleware, allow_origins=["*"])
+
+async def get_agent():
+    """Get or create Agent instance"""
+    print("🤖 get_agent() called...")
+    global agent_instance
+    if agent_instance is None and HF_TOKEN:
+        print("🔧 Creating new Agent instance...")
+        print(f"🔑 HF_TOKEN present: {bool(HF_TOKEN)}")
+        print(f"🤖 Model: {HF_MODEL}")
+        print(f"🔗 Provider: {DEFAULT_PROVIDER}")
+
+        try:
+            agent_instance = Agent(
+                model=HF_MODEL,
+                provider=DEFAULT_PROVIDER,
+                api_key=HF_TOKEN,
+                servers=[
+                    {
+                        "type": "http",
+                        "config": {
+                            "url": "https://betki-unit3-mcp-pr-agent.hf.space/mcp-server/mcp",
+                            "env": {"HF_TOKEN": HF_TOKEN} if HF_TOKEN else {},
+                        },
+                    }
+                ],
+            )
+            print("✅ Agent instance created successfully")
+            print("🔧 Loading tools...")
+            await agent_instance.load_tools()
+            print("✅ Tools loaded successfully")
+        except Exception as e:
+            print(f"❌ Error creating/loading agent: {str(e)}")
+            agent_instance = None
+    elif agent_instance is None:
+        print("❌ No HF_TOKEN available, cannot create agent")
+    else:
+        print("✅ Using existing agent instance")
+
+    return agent_instance
+
+def extract_tags_from_text(text: str) -> List[str]:
+    """Extract potential tags from discussion text"""
+    text_lower = text.lower()
+
+    # Look for explicit tag mentions like "tag: pytorch" or "#pytorch"
+    explicit_tags = []
+
+    # Pattern 1: "tag: something" or "tags: something"
+    tag_pattern = r"tags?:\s*([a-zA-Z0-9-_,\s]+)"
+    matches = re.findall(tag_pattern, text_lower)
+    for match in matches:
+        # Split by comma and clean up
+        tags = [tag.strip() for tag in match.split(",")]
+        explicit_tags.extend(tags)
+
+    # Pattern 2: "#hashtag" style
+    hashtag_pattern = r"#([a-zA-Z0-9-_]+)"
+    hashtag_matches = re.findall(hashtag_pattern, text_lower)
+    explicit_tags.extend(hashtag_matches)
+
+    # Pattern 3: Look for recognized tags mentioned in natural text
+    mentioned_tags = []
+    for tag in RECOGNIZED_TAGS:
+        if tag in text_lower:
+            mentioned_tags.append(tag)
+
+    # Combine and deduplicate
+    all_tags = list(set(explicit_tags + mentioned_tags))
+
+    # Filter to only include recognized tags or explicitly mentioned ones
+    valid_tags = []
+    for tag in all_tags:
+        if tag in RECOGNIZED_TAGS or tag in explicit_tags:
+            valid_tags.append(tag)
+
+    return valid_tags
+
+
+async def process_webhook_comment(webhook_data: Dict[str, Any]):
+    """Process webhook to detect and add tags"""
+    print("🏷️ Starting process_webhook_comment...")
+
+    try:
+        comment_content = webhook_data["comment"]["content"]
+        discussion_title = webhook_data["discussion"]["title"]
+        repo_name = webhook_data["repo"]["name"]
+        discussion_num = webhook_data["discussion"]["num"]
+        # Author is an object with "id" field
+        comment_author = webhook_data["comment"]["author"].get("id", "unknown")
+
+        print(f"📝 Comment content: {comment_content}")
+        print(f"📰 Discussion title: {discussion_title}")
+        print(f"📦 Repository: {repo_name}")
+
+        # Extract potential tags from the comment and discussion title
+        comment_tags = extract_tags_from_text(comment_content)
+        title_tags = extract_tags_from_text(discussion_title)
+        all_tags = list(set(comment_tags + title_tags))
+
+        print(f"🔍 Comment tags found: {comment_tags}")
+        print(f"🔍 Title tags found: {title_tags}")
+        print(f"🏷️ All unique tags: {all_tags}")
+
+        result_messages = []
+
+        if not all_tags:
+            msg = "No recognizable tags found in the discussion."
+            print(f"❌ {msg}")
+            result_messages.append(msg)
+        else:
+            print("🤖 Getting agent instance...")
+            agent = await get_agent()
+            if not agent:
+                msg = "Error: Agent not configured (missing HF_TOKEN)"
+                print(f"❌ {msg}")
+                result_messages.append(msg)
+            else:
+                print("✅ Agent instance obtained successfully")
+
+                # Process all tags in a single conversation with the agent
+                try:
+                    # Create a comprehensive prompt for the agent
+                    user_prompt = f"""
+I need to add the following tags to the repository '{repo_name}': {", ".join(all_tags)}
+
+For each tag, please:
+1. Check if the tag already exists on the repository using get_current_tags
+2. If the tag doesn't exist, add it using add_new_tag
+3. Provide a summary of what was done for each tag
+
+Please process all {len(all_tags)} tags: {", ".join(all_tags)}
+"""
+
+                    print("💬 Sending comprehensive prompt to agent...")
+                    print(f"📝 Prompt: {user_prompt}")
+
+                    # Let the agent handle the entire conversation
+                    conversation_result = []
+
+                    try:
+                        async for item in agent.run(user_prompt):
+                            # The agent yields different types of items
+                            item_str = str(item)
+                            conversation_result.append(item_str)
+
+                            # Log important events
+                            if (
+                                "tool_call" in item_str.lower()
+                                or "function" in item_str.lower()
+                            ):
+                                print(f"🔧 Agent using tools: {item_str[:200]}...")
+                            elif "content" in item_str and len(item_str) < 500:
+                                print(f"💭 Agent response: {item_str}")
+
+                        # Extract the final response from the conversation
+                        full_response = " ".join(conversation_result)
+                        print(f"📋 Agent conversation completed successfully")
+
+                        # Try to extract meaningful results for each tag
+                        for tag in all_tags:
+                            tag_mentioned = tag.lower() in full_response.lower()
+
+                            if (
+                                "already exists" in full_response.lower()
+                                and tag_mentioned
+                            ):
+                                msg = f"Tag '{tag}': Already exists"
+                            elif (
+                                "pr" in full_response.lower()
+                                or "pull request" in full_response.lower()
+                            ):
+                                if tag_mentioned:
+                                    msg = f"Tag '{tag}': PR created successfully"
+                                else:
+                                    msg = (
+                                        f"Tag '{tag}': Processed "
+                                        "(PR may have been created)"
+                                    )
+                            elif "success" in full_response.lower() and tag_mentioned:
+                                msg = f"Tag '{tag}': Successfully processed"
+                            elif "error" in full_response.lower() and tag_mentioned:
+                                msg = f"Tag '{tag}': Error during processing"
+                            else:
+                                msg = f"Tag '{tag}': Processed by agent"
+
+                            print(f"✅ Result for tag '{tag}': {msg}")
+                            result_messages.append(msg)
+
+                    except Exception as agent_error:
+                        print(f"⚠️ Agent streaming failed: {str(agent_error)}")
+                        print("🔄 Falling back to direct MCP tool calls...")
+
+                        # Import the MCP server functions directly as fallback
+                        try:
+                            import sys
+                            import importlib.util
+
+                            # Load the MCP server module
+                            spec = importlib.util.spec_from_file_location(
+                                "mcp_server", "./mcp_server.py"
+                            )
+                            mcp_module = importlib.util.module_from_spec(spec)
+                            spec.loader.exec_module(mcp_module)
+
+                            # Use the MCP tools directly for each tag
+                            for tag in all_tags:
+                                try:
+                                    print(
+                                        f"🔧 Directly calling get_current_tags for '{tag}'"
+                                    )
+                                    current_tags_result = mcp_module.get_current_tags(
+                                        repo_name
+                                    )
+                                    print(
+                                        f"📄 Current tags result: {current_tags_result}"
+                                    )
+
+                                    # Parse the JSON result
+                                    import json
+
+                                    tags_data = json.loads(current_tags_result)
+
+                                    if tags_data.get("status") == "success":
+                                        current_tags = tags_data.get("current_tags", [])
+                                        if tag in current_tags:
+                                            msg = f"Tag '{tag}': Already exists"
+                                            print(f"✅ {msg}")
+                                        else:
+                                            print(
+                                                f"🔧 Directly calling add_new_tag for '{tag}'"
+                                            )
+                                            add_result = mcp_module.add_new_tag(
+                                                repo_name, tag
+                                            )
+                                            print(f"📄 Add tag result: {add_result}")
+
+                                            add_data = json.loads(add_result)
+                                            if add_data.get("status") == "success":
+                                                pr_url = add_data.get("pr_url", "")
+                                                msg = f"Tag '{tag}': PR created - {pr_url}"
+                                            elif (
+                                                add_data.get("status")
+                                                == "already_exists"
+                                            ):
+                                                msg = f"Tag '{tag}': Already exists"
+                                            else:
+                                                msg = f"Tag '{tag}': {add_data.get('message', 'Processed')}"
+                                            print(f"✅ {msg}")
+                                    else:
+                                        error_msg = tags_data.get(
+                                            "error", "Unknown error"
+                                        )
+                                        msg = f"Tag '{tag}': Error - {error_msg}"
+                                        print(f"❌ {msg}")
+
+                                    result_messages.append(msg)
+
+                                except Exception as direct_error:
+                                    error_msg = f"Tag '{tag}': Direct call error - {str(direct_error)}"
+                                    print(f"❌ {error_msg}")
+                                    result_messages.append(error_msg)
+
+                        except Exception as fallback_error:
+                            error_msg = (
+                                f"Fallback approach failed: {str(fallback_error)}"
+                            )
+                            print(f"❌ {error_msg}")
+                            result_messages.append(error_msg)
+
+                except Exception as e:
+                    error_msg = f"Error during agent processing: {str(e)}"
+                    print(f"❌ {error_msg}")
+                    result_messages.append(error_msg)
+
+        # Store the interaction
+        base_url = "https://huggingface.co"
+        discussion_url = f"{base_url}/{repo_name}/discussions/{discussion_num}"
+
+        interaction = {
+            "timestamp": datetime.now().isoformat(),
+            "repo": repo_name,
+            "discussion_title": discussion_title,
+            "discussion_num": discussion_num,
+            "discussion_url": discussion_url,
+            "original_comment": comment_content,
+            "comment_author": comment_author,
+            "detected_tags": all_tags,
+            "results": result_messages,
+        }
+
+        tag_operations_store.append(interaction)
+        final_result = " | ".join(result_messages)
+        print(f"💾 Stored interaction and returning result: {final_result}")
+        return final_result
+
+    except Exception as e:
+        error_msg = f"❌ Fatal error in process_webhook_comment: {str(e)}"
+        print(error_msg)
+        return error_msg
+
+
+@app.post("/webhook")
+async def webhook_handler(request: Request, background_tasks: BackgroundTasks):
+    """Handle HF Hub webhooks"""
+    webhook_secret = request.headers.get("X-Webhook-Secret")
+    if webhook_secret != WEBHOOK_SECRET:
+        print("❌ Invalid webhook secret")
+        return {"error": "Invalid webhook secret"}
+
+    payload = await request.json()
+    print(f"📥 Received webhook payload: {json.dumps(payload, indent=2)}")
+
+    event = payload.get("event", {})
+    scope = event.get("scope")
+    action = event.get("action")
+
+    print(f"🔍 Event details - scope: {scope}, action: {action}")
+
+    # Check if this is a discussion comment creation
+    scope_check = scope == "discussion"
+    action_check = action == "create"
+    not_pr = not payload["discussion"]["isPullRequest"]
+    scope_check = scope_check and not_pr
+    print(f"✅ not_pr: {not_pr}")
+    print(f"✅ scope_check: {scope_check}")
+    print(f"✅ action_check: {action_check}")
+
+    if scope_check and action_check:
+        # Verify we have the required fields
+        required_fields = ["comment", "discussion", "repo"]
+        missing_fields = [field for field in required_fields if field not in payload]
+
+        if missing_fields:
+            error_msg = f"Missing required fields: {missing_fields}"
+            print(f"❌ {error_msg}")
+            return {"error": error_msg}
+
+        print(f"🚀 Processing webhook for repo: {payload['repo']['name']}")
+        background_tasks.add_task(process_webhook_comment, payload)
+        return {"status": "processing"}
+
+    print(f"⏭️ Ignoring webhook - scope: {scope}, action: {action}")
+    return {"status": "ignored"}
+
+
+async def simulate_webhook(
+    repo_name: str, discussion_title: str, comment_content: str
+) -> str:
+    """Simulate webhook for testing"""
+    if not all([repo_name, discussion_title, comment_content]):
+        return "Please fill in all fields."
+
+    mock_payload = {
+        "event": {"action": "create", "scope": "discussion"},
+        "comment": {
+            "content": comment_content,
+            "author": {"id": "test-user-id"},
+            "id": "mock-comment-id",
+            "hidden": False,
+        },
+        "discussion": {
+            "title": discussion_title,
+            "num": len(tag_operations_store) + 1,
+            "id": "mock-discussion-id",
+            "status": "open",
+            "isPullRequest": False,
+        },
+        "repo": {
+            "name": repo_name,
+            "type": "model",
+            "private": False,
+        },
+    }
+
+    response = await process_webhook_comment(mock_payload)
+    return f"✅ Processed! Results: {response}"
+
+
+def create_gradio_app():
+    """Create Gradio interface"""
+    with gr.Blocks(title="HF Tagging Bot", theme=gr.themes.Soft()) as demo:
+        gr.Markdown("# 🏷️ HF Tagging Bot Dashboard")
+        gr.Markdown("*Automatically adds tags to models when mentioned in discussions*")
+
+        gr.Markdown("""
+        ## How it works:
+        - Monitors HuggingFace Hub discussions
+        - Detects tag mentions in comments (e.g., "tag: pytorch",
+          "#transformers")
+        - Automatically adds recognized tags to the model repository
+        - Supports common ML tags like: pytorch, tensorflow,
+          text-generation, etc.
+        """)
+
+        with gr.Column():
+            sim_repo = gr.Textbox(
+                label="Repository",
+                value="betki/MCP-Course-Model",
+                placeholder="username/model-name",
+            )
+            sim_title = gr.Textbox(
+                label="Discussion Title",
+                value="Add pytorch tag",
+                placeholder="Discussion title",
+            )
+            sim_comment = gr.Textbox(
+                label="Comment",
+                lines=3,
+                value="This model should have tags: pytorch, text-generation",
+                placeholder="Comment mentioning tags...",
+            )
+            sim_btn = gr.Button("🏷️ Test Tag Detection")
+
+        with gr.Column():
+            sim_result = gr.Textbox(label="Result", lines=8)
+
+        sim_btn.click(
+            fn=simulate_webhook,
+            inputs=[sim_repo, sim_title, sim_comment],
+            outputs=sim_result,
+        )
+
+        gr.Markdown(f"""
+        ## Recognized Tags:
+        {", ".join(sorted(RECOGNIZED_TAGS))}
+        """)
+
+    return demo
+
+@app.get("/")
+async def root():
+    """Root endpoint with basic information"""
+    return {
+        "name": "HF Tagging Bot",
+        "status": "running",
+        "description": "Webhook listener for automatic model tagging",
+        "endpoints": {
+            "webhook": "/webhook",
+            "health": "/health",
+            "operations": "/operations"
+        }
+    }
+
+@app.get("/health")
+async def health_check():
+    """Health check endpoint for monitoring"""
+    agent = await get_agent()
+
+    return {
+        "status": "healthy",
+        "timestamp": datetime.now().isoformat(),
+        "components": {
+            "webhook_secret": "configured" if WEBHOOK_SECRET else "missing",
+            "hf_token": "configured" if HF_TOKEN else "missing",
+            "mcp_agent": "ready" if agent else "not_ready"
+        }
+    }
+
+@app.get("/operations")
+async def get_operations():
+    """Get recent tag operations for monitoring"""
+    # Return last 50 operations
+    recent_ops = tag_operations_store[-50:] if tag_operations_store else []
+    return {
+        "total_operations": len(tag_operations_store),
+        "recent_operations": recent_ops
+    }
+
+# Mount Gradio app
+gradio_app = create_gradio_app()
+app = gr.mount_gradio_app(app, gradio_app, path="/gradio")
+
+
+if __name__ == "__main__":
+    print("🚀 Starting HF Tagging Bot...")
+    print("📊 Dashboard: http://localhost:7860/gradio")
+    print("🔗 Webhook: http://localhost:7860/webhook")
+    uvicorn.run("app:app", host="0.0.0.0", port=7860, reload=True)

mcp_server.py

@@ -0,0 +1,184 @@
+#!/usr/bin/env python3
+"""
+Simplified MCP Server for HuggingFace Hub Tagging Operations using FastMCP
+"""
+
+import asyncio
+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()
+
+# 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")
+
+@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
+
+    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
+
+
+@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)
+
+        # 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)}
+"""
+
+        print(f"🚀 Creating PR with title: {pr_title}")
+
+        # 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
+
+    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
+
+# Add a simple main block for direct execution
+if __name__ == "__main__":
+    mcp.run()

pyproject.toml

@@ -0,0 +1,22 @@
+[project]
+name = "unit3-hf-tagging-bot"
+version = "0.1.0"
+description = "FastAPI and Gradio app for Hugging Face Hub discussion webhooks"
+readme = "README.md"
+requires-python = ">=3.13"
+dependencies = [
+    "fastapi",
+    "fastmcp",
+    "gradio",
+    "huggingface-hub[mcp]",
+    "pydantic",
+    "python-multipart",
+    "requests"
+]
+
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"
+
+[tool.hatch.build.targets.wheel]
+packages = ["src"]