Upload 6 files

nexus_hf_mcp_bridge/README.md

@@ -0,0 +1,83 @@
+# NEXUS HF MCP Bridge
+
+Clean, sovereign Hugging Face integration for the **NEXUS Visual Weaver** Space.
+
+## Features
+
+- **OAuth Authentication**: Device Code + PKCE flow using the "X - GROK integration for HF" app
+- **Token Management**: Space-friendly token storage with environment variable support
+- **Repository Operations**: Upload, create/update files, list repos using `huggingface_hub`
+- **MCP Tool Exposure**: Ready to be used as MCP tools inside Gradio Spaces (`mcp_server=True`)
+
+## Folder Structure
+
+```
+nexus_hf_mcp_bridge/
+├── __init__.py
+├── hf_oauth.py          # OAuth Device Code + PKCE
+├── token_manager.py     # Token storage & refresh
+├── hf_repo_tools.py     # huggingface_hub operations
+├── mcp_bridge.py        # Main bridge + MCP tools
+└── README.md
+```
+
+## Quick Start (Inside the Space)
+
+```python
+from nexus_hf_mcp_bridge import nexus_hf_bridge
+
+# 1. Authenticate (one-time)
+result = nexus_hf_bridge.tool_authenticate()
+
+# 2. Use tools
+nexus_hf_bridge.tool_create_or_update_file(
+    repo_id="your-username/your-repo",
+    path_in_repo="example.txt",
+    content="Hello from NEXUS HF MCP Bridge!"
+)
+```
+
+## Integration into `app.py`
+
+```python
+from nexus_hf_mcp_bridge import nexus_hf_bridge
+import gradio as gr
+
+with gr.Blocks() as demo:
+    gr.Markdown("# NEXUS Visual Weaver - HF Bridge")
+
+    with gr.Tab("HF Tools"):
+        repo = gr.Textbox(label="Repository ID")
+        path = gr.Textbox(label="Path in Repo")
+        content = gr.Textbox(label="Content", lines=8)
+        btn = gr.Button("Create / Update File")
+        out = gr.JSON()
+
+        btn.click(
+            nexus_hf_bridge.tool_create_or_update_file,
+            inputs=[repo, path, content],
+            outputs=out
+        )
+
+demo.launch(mcp_server=True)
+```
+
+## Authentication
+
+The bridge supports two modes:
+
+1. **Interactive** (Device Code flow) — Good for development
+2. **Environment Variable** (recommended for Spaces):
+   ```bash
+   HF_OAUTH_ACCESS_TOKEN=hf_xxxxxxxxxxxxxxxx
+   ```
+
+## Security Notes
+
+- Never commit tokens
+- Use Space Secrets for production
+- Token refresh logic is prepared for future extension
+
+## License
+
+Internal NEXUS OS Project

nexus_hf_mcp_bridge/__init__.py

@@ -0,0 +1,29 @@
+"""
+NEXUS HF MCP Bridge
+
+A clean, sovereign Hugging Face integration layer for the NEXUS Visual Weaver Space.
+
+Provides:
+- OAuth Device Code + PKCE authentication
+- Token management (Space-friendly)
+- High-level huggingface_hub operations
+- MCP tool exposure for Gradio Spaces
+
+Author: NEXUS OS Team
+"""
+
+from .hf_oauth import HFOAuthClient, HFOAuthToken, authenticate_hf
+from .token_manager import HFTokenManager
+from .hf_repo_tools import HFRepoTools
+from .mcp_bridge import NEXUS_HF_MCP_Bridge, nexus_hf_bridge
+
+__version__ = "0.1.0"
+__all__ = [
+    "HFOAuthClient",
+    "HFOAuthToken",
+    "authenticate_hf",
+    "HFTokenManager",
+    "HFRepoTools",
+    "NEXUS_HF_MCP_Bridge",
+    "nexus_hf_bridge",
+]

nexus_hf_mcp_bridge/hf_oauth.py

@@ -0,0 +1,121 @@
+"""
+hf_oauth.py
+Custom HF MCP Bridge - OAuth Layer
+Handles Device Code + PKCE flow for the "X - GROK integration for HF" OAuth app.
+
+This module is designed to run inside the NEXUS Visual Weaver Space
+(which has full internet access).
+"""
+
+import requests
+import time
+from typing import Optional, Dict, Any
+from dataclasses import dataclass
+
+CLIENT_ID = "6b41a3bf-689f-4b48-bbce-b39adc7b515d"
+DEVICE_CODE_URL = "https://huggingface.co/oauth/device"
+TOKEN_URL = "https://huggingface.co/oauth/token"
+
+
+@dataclass
+class HFOAuthToken:
+    access_token: str
+    token_type: str = "bearer"
+    expires_in: int = 0
+    refresh_token: Optional[str] = None
+    scope: Optional[str] = None
+
+
+class HFOAuthClient:
+    """
+    Handles OAuth authentication for Hugging Face using Device Code flow.
+    Suitable for CLI agents and Spaces.
+    """
+
+    def __init__(self, client_id: str = CLIENT_ID):
+        self.client_id = client_id
+        self._current_token: Optional[HFOAuthToken] = None
+
+    def start_device_flow(self) -> Dict[str, Any]:
+        """Step 1: Request device code and user verification URL."""
+        response = requests.post(
+            DEVICE_CODE_URL,
+            data={"client_id": self.client_id},
+            timeout=10
+        )
+        response.raise_for_status()
+        return response.json()
+
+    def poll_for_token(self, device_code: str, interval: int = 5, max_wait: int = 300) -> Optional[HFOAuthToken]:
+        """
+        Step 2: Poll the token endpoint until the user authorizes.
+        """
+        start_time = time.time()
+        while time.time() - start_time < max_wait:
+            response = requests.post(
+                TOKEN_URL,
+                data={
+                    "grant_type": "urn:ietf:params:oauth:grant-type:device_code",
+                    "device_code": device_code,
+                    "client_id": self.client_id,
+                },
+                timeout=10
+            )
+
+            if response.status_code == 200:
+                data = response.json()
+                token = HFOAuthToken(
+                    access_token=data["access_token"],
+                    token_type=data.get("token_type", "bearer"),
+                    expires_in=data.get("expires_in", 0),
+                    refresh_token=data.get("refresh_token"),
+                    scope=data.get("scope"),
+                )
+                self._current_token = token
+                return token
+
+            elif response.status_code == 400:
+                error = response.json().get("error")
+                if error == "authorization_pending":
+                    time.sleep(interval)
+                    continue
+                elif error == "slow_down":
+                    time.sleep(interval + 5)
+                    continue
+                else:
+                    print(f"[OAuth] Error: {response.json()}")
+                    return None
+            else:
+                print(f"[OAuth] Unexpected status: {response.status_code}")
+                return None
+
+        print("[OAuth] Timeout waiting for user authorization.")
+        return None
+
+    def get_token(self) -> Optional[HFOAuthToken]:
+        """Return the current valid token (if any)."""
+        return self._current_token
+
+    def is_authenticated(self) -> bool:
+        return self._current_token is not None and bool(self._current_token.access_token)
+
+
+# Convenience function for quick usage
+def authenticate_hf() -> Optional[HFOAuthToken]:
+    """
+    Full Device Code flow in one call.
+    Returns HFOAuthToken or None.
+    """
+    client = HFOAuthClient()
+    device_data = client.start_device_flow()
+
+    print("\n=== Hugging Face OAuth ===")
+    print(f"Open this URL: {device_data['verification_uri']}")
+    print(f"Enter code: {device_data['user_code']}\n")
+
+    input("Press Enter after you have authorized in the browser...")
+
+    token = client.poll_for_token(device_data["device_code"])
+    if token:
+        print("[+] Successfully authenticated with Hugging Face!")
+    return token

nexus_hf_mcp_bridge/hf_repo_tools.py

@@ -0,0 +1,104 @@
+"""
+hf_repo_tools.py
+Custom HF MCP Bridge - Repository Operations
+
+Core Hugging Face Hub operations (push, update, list, etc.)
+using huggingface_hub library + our OAuth token.
+"""
+
+from typing import Optional, List, Dict, Any
+from huggingface_hub import HfApi, upload_file, create_repo, list_repo_files
+import os
+
+
+class HFRepoTools:
+    """
+    High-level wrapper around huggingface_hub for common operations.
+    Uses token from HFTokenManager.
+    """
+
+    def __init__(self, token: str):
+        self.token = token
+        self.api = HfApi(token=token)
+
+    def upload_file_to_repo(
+        self,
+        local_path: str,
+        repo_id: str,
+        path_in_repo: str,
+        repo_type: str = "model",
+        commit_message: Optional[str] = None
+    ) -> bool:
+        """Upload a single file to a Hugging Face repo."""
+        try:
+            upload_file(
+                path_or_fileobj=local_path,
+                path_in_repo=path_in_repo,
+                repo_id=repo_id,
+                repo_type=repo_type,
+                token=self.token,
+                commit_message=commit_message or f"Upload {path_in_repo}"
+            )
+            return True
+        except Exception as e:
+            print(f"[HFRepoTools] Upload failed: {e}")
+            return False
+
+    def create_or_update_file(
+        self,
+        repo_id: str,
+        path_in_repo: str,
+        content: str,
+        commit_message: str = "Update file via NEXUS MCP Bridge"
+    ) -> bool:
+        """Create or update a text file in a repo."""
+        try:
+            self.api.upload_file(
+                path_or_fileobj=content.encode("utf-8"),
+                path_in_repo=path_in_repo,
+                repo_id=repo_id,
+                token=self.token,
+                commit_message=commit_message
+            )
+            return True
+        except Exception as e:
+            print(f"[HFRepoTools] create_or_update_file failed: {e}")
+            return False
+
+    def list_files(self, repo_id: str, revision: str = "main") -> List[str]:
+        """List files in a repository."""
+        try:
+            files = list_repo_files(repo_id=repo_id, revision=revision, token=self.token)
+            return files
+        except Exception as e:
+            print(f"[HFRepoTools] list_files failed: {e}")
+            return []
+
+    def repo_exists(self, repo_id: str) -> bool:
+        """Check if a repository exists."""
+        try:
+            self.api.repo_info(repo_id=repo_id, token=self.token)
+            return True
+        except:
+            return False
+
+    def create_repo_if_not_exists(
+        self,
+        repo_id: str,
+        repo_type: str = "model",
+        private: bool = False
+    ) -> bool:
+        """Create a repository if it doesn't exist."""
+        if self.repo_exists(repo_id):
+            return True
+        try:
+            create_repo(
+                repo_id=repo_id,
+                repo_type=repo_type,
+                private=private,
+                token=self.token
+            )
+            return True
+        except Exception as e:
+            print(f"[HFRepoTools] Failed to create repo: {e}")
+            return False

nexus_hf_mcp_bridge/mcp_bridge.py

@@ -0,0 +1,139 @@
+"""
+mcp_bridge.py
+Custom HF MCP Bridge - Main Integration Layer
+
+Exposes Hugging Face operations as MCP tools for Gradio Spaces.
+This is the main entry point for the NEXUS Visual Weaver Space.
+"""
+
+from typing import Dict, Any, List, Optional
+from .hf_oauth import HFOAuthClient, authenticate_hf
+from .token_manager import HFTokenManager
+from .hf_repo_tools import HFRepoTools
+
+
+class NEXUS_HF_MCP_Bridge:
+    """
+    Main bridge class that exposes HF capabilities as MCP tools.
+    Designed to be used inside a Gradio Space with mcp_server=True.
+    """
+
+    def __init__(self):
+        self.token_manager = HFTokenManager()
+        self.oauth_client = HFOAuthClient()
+        self.repo_tools: Optional[HFRepoTools] = None
+        self._authenticated = False
+
+    def authenticate(self) -> Dict[str, Any]:
+        """Start OAuth Device Code flow and store token."""
+        token = authenticate_hf()
+        if token:
+            self.token_manager.save_token(token)
+            self.repo_tools = HFRepoTools(token.access_token)
+            self._authenticated = True
+            return {"status": "success", "message": "Authenticated successfully"}
+        return {"status": "error", "message": "Authentication failed"}
+
+    def ensure_ready(self) -> bool:
+        """Make sure we have a valid token and repo_tools ready."""
+        if self._authenticated and self.repo_tools:
+            return True
+
+        token_str = self.token_manager.get_valid_token()
+        if token_str:
+            self.repo_tools = HFRepoTools(token_str)
+            self._authenticated = True
+            return True
+
+        return False
+
+    # ===================== MCP TOOLS =====================
+
+    def tool_authenticate(self) -> Dict[str, Any]:
+        """MCP Tool: Start Hugging Face OAuth authentication."""
+        return self.authenticate()
+
+    def tool_upload_file(
+        self,
+        local_path: str,
+        repo_id: str,
+        path_in_repo: str,
+        commit_message: Optional[str] = None
+    ) -> Dict[str, Any]:
+        """MCP Tool: Upload a file to a Hugging Face repository."""
+        if not self.ensure_ready():
+            return {"status": "error", "message": "Not authenticated. Call authenticate first."}
+
+        success = self.repo_tools.upload_file_to_repo(
+            local_path=local_path,
+            repo_id=repo_id,
+            path_in_repo=path_in_repo,
+            commit_message=commit_message
+        )
+        return {"status": "success" if success else "error"}
+
+    def tool_create_or_update_file(
+        self,
+        repo_id: str,
+        path_in_repo: str,
+        content: str,
+        commit_message: str = "Update via NEXUS HF MCP Bridge"
+    ) -> Dict[str, Any]:
+        """MCP Tool: Create or update a text file in a repo."""
+        if not self.ensure_ready():
+            return {"status": "error", "message": "Not authenticated."}
+
+        success = self.repo_tools.create_or_update_file(
+            repo_id=repo_id,
+            path_in_repo=path_in_repo,
+            content=content,
+            commit_message=commit_message
+        )
+        return {"status": "success" if success else "error"}
+
+    def tool_list_repo_files(self, repo_id: str) -> Dict[str, Any]:
+        """MCP Tool: List files in a repository."""
+        if not self.ensure_ready():
+            return {"status": "error", "message": "Not authenticated."}
+
+        files = self.repo_tools.list_files(repo_id=repo_id)
+        return {"status": "success", "files": files}
+
+    def tool_repo_exists(self, repo_id: str) -> Dict[str, Any]:
+        """MCP Tool: Check if a repository exists."""
+        if not self.ensure_ready():
+            return {"status": "error", "message": "Not authenticated."}
+
+        exists = self.repo_tools.repo_exists(repo_id=repo_id)
+        return {"status": "success", "exists": exists}
+
+
+# Global bridge instance (for use in Gradio)
+nexus_hf_bridge = NEXUS_HF_MCP_Bridge()
+
+
+# Example of how to expose in Gradio app.py:
+"""
+from mcp_bridge import nexus_hf_bridge
+
+with gr.Blocks() as demo:
+    gr.Markdown("# NEXUS Visual Weaver - HF MCP Bridge")
+
+    with gr.Tab("HF Authentication"):
+        auth_btn = gr.Button("Authenticate with Hugging Face")
+        auth_output = gr.JSON()
+        auth_btn.click(nexus_hf_bridge.tool_authenticate, outputs=auth_output)
+
+    with gr.Tab("Repository Tools"):
+        repo_id = gr.Textbox(label="Repository ID (e.g. username/my-model)")
+        file_path = gr.Textbox(label="Path in repo")
+        content = gr.Textbox(label="Content", lines=10)
+        upload_btn = gr.Button("Create / Update File")
+        result = gr.JSON()
+
+        upload_btn.click(
+            nexus_hf_bridge.tool_create_or_update_file,
+            inputs=[repo_id, file_path, content],
+            outputs=result
+        )
+"""

nexus_hf_mcp_bridge/token_manager.py

@@ -0,0 +1,108 @@
+"""
+token_manager.py
+Custom HF MCP Bridge - Token Management Layer
+
+Handles secure storage and refresh of Hugging Face OAuth tokens
+in a Space-friendly way (environment variables + optional file fallback).
+"""
+
+import os
+import json
+import time
+from typing import Optional
+from datetime import datetime, timedelta
+from .hf_oauth import HFOAuthToken, HFOAuthClient
+
+
+class HFTokenManager:
+    """
+    Manages Hugging Face OAuth tokens with refresh support.
+    Designed to work well inside Hugging Face Spaces.
+    """
+
+    def __init__(self, token_file: str = "/tmp/hf_oauth_token.json"):
+        self.token_file = token_file
+        self._token: Optional[HFOAuthToken] = None
+        self._client = HFOAuthClient()
+
+    def load_token(self) -> Optional[HFOAuthToken]:
+        """Load token from environment or file."""
+        # Priority 1: Environment variable (recommended for Spaces)
+        env_token = os.environ.get("HF_OAUTH_ACCESS_TOKEN")
+        if env_token:
+            self._token = HFOAuthToken(access_token=env_token)
+            return self._token
+
+        # Priority 2: Token file
+        if os.path.exists(self.token_file):
+            try:
+                with open(self.token_file, "r") as f:
+                    data = json.load(f)
+                    self._token = HFOAuthToken(**data)
+                    return self._token
+            except Exception as e:
+                print(f"[TokenManager] Failed to load token file: {e}")
+
+        return None
+
+    def save_token(self, token: HFOAuthToken) -> bool:
+        """Save token to file (and optionally env for current process)."""
+        self._token = token
+        try:
+            with open(self.token_file, "w") as f:
+                json.dump({
+                    "access_token": token.access_token,
+                    "token_type": token.token_type,
+                    "expires_in": token.expires_in,
+                    "refresh_token": token.refresh_token,
+                    "scope": token.scope,
+                }, f, indent=2)
+            return True
+        except Exception as e:
+            print(f"[TokenManager] Failed to save token: {e}")
+            return False
+
+    def get_valid_token(self) -> Optional[str]:
+        """
+        Return a valid access token.
+        If expired and refresh_token exists, attempt refresh.
+        """
+        if not self._token:
+            self.load_token()
+
+        if not self._token:
+            return None
+
+        # For simplicity in Phase 1, we assume the token is still valid
+        # In production we would check expiry and refresh here.
+        return self._token.access_token
+
+    def ensure_authenticated(self) -> bool:
+        """
+        Ensure we have a valid token.
+        If not, start Device Code flow.
+        """
+        token = self.get_valid_token()
+        if token:
+            return True
+
+        print("[TokenManager] No valid token found. Starting OAuth flow...")
+        new_token = self._client.poll_for_token(
+            # This assumes the device flow was already started elsewhere
+            # In real usage we would combine with hf_oauth.start_device_flow()
+            device_code=""  # Placeholder - real implementation would handle full flow
+        )
+
+        if new_token:
+            self.save_token(new_token)
+            return True
+        return False
+
+    def clear_token(self):
+        """Remove stored token."""
+        self._token = None
+        if os.path.exists(self.token_file):
+            try:
+                os.remove(self.token_file)
+            except:
+                pass