server.py

from mcp.server.fastmcp import FastMCP
from push_to_hf import push_files_to_bucket, create_hf_repo
from data_handler import query_hf_data
import hffs_ops
import os
import httpx
from gradio_client import Client

# Initialize the MCP server
mcp = FastMCP("hf-mcp-server")

@mcp.tool()
def inspect_space_api(space_subdomain: str) -> str:
    """Fetches the OpenAPI schema for a Gradio Space to discover API endpoints."""
    url = f"https://{space_subdomain}.hf.space/gradio_api/openapi.json"
    try:
        response = httpx.get(url)
        response.raise_for_status()
        return response.text
    except Exception as e:
        return f"Error fetching schema: {e}"

@mcp.tool()
def predict_space(space_id: str, input_data: str, api_name: str = "/predict") -> str:
    """Predicts using a Hugging Face Gradio Space."""
    token = os.environ.get("HF_TOKEN")
    try:
        client = Client(space_id, token=token)
        result = client.predict(input_data, api_name=api_name)
        return str(result)
    except Exception as e:
        return f"Error: {e}"

@mcp.tool()
def create_space(repo_id: str, space_sdk: str = "static") -> str:
    """Creates a new Hugging Face Space."""
    try:
        return create_hf_repo(repo_id, repo_type="space", space_sdk=space_sdk)
    except Exception as e:
        return f"Error: {e}"

@mcp.tool()
def push_files(local_path: str, repo_id: str, repo_type: str = "dataset") -> str:
    """Pushes files from a local directory to a Hugging Face repository."""
    try:
        push_files_to_bucket(local_path, repo_id, repo_type)
        return f"Successfully pushed files from {local_path} to {repo_id} ({repo_type})"
    except Exception as e:
        return f"Error: {e}"

@mcp.tool()
def query_hf(query: str) -> str:
    """Executes a SQL query against Hugging Face data using DuckDB."""
    try:
        result = query_hf_data(query)
        return result.to_string() if result is not None else "No data returned."
    except Exception as e:
        return f"Error: {e}"

@mcp.tool()
def hffs_write(path: str, content: str) -> str:
    """Writes content to a file on Hugging Face Filesystem."""
    try:
        hffs_ops.hffs_write_file(path, content)
        return f"Successfully wrote to {path}"
    except Exception as e:
        return f"Error: {e}"

@mcp.tool()
def hffs_copy(src: str, dst: str) -> str:
    """Copies a file on Hugging Face Filesystem."""
    try:
        hffs_ops.hffs_copy_file(src, dst)
        return f"Successfully copied {src} to {dst}"
    except Exception as e:
        return f"Error: {e}"

@mcp.tool()
def hffs_remove(path: str) -> str:
    """Removes a file on Hugging Face Filesystem."""
    try:
        hffs_ops.hffs_remove_file(path)
        return f"Successfully removed {path}"
    except Exception as e:
        return f"Error: {e}"

@mcp.tool()
def hffs_list(path: str) -> str:
    """Lists files in a directory on Hugging Face Filesystem."""
    try:
        files = hffs_ops.hffs_list_dir(path)
        return str(files)
    except Exception as e:
        return f"Error: {e}"

@mcp.tool()
def hffs_glob(pattern: str) -> str:
    """Glob files on Hugging Face Filesystem."""
    try:
        files = hffs_ops.hffs_glob_files(pattern)
        return str(files)
    except Exception as e:
        return f"Error: {e}"

if __name__ == "__main__":
    mcp.run()