File size: 3,349 Bytes
5da5531 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 | 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()
|