specimba commited on
Commit
5fb2355
·
verified ·
1 Parent(s): 1f5af58

Update app.py

Browse files
Files changed (1) hide show
  1. app.py +47 -58
app.py CHANGED
@@ -1,82 +1,71 @@
1
  import gradio as gr
2
  import spaces
3
- from nexus_hf_mcp_bridge import nexus_hf_bridge
4
- from nexus_hf_mcp_bridge.hf_oauth import HFOAuthClient
5
- from nexus_hf_mcp_bridge.token_manager import HFTokenManager
6
 
7
  @spaces.GPU
8
  def _init_gpu():
9
  pass
10
 
11
- client = HFOAuthClient()
12
- device_data = gr.State({}) # Better than global for Gradio
13
 
14
- def start_auth():
15
- try:
16
- data = client.start_device_flow()
17
- return data, {"status": "pending", "message": "Open the URL and enter the code"}
18
- except Exception as e:
19
- return {}, {"status": "error", "message": str(e)}
 
 
 
 
20
 
21
- def complete_auth(data):
22
- if not data:
23
- return {"status": "error", "message": "Please start authentication first"}
24
  try:
25
- token = client.poll_for_token(data["device_code"])
26
- if token:
27
- tm = HFTokenManager()
28
- tm.save_token(token)
29
- return {"status": "success", "message": "✅ Authentication successful!"}
30
- else:
31
- return {"status": "error", "message": "Not authorized yet or timed out"}
 
32
  except Exception as e:
33
  return {"status": "error", "message": str(e)}
34
 
35
- def get_status():
36
- tm = HFTokenManager()
37
- return " Authenticated" if tm.get_valid_token() else "Not authenticated"
38
-
39
- def safe_write(repo_id, path, content):
40
- tm = HFTokenManager()
41
- if not tm.get_valid_token():
42
- return {"error": "Please authenticate first"}
43
- if not all([repo_id, path, content]):
44
- return {"error": "Please fill all fields"}
45
- return nexus_hf_bridge.tool_create_or_update_file(repo_id, path, content)
46
-
47
- def safe_list(repo_id):
48
- tm = HFTokenManager()
49
- if not tm.get_valid_token():
50
- return {"error": "Please authenticate first"}
51
  if not repo_id:
52
- return {"error": "Please provide repository ID"}
53
- return nexus_hf_bridge.tool_list_repo_files(repo_id)
 
 
 
 
 
 
54
 
 
55
  with gr.Blocks(title="NEXUS Visual Weaver") as demo:
56
  gr.Markdown("# NEXUS Visual Weaver - HF MCP Bridge")
57
-
58
- with gr.Tab("Authentication"):
59
- gr.Markdown("### Step 1: Start Authentication")
60
- start_btn = gr.Button("Start Authentication", variant="primary")
61
- device_info = gr.JSON(label="Device Code Info")
62
-
63
- gr.Markdown("### Step 2: Complete Authentication")
64
- complete_btn = gr.Button("I have authorized in browser", variant="secondary")
65
- result = gr.JSON(label="Result")
66
-
67
- start_btn.click(start_auth, outputs=[device_info, result])
68
- complete_btn.click(complete_auth, inputs=[device_info], outputs=result)
69
-
70
- gr.Markdown("### Status")
71
- gr.Button("Check Authentication Status").click(get_status, outputs=gr.Textbox())
72
 
73
  with gr.Tab("Repository Tools"):
74
- repo = gr.Textbox(label="Repository ID")
75
- path = gr.Textbox(label="Path in Repo")
76
  content = gr.Textbox(label="Content", lines=6)
77
 
78
  with gr.Row():
79
- gr.Button("Create / Update File").click(safe_write, inputs=[repo, path, content], outputs=gr.JSON())
80
- gr.Button("List Files").click(safe_list, inputs=[repo], outputs=gr.JSON())
 
 
 
 
 
 
 
 
81
 
82
  demo.launch(mcp_server=True)
 
1
  import gradio as gr
2
  import spaces
3
+ import os
4
+ from huggingface_hub import HfApi
 
5
 
6
  @spaces.GPU
7
  def _init_gpu():
8
  pass
9
 
10
+ # Read token from Space Secret (recommended) or environment
11
+ HF_TOKEN = os.environ.get("HF_TOKEN")
12
 
13
+ def get_status():
14
+ if HF_TOKEN:
15
+ return "✅ Token loaded from Space Secrets"
16
+ return " No HF_TOKEN found. Please add it in Space Settings → Secrets"
17
+
18
+ def create_or_update_file(repo_id: str, path_in_repo: str, content: str):
19
+ if not HF_TOKEN:
20
+ return {"status": "error", "message": "HF_TOKEN not found. Add it in Space Secrets."}
21
+ if not repo_id or not path_in_repo or not content:
22
+ return {"status": "error", "message": "Please fill all fields"}
23
 
 
 
 
24
  try:
25
+ api = HfApi(token=HF_TOKEN)
26
+ api.upload_file(
27
+ path_or_fileobj=content.encode("utf-8"),
28
+ path_in_repo=path_in_repo,
29
+ repo_id=repo_id,
30
+ commit_message="Updated via NEXUS HF Bridge"
31
+ )
32
+ return {"status": "success", "message": f"File '{path_in_repo}' updated successfully"}
33
  except Exception as e:
34
  return {"status": "error", "message": str(e)}
35
 
36
+ def list_repo_files(repo_id: str):
37
+ if not HF_TOKEN:
38
+ return {"status": "error", "message": "HF_TOKEN not found"}
 
 
 
 
 
 
 
 
 
 
 
 
 
39
  if not repo_id:
40
+ return {"status": "error", "message": "Please provide repository ID"}
41
+
42
+ try:
43
+ api = HfApi(token=HF_TOKEN)
44
+ files = api.list_repo_files(repo_id=repo_id)
45
+ return {"status": "success", "files": files}
46
+ except Exception as e:
47
+ return {"status": "error", "message": str(e)}
48
 
49
+ # ===================== UI =====================
50
  with gr.Blocks(title="NEXUS Visual Weaver") as demo:
51
  gr.Markdown("# NEXUS Visual Weaver - HF MCP Bridge")
52
+ gr.Markdown("**Using Space Secrets (secure)**")
 
 
 
 
 
 
 
 
 
 
 
 
 
 
53
 
54
  with gr.Tab("Repository Tools"):
55
+ repo_id = gr.Textbox(label="Repository ID", placeholder="username/model-name")
56
+ path = gr.Textbox(label="Path in Repo", placeholder="folder/file.txt")
57
  content = gr.Textbox(label="Content", lines=6)
58
 
59
  with gr.Row():
60
+ write_btn = gr.Button("Create / Update File", variant="primary")
61
+ list_btn = gr.Button("List Files")
62
+
63
+ result = gr.JSON(label="Result")
64
+
65
+ write_btn.click(create_or_update_file, inputs=[repo_id, path, content], outputs=result)
66
+ list_btn.click(list_repo_files, inputs=[repo_id], outputs=result)
67
+
68
+ with gr.Tab("Status"):
69
+ gr.Button("Check Token Status").click(get_status, outputs=gr.Textbox())
70
 
71
  demo.launch(mcp_server=True)