Refactor app.py and update requirements.txt

- Removed unused imports and refactored environment variable handling in app.py.
- Updated gradio version in requirements.txt for compatibility.

app.py

@@ -1,18 +1,14 @@
-import argparse
+import os
 import queue
+import random
 import time
 from threading import Thread
-from typing import Callable, Literal, override
-import os
+from typing import Any, Callable, Literal, override
 
 import fastrtc
-from fastrtc import get_cloudflare_turn_credentials_async
 import gradio as gr
 import httpx
 import numpy as np
-from pydantic import BaseModel
-import random
-
 
 from api_schema import (
     AbortController,
@@ -28,61 +24,66 @@ from api_schema import (
 )
 
 HF_TOKEN = os.getenv("HF_TOKEN")
-if HF_TOKEN is None:
-    print(
-        "⚠️ [WARNING] HF_TOKEN environment variable not found.\n"
-        "WebRTC connections may fail on Hugging Face Spaces because TURN service cannot be used.\n"
-        "💡 Solution: Go to your Hugging Face Space → Settings → Secrets, "
-        "add a variable named HF_TOKEN or HF_ACCESS_TOKEN with your personal access token (with at least 'read' permission)."
-    )
-else:
-    print("✅ [INFO] HF_TOKEN detected. WebRTC will use Hugging Face TURN service for connectivity.")
-
-
-url_prefix = os.getenv("URL_PREFIX")
-server_number = int(os.getenv("NUM_SERVER"))
-
-deployment_server = []
-for i in range(1, server_number+1):
-    url = url_prefix + str(i) + ".hf.space"
-    deployment_server.append(url)
-
-
-class Args(BaseModel):
-    host: str
-    port: int
-    concurrency_limit: int
-    share: bool
-    debug: bool
-    chat_server: str
-    tag: str | None = None
-
-    @classmethod
-    def parse_args(cls):
-        parser = argparse.ArgumentParser(description="Xiaomi MiMo-Audio Chat")
-        parser.add_argument("--host", default="0.0.0.0")
-        parser.add_argument("--port", type=int, default=8087)
-        parser.add_argument("--concurrency-limit", type=int, default=32)
-        parser.add_argument("--share", action="store_true")
-        parser.add_argument("--debug", action="store_true")
-        parser.add_argument(
-            "-S",
-            "--chat-server",
-            dest="chat_server",
-            type=str,
-            default="deployment_docker_1",
-        )
-        parser.add_argument("--tag", type=str)
+SERVER_LIST = os.getenv("SERVER_LIST")
+TURN_KEY_ID = os.getenv("TURN_KEY_ID")
+TURN_KEY_API_TOKEN = os.getenv("TURN_KEY_API_TOKEN")
+CONCURRENCY_LIMIT = os.getenv("CONCURRENCY_LIMIT")
+
+
+assert SERVER_LIST is not None, "SERVER_LIST environment variable is required."
+assert TURN_KEY_ID is not None and TURN_KEY_API_TOKEN is not None, (
+    "TURN_KEY_ID and TURN_KEY_API_TOKEN environment variables are required "
+)
+
+deployment_server = [
+    server_url.strip() for server_url in SERVER_LIST.split(",") if server_url.strip()
+]
 
-        args = parser.parse_args()
-        return cls.model_validate(vars(args))
+assert len(deployment_server) > 0, "SERVER_LIST must contain at least one server URL."
 
-    def chat_server_url(self):
-        return deployment_server[random.randint(0,server_number-1)]
-        # if self.chat_server in global_chat_server_map:
-        #     return global_chat_server_map[self.chat_server]
+default_concurrency_limit = 32
+try:
+    concurrency_limit = (
+        int(CONCURRENCY_LIMIT)
+        if CONCURRENCY_LIMIT is not None
+        else default_concurrency_limit
+    )
+except ValueError:
+    concurrency_limit = default_concurrency_limit
+
+
+def chat_server_url(pathname: str = "/") -> httpx.URL:
+    n = len(deployment_server)
+    server_idx = random.randint(0, n - 1)
+    host = deployment_server[server_idx]
+    return httpx.URL(host).join(pathname)
+
+
+def auth_headers() -> dict[str, str]:
+    if HF_TOKEN is None:
+        return {}
+    return {"Authorization": f"Bearer {HF_TOKEN}"}
+
+
+def get_cloudflare_turn_credentials(
+    ttl: int = 1200,  # 20 minutes
+) -> dict[str, Any]:
+    with httpx.Client() as client:
+        response = client.post(
+            f"https://rtc.live.cloudflare.com/v1/turn/keys/{TURN_KEY_ID}/credentials/generate-ice-servers",
+            headers={
+                "Authorization": f"Bearer {TURN_KEY_API_TOKEN}",
+                "Content-Type": "application/json",
+            },
+            json={"ttl": ttl},
+        )
+        if response.is_success:
+            return response.json()
+        else:
+            raise Exception(
+                f"Failed to get TURN credentials: {response.status_code} {response.text}"
+            )
 
-        # return self.chat_server
 
 class NeverVAD(fastrtc.PauseDetectionModel):
     def vad(self, *_args, **_kwargs):
@@ -152,7 +153,6 @@ class ReplyOnMuted(fastrtc.ReplyOnPause):
         return False
 
 
-
 class ConversationManager:
     def __init__(self, assistant_style: AssistantStyle | None = None):
         self.conversation = TokenizedConversation(messages=[])
@@ -269,6 +269,7 @@ class ConversationManager:
             except queue.Empty:
                 yield None
 
+
 def get_microphone_svg(muted: bool | None = None):
     muted_svg = '<line x1="1" y1="1" x2="23" y2="23"></line>' if muted else ""
     return f"""
@@ -309,8 +310,6 @@ def new_chat_id():
 
 
 def main():
-    args = Args.parse_args()
-
     print("Starting WebRTC server")
 
     conversations: dict[str, ConversationManager] = {}
@@ -330,23 +329,17 @@ def main():
     Thread(target=cleanup_idle_conversations, daemon=True).start()
 
     def get_preset_list(category: Literal["character", "voice"]) -> list[str]:
-        url = httpx.URL(args.chat_server_url()).join(f"/preset/{category}")
-        headers = {
-            "Authorization": f"Bearer {HF_TOKEN}"  # <-- 加上 token
-        }
+        url = chat_server_url(f"/preset/{category}")
         with httpx.Client() as client:
-            response = client.get(url, headers=headers)
+            response = client.get(url, headers=auth_headers())
             if response.status_code == 200:
                 return PresetOptions.model_validate_json(response.text).options
             return ["[default]"]
 
     def get_model_name() -> str:
-        url = httpx.URL(args.chat_server_url()).join("/model-name")
-        headers = {
-            "Authorization": f"Bearer {HF_TOKEN}"  # <-- 加上 token
-        }
+        url = chat_server_url("/model-name")
         with httpx.Client() as client:
-            response = client.get(url, headers=headers)
+            response = client.get(url, headers=auth_headers())
             if response.status_code == 200:
                 return ModelNameResponse.model_validate_json(response.text).model_name
             return "unknown"
@@ -354,8 +347,6 @@ def main():
     def load_initial_data():
         model_name = get_model_name()
         title = f"Xiaomi MiMo-Audio WebRTC (model: {model_name})"
-        if args.tag is not None:
-            title = f"{args.tag} - {title}"
         character_choices = get_preset_list("character")
         voice_choices = get_preset_list("voice")
         return (
@@ -371,12 +362,6 @@ def main():
         preset_voice: str | None,
         custom_character_prompt: str | None,
     ):
-        headers = {
-            "Authorization": f"Bearer {HF_TOKEN}"  # <-- 加上 token
-        }
-        # deprecate gc
-        # with httpx.Client() as client:
-        #     client.get(httpx.URL(args.chat_server_url()).join("/gc"), headers=headers)
         nonlocal conversations
 
         if webrtc_id not in conversations:
@@ -416,7 +401,7 @@ def main():
         yield additional_outputs()
 
         try:
-            url = httpx.URL(args.chat_server_url()).join("/audio-chat")
+            url = chat_server_url("/audio-chat")
             for chunk in manager.chat(
                 url,
                 chat_id,
@@ -463,8 +448,6 @@ def main():
         yield additional_outputs()
 
     title = "Xiaomi MiMo-Audio WebRTC"
-    if args.tag is not None:
-        title = f"{args.tag} - {title}"
 
     with gr.Blocks(title=title) as demo:
         title_markdown = gr.Markdown(f"# {title}")
@@ -482,9 +465,7 @@ def main():
                     modality="audio",
                     mode="send-receive",
                     full_screen=False,
-                    rtc_configuration=get_cloudflare_turn_credentials_async
-                    # server_rtc_configuration=get_hf_turn_credentials(ttl=600 * 1000),
-                    # rtc_configuration=get_hf_turn_credentials,
+                    rtc_configuration=get_cloudflare_turn_credentials,
                 )
                 output_text = gr.Textbox(label="Output", lines=3, interactive=False)
                 status_text = gr.Textbox(label="Status", lines=1, interactive=False)
@@ -529,13 +510,13 @@ def main():
                 preset_voice_dropdown,
                 custom_character_prompt,
             ],
-            concurrency_limit=args.concurrency_limit,
+            concurrency_limit=concurrency_limit,
             outputs=[chat],
         )
         chat.on_additional_outputs(
             lambda *args: args,
             outputs=[output_text, status_text, collected_audio],
-            concurrency_limit=args.concurrency_limit,
+            concurrency_limit=concurrency_limit,
             show_progress="hidden",
         )
 
@@ -545,9 +526,9 @@ def main():
             outputs=[title_markdown, preset_character_dropdown, preset_voice_dropdown],
         )
         demo.queue(
-            default_concurrency_limit=args.concurrency_limit,
+            default_concurrency_limit=concurrency_limit,
         )
-        
+
     demo.launch()
 
 

requirements.txt

@@ -1,5 +1,5 @@
 fastapi==0.116.1
 pydantic==2.11.7
-fastrtc[vad]==0.0.33
-gradio==5.35.0
-httpx==0.28.1
+fastrtc==0.0.33
+gradio==5.44.1
+httpx==0.28.1