import sys
import threading
from http.server import HTTPServer, BaseHTTPRequestHandler
import time
import requests
from dotenv import load_dotenv
load_dotenv()
# ── Simple status web server (HF Spaces requires port 7860) ──────────────────
STATUS_HTML = """
Habari Jarvis — Swahili AI Agent
🤖
Habari Jarvis
Msaidizi wa Akili Bandia • Swahili AI Voice Agent
Connect via LiveKit to start a voice conversation.
The agent listens and responds entirely in Kiswahili.
Powered by Groq Whisper • LLaMA 3.3 70B • MMS/API TTS
"""
class StatusHandler(BaseHTTPRequestHandler):
def do_GET(self):
self.send_response(200)
self.send_header("Content-type", "text/html; charset=utf-8")
self.end_headers()
self.wfile.write(STATUS_HTML.encode("utf-8"))
def log_message(self, *args): # silence access logs
pass
def run_status_server():
server = HTTPServer(("0.0.0.0", 7860), StatusHandler)
server.serve_forever()
# ── Auto-ping mechanism to keep HF Space awake ───────────────────────────────
def keep_awake():
# Wait for the server to start
time.sleep(10)
url = "https://huggingface.co/spaces/Mapmaven/duol-agent"
while True:
try:
requests.get(url, timeout=10)
print(f"[OK] Auto-pinged {url} to prevent sleep")
except Exception as e:
print(f"[WARN] Auto-ping failed: {e}")
time.sleep(600) # Ping every 10 minutes
# ── Start LiveKit agent worker ────────────────────────────────────────────────
from livekit.agents import WorkerOptions, cli
from agent import entrypoint
if __name__ == "__main__":
# ── Start servers in background threads ───────────────────────────────────────
t1 = threading.Thread(target=run_status_server, daemon=True)
t1.start()
t2 = threading.Thread(target=keep_awake, daemon=True)
t2.start()
print("[OK] Status server started on port 7860")
cli.run_app(WorkerOptions(entrypoint_fnc=entrypoint, agent_name="DUOL"))