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

Agent Running

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"))