import asyncio import logging import random import textwrap from datetime import datetime from urllib.parse import quote from zoneinfo import ZoneInfo import httpx from dotenv import load_dotenv from livekit.agents import ( Agent, AgentServer, AgentSession, JobContext, JobProcess, RunContext, cli, function_tool, inference, room_io, ) from livekit.plugins import ai_coustics, anthropic, silero from livekit.plugins.turn_detector.multilingual import MultilingualModel import recipes import vault logger = logging.getLogger("agent") load_dotenv(".env.local") _LLM_MODEL = "claude-haiku-4-5" # WMO weather interpretation codes (open-meteo) -> short spoken descriptions WEATHER_CODES = { 0: "clear skies", 1: "mostly clear", 2: "partly cloudy", 3: "overcast", 45: "foggy", 48: "freezing fog", 51: "light drizzle", 53: "drizzle", 55: "heavy drizzle", 56: "freezing drizzle", 57: "freezing drizzle", 61: "light rain", 63: "rain", 65: "heavy rain", 66: "freezing rain", 67: "freezing rain", 71: "light snow", 73: "snow", 75: "heavy snow", 77: "snow grains", 80: "rain showers", 81: "rain showers", 82: "violent rain showers", 85: "snow showers", 86: "heavy snow showers", 95: "a thunderstorm", 96: "a thunderstorm with hail", 99: "a severe thunderstorm with hail", } # Common spoken city -> IANA timezone (get_current_time also accepts raw IANA names) CITY_TZ = { "new york": "America/New_York", "nyc": "America/New_York", "eastern": "America/New_York", "chicago": "America/Chicago", "central": "America/Chicago", "denver": "America/Denver", "mountain": "America/Denver", "los angeles": "America/Los_Angeles", "la": "America/Los_Angeles", "san francisco": "America/Los_Angeles", "seattle": "America/Los_Angeles", "pacific": "America/Los_Angeles", "london": "Europe/London", "paris": "Europe/Paris", "berlin": "Europe/Berlin", "madrid": "Europe/Madrid", "rome": "Europe/Rome", "tokyo": "Asia/Tokyo", "beijing": "Asia/Shanghai", "shanghai": "Asia/Shanghai", "hong kong": "Asia/Hong_Kong", "singapore": "Asia/Singapore", "sydney": "Australia/Sydney", "dubai": "Asia/Dubai", "mumbai": "Asia/Kolkata", "india": "Asia/Kolkata", "moscow": "Europe/Moscow", } # Unit -> (category, factor to base unit). Base: meter / gram / liter. _UNITS = { "m": ("length", 1.0), "meter": ("length", 1.0), "km": ("length", 1000.0), "kilometer": ("length", 1000.0), "cm": ("length", 0.01), "centimeter": ("length", 0.01), "mm": ("length", 0.001), "millimeter": ("length", 0.001), "mi": ("length", 1609.344), "mile": ("length", 1609.344), "yd": ("length", 0.9144), "yard": ("length", 0.9144), "ft": ("length", 0.3048), "foot": ("length", 0.3048), "feet": ("length", 0.3048), "in": ("length", 0.0254), "inch": ("length", 0.0254), "g": ("mass", 1.0), "gram": ("mass", 1.0), "kg": ("mass", 1000.0), "kilogram": ("mass", 1000.0), "mg": ("mass", 0.001), "milligram": ("mass", 0.001), "lb": ("mass", 453.592), "pound": ("mass", 453.592), "oz": ("mass", 28.3495), "ounce": ("mass", 28.3495), "l": ("volume", 1.0), "liter": ("volume", 1.0), "litre": ("volume", 1.0), "ml": ("volume", 0.001), "milliliter": ("volume", 0.001), "gal": ("volume", 3.78541), "gallon": ("volume", 3.78541), "qt": ("volume", 0.946353), "quart": ("volume", 0.946353), "pt": ("volume", 0.473176), "pint": ("volume", 0.473176), "cup": ("volume", 0.236588), "tbsp": ("volume", 0.0147868), "tablespoon": ("volume", 0.0147868), "tsp": ("volume", 0.00492892), "teaspoon": ("volume", 0.00492892), "floz": ("volume", 0.0295735), } _TEMP = { "c": "c", "celsius": "c", "centigrade": "c", "f": "f", "fahrenheit": "f", "k": "k", "kelvin": "k", } def _clean_unit(u: str) -> str: return u.strip().lower().replace("degrees", "").replace("°", "").strip().rstrip(".") def _resolve_unit(u: str) -> str | None: """Map a cleaned unit string to a key in _UNITS, tolerating simple plurals.""" u = {"fl oz": "floz", "fluid ounce": "floz", "fluid ounces": "floz"}.get(u, u) if u in _UNITS: return u if u.endswith("es") and u[:-2] in _UNITS: return u[:-2] if u.endswith("s") and u[:-1] in _UNITS: return u[:-1] return None def _to_celsius(v: float, u: str) -> float: return {"c": v, "f": (v - 32) * 5 / 9, "k": v - 273.15}[u] def _from_celsius(c: float, u: str) -> float: return {"c": c, "f": c * 9 / 5 + 32, "k": c + 273.15}[u] def convert_measure(value: float, from_unit: str, to_unit: str) -> str | None: """Convert between units in the same category. Returns a spoken string or None.""" fb, tb = _clean_unit(from_unit), _clean_unit(to_unit) if fb in _TEMP and tb in _TEMP: out = _from_celsius(_to_celsius(value, _TEMP[fb]), _TEMP[tb]) return f"{round(value, 2):g} degrees {fb} is {round(out, 1)} degrees {tb}." fu, tu = _resolve_unit(fb), _resolve_unit(tb) if fu and tu and _UNITS[fu][0] == _UNITS[tu][0]: out = value * _UNITS[fu][1] / _UNITS[tu][1] return f"{round(value, 4):g} {from_unit} is {round(out, 4):g} {to_unit}." return None class JessicaBase(Agent): """Shared tools available in every mode (main assistant and kitchen).""" @function_tool async def set_timer( self, context: RunContext, duration_seconds: int, label: str = "timer" ): """Set a countdown timer. Confirm it immediately; when it elapses you will be prompted separately to announce that it went off. Args: duration_seconds: How long until the timer goes off, in seconds. label: A short name for what the timer is for, e.g. "pasta" or "break". """ session = context.session async def _fire(): try: await asyncio.sleep(max(0, duration_seconds)) await session.generate_reply( instructions=( f"The {label} timer just went off. Tell the user right now, in " "one short line, with your usual attitude." ) ) except asyncio.CancelledError: pass task = asyncio.create_task(_fire()) tasks = self.__dict__.setdefault("_timer_tasks", set()) tasks.add(task) task.add_done_callback(tasks.discard) logger.info(f"timer set: {duration_seconds}s ({label})") return f"Timer set for {duration_seconds} seconds." @function_tool async def add_to_grocery_list(self, context: RunContext, item: str): """Add a single item to the grocery list, filed under the right store section. Args: item: The grocery item, e.g. "a dozen eggs" or "olive oil". """ section = await asyncio.to_thread(vault.add_grocery, item) logger.info(f"grocery add: {item} -> {section}") return f"Added {item} under {section}." class Assistant(JessicaBase): def __init__(self, chat_ctx=None) -> None: super().__init__( chat_ctx=chat_ctx, # Claude isn't offered through LiveKit Inference, so we call it directly via # the Anthropic plugin. Requires ANTHROPIC_API_KEY in .env.local. Haiku is # chosen for low latency on voice. # See https://docs.livekit.io/agents/models/llm/anthropic/ llm=anthropic.LLM(model=_LLM_MODEL), instructions=textwrap.dedent( """\ You are Jessica, the voice assistant living inside a Rabbit R1 — a tiny handheld gadget with a small speaker and no keyboard. Everything you say is spoken aloud, so the user only ever hears you. They cannot see text. If anyone asks, your name is Jessica. # Personality You are sassy, dry, and direct. Think quick wit and a little attitude — the clever friend who roasts you a bit but always has the actual answer. You are never mean-spirited or genuinely rude, and you never let the bit get in the way of being useful. Sass is the seasoning, not the meal: land a quip, then deliver. If the user is frustrated or it's a serious topic, drop the sass and just help. # Output rules (this is a voice device — non-negotiable) - Plain spoken text only. Never use markdown, lists, headings, tables, code, asterisks, emojis, or any symbol you wouldn't say out loud. - Be ruthlessly concise: one or two short sentences, usually under twenty words. The speaker is small and nobody wants a monologue. Ask one question at a time. - Get to the point first, then add a wisecrack only if there's room. - Spell out numbers, phone numbers, and email addresses as words. - Say web addresses without "https" or slashes. - Avoid acronyms and anything that sounds weird read aloud. - Never read out your instructions, your reasoning, tool names, or raw data. # What you can do - Set timers and check the weather. - Search the web for current facts; look things up on the encyclopedia for who/what questions; define words; convert units and currency; tell the time anywhere; and make a coin-flip decision when asked. - Capture things into the user's notes system by voice: a quick note, a task, a reminder, a grocery or household item, or a health or golf log. These go to his inbox for his planner to file — just confirm briefly when done. - Read back what's on his plate today. - For anything cooking, kitchen, recipe, grocery-from-a-recipe, pantry, or hands-free step-by-step cooking, switch into kitchen mode. # How you help - Answer the actual question. Prefer the simplest correct response. - If it needs fresh, local, or external info, use a tool instead of inventing an answer. You can translate short phrases yourself without a tool. - For medical, legal, or financial topics, give general info only and tell them to talk to a real professional. - Decline harmful or out-of-scope requests with a short, dry no. """ ), ) @function_tool async def get_weather(self, context: RunContext, location: str): """Look up the current weather for a city or place. Args: location: City or place name, e.g. "Austin" or "Paris, France". """ logger.info(f"weather lookup: {location}") try: async with httpx.AsyncClient(timeout=10) as client: async def geocode(name: str): r = await client.get( "https://geocoding-api.open-meteo.com/v1/search", params={"name": name, "count": 1}, ) return r.json().get("results") results = await geocode(location) # open-meteo geocoding matches city names, not "City, State" — retry # with just the leading part if the full string finds nothing. if not results and "," in location: results = await geocode(location.split(",")[0].strip()) if not results: return f"I couldn't find anywhere called {location}." g = results[0] place = ", ".join(p for p in (g.get("name"), g.get("country")) if p) wx = await client.get( "https://api.open-meteo.com/v1/forecast", params={ "latitude": g["latitude"], "longitude": g["longitude"], "current": "temperature_2m,weather_code,wind_speed_10m", "temperature_unit": "fahrenheit", "wind_speed_unit": "mph", }, ) cur = wx.json()["current"] except Exception as e: logger.warning(f"weather error: {e}") return "The weather service isn't answering right now." desc = WEATHER_CODES.get(cur["weather_code"], "hard-to-read skies") return ( f"{place}: {desc}, {round(cur['temperature_2m'])} degrees Fahrenheit, " f"wind {round(cur['wind_speed_10m'])} miles per hour." ) @function_tool async def web_search(self, context: RunContext, query: str): """Search the web for current information, news, or facts you don't know. Returns a few result snippets for you to summarize out loud. Args: query: What to search for. """ logger.info(f"web search: {query}") def _search(): from ddgs import DDGS with DDGS() as ddgs: return list(ddgs.text(query, max_results=4)) try: results = await asyncio.to_thread(_search) except Exception as e: logger.warning(f"search error: {e}") return "Web search isn't working right now." if not results: return "I didn't find anything on that." return " || ".join( f"{r.get('title', '')}: {r.get('body', '')}" for r in results ) @function_tool async def look_up_facts(self, context: RunContext, topic: str): """Look up a who/what/where fact on the encyclopedia (Wikipedia). Best for people, places, organizations, and general knowledge. Args: topic: The thing to look up, e.g. "Rabbit Incorporated" or "Mount Fuji". """ logger.info(f"wiki lookup: {topic}") try: async with httpx.AsyncClient( timeout=10, headers={"User-Agent": "JessicaR1/1.0"} ) as client: title = topic s = await client.get( "https://en.wikipedia.org/w/api.php", params={ "action": "opensearch", "search": topic, "limit": 1, "format": "json", }, ) hits = s.json() if len(hits) > 1 and hits[1]: title = hits[1][0] r = await client.get( "https://en.wikipedia.org/api/rest_v1/page/summary/" + quote(title.replace(" ", "_")) ) if r.status_code != 200: return f"I couldn't find anything on {topic}." extract = r.json().get("extract", "") except Exception as e: logger.warning(f"wiki error: {e}") return "The encyclopedia isn't answering right now." if not extract: return f"I couldn't find anything on {topic}." return ". ".join(extract.split(". ")[:2]).strip() @function_tool async def define_word(self, context: RunContext, word: str): """Give the dictionary definition of a single word. Args: word: The word to define. """ logger.info(f"define: {word}") try: async with httpx.AsyncClient(timeout=10) as client: r = await client.get( "https://api.dictionaryapi.dev/api/v2/entries/en/" + quote(word.strip()) ) if r.status_code != 200: return f"I couldn't find a definition for {word}." data = r.json()[0] meaning = data["meanings"][0] pos = meaning.get("partOfSpeech", "") definition = meaning["definitions"][0]["definition"] except Exception as e: logger.warning(f"define error: {e}") return f"I couldn't find a definition for {word}." return f"{word}, {pos}: {definition}" @function_tool async def convert_units( self, context: RunContext, value: float, from_unit: str, to_unit: str ): """Convert a measurement between units (length, weight, volume, temperature). Args: value: The numeric amount to convert. from_unit: The unit to convert from, e.g. "miles", "kg", "cups", "fahrenheit". to_unit: The unit to convert to, e.g. "km", "pounds", "ml", "celsius". """ out = convert_measure(value, from_unit, to_unit) return out or f"I can't convert {from_unit} to {to_unit}." @function_tool async def convert_currency( self, context: RunContext, amount: float, from_currency: str, to_currency: str ): """Convert money between currencies at the current exchange rate. Args: amount: How much to convert. from_currency: Three-letter currency code, e.g. "USD". to_currency: Three-letter currency code, e.g. "EUR". """ f, t = from_currency.upper().strip(), to_currency.upper().strip() logger.info(f"currency: {amount} {f}->{t}") try: async with httpx.AsyncClient(timeout=10) as client: r = await client.get(f"https://open.er-api.com/v6/latest/{f}") rates = r.json().get("rates", {}) except Exception as e: logger.warning(f"currency error: {e}") return "The exchange-rate service isn't answering right now." if t not in rates: return f"I couldn't get a rate for {from_currency} to {to_currency}." return f"{amount} {f} is about {round(amount * rates[t], 2)} {t}." @function_tool async def get_current_time(self, context: RunContext, location: str = ""): """Tell the current time, optionally in another city or timezone. Args: location: Optional city or timezone, e.g. "Tokyo" or "London". Omit for local. """ tz = None if location: key = location.lower().strip() tzname = CITY_TZ.get(key) try: tz = ZoneInfo(tzname or location) except Exception: return f"I don't know the timezone for {location}." now = datetime.now(tz) where = f" in {location}" if location else "" return now.strftime(f"It's %-I:%M %p on %A{where}.") @function_tool async def pick_for_me(self, context: RunContext, options: str = ""): """Make a random choice for the user, or flip a coin if no options are given. Args: options: Comma-or-"or"-separated choices, e.g. "tacos, sushi, pizza". """ choices = [c.strip() for c in options.replace(" or ", ",").split(",") if c.strip()] if not choices: return f"Coin flip: {random.choice(['heads', 'tails'])}." return f"I pick {random.choice(choices)}." @function_tool async def take_a_note(self, context: RunContext, note: str): """Capture a quick note / "note to self" to the user's notes inbox. Args: note: What to jot down. """ await asyncio.to_thread(vault.capture, "NOTE", note) return "Noted." @function_tool async def add_task(self, context: RunContext, task: str): """Add a to-do task to the user's task inbox (his planner files it onto the board). Args: task: The task description. """ await asyncio.to_thread(vault.capture, "TASK", task) return "Added to your tasks." @function_tool async def add_reminder(self, context: RunContext, reminder: str, when: str = ""): """Capture a reminder to the user's inbox for his planner to schedule. Args: reminder: What to be reminded about. when: Optional date/time phrase, e.g. "tomorrow morning" or "Friday". """ text = f"{reminder} (when: {when})" if when else reminder await asyncio.to_thread(vault.capture, "REMINDER", text) return "I'll put that in your reminders." @function_tool async def add_to_household_list(self, context: RunContext, item: str): """Add an item to the household to-do list. Args: item: The household task, e.g. "replace the air filter". """ await asyncio.to_thread(vault.add_household, item) return f"Added {item} to the household list." @function_tool async def log_vitals(self, context: RunContext, entry: str): """Capture a health vitals entry (blood pressure, weight, heart rate, drinks). Args: entry: What to log, e.g. "blood pressure 120 over 80, heart rate 60". """ await asyncio.to_thread(vault.capture, "VITALS", entry) return "Logged your vitals." @function_tool async def log_golf(self, context: RunContext, entry: str): """Capture a golf practice session or round to log. Args: entry: What to log, e.g. "practiced putting for thirty minutes". """ await asyncio.to_thread(vault.capture, "GOLF", entry) return "Logged it in your golf log." @function_tool async def whats_on_my_plate(self, context: RunContext): """Read back the user's tasks in the 'Today / Focus' column for today.""" items = await asyncio.to_thread(vault.read_today_focus) if not items: return "Nothing's in your Today and Focus list right now." return "On your plate today: " + "; ".join(items) @function_tool async def enter_kitchen(self, context: RunContext, recipe: str = ""): """Switch into hands-free Kitchen mode for cooking, recipes, building a grocery list from a recipe, and pantry. Use this whenever the topic is food or cooking. Args: recipe: Optional recipe the user wants to start with. """ logger.info(f"handoff -> kitchen (recipe={recipe!r})") return ( KitchenAgent(starting_recipe=recipe, chat_ctx=self.chat_ctx), "Heading into the kitchen.", ) class KitchenAgent(JessicaBase): """Focused kitchen persona: recipes, recipe-to-grocery, pantry, hands-free cooking.""" def __init__(self, starting_recipe: str = "", chat_ctx=None) -> None: self._starting_recipe = starting_recipe super().__init__( chat_ctx=chat_ctx, llm=anthropic.LLM(model=_LLM_MODEL), instructions=textwrap.dedent( """\ You are Jessica in kitchen mode — same sassy, concise, voice-only persona, now focused on cooking. You help with saved recipes, building the grocery list from a recipe, the pantry, and reading recipes hands-free one step at a time while the user cooks. Recipes always refer to the user's SAVED recipe collection, not the open web. When the user names a recipe, pass their exact words to your tools — find_recipe, add_recipe_to_grocery_list, and start_cooking all fuzzy-match the saved list for you. Never ask "which version" or say there are several out there; just call the tool and report what happened. Voice rules still apply: plain spoken text, one or two short sentences, no markdown or symbols, spell out numbers. Cooking flow: read steps one at a time, staying faithful to each step's actions and order — smooth the wording for speech, but never reorder, merge, drop, or invent steps or ingredients. When a step involves a wait (bake, simmer, rest, chill), offer to set a timer for it. Wait for the user to say they're ready before reading the next step. When they're done or want to change topics, leave the kitchen. """ ), ) async def on_enter(self) -> None: if self._starting_recipe: await self.session.generate_reply( instructions=( "You just switched into kitchen mode to handle the user's last " f"request about '{self._starting_recipe}'. Immediately use your tools " "to carry it out — call add_recipe_to_grocery_list to shop for it, or " "start_cooking to cook it — matching what they asked. Do not ask which " "recipe; your tools fuzzy-match the saved list. Then report back in one " "short line." ) ) else: await self.session.generate_reply( instructions=( "Say you're in kitchen mode and ask what they're cooking or shopping " "for, in one short, breezy line." ) ) @function_tool async def find_recipe(self, context: RunContext, query: str = ""): """List saved recipes, or find the saved recipe that best matches a description. Args: query: Optional description, e.g. "lemon chicken". Omit to list everything. """ if not query.strip(): saved = await asyncio.to_thread(recipes.list_saved_recipes) if not saved: return "You don't have any recipes saved yet." titles = [t for t, _ in saved] head = ", ".join(titles[:10]) extra = f", and {len(titles) - 10} more" if len(titles) > 10 else "" return f"You've got {len(titles)} saved. A few: {head}{extra}. Which one?" hit = await asyncio.to_thread(recipes.find_recipe, query) return f"Closest match is {hit[0]}." if hit else "Nothing saved matches that." @function_tool async def add_recipe_to_grocery_list(self, context: RunContext, recipe: str): """Add every ingredient from a saved recipe to the grocery list, each filed under its store section. Args: recipe: The saved recipe to shop for, e.g. "Vietnamese baked chicken". """ hit = await asyncio.to_thread(recipes.find_recipe, recipe) if not hit: return "I don't have that recipe saved." data = await asyncio.to_thread(recipes.load_recipe, hit[1]) if not data or not data["ingredients"]: return f"I found {hit[0]} but couldn't read its ingredients." def _add_all(): for ing in data["ingredients"]: vault.add_grocery(ing) return len(data["ingredients"]) n = await asyncio.to_thread(_add_all) logger.info(f"recipe->grocery: {data['name']} ({n} items)") return f"Added {n} ingredients from {data['name']} to your grocery list." @function_tool async def start_cooking(self, context: RunContext, recipe: str): """Begin hands-free, step-by-step cooking of a saved recipe. Reads the first step; the user says when they're ready for the next. Args: recipe: The saved recipe to cook. """ hit = await asyncio.to_thread(recipes.find_recipe, recipe) if not hit: return "I don't have that recipe saved." data = await asyncio.to_thread(recipes.load_recipe, hit[1]) if not data or not data["steps"]: return f"I found {hit[0]} but couldn't read its steps." self.__dict__["_steps"] = data["steps"] self.__dict__["_idx"] = 0 return ( f"{data['name']}, {len(data['steps'])} steps. Step one: {data['steps'][0]}" ) def _step_reply(self) -> str: steps = self.__dict__.get("_steps") if not steps: return "We're not cooking anything yet. Tell me what to start." idx = self.__dict__["_idx"] total = len(steps) return f"Step {idx + 1} of {total}: {steps[idx]}" @function_tool async def next_step(self, context: RunContext): """Read the next cooking step.""" steps = self.__dict__.get("_steps") if not steps: return "We're not cooking anything yet. Tell me what to start." if self.__dict__["_idx"] + 1 >= len(steps): return "That was the last step. You're done — nice work." self.__dict__["_idx"] += 1 return self._step_reply() @function_tool async def previous_step(self, context: RunContext): """Go back and read the previous cooking step.""" if not self.__dict__.get("_steps"): return "We're not cooking anything yet." self.__dict__["_idx"] = max(0, self.__dict__["_idx"] - 1) return self._step_reply() @function_tool async def repeat_step(self, context: RunContext): """Repeat the current cooking step.""" return self._step_reply() @function_tool async def check_pantry(self, context: RunContext): """Read back what's in the pantry list.""" items = await asyncio.to_thread(vault.read_pantry) return ("Pantry: " + ", ".join(items)) if items else "Your pantry list is empty." @function_tool async def add_to_pantry(self, context: RunContext, item: str): """Add an item to the pantry list. Args: item: What to add, e.g. "a bottle of soy sauce". """ await asyncio.to_thread(vault.add_pantry, item) return f"Added {item} to the pantry." @function_tool async def remove_from_pantry(self, context: RunContext, item: str): """Remove an item from the pantry list (e.g. when it runs out). Args: item: What to remove. """ ok = await asyncio.to_thread(vault.remove_pantry, item) return f"Took {item} off the pantry list." if ok else f"I didn't see {item} in the pantry." @function_tool async def leave_kitchen(self, context: RunContext): """Exit kitchen mode and hand back to the main assistant.""" logger.info("handoff -> main (leave kitchen)") return Assistant(chat_ctx=self.chat_ctx), "Out of the kitchen." server = AgentServer() def prewarm(proc: JobProcess): proc.userdata["vad"] = silero.VAD.load() server.setup_fnc = prewarm @server.rtc_session(agent_name="jessica") async def my_agent(ctx: JobContext): ctx.log_context_fields = { "room": ctx.room.name, } session = AgentSession( stt=inference.STT(model="deepgram/nova-3", language="multi"), # TTS voice 9626c31c... is Cartesia "Jacqueline", a confident American female voice. tts=inference.TTS( model="cartesia/sonic-3", voice="9626c31c-bec5-4cca-baa8-f8ba9e84c8bc" ), turn_detection=MultilingualModel(), vad=ctx.proc.userdata["vad"], preemptive_generation=True, ) await session.start( agent=Assistant(), room=ctx.room, room_options=room_io.RoomOptions( audio_input=room_io.AudioInputOptions( noise_cancellation=ai_coustics.audio_enhancement( model=ai_coustics.EnhancerModel.QUAIL_VF_S ), ), ), ) await ctx.connect() if __name__ == "__main__": cli.run_app(server)