import os import json import asyncio from google.antigravity import Agent, LocalAgentConfig from telegram_utils import send_message # --------------------------------------------------------------- # Storage Setup # --------------------------------------------------------------- BASE_DIR = os.path.dirname(os.path.abspath(__file__)) SAVE_DIR = os.path.join(BASE_DIR, "agent_data") os.makedirs(SAVE_DIR, exist_ok=True) MAPPING_FILE = os.path.join(SAVE_DIR, "user_mapping.json") MAX_RETRIES = 2 # --------------------------------------------------------------- # Custom Tools — The agent can call these Python functions! # --------------------------------------------------------------- def get_current_time() -> str: """Returns the current date and time. Use this when the user asks what time or date it is.""" from datetime import datetime now = datetime.now() return f"The current date and time is: {now.strftime('%A, %B %d, %Y at %I:%M %p')}" def calculate(expression: str) -> str: """Evaluates a simple mathematical expression and returns the result. Args: expression: A mathematical expression string, e.g. '25 * 4 + 10'. """ try: # Safe eval: only allow basic math allowed = set("0123456789+-*/()., ") if not all(c in allowed for c in expression): return "Error: Only basic math operations are allowed (+, -, *, /)." result = eval(expression) return f"The result of '{expression}' is {result}" except Exception as e: return f"Error evaluating expression: {str(e)}" def get_joke() -> str: """Returns a random programming joke. Use when the user asks for a joke.""" import random jokes = [ "Why do programmers prefer dark mode? Because light attracts bugs!", "A SQL query walks into a bar, walks up to two tables and asks... 'Can I join you?'", "Why don't scientists trust atoms? Because they make up everything!", "How many programmers does it take to change a light bulb? None — that's a hardware problem.", "Why did the developer go broke? Because he used up all his cache.", ] return random.choice(jokes) def summarize_text(text: str) -> str: """Summarizes a block of text provided by the user into key bullet points. Args: text: The text to summarize. """ # In a real scenario, you could call an API here. # For now, we return a smart prompt to help the agent do it itself. return f"Please summarize the following text into 3-5 key bullet points:\n\n{text}" # --------------------------------------------------------------- # Agent Configuration # --------------------------------------------------------------- SYSTEM_INSTRUCTIONS = """ You are Agento, a friendly and highly capable AI assistant living inside a Telegram bot. You were built by the Google DeepMind team using the Antigravity SDK. Your personality: - Warm, helpful, and conversational — like a knowledgeable friend. - You respond concisely since this is a chat environment. Keep answers short and to the point. - You use a small amount of relevant emoji to make conversations feel friendly (but don't overdo it). - You remember the full context of the current conversation. Your capabilities: - You can answer general questions, help with writing, coding, and analysis. - You have access to tools: get the current time, calculate math, tell a joke, and summarize text. - Always use your tools when they are relevant to the user's request. Important rules: - Never claim to be a human. - If you don't know something, say so honestly. - Always be respectful and helpful. """ # --------------------------------------------------------------- # Core Agent Logic # --------------------------------------------------------------- async def _chat_with_agent(chat_id: int, text: str): """ Manages conversation state, retries on harness crash, and sends the reply. Each user gets their own save_dir to prevent conversation file conflicts. """ chat_id_str = str(chat_id) # Each user gets their own isolated save directory user_save_dir = os.path.join(SAVE_DIR, f"user_{chat_id_str}") os.makedirs(user_save_dir, exist_ok=True) # Load the user conversation mapping mapping = {} if os.path.exists(MAPPING_FILE): with open(MAPPING_FILE, "r") as f: mapping = json.load(f) conversation_id = mapping.get(chat_id_str) last_exception = None for attempt in range(MAX_RETRIES): try: if attempt > 0: await asyncio.sleep(2) if conversation_id: # RESUME existing conversation config = LocalAgentConfig( model="gemini-3.5-flash", conversation_id=conversation_id, save_dir=user_save_dir, system_instructions=SYSTEM_INSTRUCTIONS, tools=[get_current_time, calculate, get_joke, summarize_text], ) async with Agent(config) as agent: response = await agent.chat(text) final_text = await response.text() send_message(chat_id, final_text) return # Success! else: # START a brand new conversation config = LocalAgentConfig( model="gemini-3.5-flash", save_dir=user_save_dir, system_instructions=SYSTEM_INSTRUCTIONS, tools=[get_current_time, calculate, get_joke, summarize_text], ) async with Agent(config) as agent: response = await agent.chat(text) final_text = await response.text() # Save the new conversation ID for future messages mapping[chat_id_str] = agent.conversation_id with open(MAPPING_FILE, "w") as f: json.dump(mapping, f, indent=2) send_message(chat_id, final_text) return # Success! except Exception as e: last_exception = e error_str = str(e) # ------------------------------------------------------- # SMART RESET: If the saved conversation no longer exists # on disk (e.g. after server move or files were deleted), # clear the stale ID and retry as a brand new conversation. # ------------------------------------------------------- if "cannot resume" in error_str or "not found" in error_str: conversation_id = None # Force a fresh start on next attempt # Clear the bad mapping so we don't try to resume it again mapping.pop(chat_id_str, None) with open(MAPPING_FILE, "w") as f: json.dump(mapping, f, indent=2) continue # Retry immediately as a new conversation # Only retry on transient harness crashes (WS close code 1006) if "1006" not in error_str and "Harness process" not in error_str: break # Non-retriable error, fail fast # All retries failed — notify the user gracefully send_message(chat_id, "Sorry, I ran into a problem processing your request. Please try again in a moment 🙏") raise last_exception def handle_message(chat_id: int, text: str): """ Synchronous entry point called from app.py. Runs the async agent logic in a fresh event loop. """ asyncio.run(_chat_with_agent(chat_id, text))