# app.py import os import re import json import requests import gradio as gr from google.oauth2.credentials import Credentials from googleapiclient.discovery import build from agentpro import create_model, ReactAgent from agentpro.tools import AresInternetTool # ────────────────────────────────────────────────────────────────────────────── # 1) READ ENVIRONMENT VARIABLES (set in HF Space Secrets) # ────────────────────────────────────────────────────────────────────────────── OPENAI_API_KEY = os.getenv("OPENAI_API_KEY") GOOGLE_CLIENT_ID = os.getenv("GOOGLE_CLIENT_ID") GOOGLE_CLIENT_SECRET= os.getenv("GOOGLE_CLIENT_SECRET") ARES_API_KEY = os.getenv("ARES_API_KEY") # if you are using AresInternetTool # Your HF Space URL (must match what you registered as redirect URI in Google Cloud Console) HF_SPACE_URL = "https://huggingface.co/spaces/case-llm-traversaal/calendar-chatbot" # ────────────────────────────────────────────────────────────────────────────── # 2) GLOBAL IN-MEMORY STORAGE FOR TOKENS (basic demo) # ────────────────────────────────────────────────────────────────────────────── # In a production ­grade multiuser app, key by user ID or IP. Here, we keep one “default” slot. user_tokens = { "default": None # Will hold a dict { "access_token": ..., "refresh_token": ..., "expiry": ... } } # ────────────────────────────────────────────────────────────────────────────── # 3) HELPERS: Build Google Auth URL / Extract & Exchange Code / Build Service # ────────────────────────────────────────────────────────────────────────────── def build_auth_url(): """ Construct the Google OAuth2 authorization URL. When the user clicks 'Connect', we open this in a new tab. Google will redirect to HF_SPACE_URL?code=XXXX. """ base = "https://accounts.google.com/o/oauth2/v2/auth" scope = "https://www.googleapis.com/auth/calendar.readonly https://www.googleapis.com/auth/calendar.events" params = { "client_id": GOOGLE_CLIENT_ID, "redirect_uri": HF_SPACE_URL, "response_type": "code", "scope": scope, "access_type": "offline", # get a refresh token "prompt": "consent" # always ask consent to receive a refresh token } # Build query string manually q = "&".join([f"{k}={requests.utils.quote(v)}" for k, v in params.items()]) return f"{base}?{q}" def extract_auth_code(full_redirect_url: str) -> str: """ The user pastes the entire redirect URL (which contains '?code=XYZ&scope=...' etc.). We pull out the 'XYZ' (authorization code) to exchange for tokens. """ match = re.search(r"[?&]code=([^&]+)", full_redirect_url) if match: return match.group(1) return None def exchange_code_for_tokens(auth_code: str): """ Given the single‐use auth_code from Google, exchange it at Google's token endpoint for access_token + refresh_token + expiry. Then store it in user_tokens["default"]. """ token_url = "https://oauth2.googleapis.com/token" data = { "code": auth_code, "client_id": GOOGLE_CLIENT_ID, "client_secret": GOOGLE_CLIENT_SECRET, "redirect_uri": HF_SPACE_URL, "grant_type": "authorization_code", } resp = requests.post(token_url, data=data) if resp.status_code != 200: return None, resp.text token_data = resp.json() # Example token_data: { "access_token": "...", "expires_in": 3599, "refresh_token": "...", "scope": "...", "token_type": "Bearer" } user_tokens["default"] = { "access_token": token_data.get("access_token"), "refresh_token": token_data.get("refresh_token"), "token_uri": token_url, "client_id": GOOGLE_CLIENT_ID, "client_secret": GOOGLE_CLIENT_SECRET, "scopes": [ "https://www.googleapis.com/auth/calendar.readonly", "https://www.googleapis.com/auth/calendar.events" ] # Note: You could also store expiry/time; google.oauth2.credentials will handle refreshing automatically. } return token_data, None def get_calendar_service(): """ Build a googleapiclient ‘service’ object using the stored tokens. If tokens exist, we can construct google.oauth2.credentials.Credentials, which will auto-refresh if expired. """ token_info = user_tokens.get("default") if not token_info: return None # Not authenticated yet creds = Credentials( token=token_info["access_token"], refresh_token=token_info["refresh_token"], token_uri=token_info["token_uri"], client_id=token_info["client_id"], client_secret=token_info["client_secret"], scopes=token_info["scopes"], ) try: service = build("calendar", "v3", credentials=creds) except Exception as e: return None return service # ────────────────────────────────────────────────────────────────────────────── # 4) IMPORT YOUR EXISTING TOOLS # (Assumes you placed them in a subfolder named ‘tools/’) # ────────────────────────────────────────────────────────────────────────────── from tools.current_datetime_tool import CurrentDateTimeTool from tools.daily_event_summary_tool import DailyEventSummaryTool from tools.weekly_event_summary_tool import WeeklyEventSummaryTool from tools.modify_event_tool import ModifyEventTool # ────────────────────────────────────────────────────────────────────────────── # 5) FUNCTION TO BUILD A FRESH AGENT FOR EACH USER # (Because the ‘service’ object is user-specific once authenticated) # ────────────────────────────────────────────────────────────────────────────── def build_agent_for_user(): """ 1. Build the OpenAI model via AgentPro. 2. Build a user-specific Google Calendar ‘service’ via stored tokens. 3. Instantiate each tool with that service (where required). 4. Return a ReactAgent ready to run queries. """ # 5.1) Create the LLM model model = create_model( provider = "openai", model_name = "gpt-3.5-turbo", api_key = OPENAI_API_KEY, temperature = 0.3 ) # 5.2) Get the user’s Google Calendar service service = get_calendar_service() if service is None: return None # Not authenticated yet # 5.3) Instantiate each tool, passing ‘service’ where needed daily_planner_tool = DailyEventSummaryTool(service=service) weekly_planner_tool = WeeklyEventSummaryTool(service=service) modify_event_tool = ModifyEventTool(service=service) current_dt_tool = CurrentDateTimeTool() ares_tool = AresInternetTool(ARES_API_KEY) tools_list = [ daily_planner_tool, weekly_planner_tool, current_dt_tool, modify_event_tool, ares_tool, ] # 5.4) Create and return the ReactAgent agent = ReactAgent(model=model, tools=tools_list) return agent # ────────────────────────────────────────────────────────────────────────────── # 6) GRADIO CALLBACKS FOR AUTHENTICATION & EXCHANGE # ────────────────────────────────────────────────────────────────────────────── def open_google_oauth(): """ Called when user clicks “Connect Google Calendar.” Returns a message telling them to copy the URL that opens. """ url = build_auth_url() # Attempt to open in a new tab. (Browsers may block this; user can copy manually.) try: import webbrowser webbrowser.open_new_tab(url) except: pass return ( "★ A new tab (or popup) should have opened for you to log in.\n\n" "If it did not, click this link manually:\n\n" f"{url}\n\n" "After granting access, you’ll end up on an error page. Copy the \ entire URL from your browser’s address bar (it contains “?code=…”) \ and paste it below." ) def handle_auth_code(full_redirect_url: str): """ Called when user pastes the entire redirect URL back into our textbox. We parse out ‘code=…’, exchange it, and report success or failure. """ code = extract_auth_code(full_redirect_url) if code is None: return "❌ Could not find ‘code’ in the URL you pasted. Please paste the exact URL you were redirected to." token_data, error = exchange_code_for_tokens(code) if error: return f"❌ Token exchange failed: {error}" return "✅ Successfully connected your Google Calendar! You can now ask about your events." # ────────────────────────────────────────────────────────────────────────────── # 7) GRADIO CHAT FUNCTION (uses user-specific agent) # ────────────────────────────────────────────────────────────────────────────── def chat_with_agent(user_input): """ This is called whenever the user sends a chat query. If not authenticated, prompt them to connect first. Otherwise, build a fresh agent for this user and get a response. """ # 7.1) Do we have a Calendar ‘service’ for this user yet? service = get_calendar_service() if service is None: return "❌ Please connect your Google Calendar first (use the button above)." # 7.2) Build a new Agent with the user’s Calendar service agent = build_agent_for_user() if agent is None: return "❌ Something went wrong building the agent. Ensure you completed Calendar authentication." # 7.3) Run the agent on the user’s input try: response = agent.run(user_input) return response.final_answer except Exception as e: return "❌ Exception in agent.run():\n" + repr(e) # ────────────────────────────────────────────────────────────────────────────── # 8) ASSEMBLE GRADIO INTERFACE # ────────────────────────────────────────────────────────────────────────────── with gr.Blocks() as demo: gr.Markdown("## 🗓️ Calendar Chatbot (Gradio + HF Spaces)") # ———————————————— # A) CONNECT GOOGLE CALENDAR SECTION # ———————————————— with gr.Row(): connect_btn = gr.Button("🔗 Connect Google Calendar") auth_message = gr.Textbox(label="Auth Instructions / Status", interactive=False) connect_btn.click(fn=open_google_oauth, outputs=auth_message) auth_code_input = gr.Textbox( lines=1, placeholder="Paste full Google redirect URL here (contains ‘code=…’)", label="Step 2: Paste full redirect URL" ) auth_status = gr.Textbox(label="Authentication Result", interactive=False) auth_code_input.submit(fn=handle_auth_code, inputs=auth_code_input, outputs=auth_status) gr.Markdown("---") # ———————————————— # B) CHAT SECTION # ———————————————— chat_input = gr.Textbox(lines=2, placeholder="E.g., What’s on my calendar today?") chat_output = gr.Textbox(label="Response", interactive=False) chat_input.submit(fn=chat_with_agent, inputs=chat_input, outputs=chat_output) # You can also add a “Send” button if you like: # send_btn = gr.Button("✉️ Send") # send_btn.click(fn=chat_with_agent, inputs=chat_input, outputs=chat_output) demo.launch()