init

agents/gmail_agent.py

@@ -0,0 +1,107 @@
+from datetime import datetime
+import json
+import base64
+from googleapiclient.discovery import build
+from utils.auth import authenticate_google_services
+
+class Gmail_Agent:
+    def __init__(self, creds=None):
+        if not creds:
+            creds = authenticate_google_services()
+        self.service = build('gmail', 'v1', credentials=creds)
+
+    def read_gmail_messages(self, queries, max_results=5):
+        query = " ".join(queries)
+        results = self.service.users().messages().list(userId='me', maxResults=max_results, q=query).execute()
+        messages = results.get('messages', [])
+        return messages
+
+    def format_messages(self, messages):
+        formatted_content = "\n\n==========\n\n"
+        for message in messages:
+            msg = self.service.users().messages().get(userId='me', id=message['id']).execute()
+            import json
+            with open('results.json', 'w', encoding='utf-8') as f:
+                json.dump(msg, f)
+            internal_Date = int(msg['internalDate'])
+            formatted_Date = datetime.fromtimestamp(internal_Date / 1000).strftime('%Y-%m-%d %H:%M:%S')
+            formatted_content += f"信件日期: {formatted_Date}\n"
+            content = self._parse_gmail_response(msg['payload']['parts'])
+            formatted_content+=f"信件內容: {content}\n\n==========\n\n"
+        return formatted_content
+
+    def send_email(self, to, subject, message_content):
+        try:
+            from email.message import EmailMessage
+            message = EmailMessage()
+            message.set_content(message_content)
+            message["To"] = ", ".join(to)
+            message["Subject"] = subject
+            encoded_message = base64.urlsafe_b64encode(message.as_bytes()).decode()
+            create_message = {"raw": encoded_message}
+            send_message = (
+                self.service.users()
+                .messages()
+                .send(userId="me", body=create_message)
+                .execute()
+            )
+            print(f'Message Id: {send_message["id"]}')
+            formatted_content = "✅ Gmail 已傳送給所有使用者，請至您的 Gmail 寄件備份確認！"
+        except Exception as error:
+            print(f"An error occurred: {error}")
+            send_message = None
+            formatted_content = f"❌ 信件未能成功寄出，以下是錯誤訊息（英文）：{error}"
+        return formatted_content
+
+    def get_all_tools(self):
+        LLM_tools = []
+        LLM_tools.append({
+            "name": "query_gmail_tool",
+            "description": "query the gmails with specified queries",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "queries": {
+                        "type": "array",
+                        "items": {"type": "string"},
+                        "description": "Queries that can be used to filter the gmails. (e.g., ['before:2025/06/01', 'after:2025/05/01', 'from:[email protected]'])",
+                    },
+                },
+                "required": ["queries"]
+            },
+        })
+        LLM_tools.append({
+            "name": "send_email_tool",
+            "description": "Send an email to a specified recipient with a specified subject and message",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "to": {
+                        "type": "array",
+                        "items": {"type": "string"},
+                        "description": "An array of email addresses of all the recipients"
+                        },
+                    "subject": {
+                        "type": "string",
+                        "description": "The subject of the email"
+                    },
+                    "message_content": {
+                        "type": "string",
+                        "description": "The content of the email"
+                    }
+                },
+                "required": ["to", "subject", "message_content"]
+            }
+        })
+        return LLM_tools
+
+    def _parse_gmail_response(self, parts):
+        for part in parts:
+            if "parts" in part:
+                content = self._parse_gmail_response(part["parts"])
+                if content:
+                    return content
+            if part['mimeType'] == 'text/plain' and part['body']['size'] != 0:
+                decoded_data = base64.urlsafe_b64decode(part['body']['data']).decode('utf-8')
+                return decoded_data
+        return None

agents/google_calendar_agent.py

@@ -0,0 +1,287 @@
+from googleapiclient.discovery import build
+from datetime import datetime, timedelta
+from utils.auth import authenticate_google_services
+
+class GoogleCalendar_Agent:
+    def __init__(self, creds=None, timezone="Asia/Taipei"):
+        if not creds:
+            creds = authenticate_google_services()
+        self.service = build("calendar", "v3", credentials=creds)
+        self.timezone = timezone
+
+    def list_events(self, n: int=5, queries=None) -> str:
+        now = datetime.utcnow().isoformat() + 'Z'
+        events_result = self.service.events().list(
+            calendarId='primary',
+            timeMin=now,
+            maxResults=n,
+            singleEvents=True,
+            orderBy='startTime'
+        ).execute()
+        events = events_result.get('items', [])
+        if not events:
+            return "🔕 找不到即將到來的行事曆事件。"
+        output = "📅 您接下來的行事曆事件如下：\n"
+        for event in events:
+            start = event['start'].get('dateTime', event['start'].get('date'))
+            summary = event.get('summary', 'No title')
+            output += f"• {summary} at {start}\n"
+        return output
+
+    def create_event(self,
+            summary: str,
+            start_time: str,
+            end_time: str,
+            location: str = None,
+            description: str = None,
+            attendees: list = None,
+            reminder_minutes: int = None) -> str:
+        event = {
+            'summary': summary,
+            'start': {'dateTime': start_time, 'timeZone': self.timezone},
+            'end': {'dateTime': end_time, 'timeZone': self.timezone}
+        }
+        if location:
+            event['location'] = location
+        if description:
+            event['description'] = description
+        if attendees:
+            event['attendees'] = [{'email': email} for email in attendees]
+        if reminder_minutes:
+            event['reminders'] = {
+                'useDefault': False,
+                'overrides': [{'method': 'popup', 'minutes': reminder_minutes}]
+            }
+        try:
+            created_event = self.service.events().insert(
+                calendarId='primary',
+                sendUpdates='all',
+                body=event
+            ).execute()
+            return f"""✅ 已成功建立事件：\n標題        : {summary}\n開始時間    : {start_time}\n結束時間    : {end_time}\n地點        : {location or '無'}\n說明        : {description or '無'}\n參與者      : {', '.join(attendees) if attendees else '無'}\n提醒時間    : {reminder_minutes if reminder_minutes is not None else '使用預設提醒'}\n時區        : {self.timezone}\n事件連結    : {created_event.get('htmlLink')}"""
+        except Exception as e:
+            return f"❌ 建立事件失敗，以下是錯誤訊息（英文）: {e}"
+
+    def find_event(self, query: str,) -> list:
+        now = datetime.utcnow().isoformat() + 'Z'
+        events_result = self.service.events().list(
+            calendarId='primary',
+            timeMin=now,
+            singleEvents=True,
+            orderBy='startTime',
+            q=query
+        ).execute()
+        return events_result.get('items', [])
+
+    def update_event(self, query: str,
+            new_summary=None,
+            new_location=None,
+            new_description=None,
+            new_start=None,
+            new_end=None,
+            new_attendees=None,
+            reminder_minutes=None):
+        events = self.find_event(query)
+        if not events:
+            return f"❌ 找不到包含「{query}」的事件"
+        if all(arg is None for arg in [
+            new_summary, new_location, new_description,
+            new_start, new_end, new_attendees, reminder_minutes
+        ]):
+            return f"⚠️ 沒有指定任何要更新的欄位，未執行更新。"
+        for event in events:
+            event_id = event['id']
+            try:
+                if new_summary:
+                    event['summary'] = new_summary
+                if new_location:
+                    event['location'] = new_location
+                if new_description:
+                    event['description'] = new_description
+                if new_attendees:
+                    event['attendees'] = [{'email': email} for email in new_attendees]
+                if reminder_minutes:
+                    event['reminders'] = {
+                        'useDefault': False,
+                        'overrides': [{'method': 'popup', 'minutes': reminder_minutes}]
+                    }
+                if new_start or new_end:
+                    event['start']['dateTime'] = new_start
+                    event['end']['dateTime'] = new_end
+                    event['start']['timeZone'] = 'Asia/Taipei'
+                    event['end']['timeZone'] = 'Asia/Taipei'
+                updated_event = self.service.events().update(
+                    calendarId='primary',
+                    eventId=event_id,
+                    sendUpdates='all',
+                    body=event
+                ).execute()
+                return f"✅ 已成功更新事件：{updated_event['summary']}"
+            except Exception as e:
+                return f"❌ 更新事件失敗，以下是錯誤訊息（英文）: {e}"
+
+    def delete_event(self, query: str):
+        now = datetime.utcnow().isoformat() + 'Z'
+        event_results = self.find_event(query)
+        if len(event_results) == 0:
+            return f"❌ 找不到包含「{query}」的事件"
+        for event in event_results:
+            target_id = event["id"]
+            try:
+                self.service.events().delete(
+                    calendarId='primary',
+                    eventId=target_id,
+                    sendUpdates='all'
+                ).execute()
+                return f"🗑️ 已刪除事件：{event['summary']}（ID: {target_id}）"
+            except Exception as e:
+                return f"❌ 刪除事件失敗（ID: {target_id}），以下是錯誤訊息（英文）: {e}"
+
+    def get_event_details(self, query: str):
+        events = self.find_event(query)
+        if not events:
+            return f"❌ 找不到符合 '{query}' 的事件！"
+        event = events[0]
+        summary = event.get('summary', 'No Title')
+        start_time = event['start'].get('dateTime', event['start'].get('date'))
+        end_time = event['end'].get('dateTime', event['end'].get('date'))
+        formatted_string = f"事件: {summary}\n開始時間: {start_time}\n結束時間: {end_time}"
+        attendees = event.get('attendees', [])
+        emails = [attendee.get('email') for attendee in attendees if attendee.get('email')]
+        if emails:
+            formatted_string += f"\n參與者: {', '.join(emails)}"
+        else:
+            formatted_string += "\n無參與者"
+        return formatted_string
+
+    def get_all_tools(self):
+        LLM_tools = []
+        LLM_tools.append({
+            "name": "create_Calendar_event",
+            "description": "Create a new Google Calendar event by specifying summary, start time, and end time, and optionally location, description, attendees, and reminder.",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "summary": {
+                        "type": "string",
+                        "description": "Title or summary of the event (e.g., 'Lunch with Alice')"
+                    },
+                    "start_time": {
+                        "type": "string",
+                        "format": "date-time",
+                        "description": "Start time of the event in ISO format (e.g., '2025-07-01T14:00:00')"
+                    },
+                    "end_time": {
+                        "type": "string",
+                        "format": "date-time",
+                        "description": "End time of the event in ISO format (e.g., '2025-07-01T15:00:00')"
+                    },
+                    "location": {
+                        "type": "string",
+                        "description": "Location of the event (e.g., 'Taipei 101')"
+                    },
+                    "description": {
+                        "type": "string",
+                        "description": "Detailed description of the event"
+                    },
+                    "attendees": {
+                        "type": "array",
+                        "items": {"type": "string"},
+                        "description": "List of email addresses to invite to the event"
+                    },
+                    "reminder_minutes": {
+                        "type": "integer",
+                        "description": "Number of minutes before the event to trigger a popup reminder"
+                    }
+                },
+                "required": ["summary", "start_time", "end_time"]
+            },
+        })
+        LLM_tools.append({
+            "name": "list_Calendar_events",
+            "description": "List your upcoming Google Calendar events.",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "n": {
+                        "type": "integer",
+                        "default": 5,
+                        "description": "Number of upcoming events to list (default: 5)"
+                    }
+                }
+            }
+        })
+        LLM_tools.append({
+            "name": "update_Calendar_event",
+            "description": "Update a Google Calendar event using a keyword query to find it. You can update summary, location, description, start/end time, attendees, or reminders.",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "query": {
+                        "type": "string",
+                        "description": "Keyword to find the event to update (e.g., 'meeting with Bob')"
+                    },
+                    "new_summary": {
+                        "type": "string",
+                        "description": "New summary/title for the event"
+                    },
+                    "new_location": {
+                        "type": "string",
+                        "description": "New location for the event"
+                    },
+                    "new_description": {
+                        "type": "string",
+                        "description": "New description for the event"
+                    },
+                    "new_start": {
+                        "type": "string",
+                        "format": "date-time",
+                        "description": "New start time in ISO format"
+                    },
+                    "new_end": {
+                        "type": "string",
+                        "format": "date-time",
+                        "description": "New end time in ISO format"
+                    },
+                    "new_attendees": {
+                        "type": "array",
+                        "items": {"type": "string"},
+                        "description": "New list of attendees (email addresses)"
+                    },
+                    "reminder_minutes": {
+                        "type": "integer",
+                        "description": "Updated popup reminder time in minutes"
+                    }
+                },
+                "required": ["query"]
+            }
+        })
+        LLM_tools.append({
+            "name": "delete_Calendar_event",
+            "description": "Delete a Google Calendar event using a keyword query to find it.",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "query": {
+                        "type": "string",
+                        "description": "Keyword to find the event to delete (e.g., 'meeting with Bob')"
+                    }
+                },
+                "required": ["query"]
+            }
+        })
+        LLM_tools.append({
+            "name": "get_event_details",
+            "description": "Find the attendee email addresses and the summary of a specific event, based on a keyword.",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "query": {
+                        "type": "string",
+                        "description": "Keyword to find the event and extract attendees and summary of that event."
+                    }
+                },
+                "required": ["query"]
+            }
+        })
+        return LLM_tools

app.py

@@ -0,0 +1,77 @@
+import os
+import gradio as gr
+from utils.auth import authenticate_google_services
+from utils.llm import Google_Gemini_LLM
+from agents.gmail_agent import Gmail_Agent
+from agents.google_calendar_agent import GoogleCalendar_Agent
+
+# Authenticate and initialize agents
+creds = authenticate_google_services()
+GMAIL_AGENT = Gmail_Agent(creds)
+CALENDAR_Agent = GoogleCalendar_Agent(creds)
+
+# Initialize LLM
+# Use os.environ.get instead of userdata.get
+GOOGLE_API_KEY = os.environ.get('GOOGLE_API_KEY')
+if not GOOGLE_API_KEY:
+    raise ValueError("GOOGLE_API_KEY environment variable not set")
+
+MODEL = "gemini-1.5-flash"
+YOUR_AI_AGENT = Google_Gemini_LLM(GOOGLE_API_KEY, MODEL)
+YOUR_AI_AGENT.append_function_tools(GMAIL_AGENT.get_all_tools())
+YOUR_AI_AGENT.append_function_tools(CALENDAR_Agent.get_all_tools())
+
+def generate_response(user_prompt, history):
+    response = YOUR_AI_AGENT.generate_content(user_prompt)
+    while(response.function_call):
+        function_name = response.function_call.name
+        function_args = response.function_call.args
+        if function_name == "query_gmail_tool":
+            messages = GMAIL_AGENT.read_gmail_messages(**function_args)
+            if not messages:
+                formatted_content = "❌ 找不到對應的 Gmail，請確認您的搜尋條件是否合理，或者請重新執行！"
+                return formatted_content
+            else:
+                formatted_content = GMAIL_AGENT.format_messages(messages)
+            user_prompt += f"\n根據使用者的搜尋條件，以下是對應的信件內容（可能有多封信件）：<信件內容開始>{formatted_content}<信件內容結束>. 請仔細閱讀以上的信件內容，並將資料整理統整給使用者！請以簡潔且新處的方式傳達內容！"
+        elif function_name == "send_email_tool":
+            formatted_content = GMAIL_AGENT.send_email(**function_args)
+            return formatted_content
+        elif function_name == "list_Calendar_events":
+            formatted_content = CALENDAR_Agent.list_events(**function_args)
+            user_prompt += f"\n根據使用者的搜尋條件，以下是從使用者的日曆中找到對應的事件內容（可能有多封信件）: <事件內容開始>{formatted_content}<事件內容結束>. 請仔細閱讀以上的事件內容，並將資料統整給使用者！請以簡潔且新處的方式傳達內容！"
+        elif function_name == "create_Calendar_event":
+            formatted_content = CALENDAR_Agent.create_event(**function_args)
+            return formatted_content
+        elif function_name == "update_Calendar_event":
+            formatted_content = CALENDAR_Agent.update_event(**function_args)
+            return formatted_content
+        elif function_name == "delete_Calendar_event":
+            formatted_content = CALENDAR_Agent.delete_event(**function_args)
+            return formatted_content
+        elif function_name == "get_event_details":
+            formatted_content = CALENDAR_Agent.get_event_details(**function_args)
+            user_prompt += f"\n根據使用者的搜尋條件，以下是從使用者的日曆中找到對應的事件內容 <事件內容開始>{formatted_content}<事件內容結束>. 請仔細閱讀以上的事件內容，並且完成使用者的指示（例如：統整事件內容、寄送提醒信件給參與事件的人...等等）!"
+        else:
+            raise NotImplementedError(f"Function Call {function_name} is not supported yet.")
+        response = YOUR_AI_AGENT.generate_content(user_prompt)
+    return response.text
+
+# It's better to get the email from the user's input
+# than to hardcode it.
+demo = gr.ChatInterface(
+    fn=generate_response,
+    chatbot=gr.Chatbot(
+        label="📬 AI 助理",
+    ),
+    title="Gmail & Calendar 智慧助理",
+    description="輸入自然語言，讓 AI 幫你管理 Gmail 或 Google 行事曆",
+    examples=[
+        "找出2025/6月標註星號的信件",
+        "請幫我列出近期的事件",
+        "建立2025/8/10早上10點的網球活動",
+    ]
+)
+
+if __name__ == "__main__":
+    demo.launch()

requirements.txt

@@ -0,0 +1,4 @@
+google-api-python-client
+google-auth-httplib2
+google-auth-oauthlib
+gradio

utils/auth.py

@@ -0,0 +1,25 @@
+
+import os, pickle
+from google_auth_oauthlib.flow import InstalledAppFlow
+from google.auth.transport.requests import Request
+
+SCOPES = [
+    'https://www.googleapis.com/auth/calendar',
+    'https://www.googleapis.com/auth/gmail.readonly',
+    'https://www.googleapis.com/auth/gmail.send'
+]
+
+def authenticate_google_services():
+    creds = None
+    if os.path.exists('token.pkl'):
+        with open('token.pkl', 'rb') as token:
+            creds = pickle.load(token)
+    if not creds or not creds.valid:
+        if creds and creds.expired and creds.refresh_token:
+            creds.refresh(Request())
+        else:
+            flow = InstalledAppFlow.from_client_secrets_file('client_secret.json', SCOPES)
+            creds = flow.run_console()
+        with open('token.pkl', 'wb') as token:
+            pickle.dump(creds, token)
+    return creds

utils/llm.py

@@ -0,0 +1,22 @@
+
+import google.generativeai as genai
+from google.generativeai import types
+
+class Google_Gemini_LLM:
+    def __init__(self, api_key, model="gemini-1.5-flash"):
+        self.api_key = api_key
+        self.model = model
+        self.client = genai.Client(api_key=api_key)
+        self.LLM_tools = []
+
+    def generate_content(self, contents):
+        tools = types.Tool(function_declarations=self.LLM_tools)
+        config = types.GenerateContentConfig(tools=[tools])
+        response = self.client.models.generate_content(
+            model=self.model, contents=contents, config=config
+        )
+        return response.candidates[0].content.parts[0]
+
+    def append_function_tools(self, tools):
+        for tool in tools:
+            self.LLM_tools.append(tool)