Introduction
Now that we have intelligence as an API call, I’ve been building my own personal AI agent and trying to get it to do things I don’t really enjoy doing, like reading email.
In this blog I’ll walk you through how to build a tool to read email and attach it to the AI agent we built in the previous post.
Step 1: Email Insights Extractor
This is the core AI part that takes an unread email as input and extracts key insights. This is where we use AI to figure out if an email requires action, how important it is, how urgent it is, etc. This not only saves time but also keeps you focused on what’s truly important.
handle_message_schema = [
{
"name": "handle_message",
"description": "You are an AI named TARS created by GP (Gurpartap Sandhu) to help handle incoming messages",
"parameters": {
"type": "object",
"properties": {
"summary": {
"description": "Summary of the message",
},
"action_needed": {
"type": "boolean",
"description": "Does this message warrant any action from me",
},
"urgency": {
"type": "string",
"description": "What is the level of urgency of taking action based on this message.",
"enum": ["high", "medium", "low", "unknown"]
},
"importance": {
"type": "string",
"description": "What is the level of importance for this message.",
"enum": ["high", "medium", "low", "unknown"]
},
"action": {
"type": "string",
"description": "What action should be performed, if any",
},
"action_instructions": {
"type": "string",
"description": "What are the step by step instructions to perform the recommended action, if any",
},
},
"required": ["summary", "action_needed", "urgency", "importance", "action", "action_instructions"],
}
}
]
# AI Gmail Handler function
def ai_gmail_handler(classifier_input):
"""
Processes the classifier input to determine actions and details about an email using a ChatOpenAI model.
Args:
classifier_input (str): The input string containing details about the email to be classified.
Returns:
dict: A dictionary containing the processed information of the email.
"""
try:
prompt = ChatPromptTemplate.from_messages([("human", "{input}")])
model = ChatOpenAI(model="gpt-4-1106-preview", temperature=0).bind(functions=handle_message_schema)
runnable = prompt | model
message_action = runnable.invoke({"input": classifier_input})
content = message_action.content
function_call = message_action.additional_kwargs.get('function_call')
if function_call is not None:
content = function_call.get('arguments', content)
# Check if content is a string and can be parsed as JSON
if isinstance(content, str):
try:
return json.loads(content)
except json.JSONDecodeError:
# If content is a string but not JSON, wrap it in a JSON-like structure
return {
"error": "Content is a string but not JSON",
"raw_content": content
}
else:
# If not a string, log and return a default dictionary
logging.error("Content is not in the expected format. Returning default dictionary.")
return {"error": "Content not in expected format", "raw_content": str(content)}
except Exception as e:
logging.error(f"Error in ai_gmail_handler: {e}")
return {"error": "Exception occurred in ai_gmail_handler", "exception_message": str(e)}Step 2: OAuth flow to read unread emails
The next part is to get credentials that can access unread Gmail programatically. The way Google allows this is creating an Oauth app.
Google Cloud Console Setup:
- First, head over to the Google Cloud Console.
- Create a new project or select an existing one.
- Go to the “APIs & Services > Dashboard” panel, and enable the Gmail API for your project.
Create Credentials:
- In the same console, go to “Credentials” and create OAuth client ID credentials.
- You’ll need to set the application type to “Desktop app” for a simple script.
- Once created, download the JSON file with your credentials. This file contains the client ID and client secret.
- This need to go into .env file so that all the creds can be in one single file
def fetch_unread_emails(service):
"""Fetch all unread emails from the user's inbox with their ID and received date.
Args:
service: Authorized Gmail API service instance.
Returns:
List of dictionaries containing the email ID, and received date of unread email messages.
"""
try:
# List all unread messages
response = service.users().messages().list(userId='me', labelIds=['UNREAD']).execute()
# Debugging: Check the type of response and print it
print(f"Response type: {type(response)}")
print(f"Response content: {response}")
# Ensure response is a dictionary before calling get
if not isinstance(response, dict):
print("Response is not a dictionary.")
return []
messages = response.get('messages', [])
all_messages = []
for msg in messages:
# Fetch only the headers of the message
msg_detail = service.users().messages().get(userId='me', id=msg['id'], format='metadata',
metadataHeaders=['Date']).execute()
# Extract the headers
headers = msg_detail.get('payload', {}).get('headers', [])
date_header = next((header['value'] for header in headers if header['name'] == 'Date'), None)
# Append a dictionary with the email ID and received date to the list
all_messages.append({
'id': msg['id'],
'received_date': date_header
})
return all_messages
except Exception as e:
print(f'An error occurred: {e}')
return []
Helper functions: https://github.com/gpsandhu23/TARS/tree/main/TARS/email_module
Step 3: Putting this together
Once we have these two steps done, we can programmatically pull unread email and pass them through the AI tool.
# Tool to handle all unread Gmails
@tool
def handle_all_unread_gmail() -> list:
"""
Processes all unread emails in the Gmail inbox and returns a list containing details of each email.
Returns:
list: A list of dictionaries, each containing details of an unread email such as sender, subject, summary,
action needed, importance, urgency, action, and action instructions.
"""
try:
service = authenticate_gmail_api()
unread_messages = fetch_unread_emails(service)
all_message_details = []
for message in unread_messages:
msg_id = message['id']
received_date = message['received_date']
mime_msg = get_mime_message(service, 'me', msg_id)
headers = mime_msg.items()
sender = next(value for name, value in headers if name == 'From')
subject = next(value for name, value in headers if name == 'Subject')
content = get_email_content(mime_msg)
classifier_input = "Sender: " + str(sender) + " Subject: " + str(subject) + " Email content: " + str(content)
ai_response = ai_gmail_handler(classifier_input)
print(type(ai_response))
email_details = {
'sender': sender,
'subject': subject,
'received_date': received_date,
'id': msg_id,
'summary': ai_response.get('summary', 'N/A'),
'action_needed': ai_response.get('action_needed', 'N/A'),
'importance': ai_response.get('importance', 'N/A'),
'urgency': ai_response.get('urgency', 'N/A'),
'action': ai_response.get('action', 'N/A'),
'action_instructions': ai_response.get('action_instructions', 'N/A')
}
logging.info(email_details)
all_message_details.append(email_details)
mark_email_as_read(service, 'me', msg_id)
return all_message_details
except Exception as e:
logging.error(f"Error in handle_all_unread_gmail: {e}")
return []Step 4: Connecting this as a tool to AI Agent
Finally, we integrate our unread email handler with the AI agent we built in the previous post. It’s a simple yet powerful connection, giving your AI agent the ability to manage your emails effectively.
from .custom_tools import handle_all_unread_gmail
# Define the LLM to use
llm = ChatOpenAI(model="gpt-4-1106-preview", temperature=0)
# Combine all tools
tools = [handle_all_unread_gmail]
llm_with_tools = llm.bind(functions=[format_tool_to_openai_function(t) for t in tools])
Conclusion
The entire code is here and you can clone it and run docker-compose up. Then you just ask your AI to check your email and let you know if anything important needs your attention.
