Asta integrates with Google Workspace (Gmail, Calendar, Drive, and Contacts) using the powerful gog CLI tool. This integration allows you to read emails, manage calendar events, search Drive files, and access contacts using natural language.
Prerequisites
The Google Workspace integration requires the gog CLI tool to be installed and authenticated.
Install gog
Install the gog CLI using Homebrew:gog is a Go-based CLI for Google Workspace. Visit gogcli.sh for installation on other platforms. Get OAuth credentials
Create a Google Cloud project and download OAuth credentials:
- Go to Google Cloud Console
- Create a new project or select an existing one
- Enable Gmail, Calendar, Drive, and Contacts APIs
- Go to Credentials > Create Credentials > OAuth 2.0 Client ID
- Download the
client_secret.json file
Configure gog credentials
Set up gog with your OAuth credentials:gog auth credentials /path/to/client_secret.json
Authenticate your account
Add your Google account with required services:This will open a browser for OAuth consent. Grant the requested permissions. Set default account (optional)
Configure your default Google account:
Gmail Integration
Search, read, and send emails using natural language.
Searching Emails
check my email
read my inbox
show unread emails
search gmail for receipts
find emails from john
Asta uses Gmail’s powerful search syntax:
# Default: emails from last 7 days
gog gmail search "newer_than:7d" --max 10 --json
# Custom queries
gog gmail search "from:[email protected]" --max 10 --json
gog gmail search "subject:invoice is:unread" --max 10 --json
Sending Emails
The skill extracts fields from your message:
And executes:
from app.skills.gog import run_gog
# Search recent emails
result = await run_gog([
"gmail", "search",
"newer_than:7d",
"--max", "10",
"--json",
"--account", "[email protected]"
])
# Parse JSON response
import json
emails = json.loads(result["stdout"])
Gmail Search Queries
Supported Gmail search syntax:
from:[email protected] - Emails from specific sender
to:[email protected] - Emails to specific recipient
subject:keyword - Search in subject line
is:unread - Unread emails only
is:starred - Starred emails
newer_than:7d - Emails from last 7 days
older_than:1m - Emails older than 1 month
has:attachment - Emails with attachments
filename:pdf - Specific attachment type
Calendar Integration
Manage calendar events and check your schedule.
Viewing Events
what's on my calendar today?
show my calendar for tomorrow
what do I have scheduled this week?
list my meetings
Asta automatically determines the date range:
if "today" in text:
start = today
end = tomorrow
elif "tomorrow" in text:
start = tomorrow
end = day_after_tomorrow
else:
start = today
end = today + 7 days # Default: next week
Executes:
gog calendar events primary \
--from 2026-03-06 \
--to 2026-03-13 \
--json \
--account [email protected]
Creating Events
create event
title: Team standup
start: 2026-03-07T09:00
end: 2026-03-07T09:30
location: Zoom
The skill parses structured fields:
title = "Team standup"
start = "2026-03-07T09:00"
end = "2026-03-07T09:30"
location = "Zoom"
And creates the event:
gog calendar create primary \
--summary "Team standup" \
--from "2026-03-07T09:00" \
--to "2026-03-07T09:30" \
--location "Zoom" \
--account [email protected]
from datetime import datetime, timedelta
# Get events for next 7 days
today = datetime.now().strftime("%Y-%m-%d")
next_week = (datetime.now() + timedelta(days=7)).strftime("%Y-%m-%d")
result = await run_gog([
"calendar", "events", "primary",
"--from", today,
"--to", next_week,
"--json",
"--account", "[email protected]"
])
events = json.loads(result["stdout"])
Supported formats for event times:
- ISO 8601:
2026-03-07T14:00
- Date only:
2026-03-07 (all-day event)
- With timezone:
2026-03-07T14:00-05:00
Drive Integration
Search for files in Google Drive.
Searching Files
search my drive for presentation
find files in google drive
query: project report
limit: 20
Executes:
gog drive search "project report" --max 20 --json
result = await run_gog([
"drive", "search",
"presentation", # Query
"--max", "10",
"--json"
])
files = json.loads(result["stdout"])
for file in files:
print(f"Name: {file['name']}")
print(f"Type: {file['mimeType']}")
print(f"URL: {file['webViewLink']}")
Drive Search Queries
Google Drive search syntax:
- Simple text:
project report
- File name:
name:'Budget 2024'
- File type:
mimeType='application/pdf'
- Modified date:
modifiedTime > '2024-01-01'
- Owned by me:
'me' in owners
- Shared with me:
sharedWithMe
List and search your Google Contacts.
show my contacts
list google contacts
limit: 50
Executes:
gog contacts list --max 20 --json
result = await run_gog([
"contacts", "list",
"--max", "20",
"--json"
])
contacts = json.loads(result["stdout"])
for contact in contacts:
print(f"Name: {contact['name']}")
print(f"Email: {contact['email']}")
print(f"Phone: {contact['phone']}")
Natural Language Processing
The Google Workspace skill uses keyword detection to route requests:
def check_eligibility(self, text: str, user_id: str) -> bool:
t = text.lower()
return any(keyword in t for keyword in [
# Gmail
"gmail", "email", "inbox", "mail", "unread",
"send email", "check email", "my emails",
# Calendar
"calendar", "event", "meeting", "schedule",
"appointment", "what do i have", "add to calendar",
# Drive
"google drive", "drive file", "my drive",
# Contacts
"google contacts", "my contacts",
# Generic
"google doc", "google sheet", "gog "
])
The skill extracts structured data from natural language:
def _extract_field(self, text: str, name: str) -> str | None:
"""Extract 'name: value' from text."""
import re
m = re.search(
rf'{re.escape(name)}[:\s]+["\']?([^"\'
]+)["\']?',
text,
re.IGNORECASE
)
return m.group(1).strip() if m else None
def _extract_number(self, text: str, word: str) -> int | None:
"""Extract 'word: 123' from text."""
import re
m = re.search(
rf'(?:{re.escape(word)})[:\s]*(\d+)',
text,
re.IGNORECASE
)
return int(m.group(1)) if m else None
Usage:
to = self._extract_field(text, "to")
subject = self._extract_field(text, "subject")
limit = self._extract_number(text, "limit") or 10
Error Handling
The skill provides helpful error messages:
gog Not Installed
{
"gog_error": "Google Workspace CLI (`gog`) not installed.",
"gog_fix": "Install with: brew install gogcli",
"gog_needs_cli": true
}
Not Authenticated
{
"gog_error": "Google Workspace not authenticated.",
"gog_fix": "1. brew install gogcli\n2. gog auth credentials /path/to/client_secret.json\n3. gog auth add [email protected] --services gmail,calendar,drive,contacts",
"gog_needs_auth": true
}
Command Failed
{
"gog_error": "stderr output from gog",
"gog_command": "gog gmail search ..."
}
Implementation Details
The Google Workspace skill is implemented in backend/app/skills/gog.py.
Binary Detection
def _gog_bin() -> str | None:
"""Find gog binary in common locations."""
for path in ["/opt/homebrew/bin/gog", "/usr/local/bin/gog"]:
if os.path.isfile(path) and os.access(path, os.X_OK):
return path
# Also check PATH
import shutil
return shutil.which("gog")
Safe Command Execution
async def run_gog(args: list[str], timeout: float = 15.0) -> dict:
"""Run gog command safely (no shell injection)."""
gog = _gog_bin()
if not gog:
return {"error": "gog not found", "returncode": 127}
proc = await asyncio.create_subprocess_exec(
gog, *args, # Safe: no shell=True
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
env=os.environ.copy()
)
stdout, stderr = await asyncio.wait_for(
proc.communicate(),
timeout=timeout
)
return {
"stdout": stdout.decode(errors="replace").strip(),
"stderr": stderr.decode(errors="replace").strip(),
"returncode": proc.returncode
}
The skill uses asyncio.create_subprocess_exec (not shell=True) to prevent shell injection vulnerabilities.
Configuration
Environment Variables
gog Binary Paths
Searched in order:
/opt/homebrew/bin/gog (Apple Silicon Mac)
/usr/local/bin/gog (Intel Mac / Linux)
- System PATH
Best Practices
Use structured input for commands
When creating events or sending emails, use clear field separators:send email
to: [email protected]
subject: Meeting Notes
body: Here are the notes from today's meeting...
Specify date ranges for calendar
Be explicit about time periods:
- ✅
show my calendar for tomorrow
- ✅
what's on my calendar today
- ❌
show calendar (defaults to 7 days)
Use Gmail search syntax
Leverage Gmail’s powerful search:
from:[email protected] is:unread
subject:invoice newer_than:7d
has:attachment filename:pdf
Set default account
Configure GOG_ACCOUNT to avoid specifying account in every command.
Troubleshooting
gog command not found
Solution:
brew install gogcli
# Verify installation
which gog
gog version
Authentication errors
Solution:
# Re-authenticate
gog auth add [email protected] --services gmail,calendar,drive,contacts
# List authenticated accounts
gog auth list
Permission denied errors
Solution:
- Check that you granted all requested scopes during OAuth
- Re-run
gog auth add to update permissions
- Ensure APIs are enabled in Google Cloud Console
Timeout errors
Solution:
- Increase timeout in
run_gog() call
- Check network connectivity
- Verify gog service status
JSON parse errors
Solution:
- Ensure
--json flag is included in command
- Check for non-JSON output in stderr
- Verify gog version supports JSON output
API Reference
Key functions in backend/app/skills/gog.py:
# Run gog command safely
await run_gog(args: list[str], timeout: float = 15.0) -> dict
# Execute Google Workspace skill
await GoogleWorkspaceSkill.execute(
user_id: str,
text: str,
extra: dict[str, Any]
) -> dict[str, Any]
# Check if message is Google-related
GoogleWorkspaceSkill.check_eligibility(
text: str,
user_id: str
) -> bool
Return format:
{
"gog_output": "...", # Raw stdout
"gog_data": [...], # Parsed JSON (if applicable)
"gog_command": "...", # Command that was run
"gog_summary": "...", # Human-readable summary
"gog_error": "...", # Error message (if failed)
}