admin/OpenJarvis
1
0
Fork 0
mirror of https://github.com/open-jarvis/OpenJarvis.git synced 2026-07-28 14:07:55 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: add Gmail connector with OAuth and mocked API sync

Browse Source 
Implements GmailConnector registered under 'gmail' in ConnectorRegistry,
a shared oauth.py helper (build_google_auth_url, load/save/delete_tokens)
reusable by Drive/Calendar/Contacts, and 7 fully mocked pytest tests
covering auth state, sync document extraction, disconnect, mcp_tools,
and registry lookup.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
krypticmouse
2026-03-25 23:56:29 +00:00
co-authored by Claude Opus 4.6
parent 5e0e8965d7
commit 56554c481a
3 changed files with 707 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/openjarvis/connectors/gmail.py
+393
View File
@@ -0,0 +1,393 @@
"""Gmail connector — bulk email sync via the Gmail REST API.
Uses OAuth 2.0 tokens stored locally (see :mod:`openjarvis.connectors.oauth`).
All network calls are isolated in module-level functions (``_gmail_api_*``)
to make them trivially mockable in tests.
"""
from __future__ import annotations
import base64
import email.utils
from datetime import datetime
from typing import Any, Dict, Iterator, List, Optional
import httpx
from openjarvis.connectors._stubs import BaseConnector, Document, SyncStatus
from openjarvis.connectors.oauth import (
build_google_auth_url,
delete_tokens,
load_tokens,
save_tokens,
)
from openjarvis.core.config import DEFAULT_CONFIG_DIR
from openjarvis.core.registry import ConnectorRegistry
from openjarvis.tools._stubs import ToolSpec
# ---------------------------------------------------------------------------
# Constants
# ---------------------------------------------------------------------------
_GMAIL_API_BASE = "https://gmail.googleapis.com/gmail/v1/users/me"
_GMAIL_SCOPE = "https://www.googleapis.com/auth/gmail.readonly"
_DEFAULT_CREDENTIALS_PATH = str(
DEFAULT_CONFIG_DIR / "connectors" / "gmail.json"
)
# ---------------------------------------------------------------------------
# Module-level API functions (easy to patch in tests)
# ---------------------------------------------------------------------------
def _gmail_api_list_messages(
token: str,
*,
page_token: Optional[str] = None,
query: str = "",
) -> Dict[str, Any]:
"""Call the Gmail ``messages.list`` endpoint.
Parameters
----------
token:
OAuth access token.
page_token:
Pagination token from a previous response's ``nextPageToken``.
query:
Gmail search query string (e.g. ``"is:unread"``).
Returns
-------
dict
Raw API response containing ``messages`` list and optional
``nextPageToken``.
"""
params: Dict[str, str] = {}
if page_token:
params["pageToken"] = page_token
if query:
params["q"] = query
resp = httpx.get(
f"{_GMAIL_API_BASE}/messages",
headers={"Authorization": f"Bearer {token}"},
params=params,
timeout=30.0,
)
resp.raise_for_status()
return resp.json()
def _gmail_api_get_message(token: str, msg_id: str) -> Dict[str, Any]:
"""Fetch a single Gmail message by ID (``full`` format).
Parameters
----------
token:
OAuth access token.
msg_id:
Gmail message ID string.
Returns
-------
dict
Raw API response for the message resource.
"""
resp = httpx.get(
f"{_GMAIL_API_BASE}/messages/{msg_id}",
headers={"Authorization": f"Bearer {token}"},
params={"format": "full"},
timeout=30.0,
)
resp.raise_for_status()
return resp.json()
# ---------------------------------------------------------------------------
# Helper functions
# ---------------------------------------------------------------------------
def _extract_header(headers: List[Dict[str, str]], name: str) -> str:
"""Return the value of the first header matching *name* (case-insensitive)."""
name_lower = name.lower()
for h in headers:
if h.get("name", "").lower() == name_lower:
return h.get("value", "")
return ""
def _decode_body(payload: Dict[str, Any]) -> str:
"""Decode the message body from a Gmail payload dict.
Handles both simple payloads (``body.data``) and multipart messages
by recursively searching for a ``text/plain`` part.
"""
mime_type: str = payload.get("mimeType", "")
if mime_type.startswith("multipart/"):
# Search parts for text/plain first, then any text/* fallback
parts: List[Dict[str, Any]] = payload.get("parts", [])
for part in parts:
if part.get("mimeType", "").startswith("text/plain"):
return _decode_body(part)
# Fallback: recurse into first part
if parts:
return _decode_body(parts[0])
return ""
body_data: str = payload.get("body", {}).get("data", "")
if not body_data:
return ""
# Gmail uses URL-safe base64 without padding
padded = body_data + "=" * (-len(body_data) % 4)
try:
return base64.urlsafe_b64decode(padded).decode("utf-8", errors="replace")
except Exception: # noqa: BLE001
return ""
def _parse_date(date_str: str) -> datetime:
"""Parse an RFC 2822 email date string into a :class:`~datetime.datetime`.
Falls back to :func:`datetime.now` if the string is unparseable.
"""
if not date_str:
return datetime.now()
try:
return email.utils.parsedate_to_datetime(date_str)
except Exception: # noqa: BLE001
return datetime.now()
# ---------------------------------------------------------------------------
# GmailConnector
# ---------------------------------------------------------------------------
@ConnectorRegistry.register("gmail")
class GmailConnector(BaseConnector):
"""Connector that syncs emails from Gmail via the REST API.
Authentication is handled through Google OAuth 2.0. Tokens are stored
locally in a JSON credentials file.
Parameters
----------
credentials_path:
Path to the JSON file where OAuth tokens are stored. Defaults to
``~/.openjarvis/connectors/gmail.json``.
"""
connector_id = "gmail"
display_name = "Gmail"
auth_type = "oauth"
def __init__(self, credentials_path: str = "") -> None:
self._credentials_path = credentials_path or _DEFAULT_CREDENTIALS_PATH
self._items_synced: int = 0
self._items_total: int = 0
self._last_sync: Optional[datetime] = None
self._last_cursor: Optional[str] = None
# ------------------------------------------------------------------
# BaseConnector interface
# ------------------------------------------------------------------
def is_connected(self) -> bool:
"""Return ``True`` if a credentials file with a valid token exists."""
tokens = load_tokens(self._credentials_path)
if tokens is None:
return False
# Accept any non-empty dict that contains at least one key
# (simplified: real impl would also check expiry / refresh token)
return bool(tokens)
def disconnect(self) -> None:
"""Delete the stored credentials file."""
delete_tokens(self._credentials_path)
def auth_url(self) -> str:
"""Return a Google OAuth consent URL requesting ``gmail.readonly`` scope."""
return build_google_auth_url(
client_id="", # placeholder — real client_id from config
scopes=[_GMAIL_SCOPE],
)
def handle_callback(self, code: str) -> None:
"""Handle the OAuth callback by persisting the authorization code.
In a full implementation this would exchange the code for tokens.
For now the code is saved directly as the token value.
"""
save_tokens(self._credentials_path, {"token": code})
def sync(
self,
*,
since: Optional[datetime] = None, # noqa: ARG002 — reserved for future use
cursor: Optional[str] = None,
) -> Iterator[Document]:
"""Yield :class:`Document` objects for Gmail messages.
Paginates through the messages.list API and fetches each message's
full payload to extract headers and body.
Parameters
----------
since:
Not yet used (Gmail API filtering is done server-side via query).
cursor:
``nextPageToken`` from a previous sync to resume pagination.
"""
tokens = load_tokens(self._credentials_path)
if not tokens:
return
token: str = tokens.get("token", tokens.get("access_token", ""))
if not token:
return
page_token: Optional[str] = cursor
synced = 0
while True:
list_resp = _gmail_api_list_messages(token, page_token=page_token)
messages: List[Dict[str, Any]] = list_resp.get("messages", [])
for msg_stub in messages:
msg_id: str = msg_stub.get("id", "")
if not msg_id:
continue
msg = _gmail_api_get_message(token, msg_id)
payload: Dict[str, Any] = msg.get("payload", {})
headers: List[Dict[str, str]] = payload.get("headers", [])
from_header = _extract_header(headers, "From")
subject = _extract_header(headers, "Subject")
date_str = _extract_header(headers, "Date")
to_header = _extract_header(headers, "To")
body = _decode_body(payload)
timestamp = _parse_date(date_str)
participants: List[str] = []
if from_header:
participants.append(from_header)
if to_header:
participants.append(to_header)
thread_id: Optional[str] = msg.get("threadId")
doc = Document(
doc_id=f"gmail:{msg_id}",
source="gmail",
doc_type="email",
content=body,
title=subject,
author=from_header,
participants=participants,
timestamp=timestamp,
thread_id=thread_id,
metadata={
"message_id": msg_id,
"labels": msg.get("labelIds", []),
},
)
synced += 1
yield doc
next_page: Optional[str] = list_resp.get("nextPageToken")
if not next_page:
self._last_cursor = None
break
page_token = next_page
self._last_cursor = next_page
self._items_synced = synced
self._last_sync = datetime.now()
def sync_status(self) -> SyncStatus:
"""Return sync progress from the most recent :meth:`sync` call."""
return SyncStatus(
state="idle",
items_synced=self._items_synced,
last_sync=self._last_sync,
cursor=self._last_cursor,
)
# ------------------------------------------------------------------
# MCP tools
# ------------------------------------------------------------------
def mcp_tools(self) -> List[ToolSpec]:
"""Expose three MCP tool specs for real-time Gmail queries."""
return [
ToolSpec(
name="gmail_search_emails",
description=(
"Search Gmail messages using a query string. "
"Supports the same syntax as the Gmail search box "
"(e.g. 'from:alice subject:report is:unread')."
),
parameters={
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "Gmail search query",
},
"max_results": {
"type": "integer",
"description": "Maximum number of emails to return",
"default": 20,
},
},
"required": ["query"],
},
category="communication",
),
ToolSpec(
name="gmail_get_thread",
description=(
"Retrieve all messages in a Gmail thread by thread ID."
),
parameters={
"type": "object",
"properties": {
"thread_id": {
"type": "string",
"description": "Gmail thread ID",
},
},
"required": ["thread_id"],
},
category="communication",
),
ToolSpec(
name="gmail_list_unread",
description=(
"List unread Gmail messages, optionally filtered by label."
),
parameters={
"type": "object",
"properties": {
"label": {
"type": "string",
"description": "Gmail label to filter by (e.g. 'INBOX')",
"default": "INBOX",
},
"max_results": {
"type": "integer",
"description": "Maximum number of messages to return",
"default": 20,
},
},
"required": [],
},
category="communication",
),
]
src/openjarvis/connectors/oauth.py
+97
View File
@@ -0,0 +1,97 @@
"""Shared OAuth 2.0 helpers for Google connectors.
Provides URL builder, token persistence, and token cleanup utilities
that are reused by gmail, drive, calendar, and contacts connectors.
"""
from __future__ import annotations
import json
import os
from pathlib import Path
from typing import Any, Dict, List, Optional
from urllib.parse import urlencode
# ---------------------------------------------------------------------------
# Google OAuth endpoints / defaults
# ---------------------------------------------------------------------------
_GOOGLE_AUTH_ENDPOINT = "https://accounts.google.com/o/oauth2/v2/auth"
_DEFAULT_REDIRECT_URI = "http://localhost:8789/callback"
_DEFAULT_SCOPES: List[str] = ["openid", "email", "profile"]
# ---------------------------------------------------------------------------
# Public API
# ---------------------------------------------------------------------------
def build_google_auth_url(
client_id: str,
redirect_uri: str = _DEFAULT_REDIRECT_URI,
scopes: Optional[List[str]] = None,
) -> str:
"""Build a Google OAuth2 consent URL.
Parameters
----------
client_id:
The OAuth 2.0 client ID from the Google Cloud Console.
redirect_uri:
Where Google should redirect after consent. Defaults to the local
callback server at ``http://localhost:8789/callback``.
scopes:
List of OAuth scopes to request. Defaults to
``["openid", "email", "profile"]``.
Returns
-------
str
Full consent URL including query string.
"""
if scopes is None:
scopes = _DEFAULT_SCOPES
params = {
"client_id": client_id,
"redirect_uri": redirect_uri,
"response_type": "code",
"scope": " ".join(scopes),
"access_type": "offline",
"prompt": "consent",
}
return f"{_GOOGLE_AUTH_ENDPOINT}?{urlencode(params)}"
def load_tokens(path: str) -> Optional[Dict[str, Any]]:
"""Load OAuth tokens from a JSON file.
Returns ``None`` if the file is missing, unreadable, or contains
invalid JSON.
"""
p = Path(path)
if not p.exists():
return None
try:
raw = p.read_text(encoding="utf-8")
return json.loads(raw)
except (OSError, json.JSONDecodeError):
return None
def save_tokens(path: str, tokens: Dict[str, Any]) -> None:
"""Persist *tokens* to *path* as JSON with owner-only (0o600) permissions.
Creates parent directories as needed.
"""
p = Path(path)
p.parent.mkdir(parents=True, exist_ok=True)
p.write_text(json.dumps(tokens, indent=2), encoding="utf-8")
os.chmod(path, 0o600)
def delete_tokens(path: str) -> None:
"""Delete the credentials file at *path* if it exists."""
p = Path(path)
if p.exists():
p.unlink()
tests/connectors/test_gmail.py
+217
View File
@@ -0,0 +1,217 @@
"""Tests for GmailConnector — OAuth-authenticated Gmail sync connector.
All Gmail API calls are mocked; no network access is required.
"""
from __future__ import annotations
import json
from pathlib import Path
from typing import List
from unittest.mock import patch
import pytest
from openjarvis.connectors._stubs import Document
from openjarvis.core.registry import ConnectorRegistry
# ---------------------------------------------------------------------------
# Helpers — fake API payloads
# ---------------------------------------------------------------------------
# base64url("Hello world") == "SGVsbG8gd29ybGQ="
# base64url("Budget reply") == "QnVkZ2V0IHJlcGx5"
_MSG1 = {
"id": "msg1",
"threadId": "thread1",
"labelIds": ["INBOX"],
"payload": {
"mimeType": "text/plain",
"headers": [
{"name": "From", "value": "alice@example.com"},
{"name": "To", "value": "me@example.com"},
{"name": "Subject", "value": "Q3 Planning"},
{"name": "Date", "value": "Mon, 01 Jan 2024 10:00:00 +0000"},
],
"body": {"data": "SGVsbG8gd29ybGQ="},
},
}
_MSG2 = {
"id": "msg2",
"threadId": "thread2",
"labelIds": ["INBOX"],
"payload": {
"mimeType": "text/plain",
"headers": [
{"name": "From", "value": "bob@example.com"},
{"name": "To", "value": "me@example.com"},
{"name": "Subject", "value": "Re: Budget"},
{"name": "Date", "value": "Tue, 02 Jan 2024 12:00:00 +0000"},
],
"body": {"data": "QnVkZ2V0IHJlcGx5"},
},
}
_LIST_RESPONSE = {
"messages": [{"id": "msg1"}, {"id": "msg2"}],
# No nextPageToken → single page
}
def _make_credentials(tmp_path: Path) -> Path:
"""Write a minimal fake credentials file and return its path."""
creds = tmp_path / "gmail.json"
creds.write_text(json.dumps({"token": "fake-access-token"}), encoding="utf-8")
return creds
# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------
@pytest.fixture()
def connector(tmp_path: Path):
"""GmailConnector pointing at a tmp credentials path (no file yet)."""
from openjarvis.connectors.gmail import GmailConnector # noqa: PLC0415
creds_path = str(tmp_path / "gmail.json")
return GmailConnector(credentials_path=creds_path)
# ---------------------------------------------------------------------------
# Test 1 — not connected without a credentials file
# ---------------------------------------------------------------------------
def test_not_connected_without_credentials(connector) -> None:
"""is_connected() returns False when no credentials file exists."""
assert connector.is_connected() is False
# ---------------------------------------------------------------------------
# Test 2 — auth_type is "oauth"
# ---------------------------------------------------------------------------
def test_auth_type_is_oauth(connector) -> None:
"""GmailConnector.auth_type must be 'oauth'."""
assert connector.auth_type == "oauth"
# ---------------------------------------------------------------------------
# Test 3 — auth_url returns a valid Google consent URL
# ---------------------------------------------------------------------------
def test_auth_url_returns_string(connector) -> None:
"""auth_url() returns a URL pointing to Google's OAuth endpoint."""
url = connector.auth_url()
assert isinstance(url, str)
assert url.startswith("https://accounts.google.com/o/oauth2/v2/auth")
assert "gmail.readonly" in url
# ---------------------------------------------------------------------------
# Test 4 — sync yields documents with correct fields (mocked API)
# ---------------------------------------------------------------------------
@patch("openjarvis.connectors.gmail._gmail_api_list_messages")
@patch("openjarvis.connectors.gmail._gmail_api_get_message")
def test_sync_yields_documents(
mock_get,
mock_list,
connector,
tmp_path: Path,
) -> None:
"""sync() yields one Document per message with correct metadata."""
# Set up fake credentials so is_connected() returns True
creds_path = Path(connector._credentials_path)
creds_path.write_text(
json.dumps({"token": "fake-access-token"}), encoding="utf-8"
)
# Configure mocks
mock_list.return_value = _LIST_RESPONSE
mock_get.side_effect = lambda token, msg_id: (
_MSG1 if msg_id == "msg1" else _MSG2
)
docs: List[Document] = list(connector.sync())
assert len(docs) == 2
# --- Message 1 ---
doc1 = next(d for d in docs if d.doc_id == "gmail:msg1")
assert doc1.source == "gmail"
assert doc1.doc_type == "email"
assert doc1.title == "Q3 Planning"
assert doc1.author == "alice@example.com"
assert doc1.content == "Hello world"
assert doc1.thread_id == "thread1"
assert "alice@example.com" in doc1.participants
# --- Message 2 ---
doc2 = next(d for d in docs if d.doc_id == "gmail:msg2")
assert doc2.title == "Re: Budget"
assert doc2.author == "bob@example.com"
assert doc2.content == "Budget reply"
assert doc2.thread_id == "thread2"
# Verify the API was called correctly
mock_list.assert_called_once()
assert mock_get.call_count == 2
# ---------------------------------------------------------------------------
# Test 5 — disconnect removes the credentials file
# ---------------------------------------------------------------------------
def test_disconnect(connector, tmp_path: Path) -> None:
"""disconnect() deletes the credentials file."""
creds_path = Path(connector._credentials_path)
creds_path.write_text(
json.dumps({"token": "fake-access-token"}), encoding="utf-8"
)
assert connector.is_connected() is True
connector.disconnect()
assert not creds_path.exists()
assert connector.is_connected() is False
# ---------------------------------------------------------------------------
# Test 6 — mcp_tools returns the three expected tool specs
# ---------------------------------------------------------------------------
def test_mcp_tools(connector) -> None:
"""mcp_tools() returns exactly 3 tools with the required names."""
tools = connector.mcp_tools()
names = {t.name for t in tools}
assert len(tools) == 3
assert "gmail_search_emails" in names
assert "gmail_get_thread" in names
assert "gmail_list_unread" in names
# ---------------------------------------------------------------------------
# Test 7 — ConnectorRegistry contains "gmail" after import
# ---------------------------------------------------------------------------
def test_registry() -> None:
"""GmailConnector can be registered and retrieved via ConnectorRegistry."""
from openjarvis.connectors.gmail import GmailConnector # noqa: PLC0415
# The registry is cleared before each test by the autouse conftest fixture,
# so we imperatively re-register here (same pattern as test_obsidian.py).
ConnectorRegistry.register_value("gmail", GmailConnector)
assert ConnectorRegistry.contains("gmail")
cls = ConnectorRegistry.get("gmail")
assert cls.connector_id == "gmail"