From 9cfcef88fb67f66e03919fd6dc583f69bc8ab1a7 Mon Sep 17 00:00:00 2001
From: Mustafa Esoofally <coolmusta@gmail.com>
Date: Mon, 18 May 2026 20:23:27 -0400
Subject: [PATCH] feat: add Gmail and Calendar context providers

- Import GmailContextProvider and GoogleCalendarContextProvider from agno
- Add factory functions supporting both service account and OAuth auth
- Gmail requires GOOGLE_DELEGATED_USER for service account mode
- Calendar works with service account or OAuth
- Both providers enable write=True by default

Test coverage:
- test_google_context_providers.py: basic status and query tests
- test_google_comprehensive.py: full CRUD tests for both providers

Note: Gmail OAuth needs all scopes (readonly + modify + compose) in one
token to avoid read/write toolkit conflicts.
---
 scout/contexts.py                |  34 ++-
 test_google_comprehensive.py     | 368 +++++++++++++++++++++++++++++++
 test_google_context_providers.py | 298 +++++++++++++++++++++++++
 3 files changed, 699 insertions(+), 1 deletion(-)
 create mode 100644 test_google_comprehensive.py
 create mode 100644 test_google_context_providers.py

diff --git a/scout/contexts.py b/scout/contexts.py
index f7db9db..639e01c 100644
--- a/scout/contexts.py
+++ b/scout/contexts.py
@@ -13,8 +13,10 @@
 from os import getenv
 from pathlib import Path
 
+from agno.context.calendar import GoogleCalendarContextProvider
 from agno.context.database import DatabaseContextProvider
 from agno.context.gdrive import GDriveContextProvider
+from agno.context.gmail import GmailContextProvider
 from agno.context.mcp import MCPContextProvider
 from agno.context.provider import ContextProvider
 from agno.context.slack import SlackContextProvider
@@ -67,7 +69,13 @@ def create_context_providers() -> list[ContextProvider]:
         _create_knowledge_wiki(),
         _create_voice_wiki(),
     ]
-    for factory in (_create_slack_provider, _create_gdrive_provider):
+    optional_factories = (
+        _create_slack_provider,
+        _create_gdrive_provider,
+        _create_gmail_provider,
+        _create_calendar_provider,
+    )
+    for factory in optional_factories:
         try:
             provider = factory()
         except Exception as exc:
@@ -248,6 +256,30 @@ def _create_gdrive_provider() -> GDriveContextProvider | None:
     return GDriveContextProvider(model=default_model())
 
 
+def _create_gmail_provider() -> GmailContextProvider | None:
+    sa_path = getenv("GOOGLE_SERVICE_ACCOUNT_FILE")
+    has_oauth = getenv("GOOGLE_CLIENT_ID") and getenv("GOOGLE_CLIENT_SECRET")
+
+    if not sa_path and not has_oauth:
+        return None
+
+    if sa_path and not getenv("GOOGLE_DELEGATED_USER"):
+        log_warning("Gmail requires GOOGLE_DELEGATED_USER for service account auth")
+        return None
+
+    return GmailContextProvider(model=default_model(), write=True)
+
+
+def _create_calendar_provider() -> GoogleCalendarContextProvider | None:
+    sa_path = getenv("GOOGLE_SERVICE_ACCOUNT_FILE")
+    has_oauth = getenv("GOOGLE_CLIENT_ID") and getenv("GOOGLE_CLIENT_SECRET")
+
+    if not sa_path and not has_oauth:
+        return None
+
+    return GoogleCalendarContextProvider(model=default_model(), write=True)
+
+
 def _create_mcp_providers() -> list[MCPContextProvider]:
     """Registered MCP servers.
 
diff --git a/test_google_comprehensive.py b/test_google_comprehensive.py
new file mode 100644
index 0000000..033998b
--- /dev/null
+++ b/test_google_comprehensive.py
@@ -0,0 +1,368 @@
+"""
+Comprehensive Google Context Provider Tests
+============================================
+
+Run: .venv/bin/python test_google_comprehensive.py
+
+Tests all read/write operations for Calendar and Gmail providers.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import os
+import time
+from datetime import datetime, timedelta, timezone
+
+os.chdir(os.path.dirname(os.path.abspath(__file__)))
+
+from dotenv import load_dotenv
+load_dotenv()
+
+from agno.context.calendar import GoogleCalendarContextProvider
+from agno.context.gmail import GmailContextProvider
+from agno.models.openai import OpenAIChat
+
+CALENDAR_TOKEN = "calendar_token.json"
+GMAIL_TOKEN = "gmail_token.json"
+
+
+def get_model():
+    return OpenAIChat(id="gpt-4o-mini")
+
+
+def header(title: str):
+    print("\n" + "=" * 70)
+    print(f"  {title}")
+    print("=" * 70)
+
+
+def subtest(name: str):
+    print(f"\n--- {name} ---")
+
+
+class CalendarTests:
+    def __init__(self):
+        self.provider = GoogleCalendarContextProvider(
+            model=get_model(),
+            token_path=CALENDAR_TOKEN,
+            write=True,
+        )
+        self.results = {}
+        self.created_event_id = None
+
+    def run_all(self):
+        header("CALENDAR PROVIDER TESTS")
+
+        # Status
+        subtest("1. Status Check")
+        status = self.provider.status()
+        print(f"OK: {status.ok}")
+        print(f"Detail: {status.detail}")
+        self.results["status"] = status.ok
+
+        if not status.ok:
+            print("SKIP: Calendar not authenticated")
+            return self.results
+
+        # Read operations
+        self.test_list_events()
+        self.test_search_events()
+        self.test_check_availability()
+        self.test_find_slots()
+
+        # Write operations
+        self.test_create_event()
+        if self.created_event_id:
+            self.test_get_event()
+            self.test_update_event()
+            self.test_delete_event()
+
+        # Async operations
+        asyncio.run(self.test_async_query())
+
+        return self.results
+
+    def test_list_events(self):
+        subtest("2. List Events (sync)")
+        try:
+            answer = self.provider.query("List my next 5 upcoming events with their times and locations")
+            print(f"Response length: {len(answer.text or '')} chars")
+            print(f"Preview: {(answer.text or '')[:300]}...")
+            self.results["list_events"] = bool(answer.text)
+        except Exception as e:
+            print(f"FAIL: {e}")
+            self.results["list_events"] = False
+
+    def test_search_events(self):
+        subtest("3. Search Events")
+        try:
+            answer = self.provider.query("Search for any meetings with 'review' or 'standup' in the title this week")
+            print(f"Response length: {len(answer.text or '')} chars")
+            print(f"Preview: {(answer.text or '')[:300]}...")
+            self.results["search_events"] = bool(answer.text)
+        except Exception as e:
+            print(f"FAIL: {e}")
+            self.results["search_events"] = False
+
+    def test_check_availability(self):
+        subtest("4. Check Availability")
+        tomorrow = (datetime.now(timezone.utc) + timedelta(days=1)).strftime("%Y-%m-%d")
+        try:
+            answer = self.provider.query(f"Am I free tomorrow ({tomorrow}) between 2pm and 4pm?")
+            print(f"Response: {(answer.text or '')[:300]}...")
+            self.results["check_availability"] = bool(answer.text)
+        except Exception as e:
+            print(f"FAIL: {e}")
+            self.results["check_availability"] = False
+
+    def test_find_slots(self):
+        subtest("5. Find Available Slots")
+        try:
+            answer = self.provider.query("Find me 3 available 30-minute slots this week for a meeting")
+            print(f"Response: {(answer.text or '')[:300]}...")
+            self.results["find_slots"] = bool(answer.text)
+        except Exception as e:
+            print(f"FAIL: {e}")
+            self.results["find_slots"] = False
+
+    def test_create_event(self):
+        subtest("6. Create Event (write)")
+        tomorrow = datetime.now(timezone.utc) + timedelta(days=1)
+        event_time = tomorrow.replace(hour=15, minute=0, second=0, microsecond=0)
+        try:
+            answer = self.provider.update(
+                f"Create a test event called 'Scout Integration Test' on {event_time.strftime('%Y-%m-%d')} "
+                f"at 3:00 PM EST for 30 minutes. Add description 'Automated test - safe to delete'"
+            )
+            print(f"Response: {(answer.text or '')[:400]}...")
+            # Try to extract event ID from response
+            text = answer.text or ""
+            if "created" in text.lower() or "event" in text.lower():
+                self.results["create_event"] = True
+                # Store that we created something (we'll search for it)
+                self.created_event_id = "pending_lookup"
+            else:
+                self.results["create_event"] = False
+        except Exception as e:
+            print(f"FAIL: {e}")
+            self.results["create_event"] = False
+
+    def test_get_event(self):
+        subtest("7. Get Event Details")
+        try:
+            answer = self.provider.query("Find the event called 'Scout Integration Test' and show me its full details")
+            print(f"Response: {(answer.text or '')[:400]}...")
+            self.results["get_event"] = bool(answer.text and "Scout Integration Test" in (answer.text or ""))
+        except Exception as e:
+            print(f"FAIL: {e}")
+            self.results["get_event"] = False
+
+    def test_update_event(self):
+        subtest("8. Update Event (write)")
+        try:
+            answer = self.provider.update(
+                "Find the event 'Scout Integration Test' and update its description to "
+                "'Updated by Scout test suite - safe to delete'"
+            )
+            print(f"Response: {(answer.text or '')[:300]}...")
+            self.results["update_event"] = bool(answer.text and ("updated" in (answer.text or "").lower()))
+        except Exception as e:
+            print(f"FAIL: {e}")
+            self.results["update_event"] = False
+
+    def test_delete_event(self):
+        subtest("9. Delete Event (write)")
+        try:
+            answer = self.provider.update(
+                "Delete the event called 'Scout Integration Test'"
+            )
+            print(f"Response: {(answer.text or '')[:300]}...")
+            self.results["delete_event"] = bool(answer.text and ("deleted" in (answer.text or "").lower() or "removed" in (answer.text or "").lower()))
+        except Exception as e:
+            print(f"FAIL: {e}")
+            self.results["delete_event"] = False
+
+    async def test_async_query(self):
+        subtest("10. Async Query")
+        try:
+            answer = await self.provider.aquery("What's on my calendar today?")
+            print(f"Response: {(answer.text or '')[:300]}...")
+            self.results["async_query"] = bool(answer.text)
+        except Exception as e:
+            print(f"FAIL: {e}")
+            self.results["async_query"] = False
+
+
+class GmailTests:
+    def __init__(self):
+        self.has_gmail = os.path.exists(GMAIL_TOKEN) or os.getenv("GOOGLE_DELEGATED_USER")
+        if self.has_gmail:
+            self.provider = GmailContextProvider(
+                model=get_model(),
+                token_path=GMAIL_TOKEN,
+                write=True,
+            )
+        else:
+            self.provider = None
+        self.results = {}
+        self.draft_id = None
+
+    def run_all(self):
+        header("GMAIL PROVIDER TESTS")
+
+        if not self.has_gmail:
+            print("SKIP: Gmail not configured (need GOOGLE_DELEGATED_USER or gmail_token.json)")
+            return {"gmail_configured": False}
+
+        # Status
+        subtest("1. Status Check")
+        try:
+            status = self.provider.status()
+            print(f"OK: {status.ok}")
+            print(f"Detail: {status.detail}")
+            self.results["status"] = status.ok
+        except Exception as e:
+            print(f"FAIL: {e}")
+            self.results["status"] = False
+            return self.results
+
+        if not self.results.get("status"):
+            print("SKIP: Gmail not authenticated")
+            return self.results
+
+        # Read operations
+        self.test_list_recent()
+        self.test_search_emails()
+        self.test_get_unread()
+
+        # Write operations
+        self.test_create_draft()
+        if self.draft_id:
+            self.test_update_draft()
+            self.test_delete_draft()
+
+        return self.results
+
+    def test_list_recent(self):
+        subtest("2. List Recent Emails")
+        try:
+            answer = self.provider.query("Show me my 3 most recent emails with sender and subject")
+            print(f"Response: {(answer.text or '')[:400]}...")
+            self.results["list_recent"] = bool(answer.text)
+        except Exception as e:
+            print(f"FAIL: {e}")
+            self.results["list_recent"] = False
+
+    def test_search_emails(self):
+        subtest("3. Search Emails")
+        try:
+            answer = self.provider.query("Search for emails from the last 7 days with attachments")
+            print(f"Response: {(answer.text or '')[:400]}...")
+            self.results["search_emails"] = bool(answer.text)
+        except Exception as e:
+            print(f"FAIL: {e}")
+            self.results["search_emails"] = False
+
+    def test_get_unread(self):
+        subtest("4. Get Unread Emails")
+        try:
+            answer = self.provider.query("How many unread emails do I have? List the first 3")
+            print(f"Response: {(answer.text or '')[:400]}...")
+            self.results["get_unread"] = bool(answer.text)
+        except Exception as e:
+            print(f"FAIL: {e}")
+            self.results["get_unread"] = False
+
+    def test_create_draft(self):
+        subtest("5. Create Draft (write)")
+        try:
+            answer = self.provider.update(
+                "Create a draft email to myself with subject 'Scout Test Draft' "
+                "and body 'This is an automated test draft - safe to delete'"
+            )
+            print(f"Response: {(answer.text or '')[:400]}...")
+            self.results["create_draft"] = bool(answer.text and "draft" in (answer.text or "").lower())
+            if self.results["create_draft"]:
+                self.draft_id = "created"
+        except Exception as e:
+            print(f"FAIL: {e}")
+            self.results["create_draft"] = False
+
+    def test_update_draft(self):
+        subtest("6. Update Draft (write)")
+        try:
+            answer = self.provider.update(
+                "Find my draft with subject 'Scout Test Draft' and update the body to "
+                "'Updated by Scout test suite'"
+            )
+            print(f"Response: {(answer.text or '')[:300]}...")
+            self.results["update_draft"] = bool(answer.text)
+        except Exception as e:
+            print(f"FAIL: {e}")
+            self.results["update_draft"] = False
+
+    def test_delete_draft(self):
+        subtest("7. Delete Draft (write)")
+        try:
+            answer = self.provider.update(
+                "Delete my draft with subject 'Scout Test Draft'"
+            )
+            print(f"Response: {(answer.text or '')[:300]}...")
+            self.results["delete_draft"] = bool(answer.text)
+        except Exception as e:
+            print(f"FAIL: {e}")
+            self.results["delete_draft"] = False
+
+
+def print_summary(calendar_results: dict, gmail_results: dict):
+    header("TEST SUMMARY")
+
+    all_results = {}
+
+    print("\nCalendar Tests:")
+    for name, passed in calendar_results.items():
+        status = "PASS" if passed else "FAIL" if passed is False else "SKIP"
+        print(f"  {name}: {status}")
+        all_results[f"calendar_{name}"] = passed
+
+    print("\nGmail Tests:")
+    for name, passed in gmail_results.items():
+        status = "PASS" if passed else "FAIL" if passed is False else "SKIP"
+        print(f"  {name}: {status}")
+        all_results[f"gmail_{name}"] = passed
+
+    passed = sum(1 for v in all_results.values() if v is True)
+    failed = sum(1 for v in all_results.values() if v is False)
+    skipped = sum(1 for v in all_results.values() if v is None or v == "SKIP")
+
+    print(f"\nTotal: {passed} passed, {failed} failed, {skipped} skipped")
+
+    if failed > 0:
+        print("\nFailed tests:")
+        for name, passed in all_results.items():
+            if passed is False:
+                print(f"  - {name}")
+
+
+def main():
+    print("=" * 70)
+    print("  COMPREHENSIVE GOOGLE CONTEXT PROVIDER TEST SUITE")
+    print("=" * 70)
+    print(f"\nCalendar token: {CALENDAR_TOKEN} (exists: {os.path.exists(CALENDAR_TOKEN)})")
+    print(f"Gmail token: {GMAIL_TOKEN} (exists: {os.path.exists(GMAIL_TOKEN)})")
+    print(f"Delegated user: {os.getenv('GOOGLE_DELEGATED_USER', '(not set)')}")
+
+    # Run tests
+    calendar_tests = CalendarTests()
+    calendar_results = calendar_tests.run_all()
+
+    gmail_tests = GmailTests()
+    gmail_results = gmail_tests.run_all()
+
+    # Summary
+    print_summary(calendar_results, gmail_results)
+
+
+if __name__ == "__main__":
+    main()
diff --git a/test_google_context_providers.py b/test_google_context_providers.py
new file mode 100644
index 0000000..643c92f
--- /dev/null
+++ b/test_google_context_providers.py
@@ -0,0 +1,298 @@
+"""
+Test Gmail and Calendar Context Providers
+==========================================
+
+Run: .venv/bin/python test_google_context_providers.py
+
+Tests:
+1. Provider status (auth validation)
+2. Gmail read operations (search, get message)
+3. Calendar read operations (list events, search)
+4. Write operations (marked skip by default)
+"""
+
+from __future__ import annotations
+
+import asyncio
+import os
+from datetime import datetime, timedelta, timezone
+
+os.chdir(os.path.dirname(os.path.abspath(__file__)))
+os.environ.setdefault("GOOGLE_SERVICE_ACCOUNT_FILE", ".scout/service-account.json")
+
+from dotenv import load_dotenv
+
+load_dotenv()
+
+from agno.context.calendar import GoogleCalendarContextProvider
+from agno.context.gmail import GmailContextProvider
+from agno.models.openai import OpenAIChat
+
+TEST_WRITE_OPS = os.getenv("TEST_WRITE_OPS", "").lower() in ("1", "true", "yes")
+
+
+def get_model():
+    return OpenAIChat(id="gpt-4o-mini")
+
+
+def test_calendar_status():
+    print("\n" + "=" * 60)
+    print("TEST: Calendar Provider Status")
+    print("=" * 60)
+
+    provider = GoogleCalendarContextProvider(model=get_model(), token_path="calendar_token.json")
+    status = provider.status()
+    print(f"OK: {status.ok}")
+    print(f"Detail: {status.detail}")
+    return status.ok
+
+
+def test_gmail_status():
+    print("\n" + "=" * 60)
+    print("TEST: Gmail Provider Status")
+    print("=" * 60)
+
+    delegated_user = os.getenv("GOOGLE_DELEGATED_USER")
+    if not delegated_user:
+        print("SKIP: GOOGLE_DELEGATED_USER not set (required for Gmail SA auth)")
+        return None
+
+    provider = GmailContextProvider(model=get_model())
+    status = provider.status()
+    print(f"OK: {status.ok}")
+    print(f"Detail: {status.detail}")
+    return status.ok
+
+
+def test_calendar_query():
+    print("\n" + "=" * 60)
+    print("TEST: Calendar Query (list events)")
+    print("=" * 60)
+
+    provider = GoogleCalendarContextProvider(model=get_model(), token_path="calendar_token.json")
+
+    # Simple query - avoid date ranges that trigger time_min/time_max bug
+    query = "List my upcoming 5 events"
+    print(f"Query: {query}")
+
+    try:
+        answer = provider.query(query)
+        text = answer.text or "(no text)"
+        print(f"\nAnswer (first 500 chars):")
+        print("-" * 40)
+        print(text[:500] if len(text) > 500 else text)
+        print("-" * 40)
+
+        # Check for common errors
+        if "API has not been used" in text or "accessNotConfigured" in text:
+            print("\nFAIL: Calendar API not enabled in GCP project")
+            print("Enable at: https://console.developers.google.com/apis/api/calendar-json.googleapis.com")
+            return False
+
+        return True
+    except Exception as exc:
+        print(f"\nFAIL: {type(exc).__name__}: {exc}")
+        return False
+
+
+def test_gmail_query():
+    print("\n" + "=" * 60)
+    print("TEST: Gmail Query (search emails)")
+    print("=" * 60)
+
+    delegated_user = os.getenv("GOOGLE_DELEGATED_USER")
+    if not delegated_user:
+        print("SKIP: GOOGLE_DELEGATED_USER not set")
+        return None
+
+    provider = GmailContextProvider(model=get_model())
+
+    query = "Show me my 3 most recent emails"
+    print(f"Query: {query}")
+
+    try:
+        answer = provider.query(query)
+        text = answer.text or "(no text)"
+        print(f"\nAnswer (first 500 chars):")
+        print("-" * 40)
+        print(text[:500] if len(text) > 500 else text)
+        print("-" * 40)
+
+        if "API has not been used" in text or "accessNotConfigured" in text:
+            print("\nFAIL: Gmail API not enabled in GCP project")
+            return False
+
+        return True
+    except Exception as exc:
+        print(f"\nFAIL: {type(exc).__name__}: {exc}")
+        return False
+
+
+async def test_calendar_async_query():
+    print("\n" + "=" * 60)
+    print("TEST: Calendar Async Query")
+    print("=" * 60)
+
+    provider = GoogleCalendarContextProvider(model=get_model(), token_path="calendar_token.json")
+
+    query = "List my next 3 events"
+    print(f"Query: {query}")
+
+    try:
+        answer = await provider.aquery(query)
+        text = answer.text or "(no text)"
+        print(f"\nAnswer (first 500 chars):")
+        print("-" * 40)
+        print(text[:500] if len(text) > 500 else text)
+        print("-" * 40)
+
+        if "API has not been used" in text or "accessNotConfigured" in text:
+            print("\nFAIL: Calendar API not enabled in GCP project")
+            return False
+
+        return True
+    except Exception as exc:
+        print(f"\nFAIL: {type(exc).__name__}: {exc}")
+        return False
+
+
+def test_calendar_update():
+    print("\n" + "=" * 60)
+    print("TEST: Calendar Update (create event)")
+    print("=" * 60)
+
+    if not TEST_WRITE_OPS:
+        print("SKIP: Set TEST_WRITE_OPS=1 to run write tests")
+        return None
+
+    provider = GoogleCalendarContextProvider(model=get_model(), token_path="calendar_token.json", write=True)
+
+    tomorrow = datetime.now(timezone.utc) + timedelta(days=1)
+    instruction = f"Create a test event called 'Scout Test Event' tomorrow ({tomorrow.date()}) at 2pm for 30 minutes"
+    print(f"Instruction: {instruction}")
+
+    try:
+        answer = provider.update(instruction)
+        text = answer.text or "(no text)"
+        print(f"\nAnswer:")
+        print("-" * 40)
+        print(text)
+        print("-" * 40)
+        return True
+    except Exception as exc:
+        print(f"\nFAIL: {type(exc).__name__}: {exc}")
+        return False
+
+
+def test_gmail_update():
+    print("\n" + "=" * 60)
+    print("TEST: Gmail Update (draft email)")
+    print("=" * 60)
+
+    delegated_user = os.getenv("GOOGLE_DELEGATED_USER")
+    if not delegated_user:
+        print("SKIP: GOOGLE_DELEGATED_USER not set")
+        return None
+
+    if not TEST_WRITE_OPS:
+        print("SKIP: Set TEST_WRITE_OPS=1 to run write tests")
+        return None
+
+    provider = GmailContextProvider(model=get_model(), write=True)
+
+    instruction = "Draft an email to myself with subject 'Scout Test' and body 'This is a test email from Scout.'"
+    print(f"Instruction: {instruction}")
+
+    try:
+        answer = provider.update(instruction)
+        text = answer.text or "(no text)"
+        print(f"\nAnswer:")
+        print("-" * 40)
+        print(text)
+        print("-" * 40)
+        return True
+    except Exception as exc:
+        print(f"\nFAIL: {type(exc).__name__}: {exc}")
+        return False
+
+
+def test_scout_contexts_integration():
+    print("\n" + "=" * 60)
+    print("TEST: Scout contexts.py integration")
+    print("=" * 60)
+
+    from scout.contexts import (
+        _create_calendar_provider,
+        _create_gmail_provider,
+    )
+
+    # Test individual factories (doesn't require database)
+    try:
+        calendar = _create_calendar_provider()
+        gmail = _create_gmail_provider()
+
+        print(f"Calendar factory: {type(calendar).__name__ if calendar else 'None'}")
+        print(f"Gmail factory: {type(gmail).__name__ if gmail else 'None'}")
+
+        calendar_ok = calendar is not None
+        gmail_ok = gmail is not None or not os.getenv("GOOGLE_DELEGATED_USER")
+
+        print(f"\nCalendar created: {calendar_ok}")
+        print(f"Gmail created: {gmail_ok} (requires GOOGLE_DELEGATED_USER)")
+
+        # Skip full registry test - requires database
+        print("\nNote: Full registry test skipped (requires database)")
+
+        return calendar_ok and gmail_ok
+    except Exception as exc:
+        print(f"\nFAIL: {type(exc).__name__}: {exc}")
+        return False
+
+
+def main():
+    print("Google Context Providers Test Suite")
+    print("=" * 60)
+    print(f"Service account: {os.environ.get('GOOGLE_SERVICE_ACCOUNT_FILE')}")
+    print(f"Delegated user: {os.environ.get('GOOGLE_DELEGATED_USER', '(not set)')}")
+    print(f"Write tests: {'ENABLED' if TEST_WRITE_OPS else 'disabled (set TEST_WRITE_OPS=1 to enable)'}")
+
+    results = {}
+
+    # Status tests
+    results["calendar_status"] = test_calendar_status()
+    results["gmail_status"] = test_gmail_status()
+
+    # Query tests
+    if results.get("calendar_status"):
+        results["calendar_query"] = test_calendar_query()
+        results["calendar_async"] = asyncio.run(test_calendar_async_query())
+
+    if results.get("gmail_status"):
+        results["gmail_query"] = test_gmail_query()
+
+    # Write tests
+    if results.get("calendar_status"):
+        results["calendar_update"] = test_calendar_update()
+
+    if results.get("gmail_status"):
+        results["gmail_update"] = test_gmail_update()
+
+    # Integration test
+    results["scout_integration"] = test_scout_contexts_integration()
+
+    # Summary
+    print("\n" + "=" * 60)
+    print("SUMMARY")
+    print("=" * 60)
+    for name, result in results.items():
+        status = "PASS" if result is True else "SKIP" if result is None else "FAIL"
+        print(f"  {name}: {status}")
+
+    passed = sum(1 for r in results.values() if r is True)
+    skipped = sum(1 for r in results.values() if r is None)
+    failed = sum(1 for r in results.values() if r is False)
+    print(f"\nTotal: {passed} passed, {skipped} skipped, {failed} failed")
+
+
+if __name__ == "__main__":
+    main()