Agent Skills: Google Chat Skill

Google Chat Skill

Unified Google Chat client for sending/reading messages in both group Spaces and DMs, with name-based addressing.

Two data sources: Live API (real-time, write) + Parquet Archive (fast bulk read, search).

File: data_sources/google_chat/04_google_chat_unified.py Class: GoogleChatUnified Auth: OAuth2 via credentials/token_write.pickle (auto-refresh)

Source Selection — MANDATORY Decision

Choose data source based on task type. When in doubt, run both in parallel.

| Task | Source | Why | |------|--------|-----| | "send message to X" | API | Write = always API | | "update/edit message" | API | Write = always API | | "what's the latest in chat X" (last few msgs) | API | Need real-time, single space | | "show my messages for N days" | Archive | Bulk read across all spaces, 0.1s | | "find messages about [topic]" | Archive | Full-text search across all spaces | | "who messaged me" / "DM summary" | Archive | Aggregation across all DMs | | "what did [person] say" | Archive first, API fallback | Archive has all spaces indexed | | "read thread in space X" | API | Thread drill-down needs live data |

Rule of thumb:

• Single space, real-time → API
• Multiple spaces, historical, search → Archive
• Ambiguous → Archive first (fast), then API if needed

Archive: Parquet Data Source (READ-ONLY)

Archiver: data_sources/google_chat/06_gchat_archiver.py Service: Runs every 60s in Miras server, auto-archives all active spaces. Storage: ~/projects/Daniel Personal/obsidian/05_Daniel_Network and Contacts/Communications/google_chat/archive/

Schema (12 columns)

| Column | Type | Notes | |--------|------|-------| | message_id | str | Unique message resource name | | space_id | str | spaces/XXXXX | | space_name | str | Display name or DM: Person Name | | space_type | str | SPACE or DIRECT_MESSAGE | | sender_id | str | users/XXXXX | | sender_name | str | Display name of sender | | text | str | Message content | | created_at | str | ISO timestamp with timezone | | thread_id | str | Thread resource name | | is_thread_reply | bool | True if reply in thread | | thread_msg_count | int | Messages in this thread | | has_attachment | bool | Has file/gif attached |

Quick Query (Python)

import pandas as pd, os

ARCHIVE = os.path.expanduser(
    "~/projects/Daniel Personal/obsidian/"
    "05_Daniel_Network and Contacts/Communications/google_chat/archive/messages"
)

# Load all weeks
dfs = [pd.read_parquet(os.path.join(ARCHIVE, f))
       for f in sorted(os.listdir(ARCHIVE)) if f.endswith('.parquet')]
df = pd.concat(dfs, ignore_index=True)
df['_dt'] = pd.to_datetime(df['created_at'], utc=True)

# Messages to Daniel in last 2 days
from datetime import datetime, timedelta, timezone
cutoff = datetime.now(timezone.utc) - timedelta(days=2)
recent = df[df['_dt'] >= cutoff].sort_values('_dt')

# DMs only
dms = recent[recent.space_type == 'DIRECT_MESSAGE']
for name, grp in dms.groupby('space_name'):
    print(f"\n--- {name} ({len(grp)} msgs) ---")
    for _, r in grp.iterrows():
        print(f"  {r.sender_name}: {r.text[:80]}")

# Search across all spaces
matches = df[df.text.str.contains('keyword', case=False, na=False)]

CLI

# Re-archive (manual trigger)
python data_sources/google_chat/06_gchat_archiver.py --days 7

# Stats
python data_sources/google_chat/06_gchat_archiver.py --stats

# Full history (~90 days)
python data_sources/google_chat/06_gchat_archiver.py --full

Archive Status API

curl http://localhost:8765/api/gchat-archive/status

MANDATORY: Session Tracking in Messages

Every message sent via this skill MUST include the Claude Code session ID at the end.

Format: append \n\n📎 Session: {session_id} to every outgoing message.

How to get session ID:

import os
session_id = os.environ.get('CLAUDE_SESSION_ID', 'unknown')

Why: Allows tracing any Google Chat message back to the Claude Code session that sent it. Critical for debugging and audit.

Example:

session_id = os.environ.get('CLAUDE_SESSION_ID', 'unknown')
message = f"Hello team!\n\n📎 Session: {session_id}"
chat.send_to_space("spaces/AAQAKNBCKZo", message)

Editing Messages

Method: update_message(message_name, new_text) API: PATCH /v1/{message.name} with updateMask=text

# message_name format: spaces/X/messages/Y.Y (note: ID repeated with dot)
chat.update_message('spaces/AAQAKNBCKZo/messages/abc123.abc123', 'Updated text')

CLI:

python data_sources/google_chat/04_google_chat_unified.py update "spaces/X/messages/Y.Y" "New text"

Use cases: appending session ID to already-sent messages, fixing typos, adding follow-up info.

Quick Start

Python (preferred for agents)

import sys, os, importlib.util
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', '..'))
spec = importlib.util.spec_from_file_location(
    'gchat_unified',
    os.path.join(os.environ.get('PROJECT_ROOT', '.'), 'data_sources/google_chat/04_google_chat_unified.py')
)
mod = importlib.util.module_from_spec(spec)
spec.loader.exec_module(mod)

chat = mod.GoogleChatUnified()

# Send to group space
chat.send("AI agent Task", "Hello from agent!")
chat.send_to_space("spaces/AAQAKNBCKZo", "Direct by ID")

# Send DM (resolves name via ClickHouse)
chat.send_dm("Ilia Nazarov", "Hi from Claude!")

# Read messages (thread replies auto-annotated by default)
messages = chat.read_messages("AI agent Task", days_back=7)
dm_msgs = chat.read_dm("Daniel Kravtsov", days_back=3)

# Each message gets: _is_thread_reply, _thread_name, _thread_msg_count
for m in messages:
    if m["_is_thread_reply"]:
        print(f"  Reply in thread: {m['_thread_name']}")

# Read all messages in a specific thread
thread_msgs = chat.read_thread("Improvado MCP", "spaces/AAQABiGai2s/threads/vYQZouzmtgw")

# All today's messages across all active spaces
today_msgs = chat.read_all_today()

# Find spaces
spaces = chat.find_space("Customer")
all_spaces = chat.list_all_spaces()

CLI

# All messages from today across all active spaces (optimized via lastActiveTime)
python data_sources/google_chat/04_google_chat_unified.py today

# Send to group space (by displayName)
python data_sources/google_chat/04_google_chat_unified.py send "AI agent Task" "Hello!"

# Send DM by person name
python data_sources/google_chat/04_google_chat_unified.py dm "Ilia Nazarov" "Hello!"

# Read from space (thread replies shown with > prefix)
python data_sources/google_chat/04_google_chat_unified.py read "AI agent Task" --days 7 --limit 50

# Read DMs
python data_sources/google_chat/04_google_chat_unified.py read-dm "Daniel Kravtsov" --days 3

# Read all messages in a specific thread
python data_sources/google_chat/04_google_chat_unified.py read-thread "Improvado MCP" "spaces/AAQABiGai2s/threads/vYQZouzmtgw"

# Find spaces by keyword
python data_sources/google_chat/04_google_chat_unified.py find "Customer"

# List all spaces
python data_sources/google_chat/04_google_chat_unified.py list

Thread Support (Default On)

All read_* methods auto-annotate messages with thread metadata:

| Field | Type | Description | |-------|------|-------------| | _is_thread_reply | bool | True if message is a reply in a thread | | _thread_name | str | Thread resource name (spaces/X/threads/Y) | | _thread_msg_count | int | Total messages in this thread |

CLI output: Thread starters show [N msgs in thread], replies show > prefix.

Read an entire thread:

msgs = chat.read_thread("Space Name", "spaces/X/threads/Y")

Disable annotations (marginal perf gain): chat.read_messages("Space", show_threads=False)

DM Resolution Chain

When sending/reading DMs by person name (always ~1 second):

1. ClickHouse lookup — stg_google_employees_cards ILIKE match → email
2. spaces.setup(email) — Google API finds existing DM space or creates one (idempotent, O(1))

No cache files, no scanning. Works instantly for any employee, first time or repeated.

Space Addressing

| Target | Example | Resolution | |--------|---------|------------| | Space ID | spaces/AAQAKNBCKZo | Direct — no lookup | | Display name | "AI agent Task" | API list_spaces() → match displayName | | Person name | "Ilia Nazarov" | ClickHouse(email) → spaces.setup() |

The send() method auto-detects: spaces/ prefix = direct ID, else tries group space name, then DM.

Other Google Chat Files (Reference)

| File | Purpose | Auth | |------|---------|------| | google_chat_client.py | Read-only client (legacy) | token.pickle (read scopes) | | gchat_sender.py | Webhook sender (17 team spaces) | Webhook URLs | | 02_send_dm.py | Service Account DM sender | SA credentials | | 03_simple_dm_sender.py | Hybrid DM (Make.com + API) | SA + webhooks | | 13_user_cache_manager.py | Legacy user cache (no longer used by unified client) | N/A | | 04_google_chat_unified.py | This skill — unified read+write | token_write.pickle |

Token Management

Token at data_sources/google_chat/credentials/token_write.pickle:

• Scopes: chat.spaces, chat.spaces.create, chat.messages, chat.messages.create, chat.memberships
• Auto-refreshes via refresh_token
• If expired beyond recovery: delete pickle, re-run to trigger OAuth flow

Known Spaces (Frequently Used)

| Space | ID | |-------|----| | Internal AI agent | spaces/AAQAKNBCKZo | | AI Dashboards | spaces/AAQAbiS6SEE | | AI Agent <> Core Product | spaces/AAQAzoIWRIw |

For customer spaces, use find "Customer" to discover.

ClickHouse Table

-- Employee lookup query (Palantir shard, is_default=False)
SELECT improvado_employee_google_full_name,
       improvado_employee_google_id,
       improvado_employee_google_email
FROM internal_analytics.stg_google_employees_cards
WHERE improvado_employee_google_full_name ILIKE '%Name%'
ORDER BY improvado_employee_is_actual DESC, improvado_employee_is_active DESC
LIMIT 1