a69b9d9160
Permission-first AI agent runtime for Google Workspace. Ports the LARC/OpenClaw governance model (disclosure chain, execution gates, queue/ingress) to Gmail, Calendar, Drive, Sheets, Tasks, and People APIs with Claude Code as the execution engine. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
392 lines
13 KiB
Python
392 lines
13 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
GARC People Helper — Google People API (Contacts & Directory)
|
|
search / list / show / create / update / delete
|
|
"""
|
|
|
|
import argparse
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).parent))
|
|
from garc_core import build_service, with_retry
|
|
|
|
PERSON_FIELDS = "names,emailAddresses,phoneNumbers,organizations,addresses,biographies"
|
|
CONTACT_SCOPES = [
|
|
"https://www.googleapis.com/auth/contacts",
|
|
"https://www.googleapis.com/auth/directory.readonly",
|
|
]
|
|
|
|
|
|
def get_service():
|
|
return build_service("people", "v1", scopes=CONTACT_SCOPES)
|
|
|
|
|
|
def _fmt_person(p: dict, short: bool = False) -> str:
|
|
"""Format a person resource as a readable string."""
|
|
name = p.get("names", [{}])[0].get("displayName", "(no name)")
|
|
emails = [e.get("value", "") for e in p.get("emailAddresses", [])]
|
|
phones = [ph.get("value", "") for ph in p.get("phoneNumbers", [])]
|
|
orgs = [o.get("name", "") for o in p.get("organizations", [])]
|
|
resource = p.get("resourceName", "")
|
|
short_id = resource.split("/")[-1] if "/" in resource else resource
|
|
|
|
if short:
|
|
email_str = emails[0] if emails else ""
|
|
org_str = f" ({orgs[0]})" if orgs else ""
|
|
return f"[{short_id[:10]}] {name}{org_str} — {email_str}"
|
|
|
|
lines = [f"Name: {name}", f"ID: {short_id}"]
|
|
for e in emails:
|
|
lines.append(f"Email: {e}")
|
|
for ph in phones:
|
|
lines.append(f"Phone: {ph}")
|
|
for o in p.get("organizations", []):
|
|
org_parts = [x for x in [o.get("name"), o.get("title"), o.get("department")] if x]
|
|
lines.append(f"Org: {' / '.join(org_parts)}")
|
|
for bio in p.get("biographies", []):
|
|
lines.append(f"Bio: {bio.get('value', '')[:80]}")
|
|
return "\n".join(lines)
|
|
|
|
|
|
# ─────────────────────────────────────────────
|
|
# Search (Directory + personal contacts)
|
|
# ─────────────────────────────────────────────
|
|
|
|
@with_retry()
|
|
def search_contacts(query: str, max_results: int = 20, format_: str = "table"):
|
|
"""Search contacts from the user's personal contacts."""
|
|
service = get_service()
|
|
result = service.people().searchContacts(
|
|
query=query,
|
|
readMask=PERSON_FIELDS,
|
|
pageSize=min(max_results, 30),
|
|
).execute()
|
|
|
|
results = result.get("results", [])
|
|
if not results:
|
|
print(f"No contacts found for: {query}")
|
|
return
|
|
|
|
if format_ == "json":
|
|
print(json.dumps([r.get("person", {}) for r in results], ensure_ascii=False, indent=2))
|
|
return
|
|
|
|
print(f"Contacts matching '{query}' ({len(results)}):")
|
|
for r in results:
|
|
print(f" {_fmt_person(r.get('person', {}), short=True)}")
|
|
|
|
|
|
@with_retry()
|
|
def search_directory(query: str, max_results: int = 20, format_: str = "table"):
|
|
"""Search the Google Workspace directory (org-wide)."""
|
|
service = get_service()
|
|
try:
|
|
result = service.people().searchDirectoryPeople(
|
|
query=query,
|
|
readMask=PERSON_FIELDS,
|
|
sources=["DIRECTORY_SOURCE_TYPE_DOMAIN_PROFILE", "DIRECTORY_SOURCE_TYPE_DOMAIN_CONTACT"],
|
|
pageSize=min(max_results, 30),
|
|
).execute()
|
|
except Exception as e:
|
|
if "403" in str(e):
|
|
print("Directory search requires Google Workspace (not personal Gmail).", file=sys.stderr)
|
|
else:
|
|
raise
|
|
return
|
|
|
|
people = result.get("people", [])
|
|
if not people:
|
|
print(f"No directory results for: {query}")
|
|
return
|
|
|
|
if format_ == "json":
|
|
print(json.dumps(people, ensure_ascii=False, indent=2))
|
|
return
|
|
|
|
print(f"Directory results for '{query}' ({len(people)}):")
|
|
for p in people:
|
|
print(f" {_fmt_person(p, short=True)}")
|
|
|
|
|
|
# ─────────────────────────────────────────────
|
|
# Contacts CRUD
|
|
# ─────────────────────────────────────────────
|
|
|
|
@with_retry()
|
|
def list_contacts(max_results: int = 50, format_: str = "table"):
|
|
"""List all personal contacts."""
|
|
service = get_service()
|
|
result = service.people().connections().list(
|
|
resourceName="people/me",
|
|
personFields=PERSON_FIELDS,
|
|
pageSize=min(max_results, 1000),
|
|
sortOrder="LAST_MODIFIED_DESCENDING",
|
|
).execute()
|
|
|
|
people = result.get("connections", [])
|
|
if not people:
|
|
print("No contacts found.")
|
|
return
|
|
|
|
if format_ == "json":
|
|
print(json.dumps(people, ensure_ascii=False, indent=2))
|
|
return
|
|
|
|
print(f"Contacts ({len(people)}):")
|
|
for p in people:
|
|
print(f" {_fmt_person(p, short=True)}")
|
|
|
|
|
|
@with_retry()
|
|
def show_contact(contact_id: str):
|
|
"""Show full details of a contact."""
|
|
service = get_service()
|
|
# Accept short ID or full resource name
|
|
if not contact_id.startswith("people/"):
|
|
contact_id = f"people/{contact_id}"
|
|
person = service.people().get(
|
|
resourceName=contact_id,
|
|
personFields=PERSON_FIELDS,
|
|
).execute()
|
|
print(_fmt_person(person))
|
|
|
|
|
|
@with_retry()
|
|
def create_contact(
|
|
name: str,
|
|
email: str = None,
|
|
phone: str = None,
|
|
company: str = None,
|
|
title: str = None,
|
|
notes: str = None,
|
|
):
|
|
"""Create a new contact."""
|
|
service = get_service()
|
|
body: dict = {}
|
|
|
|
# Name
|
|
parts = name.split(" ", 1)
|
|
body["names"] = [{
|
|
"givenName": parts[0],
|
|
"familyName": parts[1] if len(parts) > 1 else "",
|
|
}]
|
|
if email:
|
|
body["emailAddresses"] = [{"value": email, "type": "work"}]
|
|
if phone:
|
|
body["phoneNumbers"] = [{"value": phone, "type": "work"}]
|
|
if company or title:
|
|
body["organizations"] = [{
|
|
"name": company or "",
|
|
"title": title or "",
|
|
"type": "work",
|
|
}]
|
|
if notes:
|
|
body["biographies"] = [{"value": notes, "contentType": "TEXT_PLAIN"}]
|
|
|
|
result = service.people().createContact(body=body).execute()
|
|
resource_id = result.get("resourceName", "").split("/")[-1]
|
|
print(f"✅ Contact created: [{resource_id}] {name}")
|
|
if email:
|
|
print(f" Email: {email}")
|
|
|
|
|
|
@with_retry()
|
|
def update_contact(
|
|
contact_id: str,
|
|
name: str = None,
|
|
email: str = None,
|
|
phone: str = None,
|
|
company: str = None,
|
|
title: str = None,
|
|
):
|
|
"""Update fields of an existing contact."""
|
|
service = get_service()
|
|
if not contact_id.startswith("people/"):
|
|
contact_id = f"people/{contact_id}"
|
|
|
|
# Fetch current
|
|
person = service.people().get(
|
|
resourceName=contact_id,
|
|
personFields=PERSON_FIELDS,
|
|
).execute()
|
|
etag = person.get("etag")
|
|
update_fields = []
|
|
|
|
if name:
|
|
parts = name.split(" ", 1)
|
|
person["names"] = [{
|
|
"givenName": parts[0],
|
|
"familyName": parts[1] if len(parts) > 1 else "",
|
|
}]
|
|
update_fields.append("names")
|
|
|
|
if email:
|
|
person["emailAddresses"] = [{"value": email, "type": "work"}]
|
|
update_fields.append("emailAddresses")
|
|
|
|
if phone:
|
|
person["phoneNumbers"] = [{"value": phone, "type": "work"}]
|
|
update_fields.append("phoneNumbers")
|
|
|
|
if company or title:
|
|
existing_org = (person.get("organizations") or [{}])[0]
|
|
person["organizations"] = [{
|
|
"name": company or existing_org.get("name", ""),
|
|
"title": title or existing_org.get("title", ""),
|
|
"type": "work",
|
|
}]
|
|
update_fields.append("organizations")
|
|
|
|
if not update_fields:
|
|
print("No updates specified.")
|
|
return
|
|
|
|
person["etag"] = etag
|
|
service.people().updateContact(
|
|
resourceName=contact_id,
|
|
updatePersonFields=",".join(update_fields),
|
|
body=person,
|
|
).execute()
|
|
short_id = contact_id.split("/")[-1]
|
|
print(f"✅ Contact updated: [{short_id}]")
|
|
|
|
|
|
@with_retry()
|
|
def delete_contact(contact_id: str):
|
|
"""Delete a contact."""
|
|
service = get_service()
|
|
if not contact_id.startswith("people/"):
|
|
contact_id = f"people/{contact_id}"
|
|
service.people().deleteContact(resourceName=contact_id).execute()
|
|
short_id = contact_id.split("/")[-1]
|
|
print(f"🗑️ Contact deleted: [{short_id}]")
|
|
|
|
|
|
# ─────────────────────────────────────────────
|
|
# Email Lookup helper (used by gmail.sh)
|
|
# ─────────────────────────────────────────────
|
|
|
|
@with_retry()
|
|
def lookup_email(name_or_email: str):
|
|
"""Quick lookup: find email for a name. Tries contacts then directory."""
|
|
service = get_service()
|
|
|
|
# Try personal contacts first
|
|
try:
|
|
result = service.people().searchContacts(
|
|
query=name_or_email,
|
|
readMask="names,emailAddresses",
|
|
pageSize=5,
|
|
).execute()
|
|
for r in result.get("results", []):
|
|
p = r.get("person", {})
|
|
emails = p.get("emailAddresses", [])
|
|
if emails:
|
|
name = p.get("names", [{}])[0].get("displayName", "")
|
|
email = emails[0].get("value", "")
|
|
print(f"{name} <{email}>")
|
|
return
|
|
except Exception:
|
|
pass
|
|
|
|
# Try directory
|
|
try:
|
|
result = service.people().searchDirectoryPeople(
|
|
query=name_or_email,
|
|
readMask="names,emailAddresses",
|
|
sources=["DIRECTORY_SOURCE_TYPE_DOMAIN_PROFILE"],
|
|
pageSize=5,
|
|
).execute()
|
|
for p in result.get("people", []):
|
|
emails = p.get("emailAddresses", [])
|
|
if emails:
|
|
name = p.get("names", [{}])[0].get("displayName", "")
|
|
email = emails[0].get("value", "")
|
|
print(f"{name} <{email}>")
|
|
return
|
|
except Exception:
|
|
pass
|
|
|
|
print(f"Not found: {name_or_email}")
|
|
|
|
|
|
# ─────────────────────────────────────────────
|
|
# CLI
|
|
# ─────────────────────────────────────────────
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description="GARC People Helper — Google Contacts & Directory")
|
|
parser.add_argument("--format", "-f", dest="format_", default="table", choices=["table", "json"])
|
|
subparsers = parser.add_subparsers(dest="command", required=True)
|
|
|
|
# search
|
|
sp = subparsers.add_parser("search", help="Search personal contacts")
|
|
sp.add_argument("query", nargs="+")
|
|
sp.add_argument("--max", type=int, default=20)
|
|
|
|
# directory
|
|
dp = subparsers.add_parser("directory", help="Search GWS directory (org-wide)")
|
|
dp.add_argument("query", nargs="+")
|
|
dp.add_argument("--max", type=int, default=20)
|
|
|
|
# list
|
|
lp = subparsers.add_parser("list", help="List all personal contacts")
|
|
lp.add_argument("--max", type=int, default=50)
|
|
|
|
# show
|
|
shp = subparsers.add_parser("show", help="Show a contact by ID")
|
|
shp.add_argument("contact_id")
|
|
|
|
# create
|
|
cp = subparsers.add_parser("create", help="Create a new contact")
|
|
cp.add_argument("--name", required=True)
|
|
cp.add_argument("--email")
|
|
cp.add_argument("--phone")
|
|
cp.add_argument("--company")
|
|
cp.add_argument("--title")
|
|
cp.add_argument("--notes")
|
|
|
|
# update
|
|
up = subparsers.add_parser("update", help="Update a contact")
|
|
up.add_argument("contact_id")
|
|
up.add_argument("--name")
|
|
up.add_argument("--email")
|
|
up.add_argument("--phone")
|
|
up.add_argument("--company")
|
|
up.add_argument("--title")
|
|
|
|
# delete
|
|
delp = subparsers.add_parser("delete", help="Delete a contact")
|
|
delp.add_argument("contact_id")
|
|
|
|
# lookup
|
|
look = subparsers.add_parser("lookup", help="Quick email lookup by name")
|
|
look.add_argument("query", nargs="+")
|
|
|
|
args = parser.parse_args()
|
|
|
|
try:
|
|
if args.command == "search":
|
|
search_contacts(" ".join(args.query), args.max, args.format_)
|
|
elif args.command == "directory":
|
|
search_directory(" ".join(args.query), args.max, args.format_)
|
|
elif args.command == "list":
|
|
list_contacts(args.max, args.format_)
|
|
elif args.command == "show":
|
|
show_contact(args.contact_id)
|
|
elif args.command == "create":
|
|
create_contact(args.name, args.email, args.phone, args.company, args.title, args.notes)
|
|
elif args.command == "update":
|
|
update_contact(args.contact_id, args.name, args.email, args.phone, args.company, args.title)
|
|
elif args.command == "delete":
|
|
delete_contact(args.contact_id)
|
|
elif args.command == "lookup":
|
|
lookup_email(" ".join(args.query))
|
|
except KeyboardInterrupt:
|
|
sys.exit(0)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|