diff --git a/docs/v3/guides/gmail.mdx b/docs/v3/guides/gmail.mdx index 4129c641..ce5d82fb 100644 --- a/docs/v3/guides/gmail.mdx +++ b/docs/v3/guides/gmail.mdx @@ -1,47 +1,23 @@ --- -title: "Gmail Tutorial" +title: "Gmail" icon: 'envelope' description: "Load Gmail threads into Honcho to give your AI agents memory of email conversations." sidebarTitle: 'Gmail' --- -> Example code is available on [GitHub](https://github.com/plastic-labs/honcho/tree/main/examples/gmail) - -This tutorial shows how to ingest Gmail threads into Honcho using a Python script. -Each email thread becomes a session, and each participant becomes a peer — enabling Honcho to reason about email history and relationships. -It is tailored to the Gmail workspace, but the Honcho primitives are easily extended to other email/messaging platforms. +In this tutorial, we'll walk through how to ingest your Gmail emails into Honcho. By the end, each email thread will be a Honcho session and each participant will be a peer — giving your agents memory of who said what across your email history. +This guide includes a ready-to-run Python script that handles everything: Gmail OAuth, thread fetching, participant extraction, and Honcho ingestion. You can run it as-is or use the full tutorial below to understand each piece as you go. -**This is a developer-focused tutorial.** It requires creating a Google Cloud project and OAuth credentials. +The full script is available on [GitHub](https://github.com/plastic-labs/honcho/tree/main/examples/gmail). This is a developer-focused tutorial — it requires creating a Google Cloud project and OAuth credentials. -## Prerequisites +## TL;DR -- Python 3.10+ -- A [Honcho API key](https://app.honcho.dev) -- A Google account with Gmail -- Basic familiarity with [Honcho's architecture](/v3/documentation/core-concepts/architecture) (workspaces, sessions, peers, messages) +If you just want to get your emails into Honcho, here's everything you need. -## How It Works - -Key patterns: -- **Thread → Session**: Each Gmail thread maps to a Honcho session with metadata -- **Email → Peer**: Each participant is a peer, identified by their email address -- **Message Attribution**: `peer.message()` preserves who said what -- **Multi-Peer Sessions**: `session.add_peers()` links all participants to the conversation -- **Timestamp Preservation**: `created_at` maintains the original email chronology -- **Quoted Reply Stripping**: Only new content is stored per message, avoiding duplication - -The script: -1. **Authenticates** with Gmail using OAuth (opens a browser on first run) -2. **Fetches threads** based on your query/label filters -3. **Extracts participants** from each email (From, To, Cc, Bcc) -4. **Creates Honcho peers** for each unique email address -5. **Creates a session** per thread with all participants attached -6. **Stores messages** with sender attribution and timestamps - -## Step 1: Set Up Google Cloud Credentials +### 1. Set Up Google Cloud Credentials Follow Google's official [Gmail API Python Quickstart](https://developers.google.com/gmail/api/quickstart/python) to: @@ -50,11 +26,9 @@ Follow Google's official [Gmail API Python Quickstart](https://developers.google 3. Create OAuth credentials (select **Desktop app** as the application type) 4. Download the credentials JSON into the same directory as the script -The script auto-detects Google's default `client_secret_*.json` filename, so no renaming needed. You can also pass a custom path with `--credentials`. +The script auto-detects Google's default `client_secret_*.json` filename, so no renaming needed. The script only needs the `gmail.readonly` scope. -The script only needs the `gmail.readonly` scope. - -## Step 2: Install Dependencies +### 2. Install Dependencies ```bash uv @@ -66,13 +40,7 @@ pip install google-api-python-client google-auth-oauthlib honcho-ai ``` -## Step 3: Run the Script - -Make sure your `client_secret_*.json` (from Step 1) is in the same directory as the script. - -### Dry Run (Preview) - -First, test without writing to Honcho: +### 3. Preview with a Dry Run ```bash uv @@ -84,36 +52,9 @@ python honcho_gmail.py --dry-run --max-threads 5 ``` -On first run, a browser window opens for OAuth consent. After authorizing, a `token.json` file is created — future runs won't require browser interaction. +On first run, a browser window opens for OAuth consent. After authorizing, a `token.json` file is created — future runs skip this step. -Example output: - -``` -Authenticating with Gmail API... -Opening browser for OAuth consent... -Credentials saved to token.json - Authenticated successfully! - -Fetching up to 5 threads from Gmail... - Found 5 threads - Fetching thread 1/5: 18f2a3b4c5d6e7f8 - Fetching thread 2/5: 18f1b2c3d4e5f6a7 - ... - -Summary: - Threads: 5 - Messages: 12 - Unique participants: 8 - alice-example-com (Alice Smith ) - bob-company-co (Bob Jones ) - ... - -[DRY RUN] Would create the above in Honcho. -``` - -### Load into Honcho - -When ready, run without `--dry-run`: +### 4. Load into Honcho ```bash uv @@ -127,34 +68,137 @@ python honcho_gmail.py --workspace gmail-inbox --max-threads 20 ``` -### Filter by Query or Label - -Use Gmail search syntax to filter which threads to load: +You can filter threads with Gmail search syntax: ```bash uv -# Only emails from a specific sender uv run honcho_gmail.py --query "from:alice@example.com" - -# Only emails with a label uv run honcho_gmail.py --label INBOX - -# Combine filters uv run honcho_gmail.py --query "after:2024/01/01 has:attachment" --max-threads 50 ``` ```bash python -# Only emails from a specific sender python honcho_gmail.py --query "from:alice@example.com" - -# Only emails with a label python honcho_gmail.py --label INBOX - -# Combine filters python honcho_gmail.py --query "after:2024/01/01 has:attachment" --max-threads 50 ``` +That's it — your emails are now queryable in Honcho. Read on if you want to understand how the script works and the design decisions behind it. + +--- + +## Full Tutorial + +### How Gmail Maps to Honcho + +The core idea is straightforward: each Gmail thread becomes a Honcho session, and each email participant becomes a peer. Here's the full mapping: + +| Gmail Concept | Honcho Concept | Details | +|---------------|----------------|---------| +| Your Gmail account | Workspace (`gmail`) | One workspace for all email data | +| Email participant | Peer | Email address as ID for deduplication | +| Email thread | Session (`gmail-thread-{id}`) | One session per thread, all participants attached | +| Individual email | Message | Attributed to the sender with original timestamp | + +### Email as Peer ID + +The script normalizes email addresses into URL-safe peer IDs — `alice@example.com` becomes `alice-example-com`. This means the same person is automatically deduplicated across threads. If Alice emails you in 10 different threads, all of those conversations accumulate under a single peer. + +```python +def peer_id_from_email(email: str) -> str: + """Convert email to a valid Honcho peer ID.""" + return email.replace("@", "-").replace(".", "-") +``` + +This also means peers are consistent across data sources. If you import Granola meetings and Gmail threads for the same person, they merge under the same peer ID. + +### Extracting Participants + +Every email has a sender, recipients, and optionally CC/BCC addresses. The script extracts all of these to build a complete picture of who's involved in each thread: + +```python +for m in msgs: + register_peer(m["from"]) + for addr in parse_address_list(m["to"]): + register_peer(addr) + for addr in parse_address_list(m["cc"]): + register_peer(addr) + for addr in parse_address_list(m["bcc"]): + register_peer(addr) +``` + +Display names are extracted when available (e.g., `Alice Smith ` → name: "Alice Smith"). When only an email is present, the script generates a name from the local part. + +### Message Attribution and Timestamps + +Each email becomes a message attributed to its sender via `peer.message()`. The original email timestamp is preserved using `created_at`, so Honcho sees the conversation in chronological order — not the order you imported it. + +```python +honcho_msgs.append(peer.message( + content, + metadata={ + "gmail_id": m["id"], + "subject": m["subject"], + "from": m["from"], + "to": m["to"], + "labels": m["labels"], + }, + created_at=m["timestamp"], +)) +``` + +### Multi-Peer Sessions + +Each thread's session is linked to all participants using `session.add_peers()`. This means when you query Honcho about a peer, it has context not just from their messages but from the full conversations they participated in. + +```python +session = honcho.session(session_id, metadata={ + "gmail_thread_id": tid, + "subject": subject, + "source": "gmail", + "message_count": len(msgs), +}) +session.add_peers(thread_peers) +``` + +### Stripping Quoted Replies + +Email threads are full of quoted replies — each message repeats everything above it. The script strips these out so only the new content is stored per message, avoiding duplication in Honcho's memory: + +```python +def strip_quoted_replies(text: str) -> str: + """Strip quoted reply text, keeping only the new content.""" + lines = text.split("\n") + clean_lines = [] + for line in lines: + stripped = line.strip() + if re.match(r"^On .+wrote:\s*$", stripped): + break + if stripped.startswith(">"): + break + # ... other reply markers + clean_lines.append(line) + return "\n".join(clean_lines).rstrip() +``` + +### Querying After Import + +Once your emails are in Honcho, you can query any peer: + +```python +import os +from honcho import Honcho + +honcho = Honcho(workspace_id="gmail-inbox", api_key=os.environ["HONCHO_API_KEY"]) + +alice = honcho.peer("alice-example-com") +print(alice.chat("What has Alice been discussing with me?")) +print(alice.chat("What action items has Alice mentioned?")) +``` + +--- + ## CLI Reference ``` @@ -176,7 +220,7 @@ options: ### "No client_secret*.json file found" -Download OAuth credentials from Google Cloud Console (Step 1) and place the `client_secret_*.json` file in the same directory as the script. +Download OAuth credentials from Google Cloud Console and place the `client_secret_*.json` file in the same directory as the script. ### "Access blocked: This app's request is invalid" @@ -555,3 +599,13 @@ if __name__ == "__main__": main() ``` +## Next Steps + + + + See how the Granola integration maps to common Honcho patterns. + + + Source code and example script. + + diff --git a/docs/v3/guides/granola.mdx b/docs/v3/guides/granola.mdx index 14082849..97bbec59 100644 --- a/docs/v3/guides/granola.mdx +++ b/docs/v3/guides/granola.mdx @@ -5,26 +5,44 @@ description: "Import meeting notes and transcripts from Granola into Honcho" sidebarTitle: 'Granola' --- -Import your [Granola](https://granola.ai) meeting data into Honcho to build queryable representations of the people you meet with. The transfer script handles participants, transcripts, and summaries — mapping them onto Honcho's peer and session model. +In this tutorial, we'll walk through how to import your [Granola](https://granola.ai) meeting data into Honcho. By the end, your meeting participants, transcripts, and summaries will be mapped onto Honcho's peer and session model — giving your agents queryable memory of the people you meet with. + +This guide includes a ready-to-run Python script that handles everything: Granola OAuth, meeting fetching, participant detection, and interactive import. You can run it as-is or use the full tutorial below to understand each design decision. -The full code is available on [GitHub](https://github.com/plastic-labs/honcho/tree/main/examples/granola). +The full script is available on [GitHub](https://github.com/plastic-labs/honcho/tree/main/examples/granola). -## Quick Start +## TL;DR + +If you just want to get your meetings into Honcho, here's everything you need. + +### 1. Install Dependencies ```bash uv uv pip install honcho-ai httpx -export HONCHO_API_KEY="your-key-from-app.honcho.dev" - -uv run python honcho_granola.py ``` ```bash pip pip install honcho-ai httpx -export HONCHO_API_KEY="your-key-from-app.honcho.dev" +``` + +### 2. Set Your API Key + +```bash +export HONCHO_API_KEY="your-key-from-app.honcho.dev" +``` + +### 3. Run the Script + + +```bash uv +uv run python honcho_granola.py +``` + +```bash python python honcho_granola.py ``` @@ -35,7 +53,15 @@ The script will: 3. Walk you through each meeting interactively — confirm peers, choose import mode, skip meetings you don't want 4. Print a summary of what was transferred -## Honcho Mapping +That's it — your meetings are now queryable in Honcho. Read on if you want to understand how the script works and the design decisions behind it. + +--- + +## Full Tutorial + +### How Granola Maps to Honcho + +The core idea is straightforward: each Granola meeting becomes a Honcho session, and each participant becomes a peer. Here's the full mapping: | Granola Concept | Honcho Concept | Details | |-----------------|----------------|---------| @@ -45,9 +71,9 @@ The script will: | Transcript turns | Messages with attribution | Two-person calls get full speaker attribution | | Meeting summary | Message from note creator | Multi-person calls store the summary | -## Key Design Decisions +### Email as Peer ID -**Email as Peer ID.** The script uses email addresses as the basis for peer IDs, normalized to a URL-safe format (e.g., `alice@example.com` becomes `alice-example-com`). This ensures consistent identification across meetings — if you meet someone in 5 different calls, all conversations accumulate under the same peer. +The script uses email addresses as the basis for peer IDs, normalized to a URL-safe format (e.g., `alice@example.com` becomes `alice-example-com`). This ensures consistent identification across meetings — if you meet someone in 5 different calls, all conversations accumulate under the same peer. ```python # These all resolve to the same peer: @@ -55,14 +81,20 @@ honcho.peer("alice-example-com") # From Meeting A honcho.peer("alice-example-com") # From Meeting B ``` -**Auto-Detecting "Me".** Granola marks the note creator in its participant list with `(note creator)`. The script uses this to identify you automatically. +This also means peers are consistent across data sources. If you import both Granola meetings and Gmail threads for the same person, they merge under the same peer ID. + +### Auto-Detecting "Me" + +Granola marks the note creator in its participant list with `(note creator)`. The script uses this to identify you automatically — no configuration needed. ``` Participants: You (note creator) from Your Company , Alice from Acme Corp ``` -**Two-Person Calls: Full Attribution.** When exactly one other participant is present *and* the transcript contains `Them:` turns, the transcript is stored with speaker-attributed messages. Consecutive same-speaker turns are merged before storing. +### Two-Person Calls: Full Attribution + +When exactly one other participant is present *and* the transcript contains `Them:` turns, the script stores the transcript with speaker-attributed messages. Consecutive same-speaker turns are merged before storing, cleaning up the fragmentation that's common in raw transcripts. ```python session.add_messages([ @@ -71,7 +103,9 @@ session.add_messages([ ]) ``` -**Multi-Person Calls: Summary Mode.** Granola's transcript uses `Them:` for all non-creator speakers with no disambiguation — in a 4-person call, everyone else is just `Them:`. Rather than guess incorrectly, the script stores Granola's summary as your record of the meeting, with participants in metadata. +### Multi-Person Calls: Summary Mode + +Granola's transcript uses `Them:` for all non-creator speakers with no disambiguation — in a 4-person call, everyone else is just `Them:`. Rather than guess incorrectly, the script stores Granola's summary as your record of the meeting, with participants in metadata. ```python session.add_messages([ @@ -91,11 +125,17 @@ session.add_messages([ The summary is attributed to you because it's *your* record of what happened. Granola captured your notes from a meeting where those people were present. -**Interactive Confirmation.** For each meeting, you choose the import mode: two-person (full attribution), summary, or skip. For multi-person calls that are actually 1:1s (extra participants listed but didn't speak), you can override the detection and select the actual speaker. +### Interactive Confirmation -**Noisy Transcripts Preserved.** Granola's raw transcripts are often fragmented (`Me: Yeah. Them: Yeah. Me: And.`). The script merges consecutive same-speaker turns but otherwise preserves the raw content. Honcho's reasoning extracts signal from noisy data. +For each meeting, you choose the import mode: two-person (full attribution), summary, or skip. For multi-person calls that are actually 1:1s (extra participants listed but didn't speak), you can override the detection and select the actual speaker. -## Querying After Import +### Noisy Transcripts Preserved + +Granola's raw transcripts are often fragmented (`Me: Yeah. Them: Yeah. Me: And.`). The script merges consecutive same-speaker turns but otherwise preserves the raw content. Honcho's reasoning extracts signal from noisy data. + +### Querying After Import + +Once your meetings are in Honcho, you can query any peer: ```python import os @@ -112,7 +152,7 @@ me = honcho.peer("you-example-com") print(me.chat("What topics do I discuss most frequently?")) ``` -## Combining with Other Sources +### Combining with Other Sources Because meetings live in a standard Honcho workspace, you can enrich peer representations with data from other channels: @@ -131,6 +171,8 @@ discord_session.add_messages([ alice.chat("What has Alice shipped recently?") ``` +--- + ## Troubleshooting | Issue | Fix | @@ -140,6 +182,765 @@ alice.chat("What has Alice shipped recently?") | 500 errors from Honcho | Check for null bytes or control characters in transcript content. The script sanitizes these automatically. | | Rate limiting with many meetings | The script processes sequentially with delays. Honcho ingestion is async — don't poll for immediate results. | +## Full Script + + +```python +#!/usr/bin/env python3 +"""Load Granola meeting notes into Honcho. + +Uses the Granola MCP server (with OAuth) to fetch meetings and the Honcho Python SDK +to store them. Each meeting becomes a Honcho session. Two-person meetings get full +speaker attribution; multi-person meetings are stored as summaries. + +Prerequisites: + pip install honcho-ai httpx + +Environment Variables: + HONCHO_API_KEY - Your Honcho API key (get from app.honcho.dev/api-keys) + +Usage: + python honcho_granola.py +""" + +import asyncio +import base64 +import hashlib +import json +import os +import re +import secrets +import sys +import threading +import traceback +import webbrowser +from dataclasses import dataclass, field +from datetime import datetime, timezone +from http.server import HTTPServer, BaseHTTPRequestHandler +from typing import Any +from urllib.parse import parse_qs, urlencode, urlparse + +import httpx + + +@dataclass +class Participant: + name: str + email: str | None = None + org: str | None = None + + +@dataclass +class ParsedParticipants: + note_creator: Participant | None = None + others: list[Participant] = field(default_factory=list) + + +@dataclass +class TranscriptTurn: + speaker: str + text: str + + +# Granola MCP + OAuth endpoints +GRANOLA_MCP_URL = "https://mcp.granola.ai/mcp" +AUTH_BASE = "https://mcp-auth.granola.ai" +OAUTH_REDIRECT_PORT = 8765 +OAUTH_REDIRECT_URI = f"http://localhost:{OAUTH_REDIRECT_PORT}/callback" + +# Honcho message size limit (25000 max, leave headroom) +MAX_MESSAGE_LEN = 24000 + + +# --------------------------------------------------------------------------- +# OAuth callback handler (must be a class for BaseHTTPRequestHandler) +# --------------------------------------------------------------------------- + +class _OAuthCallback(BaseHTTPRequestHandler): + auth_result: dict[str, str | None] = {"code": None, "error": None} + + def do_GET(self): + params = parse_qs(urlparse(self.path).query) + if "code" in params: + _OAuthCallback.auth_result["code"] = params["code"][0] + self.send_response(200) + self.send_header("Content-Type", "text/html") + self.end_headers() + self.wfile.write(b"

Authenticated! You can close this window.

") + elif "error" in params: + _OAuthCallback.auth_result["error"] = params.get("error_description", params["error"])[0] + self.send_response(400) + self.send_header("Content-Type", "text/html") + self.end_headers() + self.wfile.write(f"

Error: {_OAuthCallback.auth_result['error']}

".encode()) + else: + self.send_response(404) + self.end_headers() + + def log_message(self, fmt, *args): + pass + + +# --------------------------------------------------------------------------- +# Granola OAuth + MCP +# --------------------------------------------------------------------------- + +async def authenticate(http_client: httpx.AsyncClient) -> str: + """Perform OAuth (DCR + PKCE) with Granola. Returns access token.""" + _OAuthCallback.auth_result = {"code": None, "error": None} + + print("\nAuthenticating with Granola...") + + # Register client (DCR) + resp = await http_client.post( + f"{AUTH_BASE}/oauth2/register", + json={ + "client_name": "Granola to Honcho Transfer", + "redirect_uris": [OAUTH_REDIRECT_URI], + "grant_types": ["authorization_code"], + "response_types": ["code"], + "token_endpoint_auth_method": "none", + }, + ) + if resp.status_code not in (200, 201): + raise RuntimeError(f"Client registration failed: {resp.status_code}") + client_id = resp.json().get("client_id") + + # PKCE + verifier = secrets.token_urlsafe(32) + challenge = base64.urlsafe_b64encode(hashlib.sha256(verifier.encode()).digest()).rstrip(b"=").decode() + + # Browser auth + auth_url = f"{AUTH_BASE}/oauth2/authorize?" + urlencode({ + "client_id": client_id, + "redirect_uri": OAUTH_REDIRECT_URI, + "response_type": "code", + "state": "granola-honcho-transfer", + "code_challenge": challenge, + "code_challenge_method": "S256", + }) + + server = HTTPServer(("localhost", OAUTH_REDIRECT_PORT), _OAuthCallback) + thread = threading.Thread(target=server.handle_request) + thread.start() + + print(" Opening browser for authentication...") + webbrowser.open(auth_url) + thread.join(timeout=120) + server.server_close() + + auth_result = _OAuthCallback.auth_result + if auth_result["error"]: + raise RuntimeError(f"Authentication failed: {auth_result['error']}") + if not auth_result["code"]: + raise RuntimeError("Authentication timed out") + + # Exchange code for token + resp = await http_client.post( + f"{AUTH_BASE}/oauth2/token", + data={ + "grant_type": "authorization_code", + "code": auth_result["code"], + "redirect_uri": OAUTH_REDIRECT_URI, + "client_id": client_id, + "code_verifier": verifier, + }, + headers={"Content-Type": "application/x-www-form-urlencoded"}, + ) + if resp.status_code != 200: + raise RuntimeError(f"Token exchange failed: {resp.status_code}") + + print(" Authenticated successfully!") + return resp.json()["access_token"] + + +async def call_mcp_tool( + http_client: httpx.AsyncClient, + access_token: str, + tool_name: str, + arguments: dict[str, Any] | None = None, +) -> dict[str, Any]: + """Call a Granola MCP tool, handling both JSON and SSE responses.""" + resp = await http_client.post( + GRANOLA_MCP_URL, + json={ + "jsonrpc": "2.0", + "id": 1, + "method": "tools/call", + "params": {"name": tool_name, "arguments": arguments or {}}, + }, + headers={ + "Authorization": f"Bearer {access_token}", + "Content-Type": "application/json", + "Accept": "application/json, text/event-stream", + }, + ) + if resp.status_code != 200: + raise RuntimeError(f"MCP call failed: {resp.status_code} - {resp.text}") + + # SSE response + if "text/event-stream" in resp.headers.get("content-type", ""): + result = None + for line in resp.text.split("\n"): + if line.strip().startswith("data: "): + try: + parsed = json.loads(line.strip()[6:]) + if "result" in parsed: + result = parsed + elif "error" in parsed: + raise RuntimeError(f"MCP error: {parsed['error']}") + except json.JSONDecodeError: + continue + if result: + final = result.get("result", {}) + return final if isinstance(final, dict) else {"result": final} + raise RuntimeError("No result in SSE response") + + # JSON response + result = resp.json() + if "error" in result: + raise RuntimeError(f"MCP error: {result['error']}") + return result.get("result", {}) + + +def extract_mcp_text(result: dict[str, Any]) -> str: + """Extract text from the first content block of an MCP result. + + Raises ValueError if the response structure is unexpected. + """ + content = result.get("content", []) + if not isinstance(content, list) or not content: + raise ValueError(f"MCP response missing content array: {list(result.keys())}") + first = content[0] + if not isinstance(first, dict) or "text" not in first: + raise ValueError(f"MCP content block missing 'text' field: {first}") + return str(first["text"]) + + +# --------------------------------------------------------------------------- +# Granola data fetching +# --------------------------------------------------------------------------- + +async def list_meetings( + http_client: httpx.AsyncClient, access_token: str, limit: int = 100, +) -> list[dict[str, Any]]: + """List meetings from Granola MCP. Parses Granola's XML-like response format.""" + result = await call_mcp_tool(http_client, access_token, "list_meetings", {"limit": limit}) + text = extract_mcp_text(result) + + meetings: list[dict[str, Any]] = [] + for match in re.finditer(r'", match.end()) + block = text[match.end():block_end] if block_end != -1 else "" + p_match = re.search(r"\s*(.*?)\s*", block, re.DOTALL) + meetings.append({ + "id": mid, + "title": title, + "date": date, + "participants": p_match.group(1).strip() if p_match else "", + }) + + return meetings + + +async def get_meeting_details( + http_client: httpx.AsyncClient, access_token: str, meeting_id: str, +) -> dict[str, Any]: + """Get full meeting details including notes.""" + result = await call_mcp_tool(http_client, access_token, "get_meetings", {"meeting_ids": [meeting_id]}) + text = extract_mcp_text(result) + return {"id": meeting_id, "raw_content": text} + + +async def get_meeting_transcript( + http_client: httpx.AsyncClient, access_token: str, meeting_id: str, + max_retries: int = 3, +) -> str | None: + """Get transcript for a meeting (paid tiers only). + + Retries on rate limit responses with exponential backoff. + """ + for attempt in range(max_retries): + try: + result = await call_mcp_tool(http_client, access_token, "get_meeting_transcript", {"meeting_id": meeting_id}) + text = extract_mcp_text(result) + except Exception as e: + print(f" Transcript unavailable: {e}") + return None + + if not text or "no transcript" in text.lower(): + return None + + # Granola returns rate limit errors as content text, not HTTP errors + if "rate limit" in text.lower(): + wait = 2 ** attempt * 3 # 3s, 6s, 12s + print(f" ⚠ Granola rate limit hit (attempt {attempt + 1}/{max_retries}), waiting {wait}s...") + await asyncio.sleep(wait) + continue + + return text + + print(f" ⚠ Transcript skipped after {max_retries} rate limit retries") + return None + + +async def fetch_all_meetings( + http_client: httpx.AsyncClient, access_token: str, +) -> list[dict[str, Any]]: + """Fetch meeting list and enrich each with transcript and details.""" + print("\nFetching meetings from Granola...") + meetings = await list_meetings(http_client, access_token, limit=500) + if not meetings: + print("No meetings found.") + return [] + print(f" Found {len(meetings)} meetings. Fetching content...\n") + + for i, m in enumerate(meetings, 1): + mid = m.get("id") + if not mid: + continue + + transcript = await get_meeting_transcript(http_client, access_token, mid) + if transcript: + m["transcript"] = transcript + + try: + m.update(await get_meeting_details(http_client, access_token, mid)) + except Exception as exc: + print(f" Failed to fetch details for {mid}: {exc}") + + has_t = "transcript" in m + has_s = bool(extract_summary(m)) + label = "transcript+summary" if has_t and has_s else "transcript only" if has_t else "summary only" if has_s else "basic only" + print(f" [{i}/{len(meetings)}] {label}: {m.get('title', 'Untitled')[:45]}") + await asyncio.sleep(1.5) # rate limit + + return meetings + + +# --------------------------------------------------------------------------- +# Parsing helpers +# --------------------------------------------------------------------------- + +def parse_participants(participants_str: str) -> ParsedParticipants: + """Parse Granola's participant string into structured participants. + + Warns on unparseable entries instead of silently dropping them. + """ + result = ParsedParticipants() + if not participants_str: + return result + + # Split on commas, but not inside angle brackets + entries, current, depth = [], [], 0 + for ch in participants_str: + if ch == "<": + depth += 1 + elif ch == ">": + depth = max(depth - 1, 0) + elif ch == "," and depth == 0: + entries.append("".join(current)) + current = [] + continue + current.append(ch) + if current: + entries.append("".join(current)) + + for entry in entries: + entry = entry.strip() + if not entry: + continue + + is_creator = "(note creator)" in entry + clean = entry.replace("(note creator)", "").strip() + + email_match = re.search(r"<([^>]+)>", clean) + email = email_match.group(1) if email_match else None + name = re.sub(r"\s*<[^>]+>", "", clean).strip() + + if not name: + print(f" Warning: could not parse participant entry: {entry!r}") + continue + + org = None + org_match = re.match(r"(.+?)\s+from\s+(.+)", name) + if org_match: + name, org = org_match.group(1).strip(), org_match.group(2).strip() + + person = Participant(name=name, email=email, org=org) + if is_creator: + result.note_creator = person + else: + result.others.append(person) + + return result + + +def parse_transcript_turns(raw: str) -> list[TranscriptTurn]: + """Split a Granola transcript into speaker turns.""" + # Unwrap JSON wrapper if present + try: + parsed = json.loads(raw) + if isinstance(parsed, dict) and "transcript" in parsed: + raw = str(parsed["transcript"]) + except (json.JSONDecodeError, TypeError): + pass + + parts = re.split(r"(?:^|\s{2,})(Me|Them):\s*", raw) + turns: list[TranscriptTurn] = [] + i = 1 + while i < len(parts) - 1: + text = parts[i + 1].strip() + if text: + turns.append(TranscriptTurn(speaker=parts[i], text=text)) + i += 2 + return turns + + +def extract_summary(meeting: dict[str, Any]) -> str: + """Extract best available summary text from meeting data.""" + candidates = [] + for key in ("summary", "notes", "note", "meeting_notes", "description"): + val = meeting.get(key) + if isinstance(val, str) and val.strip(): + candidates.append(val.strip()) + + raw = meeting.get("raw_content") + if isinstance(raw, str) and raw.strip(): + candidates.append(raw.strip()) + + for c in candidates: + for tag in ("summary", "notes"): + m = re.search(rf"<{tag}>\s*(.*?)\s*", c, re.DOTALL) + if m: + return m.group(1).strip() + + return candidates[0] if candidates else "" + + +def peer_id_from(value: str) -> str: + """Normalize a name or email into a Honcho-safe peer ID.""" + norm = re.sub(r"[^a-z0-9_-]+", "-", value.strip().lower()) + norm = re.sub(r"-{2,}", "-", norm).strip("-_") + return (norm or "peer")[:100] + + +def sanitize(text: str) -> str: + """Remove null bytes and control characters.""" + return re.sub(r"[\x00-\x08\x0b\x0c\x0e-\x1f\x7f]", "", text) + + +def parse_date(date_str: str) -> datetime: + """Parse Granola's date format into a timezone-aware datetime. + + Raises ValueError if the date string doesn't match any known format. + """ + for fmt in ["%b %d, %Y %I:%M %p", "%b %d, %Y %I:%M:%S %p", "%B %d, %Y %I:%M %p"]: + try: + return datetime.strptime(date_str, fmt).replace(tzinfo=timezone.utc) + except ValueError: + continue + raise ValueError(f"Unrecognized date format: {date_str!r}") + + +# --------------------------------------------------------------------------- +# Honcho import helpers +# --------------------------------------------------------------------------- + +def build_messages( + peer: Any, + content: str, + metadata: dict[str, object] | None, + created_at: datetime, +) -> list[Any]: + """Build chunked messages for a single peer, attaching metadata to the first chunk.""" + messages = [] + content = sanitize(content) + for start in range(0, len(content), MAX_MESSAGE_LEN): + chunk = content[start:start + MAX_MESSAGE_LEN] + msg_meta = metadata if start == 0 else None + messages.append(peer.message(chunk, metadata=msg_meta, created_at=created_at)) + return messages + + +def send_messages(session: Any, messages: list[Any]) -> None: + """Send messages to a session in batches of 100.""" + for batch_start in range(0, len(messages), 100): + session.add_messages(messages[batch_start:batch_start + 100]) + + +def import_two_person( + honcho: Any, + session: Any, + me_peer_id: str, + them_peer_id: str, + turns: list[TranscriptTurn], + metadata: dict[str, object], + created_at: datetime, +) -> None: + """Import a two-person meeting with speaker attribution.""" + me_peer = honcho.peer(me_peer_id) + them_peer = honcho.peer(them_peer_id) + + # Merge consecutive same-speaker turns + merged: list[TranscriptTurn] = [] + for t in turns: + if merged and merged[-1].speaker == t.speaker: + merged[-1].text += " " + t.text + else: + merged.append(TranscriptTurn(speaker=t.speaker, text=t.text)) + + messages: list[Any] = [] + for i, t in enumerate(merged): + peer = me_peer if t.speaker == "Me" else them_peer + msg_meta = metadata if i == 0 else None + messages.extend(build_messages(peer, t.text, msg_meta, created_at)) + + send_messages(session, messages) + print(f" -> Imported as 2-person ({me_peer_id} + {them_peer_id})") + + +def import_summary( + honcho: Any, + session: Any, + me_peer_id: str, + meeting: dict[str, Any], + metadata: dict[str, object], + created_at: datetime, +) -> None: + """Import a meeting as a summary message.""" + me_peer = honcho.peer(me_peer_id) + summary = extract_summary(meeting) + if not summary: + raw_t = meeting.get("transcript", "") + try: + parsed = json.loads(raw_t) + summary = str(parsed.get("transcript", "")) if isinstance(parsed, dict) else raw_t + except (json.JSONDecodeError, TypeError): + summary = raw_t + summary = summary or "No content available" + + title = meeting.get("title", "Untitled") + date = meeting.get("date", "") + header = f"Meeting: {title}\nDate: {date}\nParticipants: {meeting.get('participants', '')}\n\n" + + messages = build_messages(me_peer, header + summary, metadata, created_at) + send_messages(session, messages) + print(" -> Imported as summary") + + +def resolve_them_participant(others: list[Participant]) -> Participant | None: + """Ask user to pick which participant is 'Them' from a multi-person meeting.""" + for j, p in enumerate(others, 1): + email_str = f" <{p.email}>" if p.email else "" + print(f" {j}. {p.name}{email_str}") + idx_str = input(f" Who is 'Them'? [1-{len(others)}]: ").strip() + try: + return others[int(idx_str) - 1] + except (ValueError, IndexError): + print(" Invalid selection.") + return None + + +def review_meeting( + index: int, + total: int, + meeting: dict[str, Any], + participants: ParsedParticipants, + turns: list[TranscriptTurn], +) -> tuple[str, Participant | None]: + """Display meeting info and get user's import choice. + + Returns (mode, them_participant) where mode is one of: + - "two_person": import with speaker attribution using them_participant + - "summary": import as a single summary message + - "skip": skip this meeting + """ + title = meeting.get("title", "Untitled") + date = meeting.get("date", "") + creator = participants.note_creator + others = participants.others + + me_turns = sum(1 for t in turns if t.speaker == "Me") + them_turns = len(turns) - me_turns + total_words = sum(len(t.text.split()) for t in turns) + + print(f"\n{'─' * 60}") + print(f" [{index}/{total}] {title}") + print(f" Date: {date}") + if creator: + print(f" You: {creator.name} <{creator.email}>") + for j, p in enumerate(others, 1): + email_str = f" <{p.email}>" if p.email else "" + org_str = f" ({p.org})" if p.org else "" + print(f" {j}. {p.name}{email_str}{org_str}") + + has_transcript = bool(meeting.get("transcript")) + if turns: + print(f" Transcript: {me_turns} Me, {them_turns} Them, ~{total_words} words") + if them_turns == 0: + print(" ** No 'Them' turns — nobody else spoke **") + if total_words < 30: + print(" ** Very short — might be empty **") + elif has_transcript: + raw = meeting["transcript"] + print(f" Transcript: present ({len(raw)} chars) but could not parse speaker turns") + print(f" Preview: {raw[:200]!r}") + else: + print(f" Content: {'summary available' if extract_summary(meeting) else 'metadata only'}") + + # Two-person default: exactly one other participant with transcript + if len(others) == 1 and them_turns > 0: + them_label = others[0].name + (f" <{others[0].email}>" if others[0].email else "") + print(f"\n Detected: 2-person call (you + {them_label})") + choice = input(" [Enter] 2-person / [s]ummary / [k] skip: ").strip().lower() + while choice not in ("", "s", "k"): + choice = input(" [Enter] 2-person / [s]ummary / [k] skip: ").strip().lower() + if choice == "k": + return ("skip", None) + if choice == "s": + return ("summary", None) + return ("two_person", others[0]) + + # Multi-person with transcript + if len(others) > 1 and them_turns > 0: + print(f"\n {len(others)} participants") + choice = input(" [Enter] summary / [2] 2-person / [k] skip: ").strip().lower() + while choice not in ("", "2", "k"): + choice = input(" [Enter] summary / [2] 2-person / [k] skip: ").strip().lower() + if choice == "k": + return ("skip", None) + if choice == "2": + them = resolve_them_participant(others) + if them is None: + return ("summary", None) + return ("two_person", them) + return ("summary", None) + + # No transcript or no other speakers + choice = input(" [Enter] summary / [k] skip: ").strip().lower() + while choice not in ("", "k"): + choice = input(" [Enter] summary / [k] skip: ").strip().lower() + if choice == "k": + return ("skip", None) + return ("summary", None) + + +# --------------------------------------------------------------------------- +# Main +# --------------------------------------------------------------------------- + +async def main(): + print("=" * 60) + print(" Granola -> Honcho Meeting Notes Transfer") + print("=" * 60) + + if not os.environ.get("HONCHO_API_KEY"): + print("\nError: HONCHO_API_KEY not set.") + print(" Get your key at: https://app.honcho.dev/api-keys") + sys.exit(1) + + async with httpx.AsyncClient(timeout=60.0) as http_client: + try: + access_token = await authenticate(http_client) + meetings = await fetch_all_meetings(http_client, access_token) + if not meetings: + sys.exit(0) + + from honcho import Honcho + + honcho = Honcho(workspace_id="granola_test") + seen_peers: set[str] = set() + results = {"imported": 0, "skipped": 0, "failed": 0} + + print("\n" + "=" * 60) + print(" Review each meeting") + print("=" * 60) + + for i, m in enumerate(meetings, 1): + mid = m.get("id") + if not mid: + continue + + participants = parse_participants(m.get("participants", "")) + turns = parse_transcript_turns(m["transcript"]) if m.get("transcript") else [] + + mode, them = review_meeting(i, len(meetings), m, participants, turns) + + if mode == "skip": + print(" -> Skipped") + results["skipped"] += 1 + continue + + # Resolve creator peer + creator = participants.note_creator + me_source = (creator.email or creator.name) if creator else None + if not me_source: + print(" -> Skipped (no creator identifier)") + results["skipped"] += 1 + continue + + me_peer_id = peer_id_from(me_source) + if me_peer_id not in seen_peers: + print(f" New peer: {me_source} ({me_peer_id})") + seen_peers.add(me_peer_id) + + try: + created_at = parse_date(m.get("date", "")) + session = honcho.session(f"meeting-{mid}") + metadata: dict[str, object] = { + "title": m.get("title", "Untitled"), + "date": m.get("date", ""), + "granola_meeting_id": mid, + "mode": mode, + } + + if mode == "two_person" and them is not None: + them_source = them.email or them.name + them_peer_id = peer_id_from(them_source) + if them_peer_id not in seen_peers: + print(f" New peer: {them_source} ({them_peer_id})") + seen_peers.add(them_peer_id) + import_two_person(honcho, session, me_peer_id, them_peer_id, turns, metadata, created_at) + else: + import_summary(honcho, session, me_peer_id, m, metadata, created_at) + + results["imported"] += 1 + + except ValueError as e: + print(f" -> FAILED: {e}") + results["failed"] += 1 + except Exception as e: + print(f" -> FAILED: {e}") + traceback.print_exc() + results["failed"] += 1 + + # Done + print("\n" + "=" * 60) + print(" Transfer Complete!") + print("=" * 60) + print(f"\n Imported: {results['imported']}") + print(f" Skipped: {results['skipped']}") + print(f" Failed: {results['failed']}") + print(" Workspace: granola") + print(f" Peers: {sorted(seen_peers)}") + + except KeyboardInterrupt: + print("\n\nAborted.") + sys.exit(0) + except Exception as e: + print(f"\nTransfer failed: {e}") + traceback.print_exc() + sys.exit(1) + + +if __name__ == "__main__": + asyncio.run(main()) +``` + + + ## Next Steps