RomanNum3ral
/
honcho
mirror of https://github.com/plastic-labs/honcho.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

docs: adding granola documentation and cleaning up the script

This commit is contained in:
ajspig 2026-03-09 19:35:33 -04:00
parent 8b9a009f53
commit 8d184ad6d4
3 changed files with 160 additions and 127 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
docs/docs.json
View File

@ -93,6 +93,7 @@
"pages": [
"v3/guides/integrations/claude-code",
"v3/guides/integrations/crewai",
"v3/guides/integrations/granola",
"v3/guides/integrations/langgraph",
"v3/guides/integrations/mcp",
"v3/guides/integrations/n8n",

142
docs/v3/guides/integrations/granola.mdx Normal file
View File

@ -0,0 +1,142 @@
---
title: "Granola"
icon: 'microphone'
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.
<Note>
The full code is available on [GitHub](https://github.com/plastic-labs/honcho/tree/main/examples/granola).
</Note>
## Quick Start
```bash
pip install honcho-ai httpx
export HONCHO_API_KEY="your-key-from-app.honcho.dev"
python honcho_granola.py
```
The script will:
1. Open your browser for Granola OAuth authentication
2. Fetch all meetings and their content
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
| Granola Concept | Honcho Concept | Details |
|-----------------|----------------|---------|
| Your Granola account | Workspace (`granola`) | One workspace for all meetings |
| Meeting participant | Peer | Email as ID for deduplication across meetings |
| Individual meeting | Session (`meeting-{id}`) | One session per meeting |
| 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.** The script uses email addresses as the basis for peer IDs, normalized to a URL-safe format (e.g., `vince@plasticlabs.ai` becomes `vince-plasticlabs-ai`). 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:
honcho.peer("vince-plasticlabs-ai") # From Meeting A
honcho.peer("vince-plasticlabs-ai") # 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.
```
Participants: Abigail (note creator) from Plasticlabs <abigail@plasticlabs.ai>,
Vince Trost from Plasticlabs <vince@plasticlabs.ai>
```
**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.
```python
session.add_messages([
me.message("What's your timeline for the launch?"),
them.message("We're targeting Q2, but it depends on the API integration."),
])
```
**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([
me.message(
f"Meeting: Product Planning\n"
f"Date: Mar 5, 2026 2:00 PM\n"
f"Participants: Vince Trost from Plasticlabs, Jordan from Acme Corp\n\n"
f"{meeting_summary}",
metadata={
"participants": "Vince Trost from Plasticlabs, Jordan from Acme Corp",
"mode": "summary",
"granola_meeting_id": meeting_id,
}
)
])
```
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.** When the script encounters a new participant, it prompts before creating the peer. For multi-person calls that are actually 1:1s (extra participants listed but didn't speak), you can override the detection.
**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
```python
from honcho import Honcho
honcho = Honcho(workspace_id="granola", api_key=os.environ["HONCHO_API_KEY"])
# Peer IDs are normalized from emails: vince@plasticlabs.ai -> vince-plasticlabs-ai
vince = honcho.peer("vince-plasticlabs-ai")
print(vince.chat("What is Vince working on?"))
print(vince.chat("What concerns has Vince raised?"))
me = honcho.peer("abigail-plasticlabs-ai")
print(me.chat("What topics do I discuss most frequently?"))
```
## Combining with Other Sources
Because meetings live in a standard Honcho workspace, you can enrich peer representations with data from other channels:
```python
# Same workspace, same peer — data accumulates
vince = honcho.peer("vince-plasticlabs-ai")
me = honcho.peer("abigail-plasticlabs-ai")
discord_session = honcho.session("discord-general-2024-03")
discord_session.add_messages([
vince.message("Just shipped the new API version!"),
me.message("Congrats! How's the migration guide coming?"),
])
# Queries now draw from both meeting transcripts AND Discord history
vince.chat("What has Vince shipped recently?")
```
## Troubleshooting
| Issue | Fix |
|-------|-----|
| Granola OAuth fails | Ensure you have a paid Granola plan (MCP requires Pro+). Clear cached token and retry. |
| Missing transcripts | Free tier has no transcript access. The script falls back to summary content. |
| 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. |
## Next Steps
<CardGroup cols={2}>
<Card title="Design Patterns" icon="cubes" href="/v3/documentation/core-concepts/design-patterns">
See how the Granola integration maps to common Honcho patterns.
</Card>
<Card title="GitHub Repository" icon="github" href="https://github.com/plastic-labs/honcho/tree/main/examples/granola">
Source code and example script.
</Card>
</CardGroup>

144
examples/granola/honcho_granola.py
View File

@ -20,10 +20,11 @@ import json
import os
import re
import sys
import traceback
import webbrowser
from datetime import datetime
from http.server import HTTPServer, BaseHTTPRequestHandler
from urllib.parse import parse_qs, urlparse
from urllib.parse import parse_qs, urlencode, urlparse
import threading
import httpx
from typing import Any, TypedDict, cast
@ -104,9 +105,6 @@ class OAuthCallbackHandler(BaseHTTPRequestHandler):
pass # Suppress HTTP request logging
TOKEN_CACHE_FILE = os.path.expanduser("~/.granola_token.json")
class GranolaMCPClient:
"""Client for interacting with Granola MCP server."""
@ -117,26 +115,6 @@ class GranolaMCPClient:
self.resource_metadata: dict[str, Any] = {}
self.client_id: str | None = None
self.pkce_verifier: str | None = None
self._load_cached_token()
def _load_cached_token(self):
"""Load a previously cached access token if it exists."""
try:
with open(TOKEN_CACHE_FILE) as f:
data = json.load(f)
self.access_token = data.get("access_token")
self.client_id = data.get("client_id")
except (FileNotFoundError, json.JSONDecodeError):
pass
def _save_cached_token(self):
"""Cache the access token to disk."""
with open(TOKEN_CACHE_FILE, "w") as f:
json.dump({
"access_token": self.access_token,
"client_id": self.client_id,
}, f)
os.chmod(TOKEN_CACHE_FILE, 0o600) # readable only by owner
def _generate_pkce(self) -> tuple[str, str]:
"""Generate PKCE code verifier and challenge."""
@ -172,25 +150,6 @@ class GranolaMCPClient:
except Exception as e:
print(f" Could not fetch PRM: {e}")
# Alternative: Make an unauthenticated request to MCP and parse WWW-Authenticate
try:
response = await self.http_client.post(
GRANOLA_MCP_URL,
json={"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {}},
headers={"Content-Type": "application/json"}
)
if response.status_code == 401:
www_auth = response.headers.get("WWW-Authenticate", "")
print(f" Got 401 with WWW-Authenticate header")
# Parse the header to extract auth server URL
# Format: Bearer realm="...", resource="...", scope="..."
if "resource=" in www_auth:
# Extract resource metadata URL
pass
except Exception as e:
print(f" Could not probe MCP endpoint: {e}")
return self.resource_metadata
async def discover_oauth_metadata(self) -> dict[str, Any]:
@ -204,7 +163,6 @@ class GranolaMCPClient:
if auth_servers:
auth_server_url = auth_servers[0] if isinstance(auth_servers[0], str) else auth_servers[0].get("issuer")
else:
# Use the auth server URL we discovered from the error message
auth_server_url = "https://mcp-auth.granola.ai"
# Fetch authorization server metadata
@ -223,7 +181,7 @@ class GranolaMCPClient:
except Exception:
continue
# Fallback based on what we learned from the error
# Fallback to known Granola auth endpoints
self.auth_metadata = {
"authorization_endpoint": "https://mcp-auth.granola.ai/oauth2/authorize",
"token_endpoint": "https://mcp-auth.granola.ai/oauth2/token",
@ -265,28 +223,8 @@ class GranolaMCPClient:
return {}
async def _test_cached_token(self) -> bool:
"""Test if the cached token is still valid."""
if not self.access_token:
return False
try:
# Try a lightweight MCP call
await self.call_mcp_tool("list_meetings", {"limit": 1})
return True
except Exception:
self.access_token = None
return False
async def authenticate(self) -> bool:
"""Perform OAuth authentication with Granola following MCP spec."""
# Try cached token first
if self.access_token:
print("\n🔐 Testing cached Granola token...")
if await self._test_cached_token():
print("✅ Cached token is valid!")
return True
print(" Cached token expired, re-authenticating...")
global auth_result
auth_result = {"code": None, "error": None}
@ -300,9 +238,8 @@ class GranolaMCPClient:
client_id = client_info.get("client_id") or self.client_id
if not client_id:
# The error showed a client_id was already assigned, extract it
print(" Using pre-registered client flow...")
client_id = "granola-transfer-client"
print("❌ No client_id obtained from DCR. Cannot authenticate.")
return False
# Step 3: Generate PKCE (required by OAuth 2.1 / MCP spec)
self.pkce_verifier, pkce_challenge = self._generate_pkce()
@ -314,8 +251,7 @@ class GranolaMCPClient:
if supported_scopes:
scope = " ".join(supported_scopes)
else:
# Don't specify scope - let Granola provide default scopes
# The error was "invalid_scope" so we shouldn't guess
# Don't specify scope — let Granola provide default scopes
scope = None
# Build authorization URL
@ -342,7 +278,6 @@ class GranolaMCPClient:
if resource_url:
auth_params["resource"] = resource_url
from urllib.parse import urlencode
query = urlencode(auth_params)
full_auth_url = f"{auth_url}?{query}"
@ -392,7 +327,6 @@ class GranolaMCPClient:
if response.status_code == 200:
token_response = response.json()
self.access_token = token_response.get("access_token")
self._save_cached_token()
print("✅ Successfully authenticated with Granola!")
return True
else:
@ -550,7 +484,6 @@ class GranolaMCPClient:
# Return the raw text — it likely contains the notes in XML/markup format
# This is still valuable content to store in Honcho
print(f" [DEBUG] get_meetings returned non-JSON (first 500 chars): {text[:500]}")
return {"id": meeting_id, "raw_content": text}
async def get_meeting_transcript(self, meeting_id: str) -> str | None:
@ -574,17 +507,11 @@ class GranolaMCPClient:
if transcript:
return str(transcript)
print(f" [DEBUG] Transcript result structure: {json.dumps(result, default=str)[:300]}")
return None
except Exception as e:
print(f" Transcript unavailable: {e}")
return None
async def query_meetings(self, query: str) -> dict[str, Any]:
"""Query meetings with natural language."""
result = await self.call_mcp_tool("query_granola_meetings", {"query": query})
return result
async def close(self):
"""Close the HTTP client."""
await self.http_client.aclose()
@ -672,13 +599,12 @@ def parse_transcript_turns(transcript: str) -> list[TranscriptTurn]:
def analyze_transcript(transcript: str) -> dict[str, Any]:
"""Return stats about a transcript."""
turns = parse_transcript_turns(transcript)
me_turns = [t for t in turns if t["speaker"] == "Me"]
them_turns = [t for t in turns if t["speaker"] == "Them"]
me_count = sum(1 for t in turns if t["speaker"] == "Me")
them_count = len(turns) - me_count
total_words = sum(len(t["text"].split()) for t in turns)
return {
"turns": turns,
"me_count": len(me_turns),
"them_count": len(them_turns),
"me_count": me_count,
"them_count": them_count,
"total_words": total_words,
}
@ -888,48 +814,8 @@ class HonchoClient:
msg_meta = metadata if start == 0 else None
messages.append(me_peer.message(chunk, metadata=msg_meta, created_at=created_at))
print(f" [DEBUG] store_summary: {len(messages)} message(s), first chunk len={len(messages[0].content) if messages else 0}")
for i in range(0, len(messages), 100):
batch = messages[i : i + 100]
try:
session.add_messages(batch)
except Exception:
# Dump debug info for the failing batch
for j, msg in enumerate(batch):
print(f" [DEBUG] msg[{j}]: peer_id={msg.peer_id!r}, content_len={len(msg.content)}, has_metadata={msg.metadata is not None}, created_at={msg.created_at!r}")
# Check for problematic chars
bad_chars: list[str] = []
for k, ch in enumerate(msg.content):
if ord(ch) == 0 or (ord(ch) < 32 and ch not in '\n\r\t'):
bad_chars.append(f"pos {k}: {ch!r} (ord={ord(ch)})")
if bad_chars:
print(f" [DEBUG] bad chars: {bad_chars[:10]}")
else:
print(f" [DEBUG] no bad chars found")
# Raw HTTP debug — bypass SDK to see actual server response
print(f" [DEBUG] Making raw HTTP request to see full error...")
try:
import httpx as _httpx
raw_client = _httpx.Client(timeout=30.0)
base = self.client.base_url
key = os.environ.get("HONCHO_API_KEY", "")
url = f"{base}/v3/workspaces/{self.workspace_id}/sessions/{session_id}/messages"
raw_body = {
"messages": [{"content": "test", "peer_id": batch[0].peer_id}]
}
raw_resp = raw_client.post(
url,
json=raw_body,
headers={
"Authorization": f"Bearer {key}",
"Content-Type": "application/json",
},
)
print(f" [DEBUG] URL: {url}")
print(f" [DEBUG] Raw response: {raw_resp.status_code} {raw_resp.text[:1000]}")
except Exception as e2:
print(f" [DEBUG] Raw HTTP debug failed: {e2}")
raise
session.add_messages(messages[i : i + 100])
return session_id
@ -1074,6 +960,12 @@ async def main():
" [Enter] import as summary / [2] actually 2-person / [k] skip: ",
["", "2", "k"], default=""
)
else:
# No transcript or no other participants — offer summary or skip
choice = prompt_choice(
" [Enter] import as summary / [k] skip: ",
["", "k"], default=""
)
if choice == "k":
print(" -> Skipped")
@ -1177,7 +1069,6 @@ async def main():
except Exception as e:
print(f" -> FAILED: {e}")
import traceback
traceback.print_exc()
results["failed"] += 1
@ -1196,7 +1087,6 @@ async def main():
sys.exit(0)
except Exception as e:
print(f"\nTransfer failed: {e}")
import traceback
traceback.print_exc()
sys.exit(1)
finally: