Skip to content
Skip to main content

Map communication patterns between organizations

Most CRMs track what someone typed into a field. The actual relationships — who emails whom, how often, who shows up to meetings together — sit in your team’s mailboxes and calendars, untracked. This recipe scores each external contact across four signals (frequency, recency, reciprocity, meetings), rolls those scores up to organizations, and surfaces the two questions that matter most: which accounts are single-threaded (and therefore at churn risk), and who on your team is the warmest path to a given prospect.

Run a per-team-member pull from the Nylas CLI. The script below loops over connected grants, exports email and calendar to JSON, and dedupes:

#!/usr/bin/env bash
set -euo pipefail
OUT=./network-data
mkdir -p "$OUT"
for grant in $(nylas auth list --json | jq -r '.[].email'); do
nylas auth switch "$grant"
nylas email list --days 90 --limit 5000 --json > "$OUT/email-$grant.json"
nylas calendar events list --days 90 --json > "$OUT/cal-$grant.json"
done
# Dedupe and combine
jq -s 'add | unique_by(.id)' "$OUT"/email-*.json > "$OUT/email-all.json"
jq -s 'add | unique_by(.id)' "$OUT"/cal-*.json > "$OUT/cal-all.json"

Adjust --days 90 to your analysis window. Ninety days is the right default — long enough to smooth out vacation gaps, short enough that defunct relationships don’t pollute the score.

Four weighted signals produce a 0–100 score per external contact:

SignalWeightDefinition
Frequency40%Messages per month, normalized
Recency25%Time since last interaction (full credit ≤7d, linear decay to 0 at 90d)
Reciprocity20%min(sent, received) / max(sent, received) — balanced exchange scores higher
Meetings15%Shared calendar events; weighted heavier per event because a meeting is a deliberate time investment
def score(contact, msgs, events, now=datetime.utcnow()):
days_since_last = (now - contact["last_msg_at"]).days
sent = len([m for m in msgs if m["from"] == ME and contact["email"] in m["to"]])
recv = len([m for m in msgs if m["from"] == contact["email"]])
freq = min(40, ((sent + recv) / 3) * 4) # 30 msgs/3mo → 40 pts
rec = max(0, 25 * (1 - days_since_last / 90))
recip = 20 * (min(sent, recv) / max(sent, recv, 1))
meets = min(15, len(events) * 5) # 3+ meetings → 15 pts
return round(freq + rec + recip + meets)

That’s the whole formula. Tweak the weights to match your team’s reality (frontline sales might want frequency at 50%; account managers might want meetings at 25%).

Group contacts by email domain:

from collections import defaultdict
orgs = defaultdict(list)
for c in scored_contacts:
domain = c["email"].split("@")[1]
orgs[domain].append(c)
org_table = []
for domain, contacts in orgs.items():
org_table.append({
"company": domain,
"avg_score": sum(c["score"] for c in contacts) / len(contacts),
"max_score": max(c["score"] for c in contacts),
"contacts": len(contacts),
"team_owner": max(contacts, key=lambda c: c["score"])["team_owner"],
})

avg_score tells you overall relationship health with the company; max_score tells you who your best advocate there is.

A single-threaded account is one where exactly one person on your team has any meaningful relationship. Per published account-management research, single-threaded accounts churn at roughly 64% higher rates than accounts with two or more relationships. They’re the highest-priority retention work.

def single_threaded(orgs, threshold=40):
risky = []
for org in orgs:
strong_relationships = [c for c in org["contacts"] if c["score"] >= threshold]
if len(set(c["team_owner"] for c in strong_relationships)) == 1:
risky.append(org)
return risky

These show up in a weekly report. The action: introduce a second teammate to the existing strong contact, or to anyone scoring above 30 at the org.

For any target company you’re trying to break into, find the highest-scoring contact your team has there:

def warm_intro_path(target_domain: str):
contacts = orgs[target_domain]
if not contacts:
return None
best = max(contacts, key=lambda c: c["score"])
if best["score"] >= 50:
return ("strong intro", best)
elif best["score"] >= 20:
return ("warm but tepid — warm them up first", best)
return ("cold", best)

Score thresholds are heuristic — calibrate against your own team’s hit rates.

A contact with high historical email volume but low recent score is showing relationship decay:

def declining(scored_contacts):
return [c for c in scored_contacts
if c["historical_messages"] > 50 and c["recent_score"] < 30]

These are early warning signs of churn. Surface them to the account owner with a “haven’t heard from X in N days” note.

Three formats cover most downstream needs:

  • CSV — drop into a spreadsheet or upload to a CRM custom field
  • JSON — for any further programmatic processing
  • DOT (Graphviz) — for actual visualization
import csv, json
with open("relationships.csv", "w") as f:
w = csv.DictWriter(f, fieldnames=["email", "score", "company", "team_owner"])
w.writeheader()
w.writerows(scored_contacts)
with open("relationships.json", "w") as f:
json.dump(scored_contacts, f, indent=2)

For a graph view:

print("digraph G {")
for c in scored_contacts:
print(f' "me" -> "{c["email"]}" [weight={c["score"]}];')
print("}")

dot -Tpng relationships.dot -o relationships.png and you have a relationship map you can show the team.

  • Privacy. This script reads team mailboxes. Get explicit consent and document the analysis in your data-handling policy. The output should be access-controlled — relationship data is sensitive.
  • Calendar weighting. A 30-minute 1:1 and a 60-person all-hands shouldn’t count the same. Filter calendar events by attendee count (≤10 typically) before scoring.
  • Auto-replies. Out-of-office responders inflate recv counts. Filter them out before scoring.