Guide
Visualize Communication Patterns Between Organizations
CRMs track what people entered. Your inbox tracks what actually happened. This guide aggregates email and calendar data across team accounts to map the real communication network between your organization and external companies. Identify key stakeholders, find warm introduction paths, and flag single-threaded risk. Works with Gmail, Outlook, Exchange, Yahoo, iCloud, and IMAP.
By Pouya Sanooei
What a communication network map reveals
According to Gartner’s 2024 B2B Buying Report, the average B2B deal involves 6-10 decision makers. A contact map built from actual communication data answers four questions no CRM can:
- Who actually talks to whom? Not what someone typed into a CRM field, but verified by email threads and calendar invites. If Alice exchanged 47 emails with Bob at Acme, that relationship exists whether or not it was logged.
- How engaged is each relationship? A contact you emailed once six months ago is different from one you meet weekly. The strength score captures active vs. dormant, one-way vs. reciprocal.
- Where are the single-threaded risks? Accounts where only one person on your team has a relationship churn at 64% higher rates (Gartner, 2024). Flag them before the person changes roles.
- Who can make a warm introduction? For any target company, see which team member already has the strongest path in.
Aggregate across team accounts
A map for one person is useful. A map across your team is transformative. Export email data from each team member's account and combine:
# Export each team member's recent email data
for grant in alice@company.com bob@company.com carol@company.com; do
nylas email list --grant-id "$grant" --json --limit 500 > "$(echo $grant | cut -d@ -f1)_emails.json"
nylas calendar events list --grant-id "$grant" --json --limit 200 > "$(echo $grant | cut -d@ -f1)_events.json"
done
# Combine and deduplicate by message ID
jq -s 'add | group_by(.id) | map(.[0])' *_emails.json > all_emails.json
jq -s 'add' *_events.json > all_events.json
echo "Combined: $(jq length all_emails.json) unique emails, $(jq length all_events.json) events"Score relationship strength
A binary “knows / doesn’t know” isn’t enough. The scoring formula weights four signals:
- Frequency (40%): Messages per month. 20+ emails/month = max score.
- Recency (25%): Days since last interaction. Full marks within 7 days, zero after 90.
- Reciprocity (20%): Ratio of sent to received. 1:1 is ideal. One-way outreach scores low.
- Meetings (15%): Shared calendar events. 5+ meetings = max score. Meetings count 3x because they represent deliberate time investment.
#!/usr/bin/env python3
"""Score inter-org relationship strength from email and calendar data."""
import json
import subprocess
from datetime import datetime, timezone
from collections import defaultdict
def load_json(path: str) -> list[dict]:
with open(path) as f:
return json.load(f)
FREEMAIL = {"gmail.com", "yahoo.com", "outlook.com", "hotmail.com", "icloud.com",
"aol.com", "protonmail.com", "live.com"}
def score_network(emails: list[dict], events: list[dict],
our_domain: str) -> list[dict]:
contacts: dict[str, dict] = defaultdict(lambda: {
"sent": 0, "received": 0, "meetings": 0,
"last_date": None, "name": "", "domain": ""
})
now = datetime.now(timezone.utc)
for msg in emails:
sender = msg["from"][0]["email"]
sender_domain = sender.split("@")[1]
recipients = [r["email"] for r in msg.get("to", [])]
msg_date = msg.get("date")
for addr in [sender] + recipients:
domain = addr.split("@")[1]
if domain == our_domain or domain in FREEMAIL:
continue
c = contacts[addr]
c["domain"] = domain
if sender_domain == our_domain:
c["sent"] += 1
else:
c["received"] += 1
c["name"] = c["name"] or msg["from"][0].get("name", "")
if msg_date:
c["last_date"] = msg_date
for event in events:
for p in event.get("participants", []):
addr = p.get("email", "")
if addr and addr.split("@")[1] not in (FREEMAIL | {our_domain}):
contacts[addr]["meetings"] += 1
scored = []
for email, d in contacts.items():
total = d["sent"] + d["received"]
if total == 0:
continue
freq = min(total / 20, 1.0) * 40
recency = 0
if d["last_date"]:
try:
last = datetime.fromisoformat(d["last_date"].replace("Z", "+00:00"))
days = (now - last).days
recency = max(0, 1 - days / 90) * 25
except (ValueError, TypeError):
pass
reciprocity = min(d["sent"], d["received"]) / max(d["sent"], d["received"], 1) * 20
meetings = min(d["meetings"] / 5, 1.0) * 15
strength = min(round(freq + recency + reciprocity + meetings), 100)
scored.append({
"email": email, "name": d["name"], "domain": d["domain"],
"strength": strength, "sent": d["sent"], "received": d["received"],
"meetings": d["meetings"]
})
return sorted(scored, key=lambda x: -x["strength"])
emails = load_json("all_emails.json")
events = load_json("all_events.json")
results = score_network(emails, events, "yourcompany.com")
with open("scored_network.json", "w") as f:
json.dump(results, f, indent=2)
print(f"Scored {len(results)} external contacts")Group by company for account-level insights
Individual contact scores are useful. Company-level aggregation is where the strategic insights live:
cat scored_network.json | jq '
[.[] | select(.domain | IN("gmail.com","yahoo.com","outlook.com") | not)] |
group_by(.domain) |
map({
company: .[0].domain,
contacts: length,
avg_strength: ([.[].strength] | add / length | round),
strongest_contact: (sort_by(-.strength) | .[0].email),
peak_strength: (sort_by(-.strength) | .[0].strength),
total_meetings: ([.[].meetings] | add),
people: [.[] | {email, name, strength}] | sort_by(-.strength)
}) |
sort_by(-.avg_strength)'Detect single-threaded accounts
Single-threaded risk is when only one person on your team has a relationship with a key account. If that person goes on leave, changes roles, or leaves entirely, you lose the entire relationship. Flag these accounts early.
# Find accounts where only 1 internal person has communicated
cat scored_network.json | jq '
[.[] | select(.strength >= 20)] |
group_by(.domain) |
map({
company: .[0].domain,
contacts: length,
risk: (if length == 1 then "SINGLE-THREADED"
elif length == 2 then "at-risk"
else "multi-threaded" end),
key_contact: (sort_by(-.strength) | .[0].email),
strength: (sort_by(-.strength) | .[0].strength)
}) |
[.[] | select(.risk == "SINGLE-THREADED")] |
sort_by(-.strength)'The threshold of strength >= 20 filters out contacts you barely know. Adjust based on your volume. For high-value accounts, even a strength of 15 should trigger the flag.
Find warm introduction paths
For any target company, find which team member has the strongest existing relationship and can make an introduction:
#!/usr/bin/env python3
"""Find warm introduction paths to a target company."""
import json
import sys
def find_paths(scored_file: str, target_domain: str):
with open(scored_file) as f:
contacts = json.load(f)
matches = [c for c in contacts if c["domain"] == target_domain]
if not matches:
print(f"No existing contacts at {target_domain}")
return
print(f"\nWarm paths to {target_domain}:")
print(f"{'Contact':<35} {'Strength':>8} {'Emails':>7} {'Meetings':>8}")
print("-" * 65)
for c in sorted(matches, key=lambda x: -x["strength"]):
total = c["sent"] + c["received"]
print(f"{c['name'] or c['email']:<35} {c['strength']:>8} "
f"{total:>7} {c['meetings']:>8}")
best = max(matches, key=lambda x: x["strength"])
if best["strength"] >= 50:
print(f"\nStrong path through {best['name'] or best['email']} "
f"(strength {best['strength']})")
elif best["strength"] >= 20:
print(f"\nWeak path: {best['name'] or best['email']} "
f"(strength {best['strength']}) — warm up before asking for intro")
else:
print(f"\nCold territory: strongest contact is only {best['strength']}/100")
if __name__ == "__main__":
find_paths("scored_network.json", sys.argv[1] if len(sys.argv) > 1 else "acme.com")Spot declining relationships
Relationships that were once active but have gone quiet may indicate churn risk. Find contacts with high historical email volume but low recent activity:
# Contacts with lots of history but low recent engagement
cat scored_network.json | jq '
[.[] | select(
(.sent + .received) > 10
and .strength < 25
)] |
group_by(.domain) |
map({
company: .[0].domain,
fading: length,
contacts: [.[] | {email, strength, emails: (.sent + .received)}]
}) |
sort_by(-.fading)'Export formats
Export the network map in the format your tools need:
# CSV for spreadsheets and CRM import
cat scored_network.json | jq -r '
["email","name","domain","strength","sent","received","meetings"],
(.[] | [.email, .name, .domain, .strength, .sent, .received, .meetings])
| @csv' > network_map.csv
# DOT graph for Graphviz visualization
cat scored_network.json | jq -r '
"digraph network {",
" rankdir=LR;",
" node [shape=box, style=filled, fillcolor=lightblue];",
(.[] | select(.strength >= 30) |
" \"you\" -> \"\(.name // .email)\" [label=\"\(.strength)\", penwidth=\((.strength / 25) | tostring)];"),
"}"' > network.dot
# Render: dot -Tpng network.dot -o network.pngNext steps
- Reconstruct org charts from CC patterns — identify decision-makers within the companies you’ve mapped
- Import email into a graph database — query introduction paths and communication clusters with Cypher
- Personalize outbound email — use relationship scores to tailor messaging per contact
- Command reference — every flag, subcommand, and example