export KERNEL_API_KEY=your-key-here
pip install requests
python3 climb_to_ultimate_parent.py 7913528487 # by Kernel ID
python3 climb_to_ultimate_parent.py "Ben & Jerry's" --country "United States" # by name#!/usr/bin/env python3
"""Climb the Kernel ownership tree from any account to its ultimate parent.
Walks /v1/resolve-parent hop by hop until there is no parent left. The entity
it stops at is the verified apex of the tree; the chain holds every entity in
between. Takes a Kernel ID, or a company name (resolved first).
"""
import argparse
import os
import re
import sys
import time
import requests
BASE = os.environ.get("KERNEL_API_BASE_URL", "https://api.kernel.ai/rest")
MAX_DEPTH = 25 # safety cap; real trees converge well before this
KERN_RE = re.compile(r"^\d{6,12}$") # Kernel IDs are numeric
def run_job(path, body, headers):
"""Kernel jobs are async: submit, then poll until the job completes."""
job_id = requests.post(f"{BASE}/{path}", headers=headers, json=body).json()["id"]
delay = 2
while True:
time.sleep(delay)
data = requests.get(f"{BASE}/{path}/{job_id}", headers=headers).json()
if data["status"] == "failed":
sys.exit(f"{path} job failed: {data}")
if data["status"] == "completed":
return data.get("record", {})
delay = min(delay + 5, 30)
def name_of(node):
if not node:
return None
return node.get("trading_name") or node.get("legal_name") or node.get("kernel_id")
def climb(kernel_id, headers):
"""Walk parent links up until there is no parent left."""
chain, seen, cur = [], {str(kernel_id)}, str(kernel_id)
shortcut, warning = None, None
for _ in range(MAX_DEPTH):
rec = run_job("v1/resolve-parent", {"kernel_id": cur}, headers)
if shortcut is None:
shortcut = rec.get("top_parent") # the one-call answer, kept as a cross-check
parent = rec.get("parent")
if not parent: # no parent -> cur is the ultimate parent
break
pid = str(parent["kernel_id"])
if pid in seen: # cycle guard (fragmented data)
warning = "cycle detected"
break
seen.add(pid)
chain.append(parent)
cur = pid
else:
warning = "max depth reached"
return {"ultimate_parent_kernel_id": cur, "chain": chain,
"top_parent_shortcut": shortcut, "warning": warning}
def main():
ap = argparse.ArgumentParser(description="Climb the Kernel tree to the ultimate parent.")
ap.add_argument("query", help="A Kernel ID, or a company name to resolve first")
ap.add_argument("--website", help="Sharpens the match when resolving by name")
ap.add_argument("--country", help="Sharpens the match when resolving by name")
args = ap.parse_args()
key = os.environ.get("KERNEL_API_KEY")
if not key:
sys.exit("Set KERNEL_API_KEY first (see 'Setting up your API key')")
headers = {"x-api-key": key, "Content-Type": "application/json"}
if KERN_RE.match(args.query.strip()):
kid = args.query.strip()
else:
body = {k: v for k, v in {"legal_name": args.query, "website": args.website,
"country": args.country}.items() if v}
kid = run_job("v1/entity-resolution", body, headers).get("kernel_id")
if not kid:
sys.exit(f"Could not resolve '{args.query}' to a Kernel entity")
print(f"Resolved to kernel_id {kid}")
result = climb(kid, headers)
print(f"Chain (bottom -> top), starting from {kid}:")
for node in result["chain"]:
print(f" ^ {name_of(node)} ({node.get('entity_sub_category')}, "
f"{node.get('website') or '-'}, kernel_id {node['kernel_id']})")
apex = result["chain"][-1] if result["chain"] else None
print(f"Ultimate parent: {name_of(apex) or kid} "
f"(kernel_id {result['ultimate_parent_kernel_id']})")
shortcut = result["top_parent_shortcut"]
if shortcut and str(shortcut.get("kernel_id")) != result["ultimate_parent_kernel_id"]:
print(f"NOTE: top_parent shortcut said {name_of(shortcut)} - "
f"the climb went higher; trust the climb.")
if result["warning"]:
print(f"WARNING: {result['warning']}")
if __name__ == "__main__":
main()export KERNEL_API_KEY=your-key-here
pip install requests
python3 top_operating_parent.py 7913528487 # by Kernel ID
python3 top_operating_parent.py "Ben & Jerry's" --country "United States" # by name#!/usr/bin/env python3
"""Find the top operating parent for any account — the highest operating company
in its ownership tree, skipping holding companies and investment vehicles.
One /v1/resolve-parent call answers it; the walk up fills in the operating-only
chain and the non-operating entities skipped along the way. Takes a Kernel ID,
or a company name (resolved first).
"""
import argparse
import os
import re
import sys
import time
import requests
BASE = os.environ.get("KERNEL_API_BASE_URL", "https://api.kernel.ai/rest")
MAX_DEPTH = 25 # safety cap; real trees converge well before this
KERN_RE = re.compile(r"^\d{6,12}$") # Kernel IDs are numeric
NON_OPERATING = {"HoldCo/Investment", "Business Unit", "Academic Unit", "Establishment"}
def run_job(path, body, headers):
"""Kernel jobs are async: submit, then poll until the job completes."""
job_id = requests.post(f"{BASE}/{path}", headers=headers, json=body).json()["id"]
delay = 2
while True:
time.sleep(delay)
data = requests.get(f"{BASE}/{path}/{job_id}", headers=headers).json()
if data["status"] == "failed":
sys.exit(f"{path} job failed: {data}")
if data["status"] == "completed":
return data.get("record", {})
delay = min(delay + 5, 30)
def is_operating(node):
return bool(node) and node.get("entity_category") == "Company" \
and node.get("entity_sub_category") not in NON_OPERATING
def name_of(node):
if not node:
return None
return node.get("trading_name") or node.get("legal_name") or node.get("kernel_id")
def top_operating_parent(kernel_id, headers):
"""One call answers it; the walk up fills in the operating chain."""
kid = str(kernel_id)
root = run_job("v1/resolve-parent", {"kernel_id": kid}, headers)
top_op = root.get("top_operating_parent")
top_op_id = str(top_op["kernel_id"]) if top_op else None
if top_op_id in (None, kid): # the account IS the top of its operating tree
return {"kernel_id": kid, "is_own_top_operating": top_op_id == kid,
"top_operating_parent": top_op, "operating_chain": [], "skipped": [],
"ultimate_parent": root.get("top_parent")}
chain, skipped, seen, cur = [], [], {kid}, kid
for _ in range(MAX_DEPTH):
rec = root if cur == kid else run_job("v1/resolve-parent", {"kernel_id": cur}, headers)
parent = rec.get("parent")
if not parent:
break
pid = str(parent["kernel_id"])
if pid in seen: # cycle guard (fragmented data)
break
seen.add(pid)
(chain if is_operating(parent) else skipped).append(parent)
cur = pid
if pid == top_op_id: # reached the answer - stop climbing
break
if not chain or str(chain[-1]["kernel_id"]) != top_op_id:
chain.append(top_op) # make sure the apex is present and last
return {"kernel_id": kid, "is_own_top_operating": False,
"top_operating_parent": top_op, "operating_chain": chain,
"skipped": skipped, "ultimate_parent": root.get("top_parent")}
def main():
ap = argparse.ArgumentParser(description="Find the top operating parent for an account.")
ap.add_argument("query", help="A Kernel ID, or a company name to resolve first")
ap.add_argument("--website", help="Sharpens the match when resolving by name")
ap.add_argument("--country", help="Sharpens the match when resolving by name")
args = ap.parse_args()
key = os.environ.get("KERNEL_API_KEY")
if not key:
sys.exit("Set KERNEL_API_KEY first (see 'Setting up your API key')")
headers = {"x-api-key": key, "Content-Type": "application/json"}
if KERN_RE.match(args.query.strip()):
kid = args.query.strip()
else:
body = {k: v for k, v in {"legal_name": args.query, "website": args.website,
"country": args.country}.items() if v}
kid = run_job("v1/entity-resolution", body, headers).get("kernel_id")
if not kid:
sys.exit(f"Could not resolve '{args.query}' to a Kernel entity")
print(f"Resolved to kernel_id {kid}")
r = top_operating_parent(kid, headers)
if r["is_own_top_operating"]:
print(f"{kid} is its own top operating parent - it IS the top of its operating tree")
else:
top = r["top_operating_parent"]
print(f"Top operating parent: {name_of(top)} "
f"(kernel_id {top['kernel_id']}, {top.get('website') or '-'})")
print("Operating chain (bottom -> top):")
for node in r["operating_chain"]:
print(f" ^ {name_of(node)} "
f"({node.get('entity_sub_category')}, kernel_id {node['kernel_id']})")
if r["skipped"]:
print("Skipped (non-operating): "
+ ", ".join(f"{name_of(n)} [{n.get('entity_sub_category')}]"
for n in r["skipped"]))
up, top = r["ultimate_parent"], r["top_operating_parent"]
if up and top and str(up.get("kernel_id")) != str(top.get("kernel_id")):
print(f"(Ultimate parent above it: {name_of(up)} - non-operating apex)")
if __name__ == "__main__":
main()