| """ |
| LinkedIn via the unofficial linkedin-api package, authenticated with an |
| li_at session cookie from a dedicated research account. Keep volume low |
| (<100 lookups/day) and refresh the cookie monthly. |
| """ |
| from linkedin_api import Linkedin |
| from supabase_client import require_tool_keys |
|
|
|
|
| def _get_client() -> Linkedin: |
| creds = require_tool_keys("linkedin") |
| li_at = creds["api_key"] |
| return Linkedin("", "", cookies={"li_at": li_at}) |
|
|
|
|
| def run(params: dict) -> dict: |
| mode = params.get("mode") |
| target = params.get("target") |
| if mode not in ("profile", "people_search", "company") or not target: |
| raise ValueError("mode must be 'profile' | 'people_search' | 'company', target is required") |
| limit = int(params.get("limit", 25)) |
|
|
| api = _get_client() |
|
|
| if mode == "profile": |
| profile = api.get_profile(target) |
| return { |
| "name": f"{profile.get('firstName', '')} {profile.get('lastName', '')}".strip(), |
| "headline": profile.get("headline"), |
| "location": profile.get("locationName"), |
| "current_company": (profile.get("experience") or [{}])[0].get("companyName"), |
| } |
|
|
| if mode == "people_search": |
| people = api.search_people(keywords=target, network_depths=["F", "S"]) |
| return {"count": len(people[:limit]), "results": people[:limit]} |
|
|
| company = api.get_company(target) |
| return { |
| "name": company.get("name"), |
| "employee_count": company.get("staffCount"), |
| "industry": (company.get("companyIndustries") or [{}])[0].get("localizedName"), |
| "follower_count": company.get("followingInfo", {}).get("followerCount"), |
| } |
|
|
