sync: Replace Composio with Nango for LinkedIn integration

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

backend/main.py

@@ -108,7 +108,7 @@ from services import embedding_matcher
 from services import quality_gate
 from services import scoring_engine
 from services import latex_resume
-from services import linkedin_composio
+from services import linkedin_nango
 from core import auth as auth_module
 
 logger = logging.getLogger(__name__)
@@ -656,11 +656,11 @@ def linkedin_auth_url(
     user: dict = Depends(get_current_user),
 ):
     """Return a Composio OAuth URL for the user to connect their LinkedIn account."""
-    if not linkedin_composio.COMPOSIO_API_KEY:
-        raise HTTPException(status_code=503, detail="COMPOSIO_API_KEY not configured")
+    if not linkedin_nango.NANGO_SECRET_KEY:
+        raise HTTPException(status_code=503, detail="NANGO_SECRET_KEY not configured")
     entity_id = user.get("clerk_id") or user.get("id") or "default"
     try:
-        url = linkedin_composio.get_oauth_url(entity_id=entity_id, redirect_url=redirect)
+        url = linkedin_nango.get_connect_url(user_id=entity_id, redirect_url=redirect)
         return {"ok": True, "url": url}
     except Exception as e:
         logger.error("LinkedIn auth URL error: %s", e)
@@ -670,21 +670,21 @@ def linkedin_auth_url(
 @app.get("/api/linkedin/status")
 def linkedin_status(user: dict = Depends(get_current_user)):
     """Return whether the user has LinkedIn connected via Composio."""
-    if not linkedin_composio.COMPOSIO_API_KEY:
-        return {"ok": True, "connected": False, "reason": "COMPOSIO_API_KEY not configured"}
+    if not linkedin_nango.NANGO_SECRET_KEY:
+        return {"ok": True, "connected": False, "reason": "NANGO_SECRET_KEY not configured"}
     entity_id = user.get("clerk_id") or user.get("id") or "default"
-    connected = linkedin_composio.is_connected(entity_id)
+    connected = linkedin_nango.is_connected(entity_id)
     return {"ok": True, "connected": connected}
 
 
 @app.post("/api/linkedin/import")
 def linkedin_import(user: dict = Depends(get_current_user)):
     """Import user's LinkedIn profile as a parsed resume object."""
-    if not linkedin_composio.COMPOSIO_API_KEY:
-        raise HTTPException(status_code=503, detail="COMPOSIO_API_KEY not configured")
+    if not linkedin_nango.NANGO_SECRET_KEY:
+        raise HTTPException(status_code=503, detail="NANGO_SECRET_KEY not configured")
     entity_id = user.get("clerk_id") or user.get("id") or "default"
     try:
-        data = linkedin_composio.import_profile(entity_id)
+        data = linkedin_nango.import_profile(entity_id)
         return {"ok": True, "data": data}
     except Exception as e:
         logger.error("LinkedIn import error: %s", e)
@@ -694,10 +694,10 @@ def linkedin_import(user: dict = Depends(get_current_user)):
 @app.post("/api/linkedin/enrich-companies")
 def linkedin_enrich_companies(req: LinkedInEnrichRequest, user: dict = Depends(get_current_user)):
     """Enrich a list of job dicts with LinkedIn company_info. Skips silently on failure."""
-    if not linkedin_composio.COMPOSIO_API_KEY:
+    if not linkedin_nango.NANGO_SECRET_KEY:
         return {"ok": True, "jobs": req.jobs}
     entity_id = req.entity_id or user.get("clerk_id") or user.get("id") or "default"
-    enriched = linkedin_composio.enrich_companies(req.jobs, entity_id)
+    enriched = linkedin_nango.enrich_companies(req.jobs, entity_id)
     return {"ok": True, "jobs": enriched}
 
 

backend/requirements.txt

@@ -17,4 +17,3 @@ stripe>=8.0.0
 python-dotenv>=1.0.0
 # Deploy trigger v4.9.0
 python-jobspy>=1.1.80
-composio-core>=0.7.21

backend/services/linkedin_composio.py

@@ -1,230 +0,0 @@
-"""
-LinkedIn integration via Composio.
-Provides profile import and company info enrichment.
-
-Actions used:
-  LINKEDIN_GET_MY_INFO      — fetch authenticated user's profile
-  LINKEDIN_GET_COMPANY_INFO — fetch company details by name/vanity
-
-Requires:
-  COMPOSIO_API_KEY in environment
-  User must have connected their LinkedIn account via /api/linkedin/auth-url
-"""
-import os
-import logging
-from typing import Dict, List, Optional
-
-logger = logging.getLogger(__name__)
-
-COMPOSIO_API_KEY = os.environ.get("COMPOSIO_API_KEY", "")
-
-_company_cache: Dict[str, dict] = {}
-
-
-def _toolset(entity_id: str):
-    from composio import ComposioToolSet
-    return ComposioToolSet(api_key=COMPOSIO_API_KEY, entity_id=entity_id)
-
-
-def get_oauth_url(entity_id: str, redirect_url: str) -> str:
-    """Return LinkedIn OAuth URL for the given user entity."""
-    from composio import Composio, App
-    client = Composio(api_key=COMPOSIO_API_KEY)
-    entity = client.get_entity(id=entity_id)
-    conn = entity.initiate_connection(app_name=App.LINKEDIN, redirect_url=redirect_url)
-    if not conn.redirectUrl:
-        raise RuntimeError("Composio did not return a redirect URL")
-    return conn.redirectUrl
-
-
-def is_connected(entity_id: str) -> bool:
-    """Return True if user has a LinkedIn connected account."""
-    try:
-        from composio import Composio, App
-        client = Composio(api_key=COMPOSIO_API_KEY)
-        entity = client.get_entity(id=entity_id)
-        entity.get_connection(app="linkedin")
-        return True
-    except Exception:
-        return False
-
-
-def import_profile(entity_id: str) -> dict:
-    """
-    Fetch user's LinkedIn profile and return it in the same shape
-    as resume_matcher_ai.parse_resume_structured().
-    """
-    ts = _toolset(entity_id)
-    result = ts.execute_action(
-        action="LINKEDIN_GET_MY_INFO",
-        params={},
-        entity_id=entity_id,
-    )
-
-    if not result.get("successfull") and not result.get("successful"):
-        raise RuntimeError(f"LinkedIn profile fetch failed: {result.get('error', result)}")
-
-    data = result.get("data", result)
-    return _map_profile(data)
-
-
-def enrich_companies(jobs: List[dict], entity_id: str) -> List[dict]:
-    """
-    Attach company_info to each job using LINKEDIN_GET_COMPANY_INFO.
-    Deduplicates lookups and caches results. Failures are silent — the
-    job is returned unchanged.
-    """
-    if not COMPOSIO_API_KEY:
-        return jobs
-
-    try:
-        ts = _toolset(entity_id)
-    except Exception as e:
-        logger.warning("Composio toolset init failed: %s", e)
-        return jobs
-
-    unique_companies = {j.get("company", "") for j in jobs if j.get("company")}
-
-    for company_name in unique_companies:
-        if company_name in _company_cache:
-            continue
-        try:
-            result = ts.execute_action(
-                action="LINKEDIN_GET_COMPANY_INFO",
-                params={"company_name": company_name},
-                entity_id=entity_id,
-            )
-            ok = result.get("successfull") or result.get("successful")
-            if ok:
-                raw = result.get("data", {})
-                _company_cache[company_name] = _map_company(raw)
-            else:
-                _company_cache[company_name] = {}
-        except Exception as e:
-            logger.debug("Company enrichment failed for %s: %s", company_name, e)
-            _company_cache[company_name] = {}
-
-    for job in jobs:
-        name = job.get("company", "")
-        info = _company_cache.get(name)
-        if info:
-            job["company_info"] = info
-
-    return jobs
-
-
-# ── Mappers ────────────────────────────────────────────────────────────────
-
-def _map_profile(data: dict) -> dict:
-    """Convert LinkedIn profile response to resume_matcher_ai schema."""
-    first = _deep_localized(data.get("firstName", {}))
-    last = _deep_localized(data.get("lastName", {}))
-    name = f"{first} {last}".strip() or data.get("name", "")
-
-    headline = _deep_localized(data.get("headline", {})) or data.get("headline", "")
-
-    location_data = data.get("location", {})
-    location = (
-        location_data.get("name", "") if isinstance(location_data, dict)
-        else str(location_data)
-    )
-
-    email = ""
-    for elem in (data.get("elements") or []):
-        handle = elem.get("handle~", {})
-        if handle.get("emailAddress"):
-            email = handle["emailAddress"]
-            break
-
-    positions = data.get("positions", {}).get("values", []) or data.get("positions", [])
-    work_experience = []
-    for pos in positions:
-        company_obj = pos.get("company", {})
-        company_name = company_obj.get("name", "") if isinstance(company_obj, dict) else ""
-        start = pos.get("startDate", {})
-        end = pos.get("endDate", {})
-        start_str = f"{start.get('year', '')}" if start else ""
-        end_str = f"{end.get('year', '')}" if end else "Present"
-        work_experience.append({
-            "title": pos.get("title", ""),
-            "company": company_name,
-            "start_date": start_str,
-            "end_date": end_str,
-            "description": pos.get("summary", ""),
-            "achievements": [],
-        })
-
-    educations = data.get("educations", {}).get("values", []) or data.get("educations", [])
-    education = []
-    for edu in educations:
-        start = edu.get("startDate", {})
-        end = edu.get("endDate", {})
-        education.append({
-            "degree": edu.get("degree", ""),
-            "institution": edu.get("schoolName", ""),
-            "field": edu.get("fieldOfStudy", ""),
-            "graduation_year": str(end.get("year", "")) if end else "",
-            "gpa": None,
-        })
-
-    skills_data = data.get("skills", {}).get("values", []) or data.get("skills", [])
-    skills = []
-    for s in skills_data:
-        name_val = s.get("skill", {}).get("name", "") if isinstance(s.get("skill"), dict) else s.get("name", "")
-        if name_val:
-            skills.append(name_val)
-
-    return {
-        "personal_info": {
-            "name": name,
-            "email": email,
-            "phone": "",
-            "location": location,
-            "linkedin": data.get("publicProfileUrl", ""),
-            "website": "",
-        },
-        "summary": headline,
-        "skills": skills,
-        "technical_skills": {"languages": [], "frameworks": [], "tools": [], "databases": [], "cloud": []},
-        "work_experience": work_experience,
-        "education": education,
-        "certifications": [],
-        "projects": [],
-        "total_years_experience": None,
-        "seniority_level": "unknown",
-        "_source": "linkedin",
-    }
-
-
-def _map_company(data: dict) -> dict:
-    """Extract relevant company fields from LinkedIn company response."""
-    name_obj = data.get("name", {})
-    name = _deep_localized(name_obj) if isinstance(name_obj, dict) else str(name_obj)
-
-    industries = []
-    for ind in (data.get("industries", {}).get("values", []) or data.get("industries", [])):
-        if isinstance(ind, dict):
-            industries.append(ind.get("name", ""))
-        elif isinstance(ind, str):
-            industries.append(ind)
-
-    return {
-        "name": name,
-        "size": data.get("staffCount") or data.get("employeeCount"),
-        "industry": industries[0] if industries else "",
-        "followers": data.get("followersCount"),
-        "linkedin_url": data.get("companyPageUrl") or data.get("url", ""),
-        "description": _deep_localized(data.get("description", {})) or data.get("description", ""),
-    }
-
-
-def _deep_localized(obj) -> str:
-    """Extract text from LinkedIn's localized string objects."""
-    if isinstance(obj, str):
-        return obj
-    if isinstance(obj, dict):
-        localized = obj.get("localized", {})
-        if localized:
-            return next(iter(localized.values()), "")
-        return obj.get("preferredLocale", {}).get("country", "") or ""
-    return ""

backend/services/linkedin_nango.py

@@ -0,0 +1,143 @@
+"""LinkedIn integration via Nango proxy (replaces Composio)."""
+import os
+import logging
+import httpx
+from typing import Dict, List, Optional
+
+logger = logging.getLogger(__name__)
+
+NANGO_SECRET_KEY = os.environ.get("NANGO_SECRET_KEY", "")
+NANGO_API = "https://api.nango.dev"
+PROVIDER = "linkedin"
+
+_company_cache: Dict[str, dict] = {}
+
+
+def _auth_headers() -> dict:
+    return {"Authorization": f"Bearer {NANGO_SECRET_KEY}"}
+
+
+def _proxy_headers(connection_id: str) -> dict:
+    return {
+        "Authorization": f"Bearer {NANGO_SECRET_KEY}",
+        "Provider-Config-Key": PROVIDER,
+        "Connection-Id": connection_id,
+    }
+
+
+def get_connect_url(user_id: str, redirect_url: str) -> str:
+    """Create a Nango connect session and return the hosted connect_link."""
+    resp = httpx.post(
+        f"{NANGO_API}/connect/sessions",
+        headers={**_auth_headers(), "Content-Type": "application/json"},
+        json={
+            "end_user": {"id": user_id},
+            "allowed_integrations": [PROVIDER],
+            "callback_url": redirect_url,
+        },
+        timeout=10,
+    )
+    resp.raise_for_status()
+    return resp.json()["connect_link"]
+
+
+def _get_connection_id(user_id: str) -> Optional[str]:
+    """Return the Nango connection_id for the user's LinkedIn connection."""
+    try:
+        resp = httpx.get(
+            f"{NANGO_API}/connections",
+            headers=_auth_headers(),
+            params={"tags[end_user_id]": user_id, "provider_config_key": PROVIDER},
+            timeout=10,
+        )
+        if resp.status_code != 200:
+            return None
+        connections = resp.json().get("connections", [])
+        return connections[0].get("connection_id") if connections else None
+    except Exception as e:
+        logger.debug("Nango connection lookup failed: %s", e)
+        return None
+
+
+def is_connected(user_id: str) -> bool:
+    return _get_connection_id(user_id) is not None
+
+
+def import_profile(user_id: str) -> dict:
+    conn_id = _get_connection_id(user_id)
+    if not conn_id:
+        raise RuntimeError("LinkedIn not connected — please connect via /api/linkedin/auth-url")
+
+    headers = _proxy_headers(conn_id)
+
+    # OpenID Connect userinfo — name, email, picture
+    ui_resp = httpx.get(f"{NANGO_API}/proxy/v2/userinfo", headers=headers, timeout=15)
+    ui_resp.raise_for_status()
+    userinfo = ui_resp.json()
+
+    # Basic profile — headline, location, vanityName
+    me_resp = httpx.get(
+        f"{NANGO_API}/proxy/v2/me",
+        headers=headers,
+        params={"projection": "(id,headline,location,vanityName)"},
+        timeout=15,
+    )
+    me = me_resp.json() if me_resp.status_code == 200 else {}
+
+    return _map_profile(userinfo, me)
+
+
+def enrich_companies(jobs: List[dict], user_id: str) -> List[dict]:
+    """LinkedIn company enrichment requires partner API access — returns jobs unchanged."""
+    return jobs
+
+
+# ── Mappers ────────────────────────────────────────────────────────────────
+
+def _localized(obj) -> str:
+    if isinstance(obj, str):
+        return obj
+    if isinstance(obj, dict):
+        loc = obj.get("localized", {})
+        if loc:
+            return next(iter(loc.values()), "")
+    return ""
+
+
+def _map_profile(userinfo: dict, me: dict) -> dict:
+    name = userinfo.get("name", "")
+    if not name:
+        name = f"{userinfo.get('given_name', '')} {userinfo.get('family_name', '')}".strip()
+
+    headline = _localized(me.get("headline", ""))
+
+    location_obj = me.get("location", {})
+    location = ""
+    if isinstance(location_obj, dict):
+        location = location_obj.get("geographicArea", "") or location_obj.get("country", {}).get("code", "")
+    elif isinstance(location_obj, str):
+        location = location_obj
+
+    vanity = me.get("vanityName", "")
+    linkedin_url = f"https://www.linkedin.com/in/{vanity}" if vanity else ""
+
+    return {
+        "personal_info": {
+            "name": name,
+            "email": userinfo.get("email", ""),
+            "phone": "",
+            "location": location,
+            "linkedin": linkedin_url,
+            "website": "",
+        },
+        "summary": headline,
+        "skills": [],
+        "technical_skills": {"languages": [], "frameworks": [], "tools": [], "databases": [], "cloud": []},
+        "work_experience": [],
+        "education": [],
+        "certifications": [],
+        "projects": [],
+        "total_years_experience": None,
+        "seniority_level": "unknown",
+        "_source": "linkedin_nango",
+    }