"""Apollo.io API client for company and person enrichment."""
import httpx
from typing import Optional, Any
from dataclasses import dataclass
@dataclass
class ApolloClient:
"""Client for Apollo.io API."""
api_key: Optional[str] = None
base_url: str = "https://api.apollo.io/api/v1"
_client: Optional[httpx.AsyncClient] = None
async def start(self):
"""Initialize HTTP client."""
if not self._client and self.api_key:
self._client = httpx.AsyncClient(
base_url=self.base_url,
headers={"X-Api-Key": self.api_key},
timeout=30.0,
)
async def close(self):
"""Close HTTP client."""
if self._client:
await self._client.aclose()
self._client = None
async def enrich_company(self, domain: str) -> dict[str, Any]:
"""
Enrich company data from domain.
Args:
domain: Company website domain
Returns:
Company information including size, industry, location
"""
if not self.api_key:
return {"error": "Apollo API key not configured"}
await self.start()
try:
resp = await self._client.post(
"/organizations/enrich",
json={"domain": domain},
)
resp.raise_for_status()
data = resp.json()
org = data.get("organization", {})
return {
"name": org.get("name"),
"domain": domain,
"description": org.get("short_description"),
"industry": org.get("industry"),
"employees": org.get("estimated_num_employees"),
"founded": org.get("founded_year"),
"location": {
"city": org.get("city"),
"state": org.get("state"),
"country": org.get("country"),
},
"linkedin": org.get("linkedin_url"),
"twitter": org.get("twitter_url"),
"facebook": org.get("facebook_url"),
"phone": org.get("phone"),
"technologies": org.get("technologies", []),
"keywords": org.get("keywords", []),
}
except httpx.HTTPStatusError as e:
return {"error": f"Apollo API error: {e.response.status_code}"}
except Exception as e:
return {"error": str(e)}
async def find_person(
self,
first_name: str,
last_name: str,
organization: Optional[str] = None,
domain: Optional[str] = None,
) -> dict[str, Any]:
"""
Find a person by name and company.
Args:
first_name: Person's first name
last_name: Person's last name
organization: Company name
domain: Company domain
Returns:
Person details including title, email, social profiles
"""
if not self.api_key:
return {"error": "Apollo API key not configured"}
await self.start()
try:
payload = {
"first_name": first_name,
"last_name": last_name,
}
if organization:
payload["organization_name"] = organization
if domain:
payload["domain"] = domain
resp = await self._client.post("/people/match", json=payload)
resp.raise_for_status()
data = resp.json()
person = data.get("person", {})
if not person:
return {"found": False, "name": f"{first_name} {last_name}"}
return {
"found": True,
"name": person.get("name"),
"first_name": person.get("first_name"),
"last_name": person.get("last_name"),
"title": person.get("title"),
"email": person.get("email"),
"email_status": person.get("email_status"),
"linkedin": person.get("linkedin_url"),
"twitter": person.get("twitter_url"),
"organization": person.get("organization", {}).get("name"),
"organization_domain": person.get("organization", {}).get("primary_domain"),
"city": person.get("city"),
"state": person.get("state"),
"country": person.get("country"),
}
except httpx.HTTPStatusError as e:
return {"error": f"Apollo API error: {e.response.status_code}"}
except Exception as e:
return {"error": str(e)}
async def search_people(
self,
organization: Optional[str] = None,
domain: Optional[str] = None,
titles: Optional[list[str]] = None,
limit: int = 10,
) -> dict[str, Any]:
"""
Search for people at a company.
Args:
organization: Company name
domain: Company domain
titles: Filter by job titles
limit: Max results
Returns:
List of people matching criteria
"""
if not self.api_key:
return {"error": "Apollo API key not configured"}
await self.start()
try:
payload = {
"per_page": min(limit, 25),
}
if organization:
payload["organization_name"] = organization
if domain:
payload["organization_domains"] = [domain]
if titles:
payload["person_titles"] = titles
resp = await self._client.post("/mixed_people/search", json=payload)
resp.raise_for_status()
data = resp.json()
return {
"total": data.get("pagination", {}).get("total_entries", 0),
"people": [
{
"name": p.get("name"),
"title": p.get("title"),
"email": p.get("email"),
"linkedin": p.get("linkedin_url"),
"organization": p.get("organization", {}).get("name"),
}
for p in data.get("people", [])
],
}
except httpx.HTTPStatusError as e:
return {"error": f"Apollo API error: {e.response.status_code}"}
except Exception as e:
return {"error": str(e)}
