"""Hunter.io API client for email intelligence."""
import httpx
from typing import Optional, Any
from dataclasses import dataclass
@dataclass
class HunterClient:
"""Client for Hunter.io API."""
api_key: Optional[str] = None
base_url: str = "https://api.hunter.io/v2"
_client: Optional[httpx.AsyncClient] = None
async def start(self):
"""Initialize HTTP client."""
if not self._client:
self._client = httpx.AsyncClient(
base_url=self.base_url,
timeout=30.0,
)
async def close(self):
"""Close HTTP client."""
if self._client:
await self._client.aclose()
self._client = None
async def domain_search(self, domain: str, limit: int = 10) -> dict[str, Any]:
"""
Find email addresses for a domain.
Args:
domain: Company domain
limit: Max results
Returns:
Email addresses and patterns for the domain
"""
if not self.api_key:
return {"error": "Hunter API key not configured"}
await self.start()
try:
resp = await self._client.get(
"/domain-search",
params={"domain": domain, "api_key": self.api_key, "limit": limit},
)
resp.raise_for_status()
data = resp.json().get("data", {})
return {
"domain": domain,
"organization": data.get("organization"),
"email_pattern": data.get("pattern"),
"emails": [
{
"email": e.get("value"),
"type": e.get("type"),
"first_name": e.get("first_name"),
"last_name": e.get("last_name"),
"position": e.get("position"),
"confidence": e.get("confidence"),
"linkedin": e.get("linkedin"),
}
for e in data.get("emails", [])
],
"total": data.get("total", 0),
}
except httpx.HTTPStatusError as e:
return {"error": f"Hunter API error: {e.response.status_code}"}
except Exception as e:
return {"error": str(e)}
async def email_finder(
self,
domain: str,
first_name: str,
last_name: str,
) -> dict[str, Any]:
"""
Find email address for a specific person.
Args:
domain: Company domain
first_name: Person's first name
last_name: Person's last name
Returns:
Most likely email address
"""
if not self.api_key:
return {"error": "Hunter API key not configured"}
await self.start()
try:
resp = await self._client.get(
"/email-finder",
params={
"domain": domain,
"first_name": first_name,
"last_name": last_name,
"api_key": self.api_key,
},
)
resp.raise_for_status()
data = resp.json().get("data", {})
return {
"email": data.get("email"),
"score": data.get("score"),
"domain": domain,
"first_name": first_name,
"last_name": last_name,
"position": data.get("position"),
"linkedin": data.get("linkedin_url"),
"sources": data.get("sources", []),
}
except httpx.HTTPStatusError as e:
return {"error": f"Hunter API error: {e.response.status_code}"}
except Exception as e:
return {"error": str(e)}
async def email_verify(self, email: str) -> dict[str, Any]:
"""
Verify if an email address is valid.
Args:
email: Email address to verify
Returns:
Verification status and deliverability
"""
if not self.api_key:
return {"error": "Hunter API key not configured"}
await self.start()
try:
resp = await self._client.get(
"/email-verifier",
params={"email": email, "api_key": self.api_key},
)
resp.raise_for_status()
data = resp.json().get("data", {})
return {
"email": email,
"status": data.get("status"),
"result": data.get("result"),
"score": data.get("score"),
"regexp": data.get("regexp"),
"gibberish": data.get("gibberish"),
"disposable": data.get("disposable"),
"webmail": data.get("webmail"),
"mx_records": data.get("mx_records"),
"smtp_server": data.get("smtp_server"),
"smtp_check": data.get("smtp_check"),
"accept_all": data.get("accept_all"),
"block": data.get("block"),
}
except httpx.HTTPStatusError as e:
return {"error": f"Hunter API error: {e.response.status_code}"}
except Exception as e:
return {"error": str(e)}
