Contact Search

curl "https://app.lev8.com/v1/contact-search" \
  --request POST \
  --header "Content-Type: application/json" \
  --header "Accept: application/json" \
  --header "Authorization: Bearer $LEV8_API_KEY" \
  --header "Idempotency-Key: $(uuidgen)" \
  --data '{
    "contact_type": "email",
    "objective": "Find the work email of Jane Smith, VP of Engineering at Acme Corp. Owner signals: linkedin_url=https://www.linkedin.com/in/jane-smith-acme"
  }'

const response = await fetch("https://app.lev8.com/v1/contact-search", {
  method: "POST",
  headers: {
    "Content-Type": "application/json",
    Accept: "application/json",
    Authorization: `Bearer ${process.env.LEV8_API_KEY}`,
    "Idempotency-Key": crypto.randomUUID(),
  },
  body: JSON.stringify({
    contact_type: "email",
    objective:
      "Find the work email of Jane Smith, VP of Engineering at Acme Corp. Owner signals: linkedin_url=https://www.linkedin.com/in/jane-smith-acme",
  }),
});

const requestId = response.headers.get("x-request-id");
if (!response.ok) {
  const apiError = await response.json().catch(() => null);
  throw new Error(
    apiError?.error?.message ??
      `lev8 request failed with HTTP ${response.status} (request_id=${requestId ?? "unknown"})`,
  );
}

const result = await response.json();
console.log({ requestId, content: result.result.content, reference: result.result.reference });

import asyncio
import json
import os
import uuid
import httpx


API_URL = "https://app.lev8.com/v1/contact-search"
API_KEY = os.environ["LEV8_API_KEY"]


async def contact_search() -> None:
    headers = {
        "Content-Type": "application/json",
        "Accept": "application/json",
        "Authorization": f"Bearer {API_KEY}",
        "Idempotency-Key": str(uuid.uuid4()),
    }
    payload = {
        "contact_type": "email",
        "objective": "Find the work email of Jane Smith, VP of Engineering at Acme Corp. Owner signals: linkedin_url=https://www.linkedin.com/in/jane-smith-acme",
    }

    async with httpx.AsyncClient(timeout=60) as client:
        response = await client.post(API_URL, headers=headers, json=payload)
        request_id = response.headers.get("x-request-id")
        if response.is_error:
            try:
                message = response.json().get("error", {}).get("message")
            except (json.JSONDecodeError, UnicodeDecodeError):
                message = None
            raise RuntimeError(
                message
                or f"lev8 request failed with HTTP {response.status_code} "
                f"(request_id={request_id or 'unknown'})"
            )

        result = response.json()
        print(
            {
                "request_id": request_id,
                "content": result.get("result", {}).get("content"),
                "reference": result.get("result", {}).get("reference"),
            }
        )


if __name__ == "__main__":
    asyncio.run(contact_search())