How to Build a Lead Qualification Agent (With the Claude Agent SDK)

export LEADOCEAN_API_KEY="paste-your-key-here"

import os, asyncio, httpx

BASE = "https://api.leadocean.io"
HEADERS = {"x-api-key": os.environ["LEADOCEAN_API_KEY"], "content-type": "application/json"}
SPENT = {"records": 0}          # running total for this process
RECORD_CAP = 20                 # hard ceiling for one run

async def call(method, path, params=None, body=None):
    async with httpx.AsyncClient(timeout=30) as client:
        for attempt in range(2):
            r = await client.request(method, BASE + path, headers=HEADERS, params=params, json=body)
            if r.status_code == 429 and attempt == 0:
                await asyncio.sleep(int(r.headers.get("Retry-After", "1")))
                continue
            break
    if r.status_code == 402:
        return None, "Out of records on this plan. Stop and tell the user."
    if r.status_code == 429:
        return None, "Rate limited twice. Stop and report what you have."
    if r.status_code == 404:
        return None, "Nobody matched. This lookup cost nothing."
    if r.status_code >= 400:
        return None, f"HTTP {r.status_code}: {r.text[:300]}"
    return r.json(), None

import json
from claude_agent_sdk import tool, create_sdk_mcp_server

def ok(data):  return {"content": [{"type": "text", "text": json.dumps(data)}]}
def err(msg):  return {"content": [{"type": "text", "text": msg}], "is_error": True}

@tool("count_audience", "Count mailable people for an ICP. Free. Call this first.",
      {"type": "object", "properties": {
          "jobLevel": {"type": "array", "items": {"type": "string"}},
          "jobFunction": {"type": "array", "items": {"type": "string"}},
          "country": {"type": "array", "items": {"type": "string"}},
          "industry": {"type": "array", "items": {"type": "string"}},
          "employeeRange": {"type": "array", "items": {"type": "string"}}},
       "required": []})
async def count_audience(args):
    data, e = await call("POST", "/v1/people/search", params={"count": "true"},
                         body={**args, "limit": 1})
    return err(e) if e else ok(data["meta"])

@tool("search_people", "Search people for an ICP. Costs 1 record per person returned. Max 10.",
      {"type": "object", "properties": {
          "jobLevel": {"type": "array", "items": {"type": "string"}},
          "jobFunction": {"type": "array", "items": {"type": "string"}},
          "country": {"type": "array", "items": {"type": "string"}},
          "industry": {"type": "array", "items": {"type": "string"}},
          "employeeRange": {"type": "array", "items": {"type": "string"}},
          "limit": {"type": "integer", "minimum": 1, "maximum": 10}},
       "required": ["limit"]})
async def search_people(args):
    n = min(int(args.get("limit", 10)), 10)
    if SPENT["records"] + n > RECORD_CAP:
        return err("Record cap reached for this run. Do not search again.")
    data, e = await call("POST", "/v1/people/search", body={**args, "limit": n})
    if e: return err(e)
    SPENT["records"] += len(data["data"])
    return ok(data)

@tool("enrich_person", "Full lookup for one person by person_id. Costs 1 record.",
      {"person_id": str})
async def enrich_person(args):
    if SPENT["records"] + 1 > RECORD_CAP:
        return err("Record cap reached for this run. Do not enrich again.")
    data, e = await call("POST", "/v1/people/enrich",
                         body={"person_id": args["person_id"], "reveal_email": True})
    if e: return err(e)
    SPENT["records"] += 1
    return ok(data)

leadocean = create_sdk_mcp_server(name="leadocean", version="1.0.0",
                                  tools=[count_audience, search_people, enrich_person])

SYSTEM = """You qualify leads for Acme, which sells project management software.

ICP: VP of Sales at US software companies with 51 to 200 staff.

Procedure:
1. Call count_audience with the ICP filters. Report the total.
2. Call search_people with the same filters and limit 10.
3. Score each person 0 to 3: +1 title matches "VP Sales" or "VP of Sales",
   +1 contact_data.email_status is verified, +1 the company is a software company.
4. Call enrich_person for the 3 highest scores only.
5. Return a ranked list: name, title, company, score, one-line reason.

Rules: use only fields that came back in a tool result. If a field is missing,
write "unknown". Never call a tool more than the procedure says.

Filter values are exact strings: jobLevel "VP", jobFunction "Sales & Business Development",
country "US", industry "Software Development", employeeRange "51-200"."""

import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions, ResultMessage

async def main():
    options = ClaudeAgentOptions(
        system_prompt=SYSTEM,
        mcp_servers={"leadocean": leadocean},
        allowed_tools=["mcp__leadocean__count_audience",
                       "mcp__leadocean__search_people",
                       "mcp__leadocean__enrich_person"],
        max_turns=12,
        max_budget_usd=1.00,
    )
    async for message in query(prompt="Qualify this week's leads.", options=options):
        if isinstance(message, ResultMessage) and message.subtype == "success":
            print(message.result)
    print(f"LeadOcean records spent: {SPENT['records']}")

asyncio.run(main())

curl -s https://api.leadocean.io/v1/account -H "x-api-key: $LEADOCEAN_API_KEY" | jq

{
  "total": 1649,
  "totalIsExact": true,
  "distinctCompanies": 1125,
  "totalAll": 4141,
  "credits": 0,
  "defaults": "Counted only addresses worth sending to (emailStatus: verified or catch_all_valid or catch_all)"
}

{
  "success": true,
  "data": [
    {
      "profile_data": { "...": "name, headline, location" },
      "contact_data": { "has_email": true, "email_status": "verified", "email_type": "work",
                        "contact_current_experiences": [ { "...": "current role and company" } ] },
      "meta": { "source_ids": { "person_id": "<digits>" } }
    }
  ],
  "meta": { "count": 10, "limit": 10, "nextCursor": "<opaque cursor>" }
}