How to Build a Lead Qualification Agent (LangGraph and LeadOcean)

pip install langgraph langchain langchain-anthropic httpx
export LEADOCEAN_API_KEY="..."   # from app.leadocean.io
export ANTHROPIC_API_KEY="..."

import os, time, httpx

BASE = "https://api.leadocean.io"
HEADERS = {"x-api-key": os.environ["LEADOCEAN_API_KEY"]}
MAX_RECORDS = 25
spent = 0

def call(method, path, *, params=None, json=None, cost=0):
    """Returns the body, or {"error": ...} the model can read. Never raises."""
    global spent
    if spent + cost > MAX_RECORDS:
        return {"error": f"STOP: budget cap {MAX_RECORDS} reached ({spent} spent)"}
    for attempt in range(3):
        r = httpx.request(method, BASE + path, headers=HEADERS,
                          params=params, json=json, timeout=30)
        if r.status_code == 429:
            wait = int(r.headers.get("Retry-After", "1"))
            if wait > 5:
                return {"error": f"STOP: throttled, retry in {wait}s"}
            time.sleep(wait)
            continue
        if r.status_code == 503:          # transient, never billed
            time.sleep(2 ** attempt)
            continue
        break
    if r.status_code == 402:
        return {"error": "STOP: out of records (402). Tell the user."}
    if r.status_code == 404:
        return {"found": False}
    if r.status_code >= 400:
        return {"error": f"{r.status_code}: {r.text[:200]}"}
    body = r.json()
    spent += (body.get("meta") or {}).get("credits", cost)
    return body

from langchain.tools import tool

@tool
def count_people(title: list[str], country: list[str]) -> dict:
    """Free. Count people by job title keywords and ISO country codes,
    for example ["VP Sales"] and ["CA"]. Call this before search_people."""
    body = call("POST", "/v1/people/search", params={"count": "true"},
                json={"title": title, "country": country, "limit": 1})
    return body if "error" in body else {"total": body["meta"]["total"]}

@tool
def search_people(title: list[str], country: list[str], limit: int = 5) -> dict:
    """Spends one record per person returned (max 10). Each row has a person_id.
    Use enrich_person for contact details."""
    limit = max(1, min(limit, 10))
    body = call("POST", "/v1/people/search", cost=limit,
                json={"title": title, "country": country, "limit": limit})
    return body if "error" in body else {"people": body["data"]}

@tool
def enrich_person(person_id: str) -> dict:
    """Spends one record. Enrich one person by person_id from search_people."""
    return call("POST", "/v1/people/enrich", cost=1,
                json={"person_id": person_id, "reveal_email": False})

from typing import Literal
from typing_extensions import TypedDict, Annotated
import operator
from langchain.chat_models import init_chat_model
from langchain.messages import AnyMessage, SystemMessage, ToolMessage
from langgraph.graph import StateGraph, START, END

tools = [count_people, search_people, enrich_person]
tools_by_name = {t.name: t for t in tools}
model = init_chat_model("claude-sonnet-5", temperature=0).bind_tools(tools)

SYSTEM = (
    "You qualify B2B leads. ICP: VP or head of sales at a software company. "
    "Call count_people first and report the total. Search at most 5 people, "
    "enrich at most 3. Score each enriched person 1 to 5 against the ICP with "
    "one line of reasoning. If a tool result starts with STOP, stop and say why."
)

class State(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]

def llm_call(state: State):
    return {"messages": [model.invoke([SystemMessage(content=SYSTEM)] + state["messages"])]}

def tool_node(state: State):
    out = []
    for tc in state["messages"][-1].tool_calls:
        result = tools_by_name[tc["name"]].invoke(tc["args"])
        out.append(ToolMessage(content=str(result), tool_call_id=tc["id"]))
    return {"messages": out}

def should_continue(state: State) -> Literal["tool_node", END]:
    return "tool_node" if state["messages"][-1].tool_calls else END

g = StateGraph(State)
g.add_node("llm_call", llm_call)
g.add_node("tool_node", tool_node)
g.add_edge(START, "llm_call")
g.add_conditional_edges("llm_call", should_continue, ["tool_node", END])
g.add_edge("tool_node", "llm_call")
agent = g.compile()

from langchain.messages import HumanMessage

result = agent.invoke(
    {"messages": [HumanMessage(content="Find 5 VPs of sales in Canada, "
                                       "enrich the first 3 and score them.")]},
    {"recursion_limit": 12},
)
print(result["messages"][-1].content)
print(f"LeadOcean records spent: {spent}")

{ "total": 658, "totalIsExact": true, "totalAll": 2661, "distinctCompanies": 590, "credits": 0 }

{
  "success": true,
  "data": {
    "profile_data": { "profile_full_name": "Jane Doe", "...": "headline, location" },
    "contact_data": { "has_email": true, "email_status": "verified" }
  },
  "meta": { "count": 1, "credits": 1 }
}