How to Give a LangGraph Agent B2B Data (With LeadOcean)

export LEADOCEAN_API_KEY="paste-your-key-here"
export ANTHROPIC_API_KEY="paste-your-model-key-here"
pip install langgraph langchain langchain-anthropic httpx

import os, time
import httpx

BASE = "https://api.leadocean.io"
HEADERS = {"x-api-key": os.environ["LEADOCEAN_API_KEY"]}
MAX_RECORDS = 10          # hard cap for one run
spent = 0                 # records spent in this run


class Halt(Exception):
    """Stop the graph. The message goes back to the model."""


def call(method, path, *, params=None, body=None, reserve=0):
    global spent
    if spent + reserve > MAX_RECORDS:
        raise Halt(f"Run budget of {MAX_RECORDS} records reached. Report what you have.")
    for attempt in range(4):
        r = httpx.request(method, BASE + path, params=params, json=body,
                          headers=HEADERS, timeout=30)
        if r.status_code == 429:
            if r.headers.get("X-Quota-State"):                    # paced or over-limit
                raise Halt("Account reached its fair-use pace and is held to one request "
                           f"a minute until {r.headers.get('X-Quota-Reset')}. Stop and report.")
            time.sleep(int(r.headers.get("Retry-After", "1")))   # rate limit: 1 second
            continue
        if r.status_code == 402:
            raise Halt("Free records are spent. They do not reset. Stop and report.")
        if r.status_code == 503:
            time.sleep(2 ** attempt)                              # index unreachable: back off
            continue
        if r.status_code == 404:
            return None                                           # costs nothing
        r.raise_for_status()
        return r.json()
    raise Halt("LeadOcean did not answer after 4 tries. Stop and report.")

from langchain.tools import tool
import leadocean as lo


def _filters(title, country, industry, employee_range, domain):
    f = {"title": title, "country": country, "industry": industry,
         "employeeRange": employee_range, "domain": domain}
    return {k: v for k, v in f.items() if v}


@tool
def count_people(title: list[str] | None = None, country: list[str] | None = None,
                 industry: list[str] | None = None, employee_range: list[str] | None = None,
                 domain: list[str] | None = None) -> str:
    """Count mailable people for a filter set. Free. Call this before search_people.
    country is ISO alpha-2 such as US. employee_range is one of 1-10, 11-50, 51-200,
    201-500, 501-1000, 1001-5000, 5001-10000, 10001+. industry is an exact LeadOcean string."""
    body = _filters(title, country, industry, employee_range, domain) | {"limit": 1}
    res = lo.call("POST", "/v1/people/search", params={"count": "true"}, body=body)
    return f"{res['meta']['total']} mailable people match. No records spent."


@tool
def search_people(title: list[str] | None = None, country: list[str] | None = None,
                  industry: list[str] | None = None, employee_range: list[str] | None = None,
                  domain: list[str] | None = None, limit: int = 5) -> str:
    """Search people. Costs one record per person returned, so keep limit at 5 or less.
    Returns person_id, name, headline and whether an email exists."""
    limit = max(1, min(limit, 5))
    body = _filters(title, country, industry, employee_range, domain) | {"limit": limit}
    res = lo.call("POST", "/v1/people/search", body=body, reserve=limit)
    rows = res["data"]
    lo.spent += len(rows)
    return str([{
        "person_id": r["meta"]["source_ids"]["person_id"],
        "name": r["profile_data"].get("profile_full_name"),
        "headline": r["profile_data"].get("profile_headline"),
        "has_email": r["contact_data"].get("has_email"),
    } for r in rows])


@tool
def enrich_person(person_id: str) -> str:
    """Reveal emails and phones for one person_id from search_people. Costs one record.
    Skip people whose has_email is false."""
    res = lo.call("POST", "/v1/people/enrich",
                  body={"person_id": person_id, "reveal_email": True}, reserve=1)
    if res is None:
        return "Not found. No record spent."
    lo.spent += 1
    cd = res["data"]["contact_data"]
    return str({"emails": cd.get("contact_emails"), "email_status": cd.get("email_status")})

from langchain.chat_models import init_chat_model
from langchain.messages import SystemMessage, ToolMessage, HumanMessage
from langgraph.graph import StateGraph, MessagesState, START, END
import leadocean as lo
from tools import count_people, search_people, enrich_person

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

SYSTEM = ("You find B2B contacts with LeadOcean. Always call count_people first. "
          "Search at most 5 people. Enrich at most 2 who have has_email true. "
          "Placeholders only: never invent a person.")


class State(MessagesState):
    halted: bool


def llm_call(state: State):
    m = model if state.get("halted") else model_with_tools
    return {"messages": [m.invoke([SystemMessage(content=SYSTEM)] + state["messages"])]}


def tool_node(state: State):
    out, halted = [], False
    for call in state["messages"][-1].tool_calls:
        try:
            text = by_name[call["name"]].invoke(call["args"])
        except lo.Halt as stop:
            text, halted = str(stop), True
        out.append(ToolMessage(content=text, tool_call_id=call["id"]))
    return {"messages": out, "halted": halted}


def should_continue(state: State):
    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 agent import agent, HumanMessage
import leadocean as lo

task = ("Find VP Sales or Head of Sales at US software companies with 51-200 staff. "
        "Size the audience, list 5 people, then enrich the 2 best with an email.")
result = agent.invoke({"messages": [HumanMessage(content=task)], "halted": False})
print(result["messages"][-1].content)
print(f"LeadOcean records spent: {lo.spent}")

from fastmcp.client import Client
from langchain.mcp import MCPAdapter

async def load_leadocean_tools():
    async with MCPAdapter(Client("https://api.leadocean.io/mcp", auth="oauth")) as adapter:
        return await adapter.list_tools()

leadocean_count_leads  title ["VP Sales","Head of Sales"]  country ["US"]
  industry ["Software Development"]  employeeRange ["51-200"]

2026-10-21  total: 360 (verified, catch_all_valid, catch_all)  totalAll: 863
            distinctCompanies: 317  credits: 0

{
  "profile_data": { "profile_full_name": "Jane Doe", "profile_headline": "VP Sales at Acme" },
  "contact_data": { "has_email": true, "email_status": "verified", "email_type": "work" },
  "meta": { "source_ids": { "person_id": "<digits>" } }
}