How to Give Your Agent B2B Data (With the Claude Agent SDK)

pip install claude-agent-sdk httpx
export LEADOCEAN_API_KEY="..."   # from app.leadocean.io
export ANTHROPIC_API_KEY="..."

import asyncio, os
import httpx

BASE = "https://api.leadocean.io"
HEADERS = {"x-api-key": os.environ["LEADOCEAN_API_KEY"]}
CAP = 15                      # records this process may spend
spent = {"records": 0}


class LOError(Exception):
    pass


async def call(method: str, path: str, *, params=None, body=None) -> dict:
    async with httpx.AsyncClient(base_url=BASE, headers=HEADERS, timeout=30) as c:
        for attempt in range(3):
            r = await c.request(method, path, params=params, json=body)
            if r.status_code == 429:           # above the plan's per-second limit
                await asyncio.sleep(min(int(r.headers.get("Retry-After", "1")), 5))
                continue
            if r.status_code == 503:           # transient, never billed
                await asyncio.sleep(2 ** attempt)
                continue
            break
    if r.status_code == 200:
        data = r.json()
        spent["records"] += data.get("meta", {}).get("credits", 0)
        return data
    if r.status_code == 404:                   # nobody matched, costs nothing
        return {"found": False}
    if r.status_code == 402:
        raise LOError("The free records are spent. Stop and tell the user.")
    if r.status_code == 429:
        raise LOError("Rate limited. Stop and report what you have.")
    if r.status_code == 401:
        raise LOError("Bad or missing API key. Stop.")
    raise LOError(f"HTTP {r.status_code}: {r.text[:300]}")

import json
from claude_agent_sdk import tool, create_sdk_mcp_server
from lo import call, spent, CAP, LOError


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

ARR = {"type": "array", "items": {"type": "string"}}
FILTERS = {"jobLevel": ARR, "jobFunction": ARR, "country": ARR,
           "industry": ARR, "employeeRange": ARR}


@tool("list_values", "Free. Exact values for a filter name, e.g. job_level, country.",
      {"name": str, "q": str})
async def list_values(args):
    try:
        d = await call("GET", f"/v1/enums/{args['name']}",
                       params={"limit": 50, "q": args.get("q", "")})
        return ok(d["data"])
    except LOError as e:
        return err(str(e))


@tool("count_people", "Free. Size an audience. Call this first.",
      {"type": "object", "properties": FILTERS})
async def count_people(args):
    try:
        d = await call("POST", "/v1/people/search", params={"count": "true"},
                       body={**args, "limit": 1})
        return ok(d["meta"])
    except LOError as e:
        return err(str(e))


@tool("search_people", "1 record per person returned. Max 5.",
      {"type": "object", "properties": {**FILTERS, "limit": {"type": "integer"}},
       "required": ["limit"]})
async def search_people(args):
    n = min(int(args["limit"]), 5)
    if spent["records"] + n > CAP:
        return err("Record cap reached. Do not search again.")
    try:
        d = await call("POST", "/v1/people/search", body={**args, "limit": n})
        return ok(d["data"])
    except LOError as e:
        return err(str(e))


@tool("enrich_person", "1 record. Full lookup by person_id from a search row.",
      {"person_id": str})
async def enrich_person(args):
    if spent["records"] + 1 > CAP:
        return err("Record cap reached. Do not enrich again.")
    try:
        d = await call("POST", "/v1/people/enrich",
                       body={"person_id": args["person_id"], "reveal_email": True})
        return ok(d)
    except LOError as e:
        return err(str(e))


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

import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions, ResultMessage
from lo import spent

SYSTEM = """You build small B2B prospect lists with LeadOcean.
Always call count_people first and report the total.
Then call search_people with limit 5.
Enrich at most 2 people, only if the user asks for contact details.
Filter values are exact strings. If unsure, call list_values.
Use only fields that came back in a tool result."""

TOOLS = [f"mcp__leadocean__{t}" for t in
         ("list_values", "count_people", "search_people", "enrich_person")]


async def main():
    from tools import leadocean
    options = ClaudeAgentOptions(
        system_prompt=SYSTEM,
        mcp_servers={"leadocean": leadocean},
        allowed_tools=TOOLS,
        max_turns=10,
    )
    prompt = ("VPs of Sales at US software companies with 51-200 staff. "
              "Size it, then show me 5 and enrich the first 2.")
    async for m in query(prompt=prompt, options=options):
        if isinstance(m, ResultMessage) and m.subtype == "success":
            print(m.result)
    print("LeadOcean records spent:", spent["records"])

asyncio.run(main())

claude mcp add --transport http leadocean https://api.leadocean.io/mcp
claude mcp login leadocean

{"total": 1649, "totalIsExact": true, "totalAll": 4141, "distinctCompanies": 1125, "credits": 0}

{
  "success": true,
  "data": [{
    "profile_data": { "profile_full_name": "Jane Doe" },
    "contact_data": {
      "has_email": true,
      "email_status": "verified",
      "contact_current_experiences": [
        { "job_title": "VP Sales", "company_name": "Acme", "company_domain": "acme.com" }
      ]
    },
    "meta": { "source_ids": { "person_id": "12345678" } }
  }],
  "meta": { "count": 1, "limit": 5, "credits": 1, "nextCursor": "..." }
}