How to Give a LlamaIndex Agent B2B Data (LeadOcean REST and MCP)

You wire three plain Python functions into a LlamaIndex FunctionAgent: size an audience for free, search people, enrich one person. The agent then answers requests like "find five VPs of sales in Canada" without you writing a pipeline. This guide uses the REST API because a script has no browser for sign-in. The MCP route is in Step 5.

TL;DR

• Time: about 20 minutes, including the install.
• Cost: the test run spends 8 LeadOcean records (5 people from search, 3 enrichments). Sizing is free. The free plan's 1,000 records cover about 125 such runs.
• The tools return errors as data. A 402 stops the run, a 429 waits and retries, a 404 reports "not found".

Prerequisites

• Python 3.10 or newer.
• A LeadOcean API key from app.leadocean.io, exported as $LEADOCEAN_API_KEY. The free plan includes 1,000 records, one-off, no card.
• An OpenAI key as $OPENAI_API_KEY. The LlamaIndex agent tutorial uses gpt-4o-mini, and any tool-calling model works.
• Packages: llama-index-core, llama-index-llms-openai and httpx.

pip install llama-index-core llama-index-llms-openai httpx
export LEADOCEAN_API_KEY="..."   # from app.leadocean.io
export OPENAI_API_KEY="..."

Step 1: Write one request helper with a record budget

All three tools go through one function. It sends the x-api-key header, reads meta.credits after each call and refuses to start a call that would pass your cap. A runaway agent then costs at most the cap, not the plan.

LeadOcean meters in records, not requests. One person returned is one record, and a rejected request is never metered (LeadOcean docs, October 2026). The guard therefore tracks meta.credits.

import os, asyncio, httpx

BASE = "https://api.leadocean.io"
HEADERS = {"x-api-key": os.environ["LEADOCEAN_API_KEY"]}
MAX_RECORDS = 25          # hard stop for one agent run
spent = 0                 # records spent so far

class OutOfRecords(Exception):
    """Free plan has spent its 1,000 records, or our own cap is reached."""

async def call(method, path, *, params=None, json=None, cost=0):
    global spent
    if spent + cost > MAX_RECORDS:
        raise OutOfRecords(f"budget cap {MAX_RECORDS} reached ({spent} spent)")
    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=json)
            if r.status_code == 429:
                wait = int(r.headers.get("Retry-After", "1"))
                if wait > 5:      # paid plan reached its fair-use pace: do not loop
                    raise OutOfRecords(f"throttled, retry in {wait}s")
                await asyncio.sleep(wait)
                continue
            if r.status_code == 503:          # transient, never billed
                await asyncio.sleep(2 ** attempt)
                continue
            break
    if r.status_code == 402:
        raise OutOfRecords(r.json()["error"]["message"])
    if r.status_code == 404:
        return {"found": False}
    r.raise_for_status()
    body = r.json()
    spent += (body.get("meta") or {}).get("credits", cost)
    return body

Step 2: Define the three tools

LlamaIndex turns a normal function into a tool. The agent reads the function name, the parameters and the docstring to decide when to call it, so write the docstrings for the model (LlamaIndex docs, September 2026).

The search body is flat JSON. Filters such as title, country, jobLevel and jobFunction are arrays of strings, and limit runs from 1 to 100. The free sizing call takes count=true as a query parameter and returns the total in meta.total.

async def count_people(title: list[str], country: list[str]) -> dict:
    """Free. Count people matching job title keywords and ISO country codes
    (for example ["VP Sales"] and ["CA"]). Always call this before search_people."""
    body = await call("POST", "/v1/people/search",
                      params={"count": "true"},
                      json={"title": title, "country": country, "limit": 1})
    return {"total": body["meta"]["total"], "capped_at": 100000}

async def search_people(title: list[str], country: list[str], limit: int = 5) -> dict:
    """Spends one record per person returned. Returns up to `limit` people
    (max 10) with a person_id each. Use enrich_person for contact details."""
    limit = max(1, min(limit, 10))
    body = await call("POST", "/v1/people/search", cost=limit,
                      json={"title": title, "country": country, "limit": limit})
    return {"people": body["data"], "records_spent_this_run": spent}

async def enrich_person(person_id: str, reveal_email: bool = False) -> dict:
    """Spends one record. Enrich one person by the person_id from search_people.
    Set reveal_email=true only when the user asked for contact details."""
    return await call("POST", "/v1/people/enrich", cost=1,
                      json={"person_id": person_id, "reveal_email": reveal_email})

Two details matter. Enrichment takes exactly one of linkedin_url, email, phone or person_id, and person_id is the one that always works. There is no name plus domain lookup, so the agent must search first and enrich second.

Step 3: Build the FunctionAgent

Pass the functions straight to FunctionAgent. The system prompt carries the cost rules, because the model cannot see your budget unless you tell it.

from llama_index.core.agent.workflow import FunctionAgent
from llama_index.llms.openai import OpenAI

agent = FunctionAgent(
    tools=[count_people, search_people, enrich_person],
    llm=OpenAI(model="gpt-4o-mini"),
    system_prompt=(
        "You find B2B contacts with LeadOcean tools. "
        "Call count_people first and report the total. "
        "search_people and enrich_person spend records: never call them more "
        "than needed, never search more than 5 people, enrich at most 3. "
        "If a tool reports a budget or record limit, stop and say so."
    ),
)

Step 4: Run it and handle the stop conditions

The LlamaIndex agent is async, so wrap the call in asyncio.run. Catch OutOfRecords at the top level so the run ends with a message instead of a stack trace.

async def main():
    try:
        out = await agent.run(
            user_msg="Find 5 VPs of sales in Canada. Enrich the first 3 "
                     "and tell me which have a deliverable work email."
        )
        print(out)
    except OutOfRecords as e:
        print(f"Stopped: {e}")
    print(f"Records spent this run: {spent}")

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

An expected run is one free count, one search of 5 people and three enrichments: 5 + 3 = 8 records. If the model ignores the prompt and loops, the cap in Step 1 ends it at 25.

Enrich only the people the user cares about. A search row already tells you whether an address exists: has_email, email_status and email_type are published as flags before you buy anything. LeadOcean does not refund bounced emails, so read email_status before sending.

Step 5: Use the MCP server instead of REST

The hosted MCP server exposes 13 tools, including the free count_leads, list_filters and list_enum_values. Its URL is https://api.leadocean.io/mcp, and it signs in with OAuth 2.1 rather than a pasted key.

LlamaIndex consumes MCP servers through llama-index-tools-mcp, and its client supports OAuth through BasicMCPClient.with_oauth (package README, October 2026). LeadOcean's OAuth server supports dynamic client registration at https://api.leadocean.io/oauth/register.

from llama_index.tools.mcp import BasicMCPClient, McpToolSpec

client = BasicMCPClient.with_oauth(
    "https://api.leadocean.io/mcp",
    client_name="My LlamaIndex agent",
    redirect_uris=["http://localhost:3000/callback"],
    redirect_handler=lambda url: print(f"Open this URL: {url}"),
    callback_handler=lambda: (input("Paste the code: "), None),
)
tools = await McpToolSpec(client=client).to_tool_list_async()

agent = FunctionAgent(tools=tools, llm=OpenAI(model="gpt-4o-mini"),
                      system_prompt="You find B2B contacts with LeadOcean tools.")

Pick the route by where the agent runs. A scheduled script or a server job should use REST with a key. An interactive agent a person supervises can use MCP and skip key handling. MCP costs the same: search_leads is one record per person returned, get_lead is one record.

Step 6: Check cost per run

The cost of a run is the sum of two numbers: records spent at LeadOcean and tokens spent at your model provider. LeadOcean bills records. Your model provider bills tokens, and this guide does not measure them.

Check what a run spent with the free account read, then keep the record side small with three habits.

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

• Count first. The count call is free and tells the agent whether the audience is 40 people or 40,000 before it buys a page.
• Ask for small pages. A search page costs one record per person returned, so limit: 5 costs 5 and limit: 100 costs 100.
• Enrich on demand. Search returns a thin row with no contact points, so enrichment is where the contact details come from. Enrich only the people the user picks.

Paid plans (Solo $97 a month, Agency $497 a month) have no per-record price. They have a fair-use pace, so the cap in Step 1 protects the free allowance and your pace. Past the pace a key slows down until the window resets rather than being cut off. Support raises a pace for a real workload.

Step 7: Page past the first 100 people

Page through larger audiences with cursor. Pass back exactly the meta.nextCursor string from the previous page and stop when it comes back null. One search pages through at most 10,000 rows by default, and the last page then carries meta.depthCapped: true.

async def all_people(filters: dict, max_rows: int = 200):
    rows, cursor = [], None
    while len(rows) < max_rows:
        body = await call("POST", "/v1/people/search", cost=100,
                          json={**filters, "limit": 100, **({"cursor": cursor} if cursor else {})})
        rows += body["data"]
        cursor = body["meta"].get("nextCursor")
        if not cursor:
            break
    return rows

For lists beyond a few hundred rows, do not page from an agent. Use POST /v1/exports, which takes up to 50,000 rows a request and 500,000 a day. Let the agent size the audience and set the filters, and let a job pull the file.

What you get

The free count is the first thing the agent does. This is the real output of the leadocean_count_leads MCP tool on 2026-10-21. The filters were jobLevel VP, jobFunction Sales & Business Development and country CA. It used the tool's default email statuses (verified, catch_all_valid, catch_all) and cost 0 records.

{
  "total": 5574,
  "totalIsExact": true,
  "totalAll": 18800,
  "credits": 0,
  "unsupportedFilters": []
}

A search row and an enrichment then return the documented person shape. This is the response shape from the API reference, with placeholder values.

{
  "success": true,
  "data": {
    "profile_data": { "profile_full_name": "Jane Doe", "profile_headline": "VP Sales at Acme" },
    "contact_data": {
      "has_email": true,
      "email_status": "verified",
      "contact_emails": [{ "email": "jane@acme.com", "type": "work", "status": "verified" }]
    }
  },
  "meta": { "count": 1, "credits": 1, "key_hash": "..." }
}

contact_emails and contact_phones appear only when you send reveal_email: true. The trimmed fields are illustrative: read the full schema in the API reference.

Troubleshooting

FAQ