How to Give an OpenAI Agents SDK Agent B2B Data (REST Tools and MCP)

pip install openai-agents httpx
export LEADOCEAN_API_KEY="..."   # from app.leadocean.io
export OPENAI_API_KEY="..."

import asyncio
import os
from dataclasses import dataclass, field

import httpx

BASE_URL = "https://api.leadocean.io"


@dataclass
class Ctx:
    budget: int = 25  # records this run may spend
    spent: int = 0
    http: httpx.AsyncClient = field(default_factory=lambda: httpx.AsyncClient(
        base_url=BASE_URL,
        headers={"x-api-key": os.environ["LEADOCEAN_API_KEY"]},
        timeout=30,
    ))


async def call(ctx: Ctx, method: str, path: str, **kw) -> dict:
    """One LeadOcean request. Failures return {"error": ...}, never raise."""
    for attempt in range(3):
        r = await ctx.http.request(method, path, **kw)
        wait = int(r.headers.get("Retry-After", "1"))
        if r.status_code == 429 and wait <= 5:  # above the plan's per-second limit
            await asyncio.sleep(wait)
            continue
        if r.status_code == 503:  # transient, never billed
            await asyncio.sleep(2**attempt)
            continue
        break

    if r.status_code == 200:
        body = r.json()
        ctx.spent += body.get("meta", {}).get("credits", 0)
        return body
    if r.status_code == 404:  # nobody matched, costs nothing
        return {"found": False}

    try:
        message = r.json().get("error", {}).get("message", r.text)
    except ValueError:
        message = r.text
    if r.status_code == 400:
        return {"error": f"Rejected: {message}. Check enum values with list_values."}
    if r.status_code == 402:
        return {"error": "STOP. Out of records on the Free plan. Tell the user."}
    if r.status_code == 429:
        return {"error": f"STOP. Rate limited, retry after {wait} seconds."}
    return {"error": f"LeadOcean returned {r.status_code}: {message}"}

import asyncio

from agents import Agent, RunContextWrapper, Runner, function_tool

from lo import Ctx, call


@function_tool
async def list_values(ctx: RunContextWrapper[Ctx], name: str, q: str = "") -> str:
    """Free. Exact values for a filter.

    Args:
        name: One of job_level, job_function, country, seniority, industry.
        q: Optional text to narrow the list.
    """
    params = {"limit": 50, **({"q": q} if q else {})}
    return str(await call(ctx.context, "GET", f"/v1/enums/{name}", params=params))


@function_tool
async def count_people(ctx: RunContextWrapper[Ctx], job_level: str, country: str) -> str:
    """Free. Size a segment before searching.

    Args:
        job_level: Exact value, for example VP.
        country: Two-letter code, for example CA.
    """
    body = await call(ctx.context, "POST", "/v1/people/search",
                      params={"count": "true"},
                      json={"jobLevel": [job_level], "country": [country], "limit": 1})
    return str(body["meta"]["total"]) if "meta" in body else str(body)


def trim(row: dict) -> dict:
    """Keep six fields. The model does not need the whole record."""
    contact = row.get("contact_data", {})
    job = (contact.get("contact_current_experiences") or [{}])[0]
    return {
        "person_id": row.get("meta", {}).get("source_ids", {}).get("person_id"),
        "name": row.get("profile_data", {}).get("profile_full_name"),
        "title": job.get("job_title"),
        "company": job.get("company_name"),
        "domain": job.get("company_domain"),
        "email_status": contact.get("email_status"),
    }


@function_tool
async def search_people(ctx: RunContextWrapper[Ctx], job_level: str, country: str,
                        limit: int = 5) -> str:
    """Costs 1 record per person returned. Returns name, title, company, person_id.

    Args:
        job_level: Exact value, for example VP.
        country: Two-letter code, for example CA.
        limit: People to return, 1 to 25.
    """
    c = ctx.context
    left = c.budget - c.spent
    if left <= 0:
        return "STOP. Record budget for this run is spent."
    body = await call(c, "POST", "/v1/people/search", json={
        "jobLevel": [job_level], "country": [country], "limit": min(limit, left, 25)})
    return str([trim(r) for r in body["data"]]) if "data" in body else str(body)


@function_tool
async def enrich_person(ctx: RunContextWrapper[Ctx], person_id: str) -> str:
    """Costs 1 record. Returns the profile and revealed emails for one person_id.

    Args:
        person_id: The person_id from a search row.
    """
    c = ctx.context
    if c.budget - c.spent < 1:
        return "STOP. Record budget for this run is spent."
    body = await call(c, "POST", "/v1/people/enrich",
                      json={"person_id": person_id, "reveal_email": True})
    return str(body.get("data", body))


agent = Agent(
    name="Prospector",
    instructions=(
        "You build small B2B prospect lists with LeadOcean. Always call "
        "count_people first. Search only when the count is under 100,000. "
        "Enrich at most 2 people, and only when asked for contact details. "
        "Look up enum values with list_values instead of guessing. "
        "If a tool result starts with STOP, stop and tell the user why."
    ),
    tools=[list_values, count_people, search_people, enrich_person],
)


async def main() -> None:
    ctx = Ctx(budget=25)
    try:
        result = await Runner.run(
            agent,
            "Find 5 VPs in Canada and tell me how many exist in total.",
            context=ctx,
            max_turns=10,
        )
        print(result.final_output)
        print(f"LeadOcean records spent: {ctx.spent}")
    finally:
        await ctx.http.aclose()


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

python agent.py

[mcp_servers.leadocean]
url = "https://api.leadocean.io/mcp"

{"name":"job_level","total":6,"matched":6,"values":["C-Team","Director","Manager","Other","Staff","VP"],"truncated":false}

{"total": 5574, "totalIsExact": true, "totalAll": 18800, "distinctCompanies": 3873, "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": "..." }
}