How to Build an AI SDR (With the OpenAI Agents SDK)

export LEADOCEAN_API_KEY="paste-your-key-here"

import os, asyncio, json, httpx

BASE = "https://api.leadocean.io"
HEADERS = {"x-api-key": os.environ["LEADOCEAN_API_KEY"], "content-type": "application/json"}
SPENT = {"records": 0}
RECORD_CAP = 20          # hard ceiling for one run

async def call(method, path, params=None, body=None):
    async with httpx.AsyncClient(timeout=30) as client:
        for attempt in range(2):
            r = await client.request(method, BASE + path, headers=HEADERS,
                                     params=params, json=body)
            if r.status_code == 429 and attempt == 0:
                await asyncio.sleep(int(r.headers.get("Retry-After", "1")))
                continue
            break
    if r.status_code == 402:
        return None, "The free records are spent. Stop and tell the user."
    if r.status_code == 429:
        return None, "Rate limited twice. Stop and report what you have."
    if r.status_code == 404:
        return None, "Nobody matched. This lookup cost nothing."
    if r.status_code >= 400:
        return None, f"HTTP {r.status_code}: {r.text[:300]}"
    return r.json(), None

from typing import Optional
from agents import function_tool

def fail(msg: str) -> str:
    return "ERROR: " + msg

@function_tool
async def count_audience(job_level: list[str], job_function: list[str],
                         country: list[str], industry: list[str],
                         employee_range: list[str]) -> str:
    """Count people for an ICP. Free. Call this first.

    Args:
        job_level: exact values such as "VP".
        job_function: exact values such as "Sales & Business Development".
        country: country codes such as "US".
        industry: exact values such as "Software Development".
        employee_range: bands such as "51-200".
    """
    body = {"jobLevel": job_level, "jobFunction": job_function, "country": country,
            "industry": industry, "employeeRange": employee_range, "limit": 1}
    data, e = await call("POST", "/v1/people/search", params={"count": "true"}, body=body)
    return fail(e) if e else json.dumps(data["meta"])

@function_tool
async def search_people(job_level: list[str], job_function: list[str],
                        country: list[str], industry: list[str],
                        employee_range: list[str], limit: int = 10) -> str:
    """Search people for an ICP. Costs 1 record per person returned. Max 10.

    Args:
        job_level: exact values such as "VP".
        job_function: exact values such as "Sales & Business Development".
        country: country codes such as "US".
        industry: exact values such as "Software Development".
        employee_range: bands such as "51-200".
        limit: how many people to return, 1 to 10.
    """
    n = max(1, min(limit, 10))
    if SPENT["records"] + n > RECORD_CAP:
        return fail("Record cap reached for this run. Do not search again.")
    body = {"jobLevel": job_level, "jobFunction": job_function, "country": country,
            "industry": industry, "employeeRange": employee_range, "limit": n}
    data, e = await call("POST", "/v1/people/search", body=body)
    if e:
        return fail(e)
    SPENT["records"] += len(data["data"])
    return json.dumps(data)

@function_tool
async def enrich_person(person_id: str) -> str:
    """Full lookup for one person by person_id from a search row. Costs 1 record."""
    if SPENT["records"] + 1 > RECORD_CAP:
        return fail("Record cap reached for this run. Do not enrich again.")
    data, e = await call("POST", "/v1/people/enrich",
                         body={"person_id": person_id, "reveal_email": True})
    if e:
        return fail(e)
    SPENT["records"] += 1
    return json.dumps(data)

INSTRUCTIONS = """You are an SDR for Acme, which sells project management software.

ICP: VP of Sales at US software companies with 51 to 200 staff.

Procedure:
1. Call count_audience with the ICP filters. Report the total.
2. Call search_people with the same filters and limit 10.
3. Pick the 3 best fits: title closest to "VP Sales", email_status verified.
4. Call enrich_person for those 3 only.
5. For each, write a first-touch email: subject under 6 words, body under
   90 words, one specific line taken from resume_data (a past role, school
   or certification), one question as the call to action.

Rules: use only facts that came back in a tool result. If a field is missing,
leave it out. If a tool returns ERROR, stop and report it. Do not send anything.
Return the 3 drafts as a numbered list with name, title, company, email.

Filter values are exact strings: job_level "VP", job_function
"Sales & Business Development", country "US", industry "Software Development",
employee_range "51-200"."""

from agents import Agent, Runner
from agents.exceptions import MaxTurnsExceeded

sdr = Agent(
    name="Acme SDR",
    instructions=INSTRUCTIONS,
    tools=[count_audience, search_people, enrich_person],
)

async def main():
    try:
        result = await Runner.run(sdr, "Draft this week's outreach.", max_turns=12)
        print(result.final_output)
    except MaxTurnsExceeded:
        print("Stopped: hit max_turns before the drafts were done.")
    print(f"LeadOcean records spent: {SPENT['records']}")

asyncio.run(main())

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

{
  "total": 1649,
  "totalIsExact": true,
  "distinctCompanies": 1125,
  "defaults": "Counted only addresses worth sending to (emailStatus: verified or catch_all_valid or catch_all)"
}

{
  "success": true,
  "data": [
    {
      "profile_data": { "...": "Jane Doe, VP Sales at Acme" },
      "contact_data": { "has_email": true, "email_status": "verified" },
      "meta": { "source_ids": { "person_id": "<digits>" } }
    }
  ],
  "meta": { "count": 10, "limit": 10, "nextCursor": "<opaque cursor>" }
}