How to Build an AI SDR (With Pydantic AI and LeadOcean)

You build an AI SDR in Pydantic AI by registering three typed functions as tools: count, search and enrich. LeadOcean serves all three over REST. The agent sizes the list for free, pulls ten people, enriches the best five and returns a validated list.

TL;DR

• Time: 30 to 40 minutes to a working agent.
• Records: about 15 per run. Ten from search, five from enrich. The count call is free.
• Stack: Python, pydantic-ai, httpx, and a LeadOcean key from the free plan.

Prerequisites

• A LeadOcean account on the free plan: 1,000 records for life, no card, 5 requests a second. Create a key at app.leadocean.io and export it as $LEADOCEAN_API_KEY.
• Python 3.10 or newer.
• A model key for the provider you pick, for example OPENAI_API_KEY.
• pip install pydantic-ai httpx.

LeadOcean also runs a hosted MCP server at https://api.leadocean.io/mcp with 13 tools. Pydantic AI can act as an MCP client through MCPToolset (checked September 2026). The server signs in with OAuth 2.1, so there is no key to paste. This guide uses REST because a REST wrapper is easier to cap and test.

Step 1: Wrap the API in one small client

Every call goes to https://api.leadocean.io with the x-api-key header. Put the 402 and 429 handling in one place so each tool stays short.

import os, asyncio
import httpx

BASE = "https://api.leadocean.io"
HEADERS = {"x-api-key": os.environ["LEADOCEAN_API_KEY"]}

class OutOfRecords(Exception):
    pass

async def call(method, path, params=None, body=None, retries=3):
    async with httpx.AsyncClient(base_url=BASE, headers=HEADERS, timeout=30) as c:
        for _ in range(retries):
            r = await c.request(method, path, params=params, json=body)
            if r.status_code == 429:
                await asyncio.sleep(float(r.headers.get("Retry-After", 1)))
                continue
            if r.status_code == 402:
                raise OutOfRecords("LeadOcean records are spent.")
            if r.status_code == 404:
                return None
            r.raise_for_status()
            return r.json()
    raise RuntimeError("Still rate limited after retries")

A 429 means you went above the plan's rate limit (5 requests a second on Free). Retry-After says how long to wait. A 402 means the records are gone, so retrying does nothing.

Step 2: Define the output type and the agent

Pydantic AI takes a Pydantic model as output_type and forces the final answer to match it (output docs, checked September 2026). That gives you a typed lead list, not prose.

from pydantic import BaseModel
from pydantic_ai import Agent

class Lead(BaseModel):
    person_id: str
    name: str
    title: str
    company: str
    work_email: str | None = None

class Shortlist(BaseModel):
    total_matching: int
    leads: list[Lead]
    note: str = ""

SYSTEM = """You are an SDR research agent.
1. Always call count_prospects first. If total is over 5,000, narrow the filters.
2. Search at most 10 people. Never page further.
3. Enrich at most 5 people, the best title fits only.
4. If a tool says records are spent, stop and report what you have."""

agent = Agent("openai:gpt-4o-mini", instructions=SYSTEM, output_type=Shortlist)

Swap the model string for any provider you have a key for. The prompt caps are a guide. Step 5 adds a hard stop.

Step 3: Register the three tools

The docs show @agent.tool_plain for tools that need no run context (checked September 2026). The docstring and type hints become the tool schema, so write them for the model.

The count tool sends count=true as a query parameter with limit=1. That call is free and returns meta.total, capped at 100,000. Filters are arrays of strings and enum values are exact.

@agent.tool_plain
async def count_prospects(job_level: list[str], job_function: list[str],
                          country: list[str]) -> dict:
    """Free. Count people matching filters. Call this before searching.
    job_level e.g. ["VP"]; job_function e.g. ["Sales & Business Development"];
    country e.g. ["US"]."""
    body = {"jobLevel": job_level, "jobFunction": job_function,
            "country": country, "limit": 1}
    res = await call("POST", "/v1/people/search", params={"count": "true"}, body=body)
    return res["meta"]

@agent.tool_plain
async def search_prospects(job_level: list[str], job_function: list[str],
                           country: list[str], limit: int = 10) -> list:
    """Spends 1 record per person returned. Max 100 per page."""
    body = {"jobLevel": job_level, "jobFunction": job_function,
            "country": country, "limit": min(limit, 100)}
    data = (await call("POST", "/v1/people/search", body=body))["data"]
    guard(len(data))
    return data

@agent.tool_plain
async def enrich_prospect(person_id: str) -> dict:
    """Spends 1 record. Returns profile, current role and the work email."""
    res = await call("POST", "/v1/people/enrich",
                     body={"person_id": person_id, "reveal_email": True})
    if res is None:
        return {"result": "No match. Nothing was spent."}
    guard(1)
    return res

Enrich takes exactly one of linkedin_url, email, phone or person_id. There is no name plus domain lookup, so the agent searches first and enriches by person_id, which each search row carries at meta.source_ids.person_id. Step 5 defines guard.

Step 4: Run it

from pydantic_ai import UsageLimits

async def main():
    result = await agent.run(
        "Find US VPs of Sales. Enrich the top 5.",
        usage_limits=UsageLimits(request_limit=15),
    )
    for lead in result.output.leads:
        print(lead.name, lead.title, lead.company, lead.work_email)

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

UsageLimits is Pydantic AI's run-level cap on model requests, tokens and tool calls (agents docs, checked September 2026). It limits model spend. It does not limit LeadOcean records, so Step 5 does that.

Step 5: Cap records in code

A prompt is not a budget. Add a counter so a loop cannot burn the 1,000 free records. Define it above the tools.

SPENT = {"records": 0}
LIMIT = 25

def guard(n):
    SPENT["records"] += n
    if SPENT["records"] > LIMIT:
        raise OutOfRecords(f"Run cap of {LIMIT} records reached")

One record is one person you receive. A page of 10 is 10 records, and an enrichment that finds nothing costs nothing. Ten searched plus five enriched is about 15 records, so the free plan covers around 65 runs.

Step 6: Turn 402 into a message the model can read

Pydantic AI says an exception raised inside a tool that is not ModelRetry propagates out and ends the run (tools docs, checked September 2026). ModelRetry is for a bad call the model can fix. A spent balance is neither, so catch it and return a plain result. Wrap the body of each tool, or use a small decorator.

import functools

def safe(fn):
    @functools.wraps(fn)
    async def wrapper(*a, **kw):
        try:
            return await fn(*a, **kw)
        except OutOfRecords as e:
            return {"stop": True, "reason": str(e)}
        except RuntimeError as e:
            return {"stop": True, "reason": str(e)}
    return wrapper

Apply @agent.tool_plain on top of @safe for each tool. For an empty result caused by a bad enum value, raise ModelRetry("jobLevel must be one of C-Team, VP, Director, Manager, Staff, Other") so the model corrects itself.

What you get

The count call is free. On 2026-10-21 leadocean_count_leads returned this for VPs in Sales & Business Development in the US. It counts verified, catch_all_valid and catch_all emails by default:

{
  "total": 129494,
  "totalIsExact": true,
  "totalAll": 350517,
  "distinctCompanies": 58121
}

That is 129,494 people with a mailable address. Without an email filter the same filters match 350,517. Over REST, meta.total stops at 100,000, so the agent sees 100,000 for this list and should narrow it before it searches.

A search page follows the documented shape. This is the response shape from the API reference, with placeholder values:

{
  "data": [{
    "profile_data": { "...": "Jane Doe, VP Sales, Acme" },
    "meta": { "source_ids": { "person_id": "per_000000" } }
  }],
  "meta": { "total": 1, "credits": 1, "nextCursor": "..." }
}

The agent's final answer is a Shortlist with leads filled from enrich. Enrich returns the current role in contact_current_experiences[], phones in contact_phones[].phone and, since October 2026, career history in resume_data.

Troubleshooting

FAQ