How to Build a Lead Qualification Agent (Pydantic AI and LeadOcean)

pip install pydantic-ai httpx
export LEADOCEAN_API_KEY="..."   # from app.leadocean.io
export ANTHROPIC_API_KEY="..."

import os, time
from dataclasses import dataclass
import httpx

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

@dataclass
class Budget:
    cap: int = 25
    spent: int = 0

def call(budget: Budget, method, path, *, params=None, json=None, cost=0):
    """Returns the body, or {"error": ...} the model can read. Never raises."""
    if budget.spent + cost > budget.cap:
        return {"error": f"STOP: budget cap {budget.cap} reached ({budget.spent} spent)"}
    for attempt in range(3):
        r = httpx.request(method, BASE + path, headers=HEADERS,
                          params=params, json=json, timeout=30)
        if r.status_code == 429:
            wait = int(r.headers.get("Retry-After", "1"))
            if wait > 5:
                return {"error": f"STOP: throttled, retry in {wait}s"}
            time.sleep(wait)
            continue
        if r.status_code == 503:          # transient, never billed
            time.sleep(2 ** attempt)
            continue
        break
    if r.status_code == 402:
        return {"error": "STOP: out of records (402). Tell the user."}
    if r.status_code == 404:
        return {"found": False}
    if r.status_code >= 400:
        return {"error": f"{r.status_code}: {r.text[:200]}"}
    body = r.json()
    budget.spent += (body.get("meta") or {}).get("credits", cost)
    return body

from pydantic import BaseModel, Field
from pydantic_ai import Agent, RunContext

class ScoredLead(BaseModel):
    person_id: str
    name: str
    title: str
    score: int = Field(ge=1, le=5)
    reason: str

class Report(BaseModel):
    audience_total: int
    leads: list[ScoredLead]
    stopped_early: str | None = None

agent = Agent(
    "anthropic:claude-sonnet-5-5",
    deps_type=Budget,
    output_type=Report,
    instructions=(
        "You qualify B2B leads. ICP: VP or head of sales at a software company. "
        "Call count_people first and report the total. Search at most 5 people, "
        "enrich at most 3. Score each enriched person 1 to 5 against the ICP "
        "with one line of reasoning. If a tool result starts with STOP, stop and "
        "put the reason in stopped_early."
    ),
)

@agent.tool
def count_people(ctx: RunContext[Budget], title: list[str], country: list[str]) -> dict:
    """Free. Count people by job title keywords and ISO country codes,
    for example ["VP Sales"] and ["CA"]. Call this before search_people."""
    body = call(ctx.deps, "POST", "/v1/people/search", params={"count": "true"},
                json={"title": title, "country": country, "limit": 1})
    return body if "error" in body else {"total": body["meta"]["total"]}

@agent.tool
def search_people(ctx: RunContext[Budget], title: list[str],
                  country: list[str], limit: int = 5) -> dict:
    """Spends one record per person returned (max 10). Each row has a person_id.
    Use enrich_person for contact details."""
    limit = max(1, min(limit, 10))
    body = call(ctx.deps, "POST", "/v1/people/search", cost=limit,
                json={"title": title, "country": country, "limit": limit})
    return body if "error" in body else {"people": body["data"]}

@agent.tool
def enrich_person(ctx: RunContext[Budget], person_id: str) -> dict:
    """Spends one record. Enrich one person by person_id from search_people."""
    return call(ctx.deps, "POST", "/v1/people/enrich", cost=1,
                json={"person_id": person_id, "reveal_email": False})

from pydantic_ai import UsageLimits

budget = Budget(cap=25)
result = agent.run_sync(
    "Find 5 VPs of sales in Canada, enrich the first 3 and score them.",
    deps=budget,
    usage_limits=UsageLimits(request_limit=12, tool_calls_limit=10),
)
print(result.output.model_dump_json(indent=2))
print(f"LeadOcean records spent: {budget.spent}")
print(result.usage())

from pydantic_ai.mcp import MCPToolset

toolset = MCPToolset("https://api.leadocean.io/mcp")  # add OAuth via a fastmcp.Client
mcp_agent = Agent("anthropic:claude-sonnet-5-5", toolsets=[toolset])

{ "total": 658, "totalIsExact": true, "totalAll": 2661, "distinctCompanies": 590, "credits": 0 }

{
  "success": true,
  "data": {
    "profile_data": { "profile_full_name": "Jane Doe", "...": "headline, location" },
    "contact_data": { "has_email": true, "email_status": "verified" }
  },
  "meta": { "count": 1, "credits": 1 }
}