How to Build an AI SDR (CrewAI and LeadOcean)

curl -s "https://api.leadocean.io/v1/people/search?count=true&limit=1" \
  -H "x-api-key: $LEADOCEAN_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"jobLevel":["VP"],"jobFunction":["Sales & Business Development"],"country":["CA"]}'

import json, os, time
import requests
from typing import Type
from pydantic import BaseModel, Field
from crewai.tools import BaseTool

API = "https://api.leadocean.io"
HEADERS = {"x-api-key": os.environ["LEADOCEAN_API_KEY"], "Content-Type": "application/json"}


def call(method, path, **kw):
    """One request with the 402 and 429 handling the agent needs."""
    for attempt in range(3):
        r = requests.request(method, API + path, headers=HEADERS, timeout=30, **kw)
        if r.status_code == 429:
            time.sleep(int(r.headers.get("Retry-After", "1")))
            continue
        if r.status_code == 402:
            return {"stop": "LeadOcean records are spent. Do not retry. Tell the user to upgrade."}
        if r.status_code == 404:
            return {"not_found": True}
        r.raise_for_status()
        return r.json()
    return {"stop": "Rate limited three times. Wait a minute and try again."}


class SearchInput(BaseModel):
    titles: list[str] = Field(..., description='Job title keywords, e.g. ["VP Sales"]')
    country: list[str] = Field(default=["US"], description="ISO country codes, e.g. ['US','CA']")
    limit: int = Field(default=10, ge=1, le=100, description="Rows to return. Each row costs one record.")


class SearchPeople(BaseTool):
    name: str = "search_people"
    description: str = "Find people by job title and country. Returns up to `limit` rows, one record each."
    args_schema: Type[BaseModel] = SearchInput

    def _run(self, titles: list[str], country: list[str] = ["US"], limit: int = 10) -> str:
        body = {"title": titles, "country": country, "limit": min(limit, 10)}
        out = call("POST", "/v1/people/search", json=body)
        return json.dumps(out)[:6000]

class EnrichInput(BaseModel):
    person_id: str = Field(..., description="person_id from a search row")


class EnrichPerson(BaseTool):
    name: str = "enrich_person"
    description: str = "Get the email, status and current role for one person_id. Costs one record."
    args_schema: Type[BaseModel] = EnrichInput

    def _run(self, person_id: str) -> str:
        out = call("POST", "/v1/people/enrich", json={"person_id": person_id, "reveal_email": True})
        return json.dumps(out)[:6000]

from crewai import Agent, Task, Crew, Process

researcher = Agent(
    role="Prospect researcher",
    goal="Find 5 VPs of Sales in Canada with a verified work email",
    backstory="You pick people, then enrich only the first 5. You never enrich twice.",
    tools=[SearchPeople(), EnrichPerson()],
    max_iter=8,
    verbose=True,
)
writer = Agent(
    role="Outbound writer",
    goal="Write one short first email per prospect",
    backstory="Plain, under 80 words, one question at the end. No invented facts.",
    verbose=True,
)

find = Task(
    description="Search for VPs of Sales in CA (limit 10). Enrich the first 5 by person_id. "
                "Keep only people whose email_status is verified or catch_all_valid.",
    expected_output="A list of name, title, company, email and email_status.",
    agent=researcher,
)
draft = Task(
    description="For each prospect, write a first email about outbound data. Use only the fields given.",
    expected_output="One email per prospect with a subject line.",
    agent=writer,
    context=[find],
)

crew = Crew(agents=[researcher, writer], tasks=[find, draft], process=Process.sequential)
print(crew.kickoff())

from crewai.mcp import MCPServerHTTP

researcher = Agent(
    role="Prospect researcher",
    goal="Find 5 VPs of Sales in Canada",
    backstory="Count first, then search.",
    mcps=[MCPServerHTTP(url="https://api.leadocean.io/mcp", streamable=True)],
)

{
  "total": 5574,
  "totalIsExact": true,
  "totalAll": 18800,
  "distinctCompanies": 3873,
  "credits": 0
}

{
  "success": true,
  "data": {
    "profile_data": { "...": "Jane Doe, VP Sales, Acme" },
    "contact_data": {
      "has_email": true,
      "email_status": "verified",
      "contact_emails": [{ "...": "jane.doe@acme.com" }],
      "contact_current_experiences": [{ "...": "current role at Acme" }]
    },
    "resume_data": { "...": "past roles, education, when held" }
  }
}