Ultimate Guide to Agentic Lead Generation in 2026 + Code Samples + GitHub Project

Most lead generation content is useless if you need to actually build the thing. A customer sought out recommendations for his problem: “I have been experimenting with NinjaPear using my AI agents to find investors and leads for my company.” In this guide, I’m going to show you how to build a full agentic lead generation system with PydanticAI, NinjaPear, and a small set of supporting tools, including the exact 4 loops, the code structure, and the GitHub project you can ship.

Build the system first

If I were a founder building outbound against Stripe and Shopify, I would not start with prompts. I would start with a system.

These are the 4 loops:

• Loop 1, customer loop: pull likely customers of Stripe and Shopify, because a competitor's customer is better evidence than a random firmographic export
• Loop 2, competitor loop: expand into adjacent companies, because most teams do not have a lead shortage, they have a market map shortage
• Loop 3, people loop: find role-alike buyers only after account qualification, because people-first prospecting usually gives you a larger pile of irrelevant contacts
• Loop 4, timing loop: enrich and monitor those accounts for changes, because without timing you do not have agentic lead generation, you have a nicer spreadsheet

That is what this guide does. It wires those four loops into one workflow and shows the code and response shapes I would use if I were trying to sell into the same market as Stripe and Shopify.

The whole workflow is this:

seed domains -> customer loop -> competitor loop -> people loop -> timing loop -> queue

And the main() is boring on purpose.

def main():
    seeds = ["https://stripe.com", "https://shopify.com"]
    accounts = pull_competitor_customers(seeds)
    accounts = expand_competitors(accounts)
    people = find_similar_people(accounts)
    enriched = enrich_and_monitor(accounts, people)
    queue = build_outreach_queue(enriched)
    return queue

That is enough structure to build around.

When I was running FluxoMetric, this was my mistake for a while. I thought better modeling would fix bad market inputs. It does not. It just gives you a cleaner wrong answer.

The stack

Keep the stack simple:

• NinjaPear for company, customer, competitor, people, funding, updates, and monitor data
• PydanticAI for orchestration, typed tools, and validated outputs
• OpenRouter as the model gateway
• Mock helpers for email verification and email sending

Install what you need:

pip install ninjapear pydantic-ai httpx pydantic python-dotenv

Minimal settings model:

from pydantic import BaseModel
import os

class Settings(BaseModel):
    ninjapear_api_key: str
    openrouter_api_key: str

settings = Settings(
    ninjapear_api_key=os.environ["NINJAPEAR_API_KEY"],
    openrouter_api_key=os.environ["OPENROUTER_API_KEY"],
)

PydanticAI through OpenRouter:

from pydantic_ai import Agent
from pydantic_ai.models.openrouter import OpenRouterModel
from pydantic_ai.providers.openrouter import OpenRouterProvider

model = OpenRouterModel(
    "anthropic/claude-sonnet-4-5",
    provider=OpenRouterProvider(api_key=settings.openrouter_api_key),
)
agent = Agent(model)

My opinion here is straightforward. The model matters less than the data and the wiring. Most agentic lead generation demos fail because they are toys sitting on bad data.

The NinjaPear docs are useful here because they tell you the operational limits up front:

• up to 300 requests/minute
• a 1,500 request burst window over 5 minutes
• trial accounts limited to 2 requests/minute
• many endpoints take 30 to 60 seconds
• recommended timeout: 100 seconds

That is enough to design the pipeline without guessing.

Loop 1: Pull competitor customers

If I were selling something that should land in the same accounts as Stripe or Shopify, this is where I would start.

A competitor’s customer is usually a better lead than a giant stale firmographic export. Not because it is magical. Because it is evidence of actual category demand.

Seed your domains

Start with companies your buyer already compares you against.

seeds = [
    "https://stripe.com",
    "https://shopify.com",
]

Do not start with 500 domains. Start with 2 to 10 and make sure the plumbing works.

Call Customer Listing

Use the Customer Listing API.

These details matter:

• Endpoint: GET /api/v1/customer/listing
• Cost: 1 credit/request + 2 credits/company returned
• Empty results can still cost you
• quality_filter=true should stay on unless you want junk
• Pagination uses next_page and cursor
• page_size supports 1 to 200, default 200

Official SDK setup:

import ninjapear

configuration = ninjapear.Configuration(
    host="https://nubela.co",
    access_token="YOUR_API_KEY",
)

with ninjapear.ApiClient(configuration) as api_client:
    api = ninjapear.CustomerAPIApi(api_client)
    response = api.get_customer_listing(
        website="https://www.stripe.com",
        quality_filter=True,
    )

Wrapper I would keep in a project:

import ninjapear
from urllib.parse import urlparse
from pydantic import BaseModel

class ProspectAccount(BaseModel):
    source_company: str
    relationship_type: str
    name: str
    website: str | None
    description: str | None = None
    specialties: list[str] = []
    x_profile: str | None = None


def normalize_domain(url: str) -> str:
    return urlparse(url).netloc.replace("www.", "")


def get_customer_listing(api_client, website: str, cursor: str | None = None, page_size: int = 200, quality_filter: bool = True):
    api = ninjapear.CustomerAPIApi(api_client)
    return api.get_customer_listing(
        website=website,
        cursor=cursor,
        page_size=page_size,
        quality_filter=quality_filter,
    )

Pagination helper:

from urllib.parse import urlparse, parse_qs


def paginate_customer_listing(api_client, website: str, page_size: int = 200):
    cursor = None
    while True:
        response = get_customer_listing(
            api_client,
            website=website,
            cursor=cursor,
            page_size=page_size,
            quality_filter=True,
        )
        yield response
        next_page = response.get("next_page") if isinstance(response, dict) else getattr(response, "next_page", None)
        if not next_page:
            break
        cursor_values = parse_qs(urlparse(next_page).query).get("cursor", [])
        cursor = cursor_values[0] if cursor_values else None
        if not cursor:
            break

Parse only what matters

For lead generation, I do not treat all relationship types the same.

Relationship type	How to use it
Customer	Primary prospect pool
Investor	Good for market mapping, weak as direct buyer signal
Partner platform	Useful for integrations and ecosystem plays

Flatten the response into one account model:

def flatten_customer_response(source_company: str, payload) -> list[ProspectAccount]:
    rows = []
    data = payload.to_dict() if hasattr(payload, "to_dict") else payload
    for key, relationship_type in {
        "customers": "customer",
        "investors": "investor",
        "partner_platforms": "partner_platform",
    }.items():
        for item in data.get(key, []) or []:
            rows.append(
                ProspectAccount(
                    source_company=source_company,
                    relationship_type=relationship_type,
                    name=item.get("name", "Unknown"),
                    website=item.get("website"),
                    description=item.get("description"),
                    specialties=item.get("specialties") or [],
                    x_profile=item.get("x_profile"),
                )
            )
    return rows

Sample response shape from the docs:

{
  "customers": [
    {
      "name": "Apple",
      "description": "Apple Inc. designs, manufactures, and markets smartphones, personal computers, tablets, wearables, and accessories worldwide.",
      "tagline": "Think different.",
      "website": "https://www.apple.com",
      "company_logo_url": "https://nubela.co/api/v1/company/logo?website=https%3A%2F%2Fwww.apple.com",
      "id": "abc123",
      "industry": 45202030,
      "specialties": ["Technology", "Consumer Electronics"],
      "x_profile": "https://x.com/Apple"
    }
  ],
  "investors": [
    {
      "name": "Sequoia Capital",
      "description": "Sequoia Capital is a venture capital firm focused on technology companies.",
      "tagline": null,
      "website": "https://www.sequoiacap.com",
      "company_logo_url": "https://nubela.co/api/v1/company/logo?website=https%3A%2F%2Fwww.sequoiacap.com",
      "id": "def456",
      "industry": 40203010,
      "specialties": ["Venture Capital", "Growth Equity"],
      "x_profile": "https://x.com/sequoia"
    }
  ],
  "partner_platforms": [
    {
      "name": "Amazon Web Services",
      "description": "Amazon Web Services provides cloud computing platforms and APIs.",
      "tagline": null,
      "website": "https://aws.amazon.com",
      "company_logo_url": "https://nubela.co/api/v1/company/logo?website=https%3A%2F%2Faws.amazon.com",
      "id": "ghi789",
      "industry": 45101010,
      "specialties": ["Cloud Computing", "Infrastructure"],
      "x_profile": "https://x.com/awscloud"
    }
  ],
  "next_page": "https://nubela.co/api/v1/customer/listing?website=https://www.stripe.com&cursor=abc123"
}

That is enough to build a real first pass.

What I would not do is dump investors and partner platforms straight into the outreach queue.

That is the practical case for starting with competitor customer data instead of a shared database.

Dedupe the account list

Dedupe by normalized domain.

from collections import OrderedDict


def dedupe_accounts(accounts: list[ProspectAccount]) -> list[ProspectAccount]:
    deduped = OrderedDict()
    for account in accounts:
        key = normalize_domain(account.website) if account.website else account.name.lower()
        if key not in deduped:
            deduped[key] = account
    return list(deduped.values())

Most teams do not have a lead generation shortage. They have a data hygiene shortage.

Loop 2: Expand competitors

If Stripe and Shopify are your starting map, this loop is how you widen the search without turning it into random noise.

Find adjacent competitors

Use the Competitor Listing endpoint.

Relevant details from the docs:

• Endpoint: GET /api/v1/competitor/listing
• Cost: 2 credits per competitor returned
• There is a minimum request cost, so be selective

Wrapper:

def get_competitors(api_client, website: str, cursor: str | None = None, page_size: int = 50):
    api = ninjapear.CompetitorAPIApi(api_client)
    return api.get_competitor_listing(website=website, cursor=cursor, page_size=page_size)

Sample response shape:

{
  "competitors": [
    {
      "company_name": "Adyen",
      "website": "https://www.adyen.com",
      "product_category_overlap": ["payment processing", "checkout", "fraud tools"],
      "organic_seo_overlap": ["payment gateway", "merchant of record"]
    },
    {
      "company_name": "Checkout.com",
      "website": "https://www.checkout.com",
      "product_category_overlap": ["payments infrastructure"],
      "organic_seo_overlap": ["online payments"]
    }
  ],
  "next_page": null
}

This is enough. You have adjacent companies and the reasons they overlap.

Expand one layer at a time

Do not recurse forever.

def expand_competitors(api_client, seed_domains, max_depth=1):
    visited = set()
    frontier = list(seed_domains)
    discovered = []
    depth = 0

    while frontier and depth < max_depth:
        next_frontier = []
        for website in frontier:
            if website in visited:
                continue
            visited.add(website)
            payload = get_competitors(api_client, website=website)
            data = payload.to_dict() if hasattr(payload, "to_dict") else payload
            for item in data.get("competitors", []) or []:
                comp_website = item.get("website")
                if comp_website and comp_website not in visited:
                    discovered.append(comp_website)
                    next_frontier.append(comp_website)
        frontier = next_frontier
        depth += 1

    return discovered

One layer is usually enough to improve coverage without turning the account set into mush.

Track provenance

Every discovered account should carry provenance:

• which seed exposed it
• which competitor chain exposed it
• what relationship type brought it into the graph

That gives you something to score later. It also gives you something real to message against.

Stop the loop early

Use three brakes:

• max_depth
• domain dedupe
• account caps

If you do not cap this, you end up exploring the whole category and calling it prospecting.

# pseudocode
async for website in frontier:
    if website not in visited and len(discovered) < account_cap:
        fetch competitor listing
        append new domains

That is the argument for market mapping. You need better discovery, not just another seat on the same database.

Loop 3: Find similar people

This is the people layer. It should happen after account filtering.

Start with the account

Only run people discovery on accounts that already passed some fit threshold.

For example:

• competitor customer
• decent headcount band
• recent public updates
• category relevance from the source chain

Do that first. Then go looking for humans.

Pull role-alike people

Use the Similar People endpoint.

class ProspectPerson(BaseModel):
    account_domain: str
    full_name: str | None = None
    role: str | None = None
    work_email: str | None = None
    x_handle: str | None = None
    score: float = 0.0

def find_similar_people(api_client, employer_website: str, role: str):
    api = ninjapear.EmployeeAPIApi(api_client)
    return api.get_similar_people(employer_website=employer_website, role=role)

If I were building this for a payments-adjacent product, I would start with one clean role like Head of Revenue Operations or VP Sales Operations.

Sample response shape:

{
  "results": [
    {
      "full_name": "Jane Doe",
      "role": "Head of Revenue Operations",
      "work_email": "[email protected]",
      "x_handle": "janedoe",
      "x_profile_url": "https://x.com/janedoe",
      "employer_website": "https://shopify.com"
    },
    {
      "full_name": "Alex Kim",
      "role": "VP Sales Operations",
      "work_email": null,
      "x_handle": null,
      "x_profile_url": null,
      "employer_website": "https://www.apple.com"
    }
  ]
}

Enrich the person record

Then use the Person Profile endpoint.

It can resolve from:

• work email
• name + company
• role + company

NinjaPear’s published match-rate guidance is useful:

Input method	Profiles found	Accuracy
Work email	10/10	100%
First name + last name + company	9/10	90%
Role + company	7/10	70%

Wrapper:

def get_person_profile(api_client, work_email=None, employer_website=None, role=None, first_name=None, last_name=None):
    api = ninjapear.EmployeeAPIApi(api_client)
    return api.get_employee_profile(
        work_email=work_email,
        employer_website=employer_website,
        role=role,
        first_name=first_name,
        last_name=last_name,
    )

Sample response shape:

{
  "id": "a3xK9mP2",
  "full_name": "Patrick Collison",
  "first_name": "Patrick",
  "last_name": "Collison",
  "bio": "Co-founder and CEO of Stripe",
  "profile_pic_url": "https://pbs.twimg.com/...",
  "country": "IE",
  "city": "IELIM",
  "x_handle": "patrickc",
  "x_profile_url": "https://x.com/patrickc",
  "personal_website": "https://patrickcollison.com",
  "work_experience": [
    {
      "role": "Co-founder & CEO",
      "company_name": "Stripe",
      "company_website": "stripe.com",
      "start_date": "2010-01",
      "end_date": null
    }
  ],
  "education": [
    {
      "major": "B.S. Mathematics",
      "school": "MIT",
      "start_date": "2006-09",
      "end_date": "2009-06"
    }
  ]
}

Build the contact queue

Turn the best accounts into a contact queue.

def build_contact_queue(api_client, accounts, target_role="Head of Revenue Operations"):
    people = []
    for account in accounts:
        if not account.website:
            continue
        payload = find_similar_people(api_client, employer_website=account.website, role=target_role)
        data = payload.to_dict() if hasattr(payload, "to_dict") else payload
        for item in data.get("results", []) or data.get("people", []) or []:
            people.append(
                ProspectPerson(
                    account_domain=normalize_domain(account.website),
                    full_name=item.get("full_name"),
                    role=item.get("role"),
                    work_email=item.get("work_email"),
                    x_handle=item.get("x_handle"),
                )
            )
    return people

That is why people discovery should follow account logic. You want relevance. Not decorative personalization.

Loop 4: Enrich and monitor

If your workflow has no timing layer, it is not agentic lead generation. It is a nicer spreadsheet.

Enrich the account

Use these NinjaPear endpoints:

• Company Details for context
• Employee Count for fit and segmentation
• Company Funding for capital signal
• Company Monitor and Company Updates for recent public changes

Endpoint map:

Endpoint	Purpose	Loop
Customer Listing	Pull customers, investors, partners	Loop 1
Competitor Listing	Expand market graph	Loop 2
Similar People	Find role-alike contacts	Loop 3
Person Profile	Enrich target person	Loop 3
Company Details	Add firmographic and narrative context	Loop 4
Employee Count	Fit and segment by size	Loop 4
Company Funding	Detect capital signal	Loop 4
Company Updates	Detect changes worth acting on	Loop 4
Monitor API	Turn updates into an ongoing feed	Loop 4

Wrappers:

def get_company_details(api_client, website: str):
    api = ninjapear.CompanyAPIApi(api_client)
    return api.get_company_details(website=website)


def get_employee_count(api_client, website: str):
    api = ninjapear.CompanyAPIApi(api_client)
    return api.get_employee_count(website=website)


def get_company_updates(api_client, website: str):
    api = ninjapear.CompanyAPIApi(api_client)
    return api.get_company_updates(website=website)

Sample Company Details shape:

{
  "name": "Stripe",
  "website": "https://stripe.com",
  "description": "Financial infrastructure platform for businesses.",
  "industry": 40205020,
  "specialties": ["payments", "billing", "fraud prevention"],
  "x_profile": "https://x.com/stripe"
}

Sample Employee Count shape:

{
  "website": "https://stripe.com",
  "employee_count": 8000
}

Sample Company Funding shape:

{
  "company_name": "Stripe",
  "funding_rounds": [
    {
      "round_type": "Series H",
      "announced_date": "2023-03-15",
      "money_raised": 6500000000,
      "lead_investors": ["Andreessen Horowitz"]
    }
  ]
}

Add company changes

This is where the system gets its timing.

NinjaPear’s update model covers:

• blog posts
• X posts
• meaningful website changes

And the output can be consumed as RSS. That part matters because it means you can push it into Slack, a CRM worker, or your own queue without much glue.

Sample Company Updates response shape:

{
  "updates": [
    {
      "title": "New Checkout Experience for Global Payments",
      "category": "blog",
      "summary": "Stripe announced a new checkout experience focused on international merchants.",
      "link": "https://stripe.com/blog/new-checkout",
      "published_at": "2026-02-27T10:00:00Z"
    },
    {
      "title": "Pricing page updated, new Enterprise tier added",
      "category": "website update",
      "summary": "A new enterprise pricing tier was detected on the public pricing page.",
      "link": "https://stripe.com/pricing",
      "published_at": "2026-02-27T07:00:00Z"
    },
    {
      "title": "We just launched Stripe Billing v3",
      "category": "x",
      "summary": "Product launch announcement on X.",
      "link": "https://x.com/stripe/status/1839200000000000000",
      "published_at": "2026-02-26T14:30:00Z"
    }
  ]
}

The published pricing examples for monitoring are a good sanity check:

Scenario	Targets	Frequency	Credits/month
VC tracking portfolio	20	Weekly	~346
Startup monitoring competitors	10	Daily	~1,203
Sales team watching prospect accounts	5	Daily	~453

That is enough to test timing-based lead generation without turning it into a big budget exercise.

Set up monitoring

Keep the wrapper thin:

def create_monitor_feed(api_client, websites: list[str], frequency: str = "daily"):
    api = ninjapear.CompanyAPIApi(api_client)
    return api.create_monitor_feed({"targets": websites, "frequency": frequency})

Sample monitor response shape:

{
  "rss_feed_url": "https://nubela.co/api/v1/monitor/feed/demo-stripe.xml",
  "targets": [
    "https://stripe.com",
    "https://shopify.com"
  ],
  "frequency": "daily"
}

Build a score

Start with a plain scoring function.

def score_account(is_competitor_customer: bool, employee_count: int | None, recent_updates: int, role_match: bool) -> float:
    score = 0.0
    if is_competitor_customer:
        score += 40
    if employee_count and 50 <= employee_count <= 2000:
        score += 20
    score += min(recent_updates * 10, 20)
    if role_match:
        score += 20
    return score

Signal table:

Output signal	Scoring impact	Why it matters
Competitor customer	+40	Strongest evidence of category demand
Employee count in target band	+20	Filters out bad-fit tiny and giant accounts
Recent updates	up to +20	Improves timing and relevance
Role match found	+20	Confirms a likely route to action

Optional person score:

def score_person(person: ProspectPerson, account: ProspectAccount) -> float:
    score = 0.0
    if person.role and any(k in person.role.lower() for k in ['revenue', 'sales', 'marketing', 'growth', 'operations']):
        score += 40
    if person.work_email:
        score += 20
    if person.x_handle:
        score += 10
    score += min(account.score / 3, 30)
    return round(score, 2)

That is basically the timing argument in one comment.

Wire the PydanticAI agent

You do not need a full PydanticAI tutorial here. You need enough to keep orchestration typed and in one place.

Typed output model:

from pydantic import BaseModel

class LeadGenResult(BaseModel):
    accounts: list[ProspectAccount]
    people: list[ProspectPerson]
    summary: str

Agent initialization:

from pydantic_ai import Agent
from pydantic_ai.models.openrouter import OpenRouterModel
from pydantic_ai.providers.openrouter import OpenRouterProvider

model = OpenRouterModel(
    settings.openrouter_model,
    provider=OpenRouterProvider(api_key=settings.openrouter_api_key),
)

agent = Agent(
    model,
    output_type=LeadGenResult,
    system_prompt="Build an agentic lead generation queue using NinjaPear data and return typed results only.",
)

Tool-wrapped workflow:

from pydantic_ai import RunContext

@agent.tool
async def build_queue(ctx: RunContext[AgentDeps], seeds: list[str], target_role: str = "Head of Revenue Operations") -> dict:
    with make_client(ctx.deps.settings.ninjapear_api_key) as api_client:
        result = run_pipeline(api_client, seeds=seeds, target_role=target_role)
    return result.model_dump()

That is enough structure for a useful agent.

Add NinjaPear tools

This is the bridge between docs and working code.

I would put the wrappers in app/tools/ninjapear.py.

Required wrappers:

• get_customer_listing
• get_competitors
• get_company_details
• get_employee_count
• get_company_updates
• get_company_funding
• get_person_profile
• find_similar_people
• create_monitor_feed

Rule of thumb:

• NinjaPear code should be real
• non-NinjaPear helpers should be mocked explicitly

Model table:

Model	Fields	Why it exists
ProspectAccount	source, relationship, website, description, specialties, score	Ranked account object for the queue
ProspectPerson	domain, name, role, email, X handle, score	Human routing layer after account qualification
CompanyUpdate	title, category, summary, link, date	Timing signal for recency-based prioritization
LeadGenResult	accounts, people, summary	Typed output for the agent and CLI

Mocked helpers:

def verify_email(email: str) -> dict:
    return {
        "email": email,
        "is_valid": email.endswith("@example.com") is False,
        "provider": "mock",
    }


def send_email(to: str, subject: str, body: str) -> dict:
    return {
        "to": to,
        "subject": subject,
        "status": "mocked-not-sent",
    }

Be explicit about mocks. It keeps the system honest.

Build the full GitHub project

The repo for this guide should live at the root, not inside a wrapper folder. The README.md should also link back to the article.

Repo tree:

agentic-lead-generation/
├── README.md
├── .env.example
├── requirements.txt
├── app/
│   ├── config.py
│   ├── models.py
│   ├── main.py
│   ├── scoring.py
│   ├── tools/
│   │   ├── ninjapear.py
│   │   └── mock_email.py
│   └── workflows/
│       └── pipeline.py
└── examples/
    └── run_demo.py

A few short excerpts.

config.py

class Settings(BaseModel):
    ninjapear_api_key: str = Field(..., alias='NINJAPEAR_API_KEY')
    openrouter_api_key: str = Field(..., alias='OPENROUTER_API_KEY')
    openrouter_model: str = Field(default='anthropic/claude-sonnet-4-5', alias='OPENROUTER_MODEL')

models.py

class ProspectAccount(BaseModel):
    source_company: str
    relationship_type: Literal['customer', 'investor', 'partner_platform', 'competitor']
    website: str | None = None
    normalized_domain: str | None = None
    name: str
    score: float = 0.0

tools/ninjapear.py

def make_client(api_key: str):
    configuration = ninjapear.Configuration(host='https://nubela.co', access_token=api_key)
    return ninjapear.ApiClient(configuration)

scoring.py

def score_account(is_competitor_customer: bool, employee_count: int | None, recent_updates: int, role_match: bool) -> float:
    score = 0.0
    if is_competitor_customer:
        score += 40

workflows/pipeline.py

def run_pipeline(api_client, seeds: list[str], target_role: str = 'Head of Revenue Operations') -> LeadGenResult:
    accounts = pull_competitor_customers(api_client, seeds)
    competitors = expand_competitors(api_client, seeds, max_depth=1)

main.py

result = agent.run_sync(
    'Build the lead generation queue for Stripe and Shopify competitors.',
    deps=AgentDeps(settings=settings)
)

examples/run_demo.py

result = run_pipeline(
    api_client,
    seeds=['https://stripe.com', 'https://shopify.com'],
    target_role='Head of Revenue Operations',
)

That repo is the lead magnet here. That is the asset people actually want.

Run an end-to-end example

Now run the workflow from the perspective of a founder using Stripe and Shopify as the wedge.

Input:

seeds = ["https://stripe.com", "https://shopify.com"]
target_role = "Head of Revenue Operations"

Runnable demo shape:

from app.config import Settings
from app.tools.ninjapear import make_client
from app.workflows.pipeline import run_pipeline


def main():
    settings = Settings.from_env()
    with make_client(settings.ninjapear_api_key) as api_client:
        result = run_pipeline(
            api_client,
            seeds=['https://stripe.com', 'https://shopify.com'],
            target_role='Head of Revenue Operations',
        )
    print(result.model_dump_json(indent=2))

What happens:

1. Pull customer lists for Stripe and Shopify.
2. Expand the competitor graph by one layer.
3. Re-run customer pull on selected adjacent competitors.
4. Keep top accounts.
5. Find similar people at those accounts.
6. Enrich with headcount and updates.
7. Score and rank.

Representative intermediate account output after Loop 1:

[
  {
    "source_company": "https://stripe.com",
    "relationship_type": "customer",
    "name": "Apple",
    "website": "https://www.apple.com",
    "normalized_domain": "apple.com",
    "description": "Apple Inc. designs, manufactures, and markets smartphones, personal computers, tablets, wearables, and accessories worldwide.",
    "score": 0
  },
  {
    "source_company": "https://shopify.com",
    "relationship_type": "customer",
    "name": "Toyota",
    "website": "https://toyota.com",
    "normalized_domain": "toyota.com",
    "description": "Global automotive manufacturer.",
    "score": 0
  }
]

Representative intermediate people output after Loop 3:

[
  {
    "account_domain": "shopify.com",
    "full_name": "Jane Doe",
    "role": "Head of Revenue Operations",
    "work_email": "[email protected]",
    "x_handle": "janedoe",
    "score": 0
  }
]

Representative final JSON shape:

{
  "accounts": [
    {
      "website": "https://example.com",
      "source_company": "https://stripe.com",
      "relationship_type": "customer",
      "score": 78
    }
  ],
  "people": [
    {
      "full_name": "Jane Doe",
      "role": "Head of Revenue Operations",
      "account_domain": "example.com"
    }
  ],
  "summary": "Found 14 high-fit accounts and 9 likely buyer profiles from 2 seed competitors."
}

If you want to generate an outreach task after that, fine. Just keep email sending mocked until the queue quality is real.

Use the NinjaPear Skill

If you already use an AI coding agent, install the NinjaPear Skill first.

It teaches the agent the boring parts that people usually get wrong:

• auth setup
• endpoint selection with cost awareness
• Python and JavaScript SDK usage
• pagination handling
• rate limit handling
• error handling for 401, 403, 404, 429, 500, 503
• timeout configuration

Install commands:

npx skills add NinjaPear/ninjapear-skill -a claude-code
npx skills add NinjaPear/ninjapear-skill -a codex
npx skills add NinjaPear/ninjapear-skill -a opencode

That is useful here because the whole point of this guide is to build an agentic lead generation system that actually ships.

Most lead generation articles still explain funnels to people who need systems.

This is the simpler frame I would use instead:

1. Who is already buying in the category?
2. Which adjacent accounts should I care about next?
3. Which humans are worth routing to?
4. Why is now the right moment?

If your stack cannot answer all four, your agentic lead generation setup is mostly branding.

So the next step is simple. Clone the repo, plug in your API keys, and run it against two competitors you already know. Start with Stripe and Shopify if you want the exact wedge from this guide.

And if you are building with Claude Code, Codex, or Opencode, install the NinjaPear Skill before you touch the code. It will save you time on auth, pagination, and endpoint selection.