diff --git a/tools/pyproject.toml b/tools/pyproject.toml index 91fc848b19..967dfec529 100644 --- a/tools/pyproject.toml +++ b/tools/pyproject.toml @@ -28,7 +28,7 @@ dependencies = [ "python-dotenv>=1.0.0", "playwright>=1.40.0", "playwright-stealth>=1.0.5", - "litellm==1.81.7", # pinned: supply chain attack in >=1.82.7 (adenhq/hive#6783) + "litellm==1.81.7" "dnspython>=2.4.0", "resend>=2.0.0", "asana>=3.2.0", diff --git a/tools/src/aden_tools/credentials/__init__.py b/tools/src/aden_tools/credentials/__init__.py index 8ee9204879..75477eeca4 100644 --- a/tools/src/aden_tools/credentials/__init__.py +++ b/tools/src/aden_tools/credentials/__init__.py @@ -99,6 +99,7 @@ from .mongodb import MONGODB_CREDENTIALS from .n8n import N8N_CREDENTIALS from .news import NEWS_CREDENTIALS +from .ninjapear import NINJAPEAR_CREDENTIALS from .notion import NOTION_CREDENTIALS from .obsidian import OBSIDIAN_CREDENTIALS from .pagerduty import PAGERDUTY_CREDENTIALS @@ -183,6 +184,7 @@ **MONGODB_CREDENTIALS, **N8N_CREDENTIALS, **NEWS_CREDENTIALS, + **NINJAPEAR_CREDENTIALS, **NOTION_CREDENTIALS, **OBSIDIAN_CREDENTIALS, **PAGERDUTY_CREDENTIALS, @@ -275,6 +277,7 @@ "MONGODB_CREDENTIALS", "N8N_CREDENTIALS", "NEWS_CREDENTIALS", + "NINJAPEAR_CREDENTIALS", "NOTION_CREDENTIALS", "OBSIDIAN_CREDENTIALS", "PAGERDUTY_CREDENTIALS", diff --git a/tools/src/aden_tools/credentials/ninjapear.py b/tools/src/aden_tools/credentials/ninjapear.py new file mode 100644 index 0000000000..9db4fb0273 --- /dev/null +++ b/tools/src/aden_tools/credentials/ninjapear.py @@ -0,0 +1,53 @@ +""" +NinjaPear (Nubela) tool credentials. + +Contains credentials for the NinjaPear API — company and people enrichment. +API reference: https://nubela.co/docs/ +""" + +from .base import CredentialSpec + +NINJAPEAR_CREDENTIALS = { + "ninjapear": CredentialSpec( + env_var="NINJAPEAR_API_KEY", + tools=[ + "ninjapear_get_person_profile", + "ninjapear_get_company_details", + "ninjapear_get_company_funding", + "ninjapear_get_company_updates", + "ninjapear_get_company_customers", + "ninjapear_get_company_competitors", + "ninjapear_get_credit_balance", + ], + required=True, + startup_required=False, + help_url="https://nubela.co/docs/", + description="NinjaPear API key for company and people enrichment", + aden_supported=False, + direct_api_key_supported=True, + api_key_instructions="""To get a NinjaPear API key: +1. Sign up at https://nubela.co/ using a WORK email address + (free email providers such as Gmail, Outlook, Yahoo are not accepted) +2. Go to your Dashboard > API section +3. Copy your API key + +Note: NinjaPear uses a credit system. Credits are only consumed on HTTP 200 responses. +- Free trial: small credit allocation (work email required to register) +- Paid: credit packs purchased as needed +- Credit costs per call: + - ninjapear_get_person_profile: 3 credits + - ninjapear_get_company_details: 2-5 credits + - ninjapear_get_company_funding: 2+ credits (1 credit per investor) + - ninjapear_get_company_updates: 2 credits + - ninjapear_get_company_customers: 1 + 2 credits/company returned + - ninjapear_get_company_competitors: 5 credits minimum + - ninjapear_get_credit_balance: FREE (0 credits) + +Set the environment variable: + export NINJAPEAR_API_KEY=your-api-key""", + health_check_endpoint="https://nubela.co/api/v1/meta/credit-balance", + health_check_method="GET", + credential_id="ninjapear", + credential_key="api_key", + ), +} diff --git a/tools/src/aden_tools/tools/__init__.py b/tools/src/aden_tools/tools/__init__.py index 8c0dadb303..f79ae95559 100644 --- a/tools/src/aden_tools/tools/__init__.py +++ b/tools/src/aden_tools/tools/__init__.py @@ -92,6 +92,7 @@ from .mongodb_tool import register_tools as register_mongodb from .n8n_tool import register_tools as register_n8n from .news_tool import register_tools as register_news +from .ninjapear_tool import register_tools as register_ninjapear from .notion_tool import register_tools as register_notion from .obsidian_tool import register_tools as register_obsidian from .pagerduty_tool import register_tools as register_pagerduty @@ -269,6 +270,7 @@ def _register_unverified( register_microsoft_graph(mcp, credentials=credentials) register_mongodb(mcp, credentials=credentials) register_n8n(mcp, credentials=credentials) + register_ninjapear(mcp, credentials=credentials) register_obsidian(mcp, credentials=credentials) register_pagerduty(mcp, credentials=credentials) register_pinecone(mcp, credentials=credentials) diff --git a/tools/src/aden_tools/tools/ninjapear_tool/__init__.py b/tools/src/aden_tools/tools/ninjapear_tool/__init__.py new file mode 100644 index 0000000000..ca4c91c609 --- /dev/null +++ b/tools/src/aden_tools/tools/ninjapear_tool/__init__.py @@ -0,0 +1,3 @@ +from .ninjapear_tool import register_tools + +__all__ = ["register_tools"] diff --git a/tools/src/aden_tools/tools/ninjapear_tool/ninjapear_tool.py b/tools/src/aden_tools/tools/ninjapear_tool/ninjapear_tool.py new file mode 100644 index 0000000000..8c1212da92 --- /dev/null +++ b/tools/src/aden_tools/tools/ninjapear_tool/ninjapear_tool.py @@ -0,0 +1,535 @@ +""" +NinjaPear Tool - Company and people enrichment via the NinjaPear API. + +Supports: +- Person/employee profile lookup (by work email, name+employer, or employer+role) +- Company details, funding, updates, customers, and competitors +- Credit balance check (free endpoint) + +API Reference: https://nubela.co/docs/ +Auth: Authorization: Bearer {api_key} +Note: API response times are 30-60s; client timeout is set to 100s. +""" + +from __future__ import annotations + +import os +from typing import TYPE_CHECKING, Any + +import httpx +from fastmcp import FastMCP + +if TYPE_CHECKING: + from aden_tools.credentials import CredentialStoreAdapter + +NINJAPEAR_API_BASE = "https://nubela.co" + + +class _NinjaPearClient: + """Internal client wrapping NinjaPear API calls.""" + + def __init__(self, api_key: str): + self._api_key = api_key + + @property + def _headers(self) -> dict[str, str]: + return { + "Authorization": f"Bearer {self._api_key}", + "Accept": "application/json", + } + + def _handle_response(self, response: httpx.Response) -> dict[str, Any]: + """Map HTTP status codes to user-friendly errors.""" + if response.status_code == 401: + return {"error": "Invalid NinjaPear API key"} + if response.status_code == 403: + return {"error": "NinjaPear: out of credits or insufficient permissions"} + if response.status_code == 404: + try: + detail = response.json() + msg = detail.get("message") or detail.get("error") or str(detail) + except Exception: + msg = response.text or "no data found" + return {"error": f"Not found: {msg}"} + if response.status_code == 410: + return {"error": "NinjaPear: this API endpoint has been deprecated"} + if response.status_code == 429: + return { + "error": "NinjaPear rate limit exceeded. Trial accounts are limited to " + "2 requests/minute. Apply exponential backoff and retry." + } + if response.status_code == 503: + return { + "error": "NinjaPear enrichment failed (503). The service is temporarily " + "unavailable. Retry the request — no credits were charged." + } + if response.status_code >= 400: + try: + detail = response.json() + except Exception: + detail = response.text + return {"error": f"NinjaPear API error (HTTP {response.status_code}): {detail}"} + try: + return response.json() + except Exception: + return {"error": f"Failed to parse NinjaPear response: {response.text[:500]}"} + + def get_person_profile( + self, + work_email: str = "", + first_name: str = "", + last_name: str = "", + middle_name: str = "", + employer_website: str = "", + role: str = "", + slug: str = "", + profile_id: str = "", + ) -> dict[str, Any]: + """Look up a person's professional profile.""" + params: dict[str, str] = {} + if work_email: + params["work_email"] = work_email + if first_name: + params["first_name"] = first_name + if last_name: + params["last_name"] = last_name + if middle_name: + params["middle_name"] = middle_name + if employer_website: + params["employer_website"] = employer_website + if role: + params["role"] = role + if slug: + params["slug"] = slug + if profile_id: + params["id"] = profile_id + + response = httpx.get( + f"{NINJAPEAR_API_BASE}/api/v1/employee/profile", + headers=self._headers, + params=params, + # API docs warn 30-60s response times; use 100s as recommended + timeout=100.0, + ) + return self._handle_response(response) + + def get_company_details( + self, + website: str, + include_employee_count: bool = False, + include_follower_count: bool = False, + ) -> dict[str, Any]: + """Get full company metadata.""" + params: dict[str, str] = {"website": website} + if include_employee_count: + params["include_employee_count"] = "true" + if include_follower_count: + params["follower_count"] = "include" + + response = httpx.get( + f"{NINJAPEAR_API_BASE}/api/v1/company/details", + headers=self._headers, + params=params, + timeout=100.0, + ) + return self._handle_response(response) + + def get_company_funding(self, website: str) -> dict[str, Any]: + """Get full funding history for a company.""" + response = httpx.get( + f"{NINJAPEAR_API_BASE}/api/v1/company/funding", + headers=self._headers, + params={"website": website}, + timeout=100.0, + ) + return self._handle_response(response) + + def get_company_updates(self, website: str) -> dict[str, Any]: + """Get latest blog posts and X/Twitter updates for a company.""" + response = httpx.get( + f"{NINJAPEAR_API_BASE}/api/v1/company/updates", + headers=self._headers, + params={"website": website}, + timeout=100.0, + ) + return self._handle_response(response) + + def get_company_customers( + self, + website: str, + cursor: str = "", + page_size: int = 200, + quality_filter: bool = True, + ) -> dict[str, Any]: + """Get customers, investors, and partner companies for a target company.""" + params: dict[str, Any] = { + "website": website, + "page_size": page_size, + } + if cursor: + params["cursor"] = cursor + if not quality_filter: + params["quality_filter"] = "false" + + response = httpx.get( + f"{NINJAPEAR_API_BASE}/api/v1/customer/listing", + headers=self._headers, + params=params, + timeout=100.0, + ) + return self._handle_response(response) + + def get_company_competitors(self, website: str) -> dict[str, Any]: + """Get competitor companies for a target company.""" + response = httpx.get( + f"{NINJAPEAR_API_BASE}/api/v1/competitor/listing", + headers=self._headers, + params={"website": website}, + timeout=100.0, + ) + return self._handle_response(response) + + def get_credit_balance(self) -> dict[str, Any]: + """Get remaining credit balance (free endpoint, 0 credits).""" + response = httpx.get( + f"{NINJAPEAR_API_BASE}/api/v1/meta/credit-balance", + headers=self._headers, + timeout=30.0, + ) + return self._handle_response(response) + + +def register_tools( + mcp: FastMCP, + credentials: CredentialStoreAdapter | None = None, +) -> None: + """Register NinjaPear enrichment tools with the MCP server.""" + + def _get_api_key(account: str = "") -> str | None: + """Get NinjaPear API key from credential manager or environment.""" + if credentials is not None: + if account: + return credentials.get_by_alias("ninjapear", account) + key = credentials.get("ninjapear") + if key is not None and not isinstance(key, str): + raise TypeError( + f"Expected string from credentials.get('ninjapear'), got {type(key).__name__}" + ) + return key + return os.getenv("NINJAPEAR_API_KEY") + + def _get_client(account: str = "") -> _NinjaPearClient | dict[str, str]: + """Get a NinjaPear client, or return an error dict if no credentials.""" + key = _get_api_key(account) + if not key: + return { + "error": "NinjaPear credentials not configured", + "help": ( + "Set the NINJAPEAR_API_KEY environment variable " + "or configure via credential store" + ), + } + return _NinjaPearClient(key) + + # --- Person --- + + @mcp.tool() + def ninjapear_get_person_profile( + work_email: str = "", + first_name: str = "", + last_name: str = "", + middle_name: str = "", + employer_website: str = "", + role: str = "", + slug: str = "", + profile_id: str = "", + account: str = "", + ) -> dict: + """ + Look up a person's professional profile via NinjaPear enrichment. + + You must provide one of the following valid input combinations: + - work_email alone + - first_name + employer_website (last_name and role improve accuracy) + - employer_website + role (returns the person currently holding that role) + - slug alone (e.g. their X/Twitter handle — free on 404) + - profile_id alone (NinjaPear internal ID — free on 404) + + Returns profile data including work experience, education, location, + X/Twitter handle, and bio. Note: 3 credits are consumed per successful + lookup. Trial accounts are limited to 2 requests/minute. + + Args: + work_email: Work email address of the person + first_name: First name (use with employer_website) + last_name: Last name (improves accuracy when used with first_name) + middle_name: Middle name (improves accuracy) + employer_website: Employer's website URL (e.g. "stripe.com") + role: Job title or role (e.g. "CTO", "Head of Engineering") + slug: NinjaPear slug (usually the person's X/Twitter handle) + profile_id: NinjaPear internal 8-character profile ID + account: Account alias for multi-account support + + Returns: + Dict with profile data (id, slug, full_name, work_experience, + education, location, x_handle, bio, follower_count) or error + """ + client = _get_client(account) + if isinstance(client, dict): + return client + + # Validate that at least one meaningful input was provided + has_email = bool(work_email) + has_name_employer = bool(first_name and employer_website) + has_role_employer = bool(role and employer_website) + has_slug = bool(slug) + has_id = bool(profile_id) + + if not any([has_email, has_name_employer, has_role_employer, has_slug, has_id]): + return { + "error": ( + "Insufficient input. Provide one of: " + "(1) work_email, " + "(2) first_name + employer_website, " + "(3) employer_website + role, " + "(4) slug, " + "(5) profile_id" + ) + } + + try: + return client.get_person_profile( + work_email=work_email, + first_name=first_name, + last_name=last_name, + middle_name=middle_name, + employer_website=employer_website, + role=role, + slug=slug, + profile_id=profile_id, + ) + except httpx.TimeoutException: + return { + "error": ( + "Request timed out after 100s. NinjaPear enrichment can take 30-60s. " + "Retry the request." + ) + } + except httpx.RequestError as e: + return {"error": f"Network error: {e}"} + + # --- Company --- + + @mcp.tool() + def ninjapear_get_company_details( + website: str, + include_employee_count: bool = False, + include_follower_count: bool = False, + account: str = "", + ) -> dict: + """ + Get detailed metadata for a company via NinjaPear. + + Returns company description, industry (GICS), type, founding year, + specialties, addresses, executives, social links, and optionally + employee count and follower counts. + + Credit cost: 2 credits base + 2 if include_employee_count=True + + 1 if include_follower_count=True. Charged even when no data found. + + Args: + website: Company website URL (e.g. "stripe.com") + include_employee_count: If True, include estimated headcount (+2 credits) + include_follower_count: If True, include X/Twitter follower count (+1 credit) + account: Account alias for multi-account support + + Returns: + Dict with company metadata or error + """ + if not website: + return {"error": "website is required"} + client = _get_client(account) + if isinstance(client, dict): + return client + try: + return client.get_company_details( + website, include_employee_count, include_follower_count + ) + except httpx.TimeoutException: + return {"error": "Request timed out after 100s. Retry the request."} + except httpx.RequestError as e: + return {"error": f"Network error: {e}"} + + @mcp.tool() + def ninjapear_get_company_funding( + website: str, + account: str = "", + ) -> dict: + """ + Get the full funding history for a company via NinjaPear. + + Returns total funds raised and a list of funding rounds with date, + round type, amount, and investors. + + Round types include: PRE_SEED, SEED, SERIES_A through SERIES_Z, + BRIDGE, VENTURE_DEBT, CONVERTIBLE_NOTE, GRANT, IPO, and more. + + Credit cost: 2 credits base + 1 credit per unique investor returned. + Base cost charged even on 404. + + Args: + website: Company website URL (e.g. "stripe.com") + account: Account alias for multi-account support + + Returns: + Dict with total_funds_raised_usd and funding_rounds list, or error + """ + if not website: + return {"error": "website is required"} + client = _get_client(account) + if isinstance(client, dict): + return client + try: + return client.get_company_funding(website) + except httpx.TimeoutException: + return {"error": "Request timed out after 100s. Retry the request."} + except httpx.RequestError as e: + return {"error": f"Network error: {e}"} + + @mcp.tool() + def ninjapear_get_company_updates( + website: str, + account: str = "", + ) -> dict: + """ + Get the latest blog posts and X/Twitter updates for a company via NinjaPear. + + Returns recent updates sorted by timestamp (newest first). Useful for + identifying hiring signals, product launches, and company news. + + Credit cost: 2 credits per request. + + Args: + website: Company website URL (e.g. "stripe.com") + account: Account alias for multi-account support + + Returns: + Dict with updates list (url, title, description, timestamp, source) + and blogs list, or error + """ + if not website: + return {"error": "website is required"} + client = _get_client(account) + if isinstance(client, dict): + return client + try: + return client.get_company_updates(website) + except httpx.TimeoutException: + return {"error": "Request timed out after 100s. Retry the request."} + except httpx.RequestError as e: + return {"error": f"Network error: {e}"} + + @mcp.tool() + def ninjapear_get_company_customers( + website: str, + cursor: str = "", + page_size: int = 50, + quality_filter: bool = True, + account: str = "", + ) -> dict: + """ + Get customers, investors, and partner companies of a target company via NinjaPear. + + Useful for understanding who uses a product (social proof) and for + expanding a sourcing universe to adjacent companies. + + Credit cost: 1 credit + 2 credits per company returned. Charged even on + empty results. + + Args: + website: Target company website URL (e.g. "stripe.com") + cursor: Pagination cursor from a previous response's next_page field + page_size: Number of results per page (1-200, default 50) + quality_filter: If True (default), filter out low-quality results + account: Account alias for multi-account support + + Returns: + Dict with customers, investors, partner_platforms lists and + next_page cursor, or error + """ + if not website: + return {"error": "website is required"} + if not 1 <= page_size <= 200: + return {"error": "page_size must be between 1 and 200"} + client = _get_client(account) + if isinstance(client, dict): + return client + try: + return client.get_company_customers(website, cursor, page_size, quality_filter) + except httpx.TimeoutException: + return {"error": "Request timed out after 100s. Retry the request."} + except httpx.RequestError as e: + return {"error": f"Network error: {e}"} + + @mcp.tool() + def ninjapear_get_company_competitors( + website: str, + account: str = "", + ) -> dict: + """ + Get competitor companies for a target company via NinjaPear. + + Returns a list of competitors with their website and the reason for + competition (organic_keyword_overlap or product_overlap). + + Useful for expanding a sourcing universe — e.g. find all companies + competing with a known target and source candidates from each. + + Credit cost: 2 credits per competitor returned. Minimum 5 credits + per request, charged even on empty results. + + Args: + website: Target company website URL (e.g. "stripe.com") + account: Account alias for multi-account support + + Returns: + Dict with competitors list (website, company_details_url, + competition_reason), or error + """ + if not website: + return {"error": "website is required"} + client = _get_client(account) + if isinstance(client, dict): + return client + try: + return client.get_company_competitors(website) + except httpx.TimeoutException: + return {"error": "Request timed out after 100s. Retry the request."} + except httpx.RequestError as e: + return {"error": f"Network error: {e}"} + + # --- Meta --- + + @mcp.tool() + def ninjapear_get_credit_balance(account: str = "") -> dict: + """ + Get the remaining NinjaPear credit balance. + + This is a free endpoint (0 credits consumed). Use it to check available + budget before running credit-intensive workflows. + + Args: + account: Account alias for multi-account support + + Returns: + Dict with credit_balance (int), or error + """ + client = _get_client(account) + if isinstance(client, dict): + return client + try: + return client.get_credit_balance() + except httpx.TimeoutException: + return {"error": "Request timed out"} + except httpx.RequestError as e: + return {"error": f"Network error: {e}"} diff --git a/tools/src/aden_tools/tools/ninjapear_tool/tests/__init__.py b/tools/src/aden_tools/tools/ninjapear_tool/tests/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/tools/src/aden_tools/tools/ninjapear_tool/tests/test_ninjapear_tool.py b/tools/src/aden_tools/tools/ninjapear_tool/tests/test_ninjapear_tool.py new file mode 100644 index 0000000000..fe1acbe05e --- /dev/null +++ b/tools/src/aden_tools/tools/ninjapear_tool/tests/test_ninjapear_tool.py @@ -0,0 +1,715 @@ +""" +Tests for the NinjaPear enrichment tool. + +Covers: +- _NinjaPearClient: headers, _handle_response (all status codes), each API method +- Tool registration: all 7 tools, no-credentials error, env var, credential manager +- ninjapear_get_person_profile: valid input combos, insufficient-input guard +- ninjapear_get_company_details: happy path, optional flags, missing website +- ninjapear_get_company_funding: happy path, missing website +- ninjapear_get_company_updates: happy path, missing website +- ninjapear_get_company_customers: happy path, page_size validation +- ninjapear_get_company_competitors: happy path, missing website +- ninjapear_get_credit_balance: happy path, no-credentials error +- Timeout / network error handling (all tools) +- CredentialSpec: env_var, tools list +- @pytest.mark.live: real API smoke tests (skipped unless NINJAPEAR_API_KEY is set) +""" + +from __future__ import annotations + +import os +from unittest.mock import MagicMock, patch + +import httpx +import pytest + +from aden_tools.tools.ninjapear_tool.ninjapear_tool import ( + NINJAPEAR_API_BASE, + _NinjaPearClient, + register_tools, +) + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +TOOL_NAMES = [ + "ninjapear_get_person_profile", + "ninjapear_get_company_details", + "ninjapear_get_company_funding", + "ninjapear_get_company_updates", + "ninjapear_get_company_customers", + "ninjapear_get_company_competitors", + "ninjapear_get_credit_balance", +] + + +def _make_response(status_code: int, body: dict | None = None) -> MagicMock: + """Build a mock httpx.Response.""" + r = MagicMock() + r.status_code = status_code + r.json.return_value = body or {} + r.text = str(body or {}) + return r + + +def _register(credentials=None) -> tuple[MagicMock, list, dict]: + """Register tools into a mock MCP and return (mcp, fns_list, fn_by_name).""" + mcp = MagicMock() + fns: list = [] + mcp.tool.return_value = lambda fn: fns.append(fn) or fn + register_tools(mcp, credentials=credentials) + by_name = {fn.__name__: fn for fn in fns} + return mcp, fns, by_name + + +# --------------------------------------------------------------------------- +# _NinjaPearClient +# --------------------------------------------------------------------------- + + +class TestNinjaPearClient: + def setup_method(self): + self.client = _NinjaPearClient("test-key") + + # --- headers --- + + def test_headers_authorization(self): + assert self.client._headers["Authorization"] == "Bearer test-key" + + def test_headers_accept(self): + assert self.client._headers["Accept"] == "application/json" + + # --- _handle_response --- + + def test_handle_response_200(self): + r = _make_response(200, {"name": "Stripe"}) + assert self.client._handle_response(r) == {"name": "Stripe"} + + @pytest.mark.parametrize( + "status_code,expected_substring", + [ + (401, "Invalid"), + (403, "credits"), + (404, "Not found"), + (410, "deprecated"), + (429, "rate limit"), + (503, "retry"), + ], + ) + def test_handle_response_errors(self, status_code, expected_substring): + r = _make_response(status_code) + result = self.client._handle_response(r) + assert "error" in result + assert expected_substring.lower() in result["error"].lower() + + def test_handle_response_generic_5xx(self): + r = _make_response(500, {"detail": "boom"}) + result = self.client._handle_response(r) + assert "error" in result + assert "500" in result["error"] + + def test_handle_response_invalid_json(self): + r = MagicMock() + r.status_code = 200 + r.json.side_effect = ValueError("not json") + r.text = "not-json-body" + result = self.client._handle_response(r) + assert "error" in result + + # --- get_person_profile --- + + @patch("aden_tools.tools.ninjapear_tool.ninjapear_tool.httpx.get") + def test_get_person_profile_by_email(self, mock_get): + mock_get.return_value = _make_response(200, {"full_name": "Jane Doe"}) + result = self.client.get_person_profile(work_email="jane@stripe.com") + assert result["full_name"] == "Jane Doe" + params = mock_get.call_args.kwargs["params"] + assert params["work_email"] == "jane@stripe.com" + + @patch("aden_tools.tools.ninjapear_tool.ninjapear_tool.httpx.get") + def test_get_person_profile_by_name_employer(self, mock_get): + mock_get.return_value = _make_response(200, {"full_name": "John Smith"}) + self.client.get_person_profile( + first_name="John", last_name="Smith", employer_website="stripe.com" + ) + params = mock_get.call_args.kwargs["params"] + assert params["first_name"] == "John" + assert params["last_name"] == "Smith" + assert params["employer_website"] == "stripe.com" + + @patch("aden_tools.tools.ninjapear_tool.ninjapear_tool.httpx.get") + def test_get_person_profile_by_role_employer(self, mock_get): + mock_get.return_value = _make_response(200, {"full_name": "Alice"}) + self.client.get_person_profile(employer_website="stripe.com", role="CTO") + params = mock_get.call_args.kwargs["params"] + assert params["role"] == "CTO" + assert params["employer_website"] == "stripe.com" + + @patch("aden_tools.tools.ninjapear_tool.ninjapear_tool.httpx.get") + def test_get_person_profile_empty_params_omitted(self, mock_get): + """Empty string params must not be sent to the API.""" + mock_get.return_value = _make_response(200, {}) + self.client.get_person_profile(slug="janesmith") + params = mock_get.call_args.kwargs["params"] + assert "work_email" not in params + assert "first_name" not in params + assert params["slug"] == "janesmith" + + @patch("aden_tools.tools.ninjapear_tool.ninjapear_tool.httpx.get") + def test_get_person_profile_timeout_100s(self, mock_get): + mock_get.return_value = _make_response(200, {}) + self.client.get_person_profile(slug="x") + assert mock_get.call_args.kwargs["timeout"] == 100.0 + + # --- get_company_details --- + + @patch("aden_tools.tools.ninjapear_tool.ninjapear_tool.httpx.get") + def test_get_company_details_basic(self, mock_get): + mock_get.return_value = _make_response(200, {"name": "Stripe"}) + result = self.client.get_company_details("stripe.com") + assert result["name"] == "Stripe" + url = mock_get.call_args.args[0] + assert url == f"{NINJAPEAR_API_BASE}/api/v1/company/details" + + @patch("aden_tools.tools.ninjapear_tool.ninjapear_tool.httpx.get") + def test_get_company_details_employee_count_flag(self, mock_get): + mock_get.return_value = _make_response(200, {}) + self.client.get_company_details("stripe.com", include_employee_count=True) + params = mock_get.call_args.kwargs["params"] + assert params["include_employee_count"] == "true" + + @patch("aden_tools.tools.ninjapear_tool.ninjapear_tool.httpx.get") + def test_get_company_details_follower_count_flag(self, mock_get): + mock_get.return_value = _make_response(200, {}) + self.client.get_company_details("stripe.com", include_follower_count=True) + params = mock_get.call_args.kwargs["params"] + assert params["follower_count"] == "include" + + @patch("aden_tools.tools.ninjapear_tool.ninjapear_tool.httpx.get") + def test_get_company_details_no_optional_flags(self, mock_get): + mock_get.return_value = _make_response(200, {}) + self.client.get_company_details("stripe.com") + params = mock_get.call_args.kwargs["params"] + assert "include_employee_count" not in params + assert "follower_count" not in params + + # --- get_company_funding --- + + @patch("aden_tools.tools.ninjapear_tool.ninjapear_tool.httpx.get") + def test_get_company_funding(self, mock_get): + body = {"total_funds_raised_usd": 2000000000, "funding_rounds": []} + mock_get.return_value = _make_response(200, body) + result = self.client.get_company_funding("stripe.com") + assert result["total_funds_raised_usd"] == 2000000000 + url = mock_get.call_args.args[0] + assert url == f"{NINJAPEAR_API_BASE}/api/v1/company/funding" + + # --- get_company_updates --- + + @patch("aden_tools.tools.ninjapear_tool.ninjapear_tool.httpx.get") + def test_get_company_updates(self, mock_get): + body = {"updates": [{"title": "We are hiring"}], "blogs": []} + mock_get.return_value = _make_response(200, body) + result = self.client.get_company_updates("stripe.com") + assert result["updates"][0]["title"] == "We are hiring" + url = mock_get.call_args.args[0] + assert url == f"{NINJAPEAR_API_BASE}/api/v1/company/updates" + + # --- get_company_customers --- + + @patch("aden_tools.tools.ninjapear_tool.ninjapear_tool.httpx.get") + def test_get_company_customers_default_params(self, mock_get): + mock_get.return_value = _make_response(200, {"customers": []}) + self.client.get_company_customers("stripe.com") + params = mock_get.call_args.kwargs["params"] + assert params["website"] == "stripe.com" + # Client default is 200; the tool function default (50) is a separate layer + assert params["page_size"] == 200 + # quality_filter=True means param is not sent (default) + assert "quality_filter" not in params + + @patch("aden_tools.tools.ninjapear_tool.ninjapear_tool.httpx.get") + def test_get_company_customers_quality_filter_false(self, mock_get): + mock_get.return_value = _make_response(200, {}) + self.client.get_company_customers("stripe.com", quality_filter=False) + params = mock_get.call_args.kwargs["params"] + assert params["quality_filter"] == "false" + + @patch("aden_tools.tools.ninjapear_tool.ninjapear_tool.httpx.get") + def test_get_company_customers_page_size_forwarded(self, mock_get): + mock_get.return_value = _make_response(200, {}) + self.client.get_company_customers("stripe.com", page_size=50) + params = mock_get.call_args.kwargs["params"] + assert params["page_size"] == 50 + + @patch("aden_tools.tools.ninjapear_tool.ninjapear_tool.httpx.get") + def test_get_company_customers_cursor(self, mock_get): + mock_get.return_value = _make_response(200, {}) + self.client.get_company_customers("stripe.com", cursor="abc123") + params = mock_get.call_args.kwargs["params"] + assert params["cursor"] == "abc123" + + # --- get_company_competitors --- + + @patch("aden_tools.tools.ninjapear_tool.ninjapear_tool.httpx.get") + def test_get_company_competitors(self, mock_get): + body = { + "competitors": [{"website": "braintree.com", "competition_reason": "product_overlap"}] + } + mock_get.return_value = _make_response(200, body) + result = self.client.get_company_competitors("stripe.com") + assert result["competitors"][0]["website"] == "braintree.com" + url = mock_get.call_args.args[0] + assert url == f"{NINJAPEAR_API_BASE}/api/v1/competitor/listing" + + # --- get_credit_balance --- + + @patch("aden_tools.tools.ninjapear_tool.ninjapear_tool.httpx.get") + def test_get_credit_balance(self, mock_get): + mock_get.return_value = _make_response(200, {"credit_balance": 9500}) + result = self.client.get_credit_balance() + assert result["credit_balance"] == 9500 + url = mock_get.call_args.args[0] + assert url == f"{NINJAPEAR_API_BASE}/api/v1/meta/credit-balance" + + @patch("aden_tools.tools.ninjapear_tool.ninjapear_tool.httpx.get") + def test_get_credit_balance_timeout_30s(self, mock_get): + mock_get.return_value = _make_response(200, {"credit_balance": 0}) + self.client.get_credit_balance() + assert mock_get.call_args.kwargs["timeout"] == 30.0 + + +# --------------------------------------------------------------------------- +# Tool registration +# --------------------------------------------------------------------------- + + +class TestToolRegistration: + def test_all_tools_registered(self): + _, _, by_name = _register() + for name in TOOL_NAMES: + assert name in by_name, f"Tool '{name}' was not registered" + + def test_tool_count(self): + mcp = MagicMock() + mcp.tool.return_value = lambda fn: fn + register_tools(mcp) + assert mcp.tool.call_count == len(TOOL_NAMES) + + def test_no_credentials_returns_error(self): + with patch.dict("os.environ", {}, clear=True): + _, _, by_name = _register(credentials=None) + result = by_name["ninjapear_get_credit_balance"]() + assert "error" in result + assert "not configured" in result["error"] + + def test_credentials_from_env_var(self): + _, _, by_name = _register(credentials=None) + with ( + patch.dict("os.environ", {"NINJAPEAR_API_KEY": "env-key"}), + patch("aden_tools.tools.ninjapear_tool.ninjapear_tool.httpx.get") as mock_get, + ): + mock_get.return_value = _make_response(200, {"credit_balance": 10}) + result = by_name["ninjapear_get_credit_balance"]() + assert result["credit_balance"] == 10 + call_headers = mock_get.call_args.kwargs["headers"] + assert call_headers["Authorization"] == "Bearer env-key" + + def test_credentials_from_credential_manager(self): + cred = MagicMock() + cred.get.return_value = "manager-key" + _, _, by_name = _register(credentials=cred) + with patch("aden_tools.tools.ninjapear_tool.ninjapear_tool.httpx.get") as mock_get: + mock_get.return_value = _make_response(200, {"credit_balance": 42}) + result = by_name["ninjapear_get_credit_balance"]() + cred.get.assert_called_with("ninjapear") + assert result["credit_balance"] == 42 + + +# --------------------------------------------------------------------------- +# ninjapear_get_person_profile +# --------------------------------------------------------------------------- + + +class TestPersonProfileTool: + def setup_method(self): + cred = MagicMock() + cred.get.return_value = "tok" + _, _, self.fns = _register(credentials=cred) + + def _fn(self): + return self.fns["ninjapear_get_person_profile"] + + def test_no_input_returns_error(self): + result = self._fn()() + assert "error" in result + assert "Insufficient input" in result["error"] + + def test_partial_input_name_without_employer_returns_error(self): + # first_name alone is not a valid combo + result = self._fn()(first_name="Jane") + assert "error" in result + + def test_partial_input_role_without_employer_returns_error(self): + result = self._fn()(role="CTO") + assert "error" in result + + @patch("aden_tools.tools.ninjapear_tool.ninjapear_tool.httpx.get") + def test_valid_work_email(self, mock_get): + mock_get.return_value = _make_response(200, {"full_name": "Jane Doe"}) + result = self._fn()(work_email="jane@stripe.com") + assert result["full_name"] == "Jane Doe" + + @patch("aden_tools.tools.ninjapear_tool.ninjapear_tool.httpx.get") + def test_valid_name_plus_employer(self, mock_get): + mock_get.return_value = _make_response(200, {"full_name": "John"}) + result = self._fn()(first_name="John", employer_website="stripe.com") + assert "full_name" in result + + @patch("aden_tools.tools.ninjapear_tool.ninjapear_tool.httpx.get") + def test_valid_role_plus_employer(self, mock_get): + mock_get.return_value = _make_response(200, {"full_name": "Alice"}) + result = self._fn()(role="CTO", employer_website="stripe.com") + assert "full_name" in result + + @patch("aden_tools.tools.ninjapear_tool.ninjapear_tool.httpx.get") + def test_valid_slug(self, mock_get): + mock_get.return_value = _make_response(200, {"slug": "janesmith"}) + result = self._fn()(slug="janesmith") + assert "slug" in result + + @patch("aden_tools.tools.ninjapear_tool.ninjapear_tool.httpx.get") + def test_valid_profile_id(self, mock_get): + mock_get.return_value = _make_response(200, {"id": "abc12345"}) + result = self._fn()(profile_id="abc12345") + assert "id" in result + + @patch("aden_tools.tools.ninjapear_tool.ninjapear_tool.httpx.get") + def test_timeout_returns_error(self, mock_get): + mock_get.side_effect = httpx.TimeoutException("timed out") + result = self._fn()(work_email="jane@stripe.com") + assert "error" in result + assert "timed out" in result["error"].lower() or "100s" in result["error"] + + @patch("aden_tools.tools.ninjapear_tool.ninjapear_tool.httpx.get") + def test_network_error_returns_error(self, mock_get): + mock_get.side_effect = httpx.RequestError("connection failed") + result = self._fn()(work_email="jane@stripe.com") + assert "error" in result + assert "Network error" in result["error"] + + +# --------------------------------------------------------------------------- +# ninjapear_get_company_details +# --------------------------------------------------------------------------- + + +class TestCompanyDetailsTool: + def setup_method(self): + cred = MagicMock() + cred.get.return_value = "tok" + _, _, self.fns = _register(credentials=cred) + + def _fn(self): + return self.fns["ninjapear_get_company_details"] + + def test_missing_website_returns_error(self): + result = self._fn()(website="") + assert "error" in result + + @patch("aden_tools.tools.ninjapear_tool.ninjapear_tool.httpx.get") + def test_happy_path(self, mock_get): + mock_get.return_value = _make_response(200, {"name": "Stripe", "founded_year": 2010}) + result = self._fn()(website="stripe.com") + assert result["name"] == "Stripe" + + @patch("aden_tools.tools.ninjapear_tool.ninjapear_tool.httpx.get") + def test_with_employee_count_flag(self, mock_get): + mock_get.return_value = _make_response(200, {"employee_count": 7000}) + result = self._fn()(website="stripe.com", include_employee_count=True) + assert result["employee_count"] == 7000 + params = mock_get.call_args.kwargs["params"] + assert params["include_employee_count"] == "true" + + @patch("aden_tools.tools.ninjapear_tool.ninjapear_tool.httpx.get") + def test_timeout_returns_error(self, mock_get): + mock_get.side_effect = httpx.TimeoutException("timed out") + result = self._fn()(website="stripe.com") + assert "error" in result + + @patch("aden_tools.tools.ninjapear_tool.ninjapear_tool.httpx.get") + def test_network_error_returns_error(self, mock_get): + mock_get.side_effect = httpx.RequestError("connection failed") + result = self._fn()(website="stripe.com") + assert "error" in result + + +# --------------------------------------------------------------------------- +# ninjapear_get_company_funding +# --------------------------------------------------------------------------- + + +class TestCompanyFundingTool: + def setup_method(self): + cred = MagicMock() + cred.get.return_value = "tok" + _, _, self.fns = _register(credentials=cred) + + def _fn(self): + return self.fns["ninjapear_get_company_funding"] + + def test_missing_website_returns_error(self): + result = self._fn()(website="") + assert "error" in result + + @patch("aden_tools.tools.ninjapear_tool.ninjapear_tool.httpx.get") + def test_happy_path(self, mock_get): + body = { + "total_funds_raised_usd": 2000000000, + "funding_rounds": [{"round_type": "SERIES_A", "amount_usd": 5000000, "investors": []}], + } + mock_get.return_value = _make_response(200, body) + result = self._fn()(website="stripe.com") + assert result["total_funds_raised_usd"] == 2000000000 + assert len(result["funding_rounds"]) == 1 + + +# --------------------------------------------------------------------------- +# ninjapear_get_company_updates +# --------------------------------------------------------------------------- + + +class TestCompanyUpdatesTool: + def setup_method(self): + cred = MagicMock() + cred.get.return_value = "tok" + _, _, self.fns = _register(credentials=cred) + + def _fn(self): + return self.fns["ninjapear_get_company_updates"] + + def test_missing_website_returns_error(self): + result = self._fn()(website="") + assert "error" in result + + @patch("aden_tools.tools.ninjapear_tool.ninjapear_tool.httpx.get") + def test_happy_path(self, mock_get): + body = { + "updates": [ + {"title": "We're hiring 500 engineers", "source": "blog", "timestamp": "2024-01-15"} + ], + "blogs": ["https://stripe.com/blog"], + } + mock_get.return_value = _make_response(200, body) + result = self._fn()(website="stripe.com") + assert len(result["updates"]) == 1 + assert "hiring" in result["updates"][0]["title"] + + +# --------------------------------------------------------------------------- +# ninjapear_get_company_customers +# --------------------------------------------------------------------------- + + +class TestCompanyCustomersTool: + def setup_method(self): + cred = MagicMock() + cred.get.return_value = "tok" + _, _, self.fns = _register(credentials=cred) + + def _fn(self): + return self.fns["ninjapear_get_company_customers"] + + def test_missing_website_returns_error(self): + result = self._fn()(website="") + assert "error" in result + + def test_page_size_too_large_returns_error(self): + result = self._fn()(website="stripe.com", page_size=999) + assert "error" in result + + def test_page_size_zero_returns_error(self): + result = self._fn()(website="stripe.com", page_size=0) + assert "error" in result + + @patch("aden_tools.tools.ninjapear_tool.ninjapear_tool.httpx.get") + def test_happy_path(self, mock_get): + body = { + "customers": [{"name": "Amazon", "website": "amazon.com"}], + "investors": [], + "partner_platforms": [], + "next_page": None, + } + mock_get.return_value = _make_response(200, body) + result = self._fn()(website="stripe.com") + assert result["customers"][0]["name"] == "Amazon" + + @patch("aden_tools.tools.ninjapear_tool.ninjapear_tool.httpx.get") + def test_pagination_cursor_forwarded(self, mock_get): + mock_get.return_value = _make_response(200, {"customers": [], "next_page": None}) + self._fn()(website="stripe.com", cursor="next-page-token") + params = mock_get.call_args.kwargs["params"] + assert params["cursor"] == "next-page-token" + + @patch("aden_tools.tools.ninjapear_tool.ninjapear_tool.httpx.get") + def test_quality_filter_false_forwarded(self, mock_get): + mock_get.return_value = _make_response(200, {}) + self._fn()(website="stripe.com", quality_filter=False) + params = mock_get.call_args.kwargs["params"] + assert params["quality_filter"] == "false" + + +# --------------------------------------------------------------------------- +# ninjapear_get_company_competitors +# --------------------------------------------------------------------------- + + +class TestCompanyCompetitorsTool: + def setup_method(self): + cred = MagicMock() + cred.get.return_value = "tok" + _, _, self.fns = _register(credentials=cred) + + def _fn(self): + return self.fns["ninjapear_get_company_competitors"] + + def test_missing_website_returns_error(self): + result = self._fn()(website="") + assert "error" in result + + @patch("aden_tools.tools.ninjapear_tool.ninjapear_tool.httpx.get") + def test_happy_path(self, mock_get): + body = { + "competitors": [ + {"website": "braintree.com", "competition_reason": "product_overlap"}, + {"website": "adyen.com", "competition_reason": "organic_keyword_overlap"}, + ] + } + mock_get.return_value = _make_response(200, body) + result = self._fn()(website="stripe.com") + assert len(result["competitors"]) == 2 + assert result["competitors"][0]["competition_reason"] == "product_overlap" + + +# --------------------------------------------------------------------------- +# ninjapear_get_credit_balance +# --------------------------------------------------------------------------- + + +class TestCreditBalanceTool: + def setup_method(self): + cred = MagicMock() + cred.get.return_value = "tok" + _, _, self.fns = _register(credentials=cred) + + def _fn(self): + return self.fns["ninjapear_get_credit_balance"] + + @patch("aden_tools.tools.ninjapear_tool.ninjapear_tool.httpx.get") + def test_happy_path(self, mock_get): + mock_get.return_value = _make_response(200, {"credit_balance": 9500}) + result = self._fn()() + assert result["credit_balance"] == 9500 + + @patch("aden_tools.tools.ninjapear_tool.ninjapear_tool.httpx.get") + def test_timeout_returns_error(self, mock_get): + mock_get.side_effect = httpx.TimeoutException("timed out") + result = self._fn()() + assert "error" in result + + @patch("aden_tools.tools.ninjapear_tool.ninjapear_tool.httpx.get") + def test_network_error_returns_error(self, mock_get): + mock_get.side_effect = httpx.RequestError("conn refused") + result = self._fn()() + assert "error" in result + assert "Network error" in result["error"] + + def test_no_credentials_returns_error(self): + with patch.dict("os.environ", {}, clear=True): + _, _, fns = _register(credentials=None) + result = fns["ninjapear_get_credit_balance"]() + assert "error" in result + assert "not configured" in result["error"] + + +# --------------------------------------------------------------------------- +# CredentialSpec +# --------------------------------------------------------------------------- + + +class TestCredentialSpec: + def test_spec_exists(self): + from aden_tools.credentials import NINJAPEAR_CREDENTIALS + + assert "ninjapear" in NINJAPEAR_CREDENTIALS + + def test_env_var(self): + from aden_tools.credentials import NINJAPEAR_CREDENTIALS + + assert NINJAPEAR_CREDENTIALS["ninjapear"].env_var == "NINJAPEAR_API_KEY" + + def test_tools_list_complete(self): + from aden_tools.credentials import NINJAPEAR_CREDENTIALS + + spec_tools = NINJAPEAR_CREDENTIALS["ninjapear"].tools + for name in TOOL_NAMES: + assert name in spec_tools, f"Tool '{name}' missing from CredentialSpec.tools" + + def test_health_check_endpoint(self): + from aden_tools.credentials import NINJAPEAR_CREDENTIALS + + spec = NINJAPEAR_CREDENTIALS["ninjapear"] + assert "credit-balance" in spec.health_check_endpoint + + def test_in_global_credential_specs(self): + from aden_tools.credentials import CREDENTIAL_SPECS + + assert "ninjapear" in CREDENTIAL_SPECS + + +# --------------------------------------------------------------------------- +# Live tests — skipped unless NINJAPEAR_API_KEY is set in environment +# --------------------------------------------------------------------------- + + +@pytest.mark.live +class TestLive: + """ + Real API calls against NinjaPear. Excluded from CI by default. + + To run: + export NINJAPEAR_API_KEY=your-key + cd tools + uv run pytest src/aden_tools/tools/ninjapear_tool/tests/ -m live -v + """ + + @pytest.fixture(autouse=True) + def require_api_key(self): + if not os.environ.get("NINJAPEAR_API_KEY"): + pytest.skip("NINJAPEAR_API_KEY not set") + + def test_live_credit_balance(self): + """Smoke test: free endpoint, 0 credits consumed.""" + _, _, fns = _register(credentials=None) + result = fns["ninjapear_get_credit_balance"]() + assert "credit_balance" in result, f"Unexpected response: {result}" + assert isinstance(result["credit_balance"], int) + + def test_live_company_details_stripe(self): + """Fetch Stripe company details. Costs 2 credits.""" + _, _, fns = _register(credentials=None) + result = fns["ninjapear_get_company_details"](website="stripe.com") + assert "error" not in result, f"API error: {result}" + assert result.get("name") is not None + + def test_live_person_profile_by_role(self): + """Look up CTO at stripe.com. Costs 3 credits.""" + _, _, fns = _register(credentials=None) + result = fns["ninjapear_get_person_profile"](role="CTO", employer_website="stripe.com") + # 404 is acceptable (person may not be indexed), anything else is a bug + if "error" in result: + assert "Not found" in result["error"], f"Unexpected error: {result}" + else: + assert "full_name" in result or "id" in result diff --git a/tools/tests/tools/test_ninjapear_tool.py b/tools/tests/tools/test_ninjapear_tool.py new file mode 100644 index 0000000000..fe1acbe05e --- /dev/null +++ b/tools/tests/tools/test_ninjapear_tool.py @@ -0,0 +1,715 @@ +""" +Tests for the NinjaPear enrichment tool. + +Covers: +- _NinjaPearClient: headers, _handle_response (all status codes), each API method +- Tool registration: all 7 tools, no-credentials error, env var, credential manager +- ninjapear_get_person_profile: valid input combos, insufficient-input guard +- ninjapear_get_company_details: happy path, optional flags, missing website +- ninjapear_get_company_funding: happy path, missing website +- ninjapear_get_company_updates: happy path, missing website +- ninjapear_get_company_customers: happy path, page_size validation +- ninjapear_get_company_competitors: happy path, missing website +- ninjapear_get_credit_balance: happy path, no-credentials error +- Timeout / network error handling (all tools) +- CredentialSpec: env_var, tools list +- @pytest.mark.live: real API smoke tests (skipped unless NINJAPEAR_API_KEY is set) +""" + +from __future__ import annotations + +import os +from unittest.mock import MagicMock, patch + +import httpx +import pytest + +from aden_tools.tools.ninjapear_tool.ninjapear_tool import ( + NINJAPEAR_API_BASE, + _NinjaPearClient, + register_tools, +) + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +TOOL_NAMES = [ + "ninjapear_get_person_profile", + "ninjapear_get_company_details", + "ninjapear_get_company_funding", + "ninjapear_get_company_updates", + "ninjapear_get_company_customers", + "ninjapear_get_company_competitors", + "ninjapear_get_credit_balance", +] + + +def _make_response(status_code: int, body: dict | None = None) -> MagicMock: + """Build a mock httpx.Response.""" + r = MagicMock() + r.status_code = status_code + r.json.return_value = body or {} + r.text = str(body or {}) + return r + + +def _register(credentials=None) -> tuple[MagicMock, list, dict]: + """Register tools into a mock MCP and return (mcp, fns_list, fn_by_name).""" + mcp = MagicMock() + fns: list = [] + mcp.tool.return_value = lambda fn: fns.append(fn) or fn + register_tools(mcp, credentials=credentials) + by_name = {fn.__name__: fn for fn in fns} + return mcp, fns, by_name + + +# --------------------------------------------------------------------------- +# _NinjaPearClient +# --------------------------------------------------------------------------- + + +class TestNinjaPearClient: + def setup_method(self): + self.client = _NinjaPearClient("test-key") + + # --- headers --- + + def test_headers_authorization(self): + assert self.client._headers["Authorization"] == "Bearer test-key" + + def test_headers_accept(self): + assert self.client._headers["Accept"] == "application/json" + + # --- _handle_response --- + + def test_handle_response_200(self): + r = _make_response(200, {"name": "Stripe"}) + assert self.client._handle_response(r) == {"name": "Stripe"} + + @pytest.mark.parametrize( + "status_code,expected_substring", + [ + (401, "Invalid"), + (403, "credits"), + (404, "Not found"), + (410, "deprecated"), + (429, "rate limit"), + (503, "retry"), + ], + ) + def test_handle_response_errors(self, status_code, expected_substring): + r = _make_response(status_code) + result = self.client._handle_response(r) + assert "error" in result + assert expected_substring.lower() in result["error"].lower() + + def test_handle_response_generic_5xx(self): + r = _make_response(500, {"detail": "boom"}) + result = self.client._handle_response(r) + assert "error" in result + assert "500" in result["error"] + + def test_handle_response_invalid_json(self): + r = MagicMock() + r.status_code = 200 + r.json.side_effect = ValueError("not json") + r.text = "not-json-body" + result = self.client._handle_response(r) + assert "error" in result + + # --- get_person_profile --- + + @patch("aden_tools.tools.ninjapear_tool.ninjapear_tool.httpx.get") + def test_get_person_profile_by_email(self, mock_get): + mock_get.return_value = _make_response(200, {"full_name": "Jane Doe"}) + result = self.client.get_person_profile(work_email="jane@stripe.com") + assert result["full_name"] == "Jane Doe" + params = mock_get.call_args.kwargs["params"] + assert params["work_email"] == "jane@stripe.com" + + @patch("aden_tools.tools.ninjapear_tool.ninjapear_tool.httpx.get") + def test_get_person_profile_by_name_employer(self, mock_get): + mock_get.return_value = _make_response(200, {"full_name": "John Smith"}) + self.client.get_person_profile( + first_name="John", last_name="Smith", employer_website="stripe.com" + ) + params = mock_get.call_args.kwargs["params"] + assert params["first_name"] == "John" + assert params["last_name"] == "Smith" + assert params["employer_website"] == "stripe.com" + + @patch("aden_tools.tools.ninjapear_tool.ninjapear_tool.httpx.get") + def test_get_person_profile_by_role_employer(self, mock_get): + mock_get.return_value = _make_response(200, {"full_name": "Alice"}) + self.client.get_person_profile(employer_website="stripe.com", role="CTO") + params = mock_get.call_args.kwargs["params"] + assert params["role"] == "CTO" + assert params["employer_website"] == "stripe.com" + + @patch("aden_tools.tools.ninjapear_tool.ninjapear_tool.httpx.get") + def test_get_person_profile_empty_params_omitted(self, mock_get): + """Empty string params must not be sent to the API.""" + mock_get.return_value = _make_response(200, {}) + self.client.get_person_profile(slug="janesmith") + params = mock_get.call_args.kwargs["params"] + assert "work_email" not in params + assert "first_name" not in params + assert params["slug"] == "janesmith" + + @patch("aden_tools.tools.ninjapear_tool.ninjapear_tool.httpx.get") + def test_get_person_profile_timeout_100s(self, mock_get): + mock_get.return_value = _make_response(200, {}) + self.client.get_person_profile(slug="x") + assert mock_get.call_args.kwargs["timeout"] == 100.0 + + # --- get_company_details --- + + @patch("aden_tools.tools.ninjapear_tool.ninjapear_tool.httpx.get") + def test_get_company_details_basic(self, mock_get): + mock_get.return_value = _make_response(200, {"name": "Stripe"}) + result = self.client.get_company_details("stripe.com") + assert result["name"] == "Stripe" + url = mock_get.call_args.args[0] + assert url == f"{NINJAPEAR_API_BASE}/api/v1/company/details" + + @patch("aden_tools.tools.ninjapear_tool.ninjapear_tool.httpx.get") + def test_get_company_details_employee_count_flag(self, mock_get): + mock_get.return_value = _make_response(200, {}) + self.client.get_company_details("stripe.com", include_employee_count=True) + params = mock_get.call_args.kwargs["params"] + assert params["include_employee_count"] == "true" + + @patch("aden_tools.tools.ninjapear_tool.ninjapear_tool.httpx.get") + def test_get_company_details_follower_count_flag(self, mock_get): + mock_get.return_value = _make_response(200, {}) + self.client.get_company_details("stripe.com", include_follower_count=True) + params = mock_get.call_args.kwargs["params"] + assert params["follower_count"] == "include" + + @patch("aden_tools.tools.ninjapear_tool.ninjapear_tool.httpx.get") + def test_get_company_details_no_optional_flags(self, mock_get): + mock_get.return_value = _make_response(200, {}) + self.client.get_company_details("stripe.com") + params = mock_get.call_args.kwargs["params"] + assert "include_employee_count" not in params + assert "follower_count" not in params + + # --- get_company_funding --- + + @patch("aden_tools.tools.ninjapear_tool.ninjapear_tool.httpx.get") + def test_get_company_funding(self, mock_get): + body = {"total_funds_raised_usd": 2000000000, "funding_rounds": []} + mock_get.return_value = _make_response(200, body) + result = self.client.get_company_funding("stripe.com") + assert result["total_funds_raised_usd"] == 2000000000 + url = mock_get.call_args.args[0] + assert url == f"{NINJAPEAR_API_BASE}/api/v1/company/funding" + + # --- get_company_updates --- + + @patch("aden_tools.tools.ninjapear_tool.ninjapear_tool.httpx.get") + def test_get_company_updates(self, mock_get): + body = {"updates": [{"title": "We are hiring"}], "blogs": []} + mock_get.return_value = _make_response(200, body) + result = self.client.get_company_updates("stripe.com") + assert result["updates"][0]["title"] == "We are hiring" + url = mock_get.call_args.args[0] + assert url == f"{NINJAPEAR_API_BASE}/api/v1/company/updates" + + # --- get_company_customers --- + + @patch("aden_tools.tools.ninjapear_tool.ninjapear_tool.httpx.get") + def test_get_company_customers_default_params(self, mock_get): + mock_get.return_value = _make_response(200, {"customers": []}) + self.client.get_company_customers("stripe.com") + params = mock_get.call_args.kwargs["params"] + assert params["website"] == "stripe.com" + # Client default is 200; the tool function default (50) is a separate layer + assert params["page_size"] == 200 + # quality_filter=True means param is not sent (default) + assert "quality_filter" not in params + + @patch("aden_tools.tools.ninjapear_tool.ninjapear_tool.httpx.get") + def test_get_company_customers_quality_filter_false(self, mock_get): + mock_get.return_value = _make_response(200, {}) + self.client.get_company_customers("stripe.com", quality_filter=False) + params = mock_get.call_args.kwargs["params"] + assert params["quality_filter"] == "false" + + @patch("aden_tools.tools.ninjapear_tool.ninjapear_tool.httpx.get") + def test_get_company_customers_page_size_forwarded(self, mock_get): + mock_get.return_value = _make_response(200, {}) + self.client.get_company_customers("stripe.com", page_size=50) + params = mock_get.call_args.kwargs["params"] + assert params["page_size"] == 50 + + @patch("aden_tools.tools.ninjapear_tool.ninjapear_tool.httpx.get") + def test_get_company_customers_cursor(self, mock_get): + mock_get.return_value = _make_response(200, {}) + self.client.get_company_customers("stripe.com", cursor="abc123") + params = mock_get.call_args.kwargs["params"] + assert params["cursor"] == "abc123" + + # --- get_company_competitors --- + + @patch("aden_tools.tools.ninjapear_tool.ninjapear_tool.httpx.get") + def test_get_company_competitors(self, mock_get): + body = { + "competitors": [{"website": "braintree.com", "competition_reason": "product_overlap"}] + } + mock_get.return_value = _make_response(200, body) + result = self.client.get_company_competitors("stripe.com") + assert result["competitors"][0]["website"] == "braintree.com" + url = mock_get.call_args.args[0] + assert url == f"{NINJAPEAR_API_BASE}/api/v1/competitor/listing" + + # --- get_credit_balance --- + + @patch("aden_tools.tools.ninjapear_tool.ninjapear_tool.httpx.get") + def test_get_credit_balance(self, mock_get): + mock_get.return_value = _make_response(200, {"credit_balance": 9500}) + result = self.client.get_credit_balance() + assert result["credit_balance"] == 9500 + url = mock_get.call_args.args[0] + assert url == f"{NINJAPEAR_API_BASE}/api/v1/meta/credit-balance" + + @patch("aden_tools.tools.ninjapear_tool.ninjapear_tool.httpx.get") + def test_get_credit_balance_timeout_30s(self, mock_get): + mock_get.return_value = _make_response(200, {"credit_balance": 0}) + self.client.get_credit_balance() + assert mock_get.call_args.kwargs["timeout"] == 30.0 + + +# --------------------------------------------------------------------------- +# Tool registration +# --------------------------------------------------------------------------- + + +class TestToolRegistration: + def test_all_tools_registered(self): + _, _, by_name = _register() + for name in TOOL_NAMES: + assert name in by_name, f"Tool '{name}' was not registered" + + def test_tool_count(self): + mcp = MagicMock() + mcp.tool.return_value = lambda fn: fn + register_tools(mcp) + assert mcp.tool.call_count == len(TOOL_NAMES) + + def test_no_credentials_returns_error(self): + with patch.dict("os.environ", {}, clear=True): + _, _, by_name = _register(credentials=None) + result = by_name["ninjapear_get_credit_balance"]() + assert "error" in result + assert "not configured" in result["error"] + + def test_credentials_from_env_var(self): + _, _, by_name = _register(credentials=None) + with ( + patch.dict("os.environ", {"NINJAPEAR_API_KEY": "env-key"}), + patch("aden_tools.tools.ninjapear_tool.ninjapear_tool.httpx.get") as mock_get, + ): + mock_get.return_value = _make_response(200, {"credit_balance": 10}) + result = by_name["ninjapear_get_credit_balance"]() + assert result["credit_balance"] == 10 + call_headers = mock_get.call_args.kwargs["headers"] + assert call_headers["Authorization"] == "Bearer env-key" + + def test_credentials_from_credential_manager(self): + cred = MagicMock() + cred.get.return_value = "manager-key" + _, _, by_name = _register(credentials=cred) + with patch("aden_tools.tools.ninjapear_tool.ninjapear_tool.httpx.get") as mock_get: + mock_get.return_value = _make_response(200, {"credit_balance": 42}) + result = by_name["ninjapear_get_credit_balance"]() + cred.get.assert_called_with("ninjapear") + assert result["credit_balance"] == 42 + + +# --------------------------------------------------------------------------- +# ninjapear_get_person_profile +# --------------------------------------------------------------------------- + + +class TestPersonProfileTool: + def setup_method(self): + cred = MagicMock() + cred.get.return_value = "tok" + _, _, self.fns = _register(credentials=cred) + + def _fn(self): + return self.fns["ninjapear_get_person_profile"] + + def test_no_input_returns_error(self): + result = self._fn()() + assert "error" in result + assert "Insufficient input" in result["error"] + + def test_partial_input_name_without_employer_returns_error(self): + # first_name alone is not a valid combo + result = self._fn()(first_name="Jane") + assert "error" in result + + def test_partial_input_role_without_employer_returns_error(self): + result = self._fn()(role="CTO") + assert "error" in result + + @patch("aden_tools.tools.ninjapear_tool.ninjapear_tool.httpx.get") + def test_valid_work_email(self, mock_get): + mock_get.return_value = _make_response(200, {"full_name": "Jane Doe"}) + result = self._fn()(work_email="jane@stripe.com") + assert result["full_name"] == "Jane Doe" + + @patch("aden_tools.tools.ninjapear_tool.ninjapear_tool.httpx.get") + def test_valid_name_plus_employer(self, mock_get): + mock_get.return_value = _make_response(200, {"full_name": "John"}) + result = self._fn()(first_name="John", employer_website="stripe.com") + assert "full_name" in result + + @patch("aden_tools.tools.ninjapear_tool.ninjapear_tool.httpx.get") + def test_valid_role_plus_employer(self, mock_get): + mock_get.return_value = _make_response(200, {"full_name": "Alice"}) + result = self._fn()(role="CTO", employer_website="stripe.com") + assert "full_name" in result + + @patch("aden_tools.tools.ninjapear_tool.ninjapear_tool.httpx.get") + def test_valid_slug(self, mock_get): + mock_get.return_value = _make_response(200, {"slug": "janesmith"}) + result = self._fn()(slug="janesmith") + assert "slug" in result + + @patch("aden_tools.tools.ninjapear_tool.ninjapear_tool.httpx.get") + def test_valid_profile_id(self, mock_get): + mock_get.return_value = _make_response(200, {"id": "abc12345"}) + result = self._fn()(profile_id="abc12345") + assert "id" in result + + @patch("aden_tools.tools.ninjapear_tool.ninjapear_tool.httpx.get") + def test_timeout_returns_error(self, mock_get): + mock_get.side_effect = httpx.TimeoutException("timed out") + result = self._fn()(work_email="jane@stripe.com") + assert "error" in result + assert "timed out" in result["error"].lower() or "100s" in result["error"] + + @patch("aden_tools.tools.ninjapear_tool.ninjapear_tool.httpx.get") + def test_network_error_returns_error(self, mock_get): + mock_get.side_effect = httpx.RequestError("connection failed") + result = self._fn()(work_email="jane@stripe.com") + assert "error" in result + assert "Network error" in result["error"] + + +# --------------------------------------------------------------------------- +# ninjapear_get_company_details +# --------------------------------------------------------------------------- + + +class TestCompanyDetailsTool: + def setup_method(self): + cred = MagicMock() + cred.get.return_value = "tok" + _, _, self.fns = _register(credentials=cred) + + def _fn(self): + return self.fns["ninjapear_get_company_details"] + + def test_missing_website_returns_error(self): + result = self._fn()(website="") + assert "error" in result + + @patch("aden_tools.tools.ninjapear_tool.ninjapear_tool.httpx.get") + def test_happy_path(self, mock_get): + mock_get.return_value = _make_response(200, {"name": "Stripe", "founded_year": 2010}) + result = self._fn()(website="stripe.com") + assert result["name"] == "Stripe" + + @patch("aden_tools.tools.ninjapear_tool.ninjapear_tool.httpx.get") + def test_with_employee_count_flag(self, mock_get): + mock_get.return_value = _make_response(200, {"employee_count": 7000}) + result = self._fn()(website="stripe.com", include_employee_count=True) + assert result["employee_count"] == 7000 + params = mock_get.call_args.kwargs["params"] + assert params["include_employee_count"] == "true" + + @patch("aden_tools.tools.ninjapear_tool.ninjapear_tool.httpx.get") + def test_timeout_returns_error(self, mock_get): + mock_get.side_effect = httpx.TimeoutException("timed out") + result = self._fn()(website="stripe.com") + assert "error" in result + + @patch("aden_tools.tools.ninjapear_tool.ninjapear_tool.httpx.get") + def test_network_error_returns_error(self, mock_get): + mock_get.side_effect = httpx.RequestError("connection failed") + result = self._fn()(website="stripe.com") + assert "error" in result + + +# --------------------------------------------------------------------------- +# ninjapear_get_company_funding +# --------------------------------------------------------------------------- + + +class TestCompanyFundingTool: + def setup_method(self): + cred = MagicMock() + cred.get.return_value = "tok" + _, _, self.fns = _register(credentials=cred) + + def _fn(self): + return self.fns["ninjapear_get_company_funding"] + + def test_missing_website_returns_error(self): + result = self._fn()(website="") + assert "error" in result + + @patch("aden_tools.tools.ninjapear_tool.ninjapear_tool.httpx.get") + def test_happy_path(self, mock_get): + body = { + "total_funds_raised_usd": 2000000000, + "funding_rounds": [{"round_type": "SERIES_A", "amount_usd": 5000000, "investors": []}], + } + mock_get.return_value = _make_response(200, body) + result = self._fn()(website="stripe.com") + assert result["total_funds_raised_usd"] == 2000000000 + assert len(result["funding_rounds"]) == 1 + + +# --------------------------------------------------------------------------- +# ninjapear_get_company_updates +# --------------------------------------------------------------------------- + + +class TestCompanyUpdatesTool: + def setup_method(self): + cred = MagicMock() + cred.get.return_value = "tok" + _, _, self.fns = _register(credentials=cred) + + def _fn(self): + return self.fns["ninjapear_get_company_updates"] + + def test_missing_website_returns_error(self): + result = self._fn()(website="") + assert "error" in result + + @patch("aden_tools.tools.ninjapear_tool.ninjapear_tool.httpx.get") + def test_happy_path(self, mock_get): + body = { + "updates": [ + {"title": "We're hiring 500 engineers", "source": "blog", "timestamp": "2024-01-15"} + ], + "blogs": ["https://stripe.com/blog"], + } + mock_get.return_value = _make_response(200, body) + result = self._fn()(website="stripe.com") + assert len(result["updates"]) == 1 + assert "hiring" in result["updates"][0]["title"] + + +# --------------------------------------------------------------------------- +# ninjapear_get_company_customers +# --------------------------------------------------------------------------- + + +class TestCompanyCustomersTool: + def setup_method(self): + cred = MagicMock() + cred.get.return_value = "tok" + _, _, self.fns = _register(credentials=cred) + + def _fn(self): + return self.fns["ninjapear_get_company_customers"] + + def test_missing_website_returns_error(self): + result = self._fn()(website="") + assert "error" in result + + def test_page_size_too_large_returns_error(self): + result = self._fn()(website="stripe.com", page_size=999) + assert "error" in result + + def test_page_size_zero_returns_error(self): + result = self._fn()(website="stripe.com", page_size=0) + assert "error" in result + + @patch("aden_tools.tools.ninjapear_tool.ninjapear_tool.httpx.get") + def test_happy_path(self, mock_get): + body = { + "customers": [{"name": "Amazon", "website": "amazon.com"}], + "investors": [], + "partner_platforms": [], + "next_page": None, + } + mock_get.return_value = _make_response(200, body) + result = self._fn()(website="stripe.com") + assert result["customers"][0]["name"] == "Amazon" + + @patch("aden_tools.tools.ninjapear_tool.ninjapear_tool.httpx.get") + def test_pagination_cursor_forwarded(self, mock_get): + mock_get.return_value = _make_response(200, {"customers": [], "next_page": None}) + self._fn()(website="stripe.com", cursor="next-page-token") + params = mock_get.call_args.kwargs["params"] + assert params["cursor"] == "next-page-token" + + @patch("aden_tools.tools.ninjapear_tool.ninjapear_tool.httpx.get") + def test_quality_filter_false_forwarded(self, mock_get): + mock_get.return_value = _make_response(200, {}) + self._fn()(website="stripe.com", quality_filter=False) + params = mock_get.call_args.kwargs["params"] + assert params["quality_filter"] == "false" + + +# --------------------------------------------------------------------------- +# ninjapear_get_company_competitors +# --------------------------------------------------------------------------- + + +class TestCompanyCompetitorsTool: + def setup_method(self): + cred = MagicMock() + cred.get.return_value = "tok" + _, _, self.fns = _register(credentials=cred) + + def _fn(self): + return self.fns["ninjapear_get_company_competitors"] + + def test_missing_website_returns_error(self): + result = self._fn()(website="") + assert "error" in result + + @patch("aden_tools.tools.ninjapear_tool.ninjapear_tool.httpx.get") + def test_happy_path(self, mock_get): + body = { + "competitors": [ + {"website": "braintree.com", "competition_reason": "product_overlap"}, + {"website": "adyen.com", "competition_reason": "organic_keyword_overlap"}, + ] + } + mock_get.return_value = _make_response(200, body) + result = self._fn()(website="stripe.com") + assert len(result["competitors"]) == 2 + assert result["competitors"][0]["competition_reason"] == "product_overlap" + + +# --------------------------------------------------------------------------- +# ninjapear_get_credit_balance +# --------------------------------------------------------------------------- + + +class TestCreditBalanceTool: + def setup_method(self): + cred = MagicMock() + cred.get.return_value = "tok" + _, _, self.fns = _register(credentials=cred) + + def _fn(self): + return self.fns["ninjapear_get_credit_balance"] + + @patch("aden_tools.tools.ninjapear_tool.ninjapear_tool.httpx.get") + def test_happy_path(self, mock_get): + mock_get.return_value = _make_response(200, {"credit_balance": 9500}) + result = self._fn()() + assert result["credit_balance"] == 9500 + + @patch("aden_tools.tools.ninjapear_tool.ninjapear_tool.httpx.get") + def test_timeout_returns_error(self, mock_get): + mock_get.side_effect = httpx.TimeoutException("timed out") + result = self._fn()() + assert "error" in result + + @patch("aden_tools.tools.ninjapear_tool.ninjapear_tool.httpx.get") + def test_network_error_returns_error(self, mock_get): + mock_get.side_effect = httpx.RequestError("conn refused") + result = self._fn()() + assert "error" in result + assert "Network error" in result["error"] + + def test_no_credentials_returns_error(self): + with patch.dict("os.environ", {}, clear=True): + _, _, fns = _register(credentials=None) + result = fns["ninjapear_get_credit_balance"]() + assert "error" in result + assert "not configured" in result["error"] + + +# --------------------------------------------------------------------------- +# CredentialSpec +# --------------------------------------------------------------------------- + + +class TestCredentialSpec: + def test_spec_exists(self): + from aden_tools.credentials import NINJAPEAR_CREDENTIALS + + assert "ninjapear" in NINJAPEAR_CREDENTIALS + + def test_env_var(self): + from aden_tools.credentials import NINJAPEAR_CREDENTIALS + + assert NINJAPEAR_CREDENTIALS["ninjapear"].env_var == "NINJAPEAR_API_KEY" + + def test_tools_list_complete(self): + from aden_tools.credentials import NINJAPEAR_CREDENTIALS + + spec_tools = NINJAPEAR_CREDENTIALS["ninjapear"].tools + for name in TOOL_NAMES: + assert name in spec_tools, f"Tool '{name}' missing from CredentialSpec.tools" + + def test_health_check_endpoint(self): + from aden_tools.credentials import NINJAPEAR_CREDENTIALS + + spec = NINJAPEAR_CREDENTIALS["ninjapear"] + assert "credit-balance" in spec.health_check_endpoint + + def test_in_global_credential_specs(self): + from aden_tools.credentials import CREDENTIAL_SPECS + + assert "ninjapear" in CREDENTIAL_SPECS + + +# --------------------------------------------------------------------------- +# Live tests — skipped unless NINJAPEAR_API_KEY is set in environment +# --------------------------------------------------------------------------- + + +@pytest.mark.live +class TestLive: + """ + Real API calls against NinjaPear. Excluded from CI by default. + + To run: + export NINJAPEAR_API_KEY=your-key + cd tools + uv run pytest src/aden_tools/tools/ninjapear_tool/tests/ -m live -v + """ + + @pytest.fixture(autouse=True) + def require_api_key(self): + if not os.environ.get("NINJAPEAR_API_KEY"): + pytest.skip("NINJAPEAR_API_KEY not set") + + def test_live_credit_balance(self): + """Smoke test: free endpoint, 0 credits consumed.""" + _, _, fns = _register(credentials=None) + result = fns["ninjapear_get_credit_balance"]() + assert "credit_balance" in result, f"Unexpected response: {result}" + assert isinstance(result["credit_balance"], int) + + def test_live_company_details_stripe(self): + """Fetch Stripe company details. Costs 2 credits.""" + _, _, fns = _register(credentials=None) + result = fns["ninjapear_get_company_details"](website="stripe.com") + assert "error" not in result, f"API error: {result}" + assert result.get("name") is not None + + def test_live_person_profile_by_role(self): + """Look up CTO at stripe.com. Costs 3 credits.""" + _, _, fns = _register(credentials=None) + result = fns["ninjapear_get_person_profile"](role="CTO", employer_website="stripe.com") + # 404 is acceptable (person may not be indexed), anything else is a bug + if "error" in result: + assert "Not found" in result["error"], f"Unexpected error: {result}" + else: + assert "full_name" in result or "id" in result