From 6b218c1a695f699c2e854f543b8b40988df4257e Mon Sep 17 00:00:00 2001 From: Giga Chkhikvadze Date: Mon, 19 Oct 2026 13:18:04 +0400 Subject: [PATCH] feat: add People Search tools (1.1.0) Add two LangChain tools for Desearch People Search: - PeopleSearchTool (people_search_tool) starts a billed search on x, linkedin, youtube or tiktok, waits up to `timeout` seconds and returns compact people (name, handle, url, platform, audience, location, score, why they matched, contacts, links). Called again with only task_id it keeps waiting on the same search without starting or billing a new one. Platform filters that the API would reject are refused before any call. - PeopleProfilesTool (people_profiles_tool) lists the free profiles library with the result filters (has, min/max_audience, min_score, location, verified) and sorting. Both are in a new people_tools group and kept out of search_tools, so the default agent never starts a billed search. The tools call the /desearch/people routes over HTTP with aiohttp because the People Search methods arrive in desearch-py 1.4.0, which is not on PyPI yet; the desearch-py requirement is unchanged. Tests run against a local aiohttp server, never the live API. Co-Authored-By: Claude Opus 5.5 --- README.md | 58 +++ langchain_desearch/__init__.py | 4 + langchain_desearch/people.py | 475 +++++++++++++++++++++++ langchain_desearch/search_tools.py | 9 + langchain_desearch/tools.py | 461 +++++++++++++++++++++++ pyproject.toml | 3 +- setup.py | 3 +- tests/unit_tests/test_imports.py | 2 + tests/unit_tests/test_people_tools.py | 520 ++++++++++++++++++++++++++ tests/unit_tests/test_tools.py | 4 +- 10 files changed, 1535 insertions(+), 4 deletions(-) create mode 100644 langchain_desearch/people.py create mode 100644 tests/unit_tests/test_people_tools.py diff --git a/README.md b/README.md index f76fe15..77e57cf 100644 --- a/README.md +++ b/README.md @@ -7,6 +7,7 @@ This project integrates the Desearch API with LangChain tools to enable various - **Grouped Tools**: - **Search Tools**: General-purpose search tools for AI, web, and Twitter searches. - **Twitter Tools**: Tools specifically for Twitter-related operations. + - **People Tools**: Find people on X, LinkedIn, YouTube and TikTok who match a plain-language description, and browse everyone your searches found. ## Installation @@ -37,6 +38,13 @@ The `twitter_tools` group contains tools specifically for Twitter-related operat - `FetchRetweetsByPostTool`: Retrieve users who retweeted a specific post. - `FetchTwitterUserTool`: Get detailed information about a Twitter user. +#### People Tools +The `people_tools` group contains the [People Search](https://www.desearch.ai/docs/guide/apis/people-search) tools: +- `PeopleSearchTool` (`people_search_tool`): Start a People Search on `x`, `linkedin`, `youtube` or `tiktok` and wait for it. Returns compact people: name, handle, url, platform, audience, location, fit score, why they matched, contacts and links. +- `PeopleProfilesTool` (`people_profiles_tool`): List the people your earlier searches found on one platform, with filters (`has=["email"]`, audience range, minimum score, location, verified) and sorting. Free. + +> **Billing and timing**: `people_search_tool` is billed. Starting a search reserves `count` profiles (5-200) at the route price; you pay only for the profiles it delivers, and a search that finds no one is free. A search runs in the background for 1-5 minutes. The tool waits up to `timeout` seconds (default 300), then returns the people found so far with the `task_id`. Call it again with only `task_id` to keep waiting: that is free and does not start a new search. `people_tools` is kept out of `search_tools`, so `create_search_agent` never starts a billed search. + ### Examples #### Using Tools @@ -75,6 +83,55 @@ result = tool._run( print(result) ``` +#### Using People Search +```python +from langchain_desearch.tools import PeopleSearchTool, PeopleProfilesTool +from dotenv import load_dotenv +load_dotenv() + +# Start a billed search and wait up to 5 minutes for it. +people = PeopleSearchTool().invoke({ + "query": "Indie hackers building in public who share their MRR", + "platform": "x", + "count": 25, + "min_score": 60, + "min_followers": 1000, + "timeout": 300, +}) +print(people["status"], len(people["people"])) +for person in people["people"]: + print(person["handle"], person.get("audience"), person.get("why")) + +# Still running? Keep waiting on the same search (free, nothing new is billed). +if not people["done"]: + people = PeopleSearchTool().invoke({"task_id": people["task_id"], "timeout": 300}) + +# Browse everyone your X searches found who has a public email (free). +library = PeopleProfilesTool().invoke({ + "platform": "x", + "has": ["email"], + "min_audience": 1000, + "sort": "score", + "limit": 25, +}) +print(library["total"], library.get("next_cursor")) +``` + +Platform filters for `people_search_tool`: + +| Platform | Filters | +| --- | --- | +| `x` | `min_followers`, `max_followers`, `verified_only`, `countries`, `keywords`, `exclude_keywords` | +| `linkedin` | `min_followers`, `max_followers`, `countries`, `keywords`, `exclude_keywords` | +| `youtube` | `min_subscribers`, `max_subscribers`, `countries`, `require_contact`, `keywords`, `exclude_keywords` | +| `tiktok` | `min_followers`, `max_followers`, `require_contact`, `keywords`, `exclude_keywords` | + +A filter that does not apply to the platform is rejected before any request is sent, so nothing is billed. Every search also takes `count`, `min_score`, `exclude` (handles to leave out) and `idempotency_key` (sending the same key again returns the first search instead of billing a new one). + +To give an agent the People tools, pass `people_tools` from `langchain_desearch.search_tools` to your agent. Like the other tools, they set `return_direct=True`; create them with `PeopleSearchTool(return_direct=False)` if the agent should read the result and keep going. + +> People Search calls the `/desearch/people` routes over HTTP with aiohttp, because the People Search methods in `desearch-py` arrive in 1.4.0, which is not on PyPI yet. The tools will switch to the SDK once it is published. + #### Using RAG (Retrieval-Augmented Generation) ```python from langchain_core.prompts import PromptTemplate, ChatPromptTemplate @@ -186,6 +243,7 @@ langchain_desearch/ ├── langchain_desearch/ │ ├── __init__.py │ ├── tools.py +│ ├── people.py │ ├── search_tools.py │ ├── agent.py ├── examples/ diff --git a/langchain_desearch/__init__.py b/langchain_desearch/__init__.py index c2425ba..b09b84f 100644 --- a/langchain_desearch/__init__.py +++ b/langchain_desearch/__init__.py @@ -2,10 +2,14 @@ DesearchTool, BasicWebSearchTool, BasicTwitterSearchTool, + PeopleSearchTool, + PeopleProfilesTool, ) __all__ = [ "DesearchTool", "BasicWebSearchTool", "BasicTwitterSearchTool", + "PeopleSearchTool", + "PeopleProfilesTool", ] diff --git a/langchain_desearch/people.py b/langchain_desearch/people.py new file mode 100644 index 0000000..84d8408 --- /dev/null +++ b/langchain_desearch/people.py @@ -0,0 +1,475 @@ +"""People Search on the Desearch API (``/desearch/people/*``). + +desearch-py adds People Search methods in 1.4.0, which is not on PyPI yet, so +this module calls the routes directly with aiohttp (installed with desearch-py) +and the same ``Authorization: `` header the SDK sends. Switch the +tools to the SDK methods once desearch-py 1.4.0 is published. + +- ``POST /desearch/people/search/{platform}`` starts a background search and is + billed: it reserves ``count`` profiles at the route price and settles to the + profiles it delivers. A search that finds no one is free. +- Reading a task, its results and the profiles library is free. +""" + +import asyncio +import json +from typing import Any, Dict, List, Optional, Tuple +from urllib.parse import quote + +import aiohttp + +PEOPLE_API_BASE_URL = "https://api.desearch.ai" + +PEOPLE_PLATFORMS = ("x", "youtube", "linkedin", "tiktok") + +# Statuses after which a task never changes again. +PEOPLE_ENDED_STATUSES = frozenset( + {"completed", "partial", "failed", "cancelled", "expired"} +) + +# Channels accepted by the ``has`` filter of the profiles library. +PEOPLE_CHANNELS = ("email", "website", "x", "linkedin", "instagram", "youtube", "tiktok") + +PEOPLE_PROFILE_SORTS = ("score", "recent", "audience") + +# Search filters each platform accepts in the request ``filters`` object. The +# API rejects other keys with HTTP 422, so they are refused before the billed +# call is made. +PEOPLE_SEARCH_FILTERS: Dict[str, Tuple[str, ...]] = { + "x": ( + "min_followers", + "max_followers", + "verified_only", + "countries", + "keywords", + "exclude_keywords", + ), + "youtube": ( + "min_subscribers", + "max_subscribers", + "countries", + "require_contact", + "keywords", + "exclude_keywords", + ), + "linkedin": ( + "min_followers", + "max_followers", + "countries", + "keywords", + "exclude_keywords", + ), + "tiktok": ( + "min_followers", + "max_followers", + "require_contact", + "keywords", + "exclude_keywords", + ), +} + +_REQUEST_TIMEOUT_SECONDS = 120 +_RESULTS_PAGE_SIZE = 200 +_MAX_RESULT_PAGES = 25 + + +class DesearchPeopleError(Exception): + """An HTTP error from a People Search route. The message never holds the API key.""" + + def __init__(self, status: int, detail: str) -> None: + self.status = status + self.detail = detail + super().__init__(f"HTTP {status}: {detail}") + + +def _error_detail(data: Any, text: str) -> str: + if isinstance(data, dict) and data.get("detail") is not None: + detail = data["detail"] + if not isinstance(detail, str): + detail = json.dumps(detail) + task = data.get("task") + if isinstance(task, dict) and isinstance(task.get("status"), str): + return f"{detail} Task status: {task['status']}." + return detail + return text if len(text) <= 500 else f"{text[:500]}..." + + +class PeopleClient: + """Minimal async client for the People Search routes.""" + + def __init__(self, api_key: str, base_url: str = PEOPLE_API_BASE_URL) -> None: + self._api_key = api_key + self._base_url = base_url.rstrip("/") + self._session: Optional[aiohttp.ClientSession] = None + + async def __aenter__(self) -> "PeopleClient": + return self + + async def __aexit__(self, *exc_info: Any) -> None: + await self.close() + + async def close(self) -> None: + if self._session is not None: + await self._session.close() + self._session = None + + async def _call( + self, + method: str, + path: str, + *, + params: Optional[Dict[str, Any]] = None, + body: Optional[Dict[str, Any]] = None, + headers: Optional[Dict[str, str]] = None, + ) -> Tuple[int, Any]: + if self._session is None: + self._session = aiohttp.ClientSession( + timeout=aiohttp.ClientTimeout(total=_REQUEST_TIMEOUT_SECONDS) + ) + request_headers = {"Authorization": self._api_key} + if headers: + request_headers.update(headers) + async with self._session.request( + method, + f"{self._base_url}{path}", + params=params, + json=body, + headers=request_headers, + ) as response: + status = response.status + text = await response.text() + data: Any = text + if text: + try: + data = json.loads(text) + except ValueError: + pass + if status >= 400: + raise DesearchPeopleError(status, _error_detail(data, text)) + return status, data + + @staticmethod + def _task_path(task_id: str, suffix: str = "") -> str: + return f"/desearch/people/tasks/{quote(str(task_id), safe='')}{suffix}" + + async def start_search( + self, + platform: str, + body: Dict[str, Any], + idempotency_key: Optional[str] = None, + ) -> Tuple[int, Dict[str, Any]]: + """Start a billed search. Returns the HTTP status (202 new, 200 replayed) and the task.""" + headers = {"Idempotency-Key": idempotency_key} if idempotency_key else None + return await self._call( + "POST", + f"/desearch/people/search/{quote(platform, safe='')}", + body=body, + headers=headers, + ) + + async def get_task(self, task_id: str) -> Dict[str, Any]: + _, data = await self._call("GET", self._task_path(task_id)) + return data + + async def get_results( + self, task_id: str, cursor: Optional[str] = None, limit: Optional[int] = None + ) -> Dict[str, Any]: + params = _drop_none({"cursor": cursor, "limit": limit}) + _, data = await self._call( + "GET", self._task_path(task_id, "/results"), params=params + ) + return data + + async def list_profiles(self, params: Dict[str, Any]) -> Dict[str, Any]: + _, data = await self._call("GET", "/desearch/people/profiles", params=params) + return data + + +def _drop_none(values: Dict[str, Any]) -> Dict[str, Any]: + return {key: value for key, value in values.items() if value is not None} + + +def people_search_body( + platform: str, + query: str, + count: Optional[int] = None, + min_score: Optional[int] = None, + exclude: Optional[List[str]] = None, + **filters: Any, +) -> Dict[str, Any]: + """Build the request body. Raises ValueError for a filter the platform does not accept.""" + if platform not in PEOPLE_SEARCH_FILTERS: + raise ValueError( + f"Unknown platform {platform!r}. Use one of: {', '.join(PEOPLE_PLATFORMS)}." + ) + allowed = PEOPLE_SEARCH_FILTERS[platform] + set_filters = _drop_none(filters) + rejected = [name for name in set_filters if name not in allowed] + if rejected: + verb = "does" if len(rejected) == 1 else "do" + raise ValueError( + f"{', '.join(rejected)} {verb} not apply to {platform}. " + f"{platform} filters: {', '.join(allowed)}. Nothing was charged." + ) + body = _drop_none( + {"query": query, "count": count, "min_score": min_score, "exclude": exclude} + ) + if set_filters: + body["filters"] = set_filters + return body + + +def profile_filter_params( + platform: str, + q: Optional[str] = None, + sort: Optional[str] = None, + limit: Optional[int] = None, + cursor: Optional[str] = None, + has: Optional[List[str]] = None, + min_audience: Optional[int] = None, + max_audience: Optional[int] = None, + min_score: Optional[int] = None, + location: Optional[str] = None, + verified: Optional[bool] = None, +) -> Dict[str, Any]: + """Query parameters for ``GET /desearch/people/profiles``.""" + params = _drop_none( + { + "platform": platform, + "q": q, + "sort": sort, + "limit": limit, + "cursor": cursor, + "min_audience": min_audience, + "max_audience": max_audience, + "min_score": min_score, + "location": location, + } + ) + if has: + params["has"] = ",".join(dict.fromkeys(str(channel) for channel in has)) + if verified is not None: + params["verified"] = "true" if verified else "false" + return params + + +def _obj(value: Any) -> Dict[str, Any]: + return value if isinstance(value, dict) else {} + + +def _list(value: Any) -> List[Any]: + return value if isinstance(value, list) else [] + + +def _number(value: Any) -> Optional[float]: + if isinstance(value, bool) or not isinstance(value, (int, float)): + return None + return value + + +def _text(value: Any) -> Optional[str]: + return value if isinstance(value, str) and value.strip() else None + + +def _compact(fields: Dict[str, Any]) -> Dict[str, Any]: + """Drop empty values so each person stays small in an LLM context.""" + return { + key: value + for key, value in fields.items() + if value is not None and value != "" and value != [] and value != {} + } + + +def compact_person(person: Any) -> Dict[str, Any]: + """One person as an agent needs them: who, audience, fit, why, and how to reach them.""" + p = _obj(person) + metrics = _obj(p.get("metrics")) + attributes = _obj(p.get("attributes")) + score = _obj(p.get("score")) + audience = _number(metrics.get("followers")) + if audience is None: + audience = _number(metrics.get("subscribers")) + reasons = [reason for reason in _list(score.get("reasons")) if _text(reason)] + why = _text(score.get("summary")) or (reasons[0] if reasons else None) + evidence_url = next( + ( + _text(item.get("url")) + for item in map(_obj, _list(p.get("evidence"))) + if _text(item.get("url")) + ), + None, + ) + found_by = p.get("found_by") + return _compact( + { + "name": _text(p.get("name")), + "handle": p.get("handle"), + "platform": p.get("platform"), + "url": _text(p.get("url")), + "audience": audience, + "location": _text(p.get("location")), + "verified": attributes.get("verified") + if isinstance(attributes.get("verified"), bool) + else None, + "headline": _text(attributes.get("headline")), + "company": _text(attributes.get("company")), + "score": _number(score.get("value")), + "why": why, + "contacts": [ + _compact({"type": item.get("type"), "value": item.get("value")}) + for item in map(_obj, _list(p.get("contacts"))) + if _text(item.get("value")) + ], + "links": [ + _compact({"type": item.get("type"), "url": item.get("url")}) + for item in map(_obj, _list(p.get("links"))) + if _text(item.get("url")) + ], + "evidence_url": evidence_url, + "found_by": _compact( + { + "task_id": found_by.get("task_id"), + "query": found_by.get("query"), + "times_found": found_by.get("times_found"), + "found_at": found_by.get("found_at"), + } + ) + if isinstance(found_by, dict) + else None, + } + ) + + +def compact_task(task: Any) -> Dict[str, Any]: + """Task fields an agent needs: status, progress counts and cost.""" + t = _obj(task) + status = t.get("status") + return _compact( + { + "task_id": t.get("task_id"), + "platform": t.get("platform"), + "status": status, + "stage": t.get("stage"), + "done": status in PEOPLE_ENDED_STATUSES if isinstance(status, str) else None, + "query": t.get("query"), + "count": t.get("count"), + "result_count": t.get("result_count"), + "reserved_cost_usd": t.get("reserved_cost_usd"), + "final_cost_usd": t.get("final_cost_usd"), + "refunded_usd": t.get("refunded_usd"), + "errors": t.get("errors"), + "created_at": t.get("created_at"), + "finished_at": t.get("finished_at"), + "results_expire_at": t.get("results_expire_at"), + } + ) + + +def people_next_step(status: Optional[str], task_id: Optional[str], tool_name: str) -> str: + """One sentence on what to do next, so an agent does not infer it from the status.""" + if status == "completed": + return "Finished. These are all the people this search found." + if status == "partial": + return ( + "Finished with partial results (some sub-searches failed). " + "These are all the people it found." + ) + if status == "failed": + return "The search failed; see errors. A search that finds no one is not charged." + if status == "cancelled": + return ( + "Cancelled. You pay only for the people it delivered; " + "the rest of the reservation is returned." + ) + if status == "expired": + return "The results expired (they are kept for 30 days after a search ends)." + if status == "paused": + return ( + f"Paused. It will not finish until it is resumed in the Desearch console " + f"or API. Task id: {task_id}." + ) + return ( + f"Still {status or 'running'}. The people found so far are listed. Call " + f"{tool_name} again with task_id='{task_id}' to keep waiting; that is free " + "and does not start or bill a new search." + ) + + +async def read_all_results(client: PeopleClient, task_id: str) -> List[Dict[str, Any]]: + """Read every person a task has found so far, following the cursor.""" + people: List[Dict[str, Any]] = [] + cursor: Optional[str] = None + for _ in range(_MAX_RESULT_PAGES): + page = _obj(await client.get_results(task_id, cursor=cursor, limit=_RESULTS_PAGE_SIZE)) + results = _list(page.get("results")) + people.extend(results) + cursor = page.get("next_cursor") + if not page.get("has_more") or not results or not cursor: + break + return people + + +async def search_people( + client: PeopleClient, + *, + tool_name: str, + timeout: float, + poll_interval: float, + platform: Optional[str] = None, + body: Optional[Dict[str, Any]] = None, + task_id: Optional[str] = None, + idempotency_key: Optional[str] = None, +) -> Dict[str, Any]: + """Start a search (or pick up ``task_id``), wait up to ``timeout`` seconds, return compact people.""" + replayed = False + if task_id: + task = _obj(await client.get_task(task_id)) + else: + if platform is None or body is None: + raise ValueError("platform and body are required to start a People Search.") + http_status, created = await client.start_search(platform, body, idempotency_key) + task = _obj(created) + # 200 means this Idempotency-Key already created the task; nothing new was charged. + replayed = http_status == 200 + task_id = task.get("task_id") + if not task_id: + raise DesearchPeopleError(http_status, "The response did not include a task_id.") + + loop = asyncio.get_running_loop() + deadline = loop.time() + max(float(timeout), 0.0) + while task.get("status") not in PEOPLE_ENDED_STATUSES: + remaining = deadline - loop.time() + if remaining <= 0: + break + await asyncio.sleep(min(poll_interval, remaining)) + task = _obj(await client.get_task(task_id)) + + people = await read_all_results(client, task_id) + output = compact_task(task) + output.setdefault("task_id", task_id) + if replayed: + output["existing_task"] = True + output["people"] = [compact_person(person) for person in people] + output["next_step"] = people_next_step(task.get("status"), task_id, tool_name) + return output + + +async def list_people_profiles( + client: PeopleClient, *, tool_name: str, **filters: Any +) -> Dict[str, Any]: + """One page of the profiles library, compacted.""" + platform = filters.get("platform") + page = _obj(await client.list_profiles(profile_filter_params(**filters))) + next_cursor = page.get("next_cursor") + output: Dict[str, Any] = { + "platform": platform, + "total": page.get("total"), + "people": [compact_person(person) for person in _list(page.get("profiles"))], + } + if next_cursor: + output["next_cursor"] = next_cursor + output["next_step"] = ( + f"More people match. Call {tool_name} again with the same filters and " + f"cursor='{next_cursor}'." + ) + return output diff --git a/langchain_desearch/search_tools.py b/langchain_desearch/search_tools.py index 5024430..7b2ef37 100644 --- a/langchain_desearch/search_tools.py +++ b/langchain_desearch/search_tools.py @@ -9,6 +9,8 @@ FetchRepliesByPostTool, FetchRetweetsByPostTool, FetchTwitterUserTool, + PeopleSearchTool, + PeopleProfilesTool, ) # Group only 3 search tools into a list @@ -29,3 +31,10 @@ FetchRetweetsByPostTool(), FetchTwitterUserTool(), ] + +# Group People Search tools into a separate list. people_search_tool is billed +# and takes minutes, so it is not part of search_tools. +people_tools = [ + PeopleSearchTool(), + PeopleProfilesTool(), +] diff --git a/langchain_desearch/tools.py b/langchain_desearch/tools.py index d6e98ba..8e93bf5 100644 --- a/langchain_desearch/tools.py +++ b/langchain_desearch/tools.py @@ -12,6 +12,14 @@ from langchain_core.tools import BaseTool from langchain_core.tools.base import ArgsSchema +from langchain_desearch.people import ( + PEOPLE_API_BASE_URL, + PeopleClient, + list_people_profiles, + people_search_body, + search_people, +) + _AI_SEARCH_MODELS = ("NOVA", "ORBIT", "HORIZON") @@ -677,3 +685,456 @@ async def _arun( """Use the tool asynchronously.""" del run_manager return await _acall_client("x_user_posts", username=user) + + +# ─── People Search ─────────────────────────────────────────────────────────── +# +# desearch-py ships People Search in 1.4.0, which is not on PyPI yet, so these +# tools call the /desearch/people routes over HTTP (see people.py). + +_PeoplePlatform = Literal["x", "youtube", "linkedin", "tiktok"] +_PeopleChannel = Literal[ + "email", "website", "x", "linkedin", "instagram", "youtube", "tiktok" +] +_PEOPLE_FILTER_FIELDS = ( + "min_followers", + "max_followers", + "min_subscribers", + "max_subscribers", + "verified_only", + "require_contact", + "countries", + "keywords", + "exclude_keywords", +) + + +def _people_error(exc: Exception) -> str: + detail = str(exc) or type(exc).__name__ + return f"An error occurred while calling Desearch: {detail}" + + +async def _invoke_people( + api_key: str, base_url: str, operation: Callable[..., Awaitable[Any]], **kwargs: Any +) -> Any: + async with PeopleClient(api_key, base_url) as client: + return await operation(client, **kwargs) + + +def _call_people( + api_key: str, base_url: str, operation: Callable[..., Awaitable[Any]], **kwargs: Any +) -> Any: + try: + return _run_coroutine( + lambda: _invoke_people(api_key, base_url, operation, **kwargs) + ) + except Exception as exc: + return _people_error(exc) + + +async def _acall_people( + api_key: str, base_url: str, operation: Callable[..., Awaitable[Any]], **kwargs: Any +) -> Any: + try: + return await _invoke_people(api_key, base_url, operation, **kwargs) + except Exception as exc: + return _people_error(exc) + + +class PeopleSearchToolInput(BaseModel): + query: Optional[str] = Field( + default=None, + min_length=3, + max_length=1000, + description=( + "Who to find, in plain language, for example 'Heads of sales at fintech " + "startups who post about outbound'. Required to start a search." + ), + ) + platform: Optional[_PeoplePlatform] = Field( + default=None, + description=( + "Where to search: 'x' (X accounts), 'linkedin' (LinkedIn members), " + "'youtube' (YouTube creators) or 'tiktok' (TikTok creators). " + "Required to start a search." + ), + ) + count: int = Field( + default=25, + ge=5, + le=200, + description=( + "How many qualified people to find (5-200). The search reserves this " + "many profiles at the route price and bills only the ones it delivers." + ), + ) + min_score: Optional[int] = Field( + default=None, + ge=0, + le=100, + description="Only return people with at least this fit score (0-100).", + ) + exclude: Optional[List[str]] = Field( + default=None, + description="Handles to leave out, for example people you already contacted.", + ) + min_followers: Optional[int] = Field( + default=None, ge=0, description="Minimum followers (x, linkedin, tiktok)." + ) + max_followers: Optional[int] = Field( + default=None, ge=0, description="Maximum followers (x, linkedin, tiktok)." + ) + min_subscribers: Optional[int] = Field( + default=None, ge=0, description="Minimum subscribers (youtube)." + ) + max_subscribers: Optional[int] = Field( + default=None, ge=0, description="Maximum subscribers (youtube)." + ) + verified_only: Optional[bool] = Field( + default=None, description="Only verified or X Premium accounts (x)." + ) + require_contact: Optional[bool] = Field( + default=None, + description="Only creators with a public email or contact link (youtube, tiktok).", + ) + countries: Optional[List[str]] = Field( + default=None, + description="Countries to prefer, for example ['US', 'GB'] (x, youtube, linkedin).", + ) + keywords: Optional[List[str]] = Field( + default=None, + description="Words that should appear in their posts, videos or bios.", + ) + exclude_keywords: Optional[List[str]] = Field( + default=None, description="Leave out people who mention these words." + ) + idempotency_key: Optional[str] = Field( + default=None, + max_length=200, + description=( + "Any unique string. Sending the same key again returns the first " + "search instead of starting and billing a new one." + ), + ) + timeout: int = Field( + default=300, + ge=0, + le=1800, + description=( + "Seconds to wait for the search to finish (default 300). If it is " + "still running, the people found so far are returned with the task_id." + ), + ) + task_id: Optional[str] = Field( + default=None, + description=( + "task_id of a search started earlier. Waits on that search and returns " + "its people instead of starting a new one. Free." + ), + ) + + @model_validator(mode="after") + def check_search_or_task(self) -> "PeopleSearchToolInput": + if self.task_id: + return self + if not self.query or not self.platform: + raise ValueError( + "Set query and platform to start a People Search, or task_id to " + "keep waiting on one." + ) + # Refuse filters the platform does not accept before anything is billed. + people_search_body( + self.platform, + self.query, + **{name: getattr(self, name) for name in _PEOPLE_FILTER_FIELDS}, + ) + return self + + +class PeopleSearchTool(BaseTool): + name: str = "people_search_tool" + description: str = ( + "Finds people on X, LinkedIn, YouTube or TikTok who match a plain-language " + "description (for example 'founders of AI developer-tool startups who post " + "about hiring') and returns compact profiles: name, handle, url, platform, " + "audience, location, fit score, why they matched, contacts and links. " + "BILLED: starting a search reserves `count` profiles at the Desearch route " + "price; you pay only for the profiles it delivers, and a search that finds " + "no one is free. SLOW: a search runs for 1-5 minutes. This tool waits up to " + "`timeout` seconds, then returns the people found so far and the task_id. " + "To keep waiting, call it again with only task_id (free; it does not start " + "or bill a new search). Check people_profiles_tool first for people earlier " + "searches already found." + ) + args_schema: ArgsSchema = PeopleSearchToolInput + return_direct: bool = True + base_url: str = PEOPLE_API_BASE_URL + poll_interval: float = 5.0 + + def _search_kwargs( + self, + query: Optional[str], + platform: Optional[str], + count: int, + min_score: Optional[int], + exclude: Optional[List[str]], + filters: Dict[str, Any], + idempotency_key: Optional[str], + timeout: int, + task_id: Optional[str], + ) -> Dict[str, Any]: + kwargs: Dict[str, Any] = { + "tool_name": self.name, + "timeout": timeout, + "poll_interval": self.poll_interval, + } + if task_id: + return {**kwargs, "task_id": task_id} + if not query or not platform: + raise ValueError( + "Set query and platform to start a People Search, or task_id to " + "keep waiting on one." + ) + body = people_search_body( + platform, + query, + count=count, + min_score=min_score, + exclude=exclude, + **filters, + ) + return { + **kwargs, + "platform": platform, + "body": body, + "idempotency_key": idempotency_key, + } + + def _run( + self, + query: Optional[str] = None, + platform: Optional[str] = None, + count: int = 25, + min_score: Optional[int] = None, + exclude: Optional[List[str]] = None, + min_followers: Optional[int] = None, + max_followers: Optional[int] = None, + min_subscribers: Optional[int] = None, + max_subscribers: Optional[int] = None, + verified_only: Optional[bool] = None, + require_contact: Optional[bool] = None, + countries: Optional[List[str]] = None, + keywords: Optional[List[str]] = None, + exclude_keywords: Optional[List[str]] = None, + idempotency_key: Optional[str] = None, + timeout: int = 300, + task_id: Optional[str] = None, + run_manager: Optional[CallbackManagerForToolRun] = None, + ) -> Any: + """Use the tool synchronously.""" + del run_manager + kwargs = self._search_kwargs( + query, + platform, + count, + min_score, + exclude, + _compact( + min_followers=min_followers, + max_followers=max_followers, + min_subscribers=min_subscribers, + max_subscribers=max_subscribers, + verified_only=verified_only, + require_contact=require_contact, + countries=countries, + keywords=keywords, + exclude_keywords=exclude_keywords, + ), + idempotency_key, + timeout, + task_id, + ) + return _call_people(_require_api_key(), self.base_url, search_people, **kwargs) + + async def _arun( + self, + query: Optional[str] = None, + platform: Optional[str] = None, + count: int = 25, + min_score: Optional[int] = None, + exclude: Optional[List[str]] = None, + min_followers: Optional[int] = None, + max_followers: Optional[int] = None, + min_subscribers: Optional[int] = None, + max_subscribers: Optional[int] = None, + verified_only: Optional[bool] = None, + require_contact: Optional[bool] = None, + countries: Optional[List[str]] = None, + keywords: Optional[List[str]] = None, + exclude_keywords: Optional[List[str]] = None, + idempotency_key: Optional[str] = None, + timeout: int = 300, + task_id: Optional[str] = None, + run_manager: Optional[AsyncCallbackManagerForToolRun] = None, + ) -> Any: + """Use the tool asynchronously.""" + del run_manager + kwargs = self._search_kwargs( + query, + platform, + count, + min_score, + exclude, + _compact( + min_followers=min_followers, + max_followers=max_followers, + min_subscribers=min_subscribers, + max_subscribers=max_subscribers, + verified_only=verified_only, + require_contact=require_contact, + countries=countries, + keywords=keywords, + exclude_keywords=exclude_keywords, + ), + idempotency_key, + timeout, + task_id, + ) + return await _acall_people( + _require_api_key(), self.base_url, search_people, **kwargs + ) + + +class PeopleProfilesToolInput(BaseModel): + platform: _PeoplePlatform = Field( + description="Whose people to list: 'x', 'youtube', 'linkedin' or 'tiktok'." + ) + q: Optional[str] = Field( + default=None, + max_length=200, + description="Text in the name, handle, bio or location.", + ) + sort: Literal["score", "recent", "audience"] = Field( + default="score", + description="'score' (best fit first), 'recent' or 'audience' (largest first).", + ) + limit: int = Field( + default=25, ge=1, le=200, description="People per page (1-200)." + ) + cursor: Optional[str] = Field( + default=None, description="next_cursor from the previous page." + ) + has: Optional[List[_PeopleChannel]] = Field( + default=None, + description=( + "Channels every person must have, for example ['email'] or " + "['email', 'instagram']." + ), + ) + min_audience: Optional[int] = Field( + default=None, + ge=0, + description="Minimum followers (subscribers on YouTube).", + ) + max_audience: Optional[int] = Field( + default=None, + ge=0, + description="Maximum followers (subscribers on YouTube).", + ) + min_score: Optional[int] = Field( + default=None, ge=0, le=100, description="Minimum fit score (0-100)." + ) + location: Optional[str] = Field( + default=None, + max_length=100, + description="Text in the location, for example 'united states'.", + ) + verified: Optional[bool] = Field( + default=None, description="True keeps only verified accounts (x, tiktok)." + ) + + +class PeopleProfilesTool(BaseTool): + name: str = "people_profiles_tool" + description: str = ( + "Lists the people that earlier Desearch People Searches already found on " + "one platform (x, youtube, linkedin or tiktok), one entry per person with " + "their best fit score and the search that found them. Filter by text, by " + "channels they have (has=email, website, x, linkedin, instagram, youtube, " + "tiktok), audience range, minimum score, location and verified (x, tiktok); " + "sort by score, recent or audience. Free and instant: it reads saved " + "results and never starts a search." + ) + args_schema: ArgsSchema = PeopleProfilesToolInput + return_direct: bool = True + base_url: str = PEOPLE_API_BASE_URL + + def _run( + self, + platform: str, + q: Optional[str] = None, + sort: str = "score", + limit: int = 25, + cursor: Optional[str] = None, + has: Optional[List[str]] = None, + min_audience: Optional[int] = None, + max_audience: Optional[int] = None, + min_score: Optional[int] = None, + location: Optional[str] = None, + verified: Optional[bool] = None, + run_manager: Optional[CallbackManagerForToolRun] = None, + ) -> Any: + """Use the tool synchronously.""" + del run_manager + return _call_people( + _require_api_key(), + self.base_url, + list_people_profiles, + tool_name=self.name, + platform=platform, + q=q, + sort=sort, + limit=limit, + cursor=cursor, + has=has, + min_audience=min_audience, + max_audience=max_audience, + min_score=min_score, + location=location, + verified=verified, + ) + + async def _arun( + self, + platform: str, + q: Optional[str] = None, + sort: str = "score", + limit: int = 25, + cursor: Optional[str] = None, + has: Optional[List[str]] = None, + min_audience: Optional[int] = None, + max_audience: Optional[int] = None, + min_score: Optional[int] = None, + location: Optional[str] = None, + verified: Optional[bool] = None, + run_manager: Optional[AsyncCallbackManagerForToolRun] = None, + ) -> Any: + """Use the tool asynchronously.""" + del run_manager + return await _acall_people( + _require_api_key(), + self.base_url, + list_people_profiles, + tool_name=self.name, + platform=platform, + q=q, + sort=sort, + limit=limit, + cursor=cursor, + has=has, + min_audience=min_audience, + max_audience=max_audience, + min_score=min_score, + location=location, + verified=verified, + ) diff --git a/pyproject.toml b/pyproject.toml index 7fdbb29..3807faa 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -10,6 +10,7 @@ classifiers = [ dependencies = [ "langchain-core>=0.1.0", "desearch-py>=1.1.0,<2.0.0", + "aiohttp>=3.8,<4.0.0", "langchain-deepseek>=0.1.0", "python-dotenv>=0.21.0", "pytest>=7.0.0", @@ -19,7 +20,7 @@ license = {text = "MIT"} name = "langchain-desearch" readme = "README.md" requires-python = ">=3.9" -version = "1.0.7" +version = "1.1.0" [project.urls] Homepage = "https://github.com/Desearch-ai/langchain-desearch" diff --git a/setup.py b/setup.py index a62fe82..c1d92bc 100644 --- a/setup.py +++ b/setup.py @@ -2,7 +2,7 @@ setup( name="langchain-desearch", # Replace with your desired package name - version="1.0.7", + version="1.1.0", author="Desearch", # Replace with your name author_email="your-email@example.com", # Replace with your email description="LangChain integration with Desearch API for search and data-fetching tools.", @@ -17,6 +17,7 @@ "pydantic==2.11.3", "python-dotenv==1.1.0", "desearch-py>=1.1.0,<2.0.0", + "aiohttp>=3.8,<4.0.0", "pytest==8.3.5", ], classifiers=[ diff --git a/tests/unit_tests/test_imports.py b/tests/unit_tests/test_imports.py index 12f916b..8269153 100644 --- a/tests/unit_tests/test_imports.py +++ b/tests/unit_tests/test_imports.py @@ -4,6 +4,8 @@ "DesearchTool", "BasicWebSearchTool", "BasicTwitterSearchTool", + "PeopleSearchTool", + "PeopleProfilesTool", ] diff --git a/tests/unit_tests/test_people_tools.py b/tests/unit_tests/test_people_tools.py new file mode 100644 index 0000000..de9309b --- /dev/null +++ b/tests/unit_tests/test_people_tools.py @@ -0,0 +1,520 @@ +"""People Search tools, tested against a local HTTP server (never the live API).""" + +import asyncio +import threading +from pathlib import Path + +import pytest +from aiohttp import web +from pydantic import ValidationError + +from langchain_desearch.people import compact_person +from langchain_desearch.search_tools import people_tools, search_tools +from langchain_desearch.tools import ( + PeopleProfilesTool, + PeopleSearchTool, + PeopleSearchToolInput, +) + +ROOT = Path(__file__).resolve().parents[2] +TASK_ID = "5b33f3ee-168b-4780-9c8e-63a626781872" +TASK_PATH = f"/desearch/people/tasks/{TASK_ID}" + + +def _task(status, result_count=0, **extra): + return { + "task_id": TASK_ID, + "search_type": "x_profile_search", + "platform": "x", + "status": status, + "stage": None if status == "completed" else "searching", + "query": "Indie hackers building in public", + "count": 25, + "result_count": result_count, + "progress": {"candidates": 40}, + "errors": [], + "reserved_cost_usd": 1.0, + "final_cost_usd": 0.2 if status == "completed" else None, + "created_at": "2026-10-19T10:00:00Z", + **extra, + } + + +PERSON_X = { + "id": "x:1", + "platform": "x", + "handle": "levelsio", + "url": "https://x.com/levelsio", + "name": "Pieter Levels", + "avatar_url": "https://img.example/levelsio.jpg", + "bio": "Building in public", + "location": "Portugal", + "metrics": {"followers": 600000, "following": 2000}, + "attributes": {"verified": True}, + "links": [ + {"type": "website", "url": "https://levels.io"}, + {"type": "website", "url": ""}, + ], + "contacts": [{"type": "email", "value": "pieter@example.com"}, {"type": "email"}], + "evidence": [ + {"type": "post", "text": "MRR update"}, + {"type": "post", "url": "https://x.com/levelsio/status/1", "text": "MRR $100k"}, + ], + "score": { + "value": 92, + "qualified": True, + "summary": "Shares MRR updates every month.", + "reasons": ["Posts revenue numbers", "Indie founder"], + }, +} + +PERSON_X_2 = { + "id": "x:2", + "platform": "x", + "handle": "marc_louvion", + "metrics": {"followers": 120000}, + "score": {"value": 81, "reasons": ["Shares revenue screenshots"]}, +} + + +class _FakePeopleAPI: + """Records each request and answers from scripted responses. + + A route keeps returning its last response, so ``add`` a sequence for + polling (running, running, completed) and the last one sticks. + """ + + def __init__(self): + self.requests = [] + self.routes = {} + self.base_url = "" + + def add(self, method, path, status, payload): + self.routes.setdefault((method, path), []).append((status, payload)) + + def calls(self, method=None, path=None): + return [ + r + for r in self.requests + if (method is None or r["method"] == method) + and (path is None or r["path"] == path) + ] + + async def handle(self, request): + body = await request.json() if request.can_read_body else None + self.requests.append( + { + "method": request.method, + "path": request.path, + "query": dict(request.query), + "headers": dict(request.headers), + "json": body, + } + ) + queue = self.routes.get((request.method, request.path)) + if not queue: + return web.json_response({"detail": "Not Found"}, status=404) + status, payload = queue.pop(0) if len(queue) > 1 else queue[0] + return web.json_response(payload, status=status) + + +@pytest.fixture +def people_api(monkeypatch): + api = _FakePeopleAPI() + app = web.Application() + app.router.add_route("*", "/{tail:.*}", api.handle) + loop = asyncio.new_event_loop() + runner = web.AppRunner(app) + loop.run_until_complete(runner.setup()) + site = web.TCPSite(runner, "127.0.0.1", 0) + loop.run_until_complete(site.start()) + api.base_url = f"http://127.0.0.1:{runner.addresses[0][1]}" + thread = threading.Thread(target=loop.run_forever, daemon=True) + thread.start() + monkeypatch.setenv("DESEARCH_API_KEY", "test-key") + try: + yield api + finally: + asyncio.run_coroutine_threadsafe(runner.cleanup(), loop).result(5) + loop.call_soon_threadsafe(loop.stop) + thread.join(5) + loop.close() + + +def _search_tool(api): + return PeopleSearchTool(base_url=api.base_url, poll_interval=0.01) + + +def _profiles_tool(api): + return PeopleProfilesTool(base_url=api.base_url) + + +def _results(people, has_more=False, cursor="c1", status="completed"): + return { + "task_id": TASK_ID, + "status": status, + "results": people, + "next_cursor": cursor, + "has_more": has_more, + } + + +def test_people_tool_names_groups_and_descriptions(): + assert PeopleSearchTool().name == "people_search_tool" + assert PeopleProfilesTool().name == "people_profiles_tool" + assert [tool.name for tool in people_tools] == [ + "people_search_tool", + "people_profiles_tool", + ] + # The billed, slow tool stays out of the default agent tool list. + assert "people_search_tool" not in [tool.name for tool in search_tools] + + search_description = PeopleSearchTool().description + assert "BILLED" in search_description + assert "1-5 minutes" in search_description + assert "task_id" in search_description + assert "Free" in PeopleProfilesTool().description + + +def test_people_search_does_not_require_unpublished_sdk(): + for name in ("pyproject.toml", "setup.py"): + text = (ROOT / name).read_text() + assert "desearch-py>=1.4" not in text + assert "desearch-py>=1.1.0,<2.0.0" in text + + +def test_people_search_starts_waits_and_returns_compact_people(people_api): + people_api.add("POST", "/desearch/people/search/x", 202, _task("queued", cost_usd=1.0)) + people_api.add("GET", TASK_PATH, 200, _task("running", 1)) + people_api.add("GET", TASK_PATH, 200, _task("completed", 2)) + people_api.add( + "GET", f"{TASK_PATH}/results", 200, _results([PERSON_X, PERSON_X_2]) + ) + + result = _search_tool(people_api).invoke( + { + "query": "Indie hackers building in public", + "platform": "x", + "count": 25, + "min_score": 60, + "exclude": ["someone"], + "min_followers": 1000, + "verified_only": True, + "keywords": ["MRR"], + "idempotency_key": "run-1", + "timeout": 30, + } + ) + + assert result["task_id"] == TASK_ID + assert result["status"] == "completed" + assert result["done"] is True + assert result["final_cost_usd"] == 0.2 + assert "existing_task" not in result + assert result["next_step"].startswith("Finished.") + assert result["people"] == [ + { + "name": "Pieter Levels", + "handle": "levelsio", + "platform": "x", + "url": "https://x.com/levelsio", + "audience": 600000, + "location": "Portugal", + "verified": True, + "score": 92, + "why": "Shares MRR updates every month.", + "contacts": [{"type": "email", "value": "pieter@example.com"}], + "links": [{"type": "website", "url": "https://levels.io"}], + "evidence_url": "https://x.com/levelsio/status/1", + }, + { + "handle": "marc_louvion", + "platform": "x", + "audience": 120000, + "score": 81, + "why": "Shares revenue screenshots", + }, + ] + + (create,) = people_api.calls("POST") + assert create["path"] == "/desearch/people/search/x" + assert create["headers"]["Authorization"] == "test-key" + assert create["headers"]["Idempotency-Key"] == "run-1" + assert create["json"] == { + "query": "Indie hackers building in public", + "count": 25, + "min_score": 60, + "exclude": ["someone"], + "filters": {"min_followers": 1000, "verified_only": True, "keywords": ["MRR"]}, + } + assert len(people_api.calls("GET", TASK_PATH)) == 2 + (results_call,) = people_api.calls("GET", f"{TASK_PATH}/results") + assert results_call["query"] == {"limit": "200"} + assert all(r["headers"]["Authorization"] == "test-key" for r in people_api.requests) + + +def test_people_search_returns_people_so_far_when_it_times_out(people_api): + people_api.add("POST", "/desearch/people/search/linkedin", 202, _task("queued")) + people_api.add("GET", f"{TASK_PATH}/results", 200, _results([PERSON_X], status="running")) + + result = _search_tool(people_api).invoke( + {"query": "Heads of sales at fintech startups", "platform": "linkedin", "timeout": 0} + ) + + assert result["status"] == "queued" + assert result["done"] is False + assert [person["handle"] for person in result["people"]] == ["levelsio"] + assert f"task_id='{TASK_ID}'" in result["next_step"] + assert "free" in result["next_step"] + # timeout=0 returns without polling the task. + assert people_api.calls("GET", TASK_PATH) == [] + + +def test_people_search_with_task_id_waits_without_starting_a_search(people_api): + people_api.add("GET", TASK_PATH, 200, _task("running", 1)) + people_api.add("GET", TASK_PATH, 200, _task("partial", 1, errors=["x timeout"])) + people_api.add("GET", f"{TASK_PATH}/results", 200, _results([PERSON_X_2])) + + result = _search_tool(people_api).invoke({"task_id": TASK_ID, "timeout": 30}) + + assert people_api.calls("POST") == [] + assert result["status"] == "partial" + assert result["errors"] == ["x timeout"] + assert result["next_step"].startswith("Finished with partial results") + assert [person["handle"] for person in result["people"]] == ["marc_louvion"] + + +def test_people_search_reads_every_results_page(people_api): + people_api.add("POST", "/desearch/people/search/x", 202, _task("completed", 2)) + people_api.add( + "GET", f"{TASK_PATH}/results", 200, _results([PERSON_X], has_more=True, cursor="p2") + ) + people_api.add("GET", f"{TASK_PATH}/results", 200, _results([PERSON_X_2], cursor="p3")) + + result = _search_tool(people_api).invoke( + {"query": "Indie hackers building in public", "platform": "x"} + ) + + assert [person["handle"] for person in result["people"]] == [ + "levelsio", + "marc_louvion", + ] + pages = people_api.calls("GET", f"{TASK_PATH}/results") + assert [page["query"] for page in pages] == [ + {"limit": "200"}, + {"limit": "200", "cursor": "p2"}, + ] + + +def test_people_search_marks_an_idempotent_replay(people_api): + people_api.add("POST", "/desearch/people/search/x", 200, _task("completed", 0)) + people_api.add("GET", f"{TASK_PATH}/results", 200, _results([])) + + result = _search_tool(people_api).invoke( + {"query": "Indie hackers", "platform": "x", "idempotency_key": "same-key"} + ) + + assert result["existing_task"] is True + assert result["people"] == [] + + +def test_people_search_rejects_filters_the_platform_does_not_accept(people_api): + with pytest.raises(ValidationError, match="verified_only does not apply to linkedin"): + _search_tool(people_api).invoke( + {"query": "Heads of sales", "platform": "linkedin", "verified_only": True} + ) + with pytest.raises(ValueError, match="min_followers does not apply to youtube"): + _search_tool(people_api)._run( + query="Cooking creators", platform="youtube", min_followers=10 + ) + assert people_api.requests == [] + + +@pytest.mark.parametrize( + "payload", + [ + {"query": "Indie hackers"}, + {"platform": "x"}, + {"query": "ab", "platform": "x"}, + {"query": "Indie hackers", "platform": "x", "count": 4}, + {"query": "Indie hackers", "platform": "x", "count": 201}, + {"query": "Indie hackers", "platform": "facebook"}, + ], +) +def test_people_search_input_is_validated_before_any_call(payload): + with pytest.raises(ValidationError): + PeopleSearchToolInput(**payload) + + +def test_people_search_http_errors_are_returned_without_the_api_key(people_api): + people_api.add( + "POST", + "/desearch/people/search/tiktok", + 402, + {"detail": "Insufficient balance for this search."}, + ) + + result = _search_tool(people_api).invoke( + {"query": "Cooking creators who share recipes", "platform": "tiktok"} + ) + + assert result == ( + "An error occurred while calling Desearch: " + "HTTP 402: Insufficient balance for this search." + ) + assert "test-key" not in result + + +def test_people_search_missing_api_key_raises(monkeypatch): + monkeypatch.delenv("DESEARCH_API_KEY", raising=False) + with pytest.raises(ValueError, match="DESEARCH_API_KEY"): + PeopleSearchTool().invoke({"query": "Indie hackers", "platform": "x"}) + + +def test_people_search_sync_invoke_works_inside_a_running_loop(people_api): + people_api.add("GET", TASK_PATH, 200, _task("completed", 1)) + people_api.add("GET", f"{TASK_PATH}/results", 200, _results([PERSON_X])) + + async def _call(): + return _search_tool(people_api).invoke({"task_id": TASK_ID}) + + result = asyncio.run(_call()) + assert result["status"] == "completed" + assert result["people"][0]["handle"] == "levelsio" + + +def test_people_search_async_invoke(people_api): + people_api.add("POST", "/desearch/people/search/youtube", 202, _task("queued")) + people_api.add("GET", TASK_PATH, 200, _task("completed", 1)) + people_api.add("GET", f"{TASK_PATH}/results", 200, _results([PERSON_X])) + + result = asyncio.run( + _search_tool(people_api).ainvoke( + { + "query": "Cooking creators who share recipes", + "platform": "youtube", + "min_subscribers": 10000, + "require_contact": True, + } + ) + ) + + assert result["status"] == "completed" + (create,) = people_api.calls("POST") + assert create["json"]["filters"] == {"min_subscribers": 10000, "require_contact": True} + + +def test_people_profiles_sends_filters_and_returns_compact_people(people_api): + profile = { + **PERSON_X, + "platform": "tiktok", + "found_by": { + "task_id": TASK_ID, + "query": "Cooking creators", + "found_at": "2026-10-19T10:05:00Z", + "times_found": 2, + }, + } + people_api.add( + "GET", + "/desearch/people/profiles", + 200, + {"profiles": [profile], "total": 31, "next_cursor": "next-page"}, + ) + + result = _profiles_tool(people_api).invoke( + { + "platform": "tiktok", + "q": "recipes", + "sort": "audience", + "limit": 10, + "has": ["email", "instagram", "email"], + "min_audience": 1000, + "max_audience": 500000, + "min_score": 70, + "location": "united states", + "verified": True, + } + ) + + (call,) = people_api.calls("GET", "/desearch/people/profiles") + assert call["headers"]["Authorization"] == "test-key" + assert call["query"] == { + "platform": "tiktok", + "q": "recipes", + "sort": "audience", + "limit": "10", + "has": "email,instagram", + "min_audience": "1000", + "max_audience": "500000", + "min_score": "70", + "location": "united states", + "verified": "true", + } + assert result["platform"] == "tiktok" + assert result["total"] == 31 + assert result["next_cursor"] == "next-page" + assert "cursor='next-page'" in result["next_step"] + (person,) = result["people"] + assert person["handle"] == "levelsio" + assert person["found_by"] == { + "task_id": TASK_ID, + "query": "Cooking creators", + "times_found": 2, + "found_at": "2026-10-19T10:05:00Z", + } + + +def test_people_profiles_defaults_and_async(people_api): + people_api.add( + "GET", + "/desearch/people/profiles", + 200, + {"profiles": [], "total": 0, "next_cursor": None}, + ) + + result = asyncio.run(_profiles_tool(people_api).ainvoke({"platform": "x"})) + + assert result == {"platform": "x", "total": 0, "people": []} + (call,) = people_api.calls("GET", "/desearch/people/profiles") + assert call["query"] == {"platform": "x", "sort": "score", "limit": "25"} + + +def test_people_profiles_errors_are_returned_as_messages(people_api): + people_api.add( + "GET", + "/desearch/people/profiles", + 422, + {"detail": [{"loc": ["query", "platform"], "msg": "bad platform"}]}, + ) + + result = _profiles_tool(people_api).invoke({"platform": "x"}) + + assert result.startswith("An error occurred while calling Desearch: HTTP 422: ") + assert "bad platform" in result + + +def test_compact_person_uses_subscribers_and_linkedin_fields(): + youtube = compact_person( + { + "platform": "youtube", + "handle": "@chef", + "metrics": {"subscribers": 54000}, + "score": {"value": 77.5, "summary": "Posts weekly recipes."}, + } + ) + assert youtube["audience"] == 54000 + assert youtube["score"] == 77.5 + + linkedin = compact_person( + { + "platform": "linkedin", + "handle": "jane-doe", + "metrics": {"followers": 0}, + "attributes": {"headline": "VP Sales", "company": "Acme", "verified": "yes"}, + } + ) + assert linkedin["audience"] == 0 + assert linkedin["headline"] == "VP Sales" + assert linkedin["company"] == "Acme" + assert "verified" not in linkedin diff --git a/tests/unit_tests/test_tools.py b/tests/unit_tests/test_tools.py index b05917c..800da53 100644 --- a/tests/unit_tests/test_tools.py +++ b/tests/unit_tests/test_tools.py @@ -115,8 +115,8 @@ def test_public_names_stay_compatible(): def test_package_metadata_pins_compatible_desearch_py(): pyproject = (ROOT / "pyproject.toml").read_text() setup_py = (ROOT / "setup.py").read_text() - assert 'version = "1.0.7"' in pyproject - assert 'version="1.0.7"' in setup_py + assert 'version = "1.1.0"' in pyproject + assert 'version="1.1.0"' in setup_py assert DESEARCH_REQUIREMENT in pyproject assert DESEARCH_REQUIREMENT in setup_py assert "desearch-py>=" in pyproject