feat(tools): add SerpAPI tools for Google Scholar & Patents search (#3986)
Implements 5 tools as proposed in #3224: - scholar_search: Search Google Scholar for academic papers - scholar_get_citations: Get citation formats (MLA, APA, Chicago, etc.) - scholar_get_author: Author profiles with h-index, i10-index, metrics - patents_search: Search Google Patents with filters - patents_get_details: Detailed patent information by publication number Follows existing tool pattern (web_search_tool, hubspot_tool, slack_tool): - register_tools(mcp, credentials) with @mcp.tool() decorators - _SerpAPIClient internal class for HTTP calls via httpx - Credential fallback: CredentialStoreAdapter -> SERPAPI_API_KEY env var - Error handling: always returns dicts, never raises 25 unit tests + live integration tests verified. 697/697 full test suite passing. Fixes #3224
This commit is contained in:
Amit Kumar
@@ -59,6 +59,7 @@ from .health_check import HealthCheckResult, check_credential_health
|
||||
from .hubspot import HUBSPOT_CREDENTIALS
|
||||
from .llm import LLM_CREDENTIALS
|
||||
from .search import SEARCH_CREDENTIALS
|
||||
from .serpapi import SERPAPI_CREDENTIALS
|
||||
from .shell_config import (
|
||||
add_env_var_to_shell_config,
|
||||
detect_shell,
|
||||
@@ -77,6 +78,7 @@ CREDENTIAL_SPECS = {
|
||||
**GITHUB_CREDENTIALS,
|
||||
**HUBSPOT_CREDENTIALS,
|
||||
**SLACK_CREDENTIALS,
|
||||
**SERPAPI_CREDENTIALS,
|
||||
}
|
||||
|
||||
__all__ = [
|
||||
@@ -108,4 +110,5 @@ __all__ = [
|
||||
"HUBSPOT_CREDENTIALS",
|
||||
"SLACK_CREDENTIALS",
|
||||
"APOLLO_CREDENTIALS",
|
||||
"SERPAPI_CREDENTIALS",
|
||||
]
|
||||
|
||||
@@ -0,0 +1,33 @@
|
||||
"""
|
||||
SerpAPI tool credentials.
|
||||
|
||||
Contains credentials for SerpAPI (Google Scholar & Patents search).
|
||||
"""
|
||||
|
||||
from .base import CredentialSpec
|
||||
|
||||
SERPAPI_CREDENTIALS = {
|
||||
"serpapi": CredentialSpec(
|
||||
env_var="SERPAPI_API_KEY",
|
||||
tools=[
|
||||
"scholar_search",
|
||||
"scholar_get_citations",
|
||||
"scholar_get_author",
|
||||
"patents_search",
|
||||
"patents_get_details",
|
||||
],
|
||||
required=True,
|
||||
startup_required=False,
|
||||
help_url="https://serpapi.com/manage-api-key",
|
||||
description="API key for SerpAPI (Google Scholar & Patents)",
|
||||
direct_api_key_supported=True,
|
||||
api_key_instructions="""To get a SerpAPI API key:
|
||||
1. Go to https://serpapi.com/users/sign_up
|
||||
2. Create an account (free tier: 100 searches/month)
|
||||
3. Go to https://serpapi.com/manage-api-key
|
||||
4. Copy your API key""",
|
||||
health_check_endpoint="https://serpapi.com/account.json",
|
||||
credential_id="serpapi",
|
||||
credential_key="api_key",
|
||||
),
|
||||
}
|
||||
@@ -44,6 +44,7 @@ from .github_tool import register_tools as register_github
|
||||
from .hubspot_tool import register_tools as register_hubspot
|
||||
from .pdf_read_tool import register_tools as register_pdf_read
|
||||
from .runtime_logs_tool import register_tools as register_runtime_logs
|
||||
from .serpapi_tool import register_tools as register_serpapi
|
||||
from .slack_tool import register_tools as register_slack
|
||||
from .web_scrape_tool import register_tools as register_web_scrape
|
||||
from .web_search_tool import register_tools as register_web_search
|
||||
@@ -78,6 +79,7 @@ def register_all_tools(
|
||||
register_email(mcp, credentials=credentials)
|
||||
register_hubspot(mcp, credentials=credentials)
|
||||
register_apollo(mcp, credentials=credentials)
|
||||
register_serpapi(mcp, credentials=credentials)
|
||||
register_slack(mcp, credentials=credentials)
|
||||
|
||||
# Register file system toolkits
|
||||
@@ -151,6 +153,12 @@ def register_all_tools(
|
||||
"query_runtime_logs",
|
||||
"query_runtime_log_details",
|
||||
"query_runtime_log_raw",
|
||||
# SerpAPI tools (Google Scholar & Patents)
|
||||
"scholar_search",
|
||||
"scholar_get_citations",
|
||||
"scholar_get_author",
|
||||
"patents_search",
|
||||
"patents_get_details",
|
||||
"slack_send_message",
|
||||
"slack_list_channels",
|
||||
"slack_get_channel_history",
|
||||
|
||||
@@ -0,0 +1,77 @@
|
||||
# SerpAPI Tool
|
||||
|
||||
Google Scholar & Google Patents search via SerpAPI.
|
||||
|
||||
## Description
|
||||
|
||||
Provides 5 tools for academic paper search, citation lookup, author profiles, and patent search. Google Scholar has no official API — SerpAPI is the only way to get structured paper metadata including citation counts and h-index data.
|
||||
|
||||
## Tools
|
||||
|
||||
### `scholar_search`
|
||||
|
||||
Search Google Scholar for academic papers.
|
||||
|
||||
| Argument | Type | Required | Default | Description |
|
||||
|----------|------|----------|---------|-------------|
|
||||
| `query` | str | Yes | - | Search query (1-500 chars) |
|
||||
| `num_results` | int | No | `10` | Results to return (1-20) |
|
||||
| `start` | int | No | `0` | Pagination offset |
|
||||
| `year_low` | int | No | - | Published after this year |
|
||||
| `year_high` | int | No | - | Published before this year |
|
||||
| `sort_by_date` | bool | No | `False` | Sort by date vs relevance |
|
||||
|
||||
### `scholar_get_citations`
|
||||
|
||||
Get citation formats (MLA, APA, Chicago, Harvard, Vancouver) for a paper.
|
||||
|
||||
| Argument | Type | Required | Description |
|
||||
|----------|------|----------|-------------|
|
||||
| `result_id` | str | Yes | The `result_id` from a `scholar_search` result |
|
||||
|
||||
### `scholar_get_author`
|
||||
|
||||
Get author profile with h-index, i10-index, total citations, and articles.
|
||||
|
||||
| Argument | Type | Required | Default | Description |
|
||||
|----------|------|----------|---------|-------------|
|
||||
| `author_id` | str | Yes | - | Google Scholar author ID |
|
||||
| `num_articles` | int | No | `20` | Articles to return (1-100) |
|
||||
| `start` | int | No | `0` | Pagination offset |
|
||||
| `sort_by` | str | No | `citedby` | Sort: `citedby` or `pubdate` |
|
||||
|
||||
### `patents_search`
|
||||
|
||||
Search Google Patents.
|
||||
|
||||
| Argument | Type | Required | Default | Description |
|
||||
|----------|------|----------|---------|-------------|
|
||||
| `query` | str | Yes | - | Search query (1-500 chars) |
|
||||
| `page` | int | No | `1` | Page number (1-indexed) |
|
||||
| `country` | str | No | - | Country code (US, EP, WO, CN) |
|
||||
| `status` | str | No | - | `GRANT` or `APPLICATION` |
|
||||
| `before_date` | str | No | - | Filed before (YYYYMMDD) |
|
||||
| `after_date` | str | No | - | Filed after (YYYYMMDD) |
|
||||
|
||||
### `patents_get_details`
|
||||
|
||||
Get full details for a specific patent.
|
||||
|
||||
| Argument | Type | Required | Description |
|
||||
|----------|------|----------|-------------|
|
||||
| `patent_id` | str | Yes | Patent publication number (e.g. `US20210012345A1`) |
|
||||
|
||||
## Environment Variables
|
||||
|
||||
| Variable | Required | Description |
|
||||
|----------|----------|-------------|
|
||||
| `SERPAPI_API_KEY` | Yes | API key from [SerpAPI Dashboard](https://serpapi.com/manage-api-key) |
|
||||
|
||||
## Error Handling
|
||||
|
||||
Returns error dicts for common issues:
|
||||
- `SerpAPI credentials not configured` - No API key set
|
||||
- `Query must be 1-500 characters` - Invalid query length
|
||||
- `Invalid SerpAPI API key` - Key rejected by API
|
||||
- `SerpAPI rate limit exceeded` - Too many requests
|
||||
- `Search request timed out` - Request exceeded 30s timeout
|
||||
@@ -0,0 +1,5 @@
|
||||
"""SerpAPI Tool - Google Scholar & Patents search via SerpAPI."""
|
||||
|
||||
from .serpapi_tool import register_tools
|
||||
|
||||
__all__ = ["register_tools"]
|
||||
@@ -0,0 +1,509 @@
|
||||
"""
|
||||
SerpAPI Tool - Google Scholar & Google Patents search via SerpAPI.
|
||||
|
||||
Supports:
|
||||
- Direct API key (SERPAPI_API_KEY)
|
||||
- Credential store via CredentialStoreAdapter
|
||||
|
||||
API Reference: https://serpapi.com/search-api
|
||||
|
||||
Tools:
|
||||
- scholar_search: Search Google Scholar for academic papers
|
||||
- scholar_get_citations: Get citation formats for a specific paper
|
||||
- scholar_get_author: Get author profile, h-index, articles
|
||||
- patents_search: Search Google Patents
|
||||
- patents_get_details: Get detailed patent information
|
||||
"""
|
||||
|
||||
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
|
||||
|
||||
SERPAPI_BASE = "https://serpapi.com/search.json"
|
||||
SERPAPI_ACCOUNT = "https://serpapi.com/account.json"
|
||||
|
||||
|
||||
class _SerpAPIClient:
|
||||
"""Internal client wrapping SerpAPI HTTP calls."""
|
||||
|
||||
def __init__(self, api_key: str):
|
||||
self._api_key = api_key
|
||||
|
||||
def _request(self, params: dict[str, Any]) -> dict[str, Any]:
|
||||
"""Make a GET request to SerpAPI."""
|
||||
params["api_key"] = self._api_key
|
||||
response = httpx.get(SERPAPI_BASE, params=params, timeout=30.0)
|
||||
|
||||
if response.status_code == 401:
|
||||
return {
|
||||
"error": "Invalid SerpAPI API key",
|
||||
"help": "Check your key at https://serpapi.com/manage-api-key",
|
||||
}
|
||||
if response.status_code == 429:
|
||||
return {"error": "SerpAPI rate limit exceeded. Try again later."}
|
||||
if response.status_code >= 400:
|
||||
try:
|
||||
detail = response.json().get("error", response.text)
|
||||
except Exception:
|
||||
detail = response.text
|
||||
return {"error": f"SerpAPI error (HTTP {response.status_code}): {detail}"}
|
||||
|
||||
data = response.json()
|
||||
if "error" in data:
|
||||
return {"error": f"SerpAPI error: {data['error']}"}
|
||||
return data
|
||||
|
||||
def scholar_search(
|
||||
self,
|
||||
query: str,
|
||||
num: int = 10,
|
||||
start: int = 0,
|
||||
year_low: int | None = None,
|
||||
year_high: int | None = None,
|
||||
sort_by_date: bool = False,
|
||||
) -> dict[str, Any]:
|
||||
"""Search Google Scholar."""
|
||||
params: dict[str, Any] = {
|
||||
"engine": "google_scholar",
|
||||
"q": query,
|
||||
"num": min(num, 20),
|
||||
"start": start,
|
||||
}
|
||||
if year_low is not None:
|
||||
params["as_ylo"] = year_low
|
||||
if year_high is not None:
|
||||
params["as_yhi"] = year_high
|
||||
if sort_by_date:
|
||||
params["scisbd"] = 1
|
||||
return self._request(params)
|
||||
|
||||
def scholar_cite(self, result_id: str) -> dict[str, Any]:
|
||||
"""Get citation formats for a scholar result."""
|
||||
return self._request({"engine": "google_scholar_cite", "q": result_id})
|
||||
|
||||
def scholar_author(
|
||||
self,
|
||||
author_id: str,
|
||||
start: int = 0,
|
||||
num: int = 20,
|
||||
sort_by: str = "citedby",
|
||||
) -> dict[str, Any]:
|
||||
"""Get author profile and articles."""
|
||||
return self._request(
|
||||
{
|
||||
"engine": "google_scholar_author",
|
||||
"author_id": author_id,
|
||||
"start": start,
|
||||
"num": min(num, 100),
|
||||
"sort": sort_by,
|
||||
}
|
||||
)
|
||||
|
||||
def patents_search(
|
||||
self,
|
||||
query: str,
|
||||
page: int = 1,
|
||||
country: str | None = None,
|
||||
status: str | None = None,
|
||||
before: str | None = None,
|
||||
after: str | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Search Google Patents."""
|
||||
params: dict[str, Any] = {
|
||||
"engine": "google_patents",
|
||||
"q": query,
|
||||
"page": page,
|
||||
}
|
||||
if country:
|
||||
params["country"] = country
|
||||
if status:
|
||||
params["status"] = status
|
||||
if before:
|
||||
params["before"] = f"priority:{before}"
|
||||
if after:
|
||||
params["after"] = f"priority:{after}"
|
||||
return self._request(params)
|
||||
|
||||
def patents_details(self, patent_id: str) -> dict[str, Any]:
|
||||
"""Get details for a specific patent by searching its ID."""
|
||||
return self._request({"engine": "google_patents", "q": patent_id})
|
||||
|
||||
|
||||
def register_tools(
|
||||
mcp: FastMCP,
|
||||
credentials: CredentialStoreAdapter | None = None,
|
||||
) -> None:
|
||||
"""Register SerpAPI tools with the MCP server."""
|
||||
|
||||
def _get_api_key() -> str | None:
|
||||
"""Get SerpAPI API key from credential store or environment."""
|
||||
if credentials is not None:
|
||||
return credentials.get("serpapi")
|
||||
return os.getenv("SERPAPI_API_KEY")
|
||||
|
||||
def _get_client() -> _SerpAPIClient | dict[str, str]:
|
||||
"""Get a SerpAPI client, or return an error dict if no credentials."""
|
||||
api_key = _get_api_key()
|
||||
if not api_key:
|
||||
return {
|
||||
"error": "SerpAPI credentials not configured",
|
||||
"help": (
|
||||
"Set SERPAPI_API_KEY environment variable or configure "
|
||||
"via credential store. Get a key at https://serpapi.com/manage-api-key"
|
||||
),
|
||||
}
|
||||
return _SerpAPIClient(api_key)
|
||||
|
||||
@mcp.tool()
|
||||
def scholar_search(
|
||||
query: str,
|
||||
num_results: int = 10,
|
||||
start: int = 0,
|
||||
year_low: int | None = None,
|
||||
year_high: int | None = None,
|
||||
sort_by_date: bool = False,
|
||||
) -> dict:
|
||||
"""
|
||||
Search Google Scholar for academic papers, articles, and citations.
|
||||
|
||||
Returns structured results with titles, authors, citation counts,
|
||||
and links. Google Scholar has no official API — this is the only way
|
||||
to get structured paper metadata including citation counts and h-index.
|
||||
|
||||
Args:
|
||||
query: Search query for academic papers (1-500 chars)
|
||||
num_results: Number of results to return (1-20, default 10)
|
||||
start: Pagination offset (0, 10, 20, etc.)
|
||||
year_low: Filter papers published after this year (e.g. 2020)
|
||||
year_high: Filter papers published before this year (e.g. 2024)
|
||||
sort_by_date: If True, sort by date instead of relevance
|
||||
|
||||
Returns:
|
||||
Dict with organic_results containing paper metadata, or error dict
|
||||
"""
|
||||
if not query or len(query) > 500:
|
||||
return {"error": "Query must be 1-500 characters"}
|
||||
|
||||
client = _get_client()
|
||||
if isinstance(client, dict):
|
||||
return client
|
||||
|
||||
try:
|
||||
data = client.scholar_search(
|
||||
query=query,
|
||||
num=num_results,
|
||||
start=start,
|
||||
year_low=year_low,
|
||||
year_high=year_high,
|
||||
sort_by_date=sort_by_date,
|
||||
)
|
||||
if "error" in data:
|
||||
return data
|
||||
|
||||
results = []
|
||||
for item in data.get("organic_results", []):
|
||||
result = {
|
||||
"title": item.get("title", ""),
|
||||
"link": item.get("link", ""),
|
||||
"snippet": item.get("snippet", ""),
|
||||
"result_id": item.get("result_id", ""),
|
||||
"publication_info": item.get("publication_info", {}).get("summary", ""),
|
||||
"cited_by_count": (
|
||||
item.get("inline_links", {}).get("cited_by", {}).get("total", 0)
|
||||
),
|
||||
"cites_id": (
|
||||
item.get("inline_links", {}).get("cited_by", {}).get("cites_id", "")
|
||||
),
|
||||
}
|
||||
authors = item.get("publication_info", {}).get("authors", [])
|
||||
if authors:
|
||||
result["authors"] = [
|
||||
{
|
||||
"name": a.get("name", ""),
|
||||
"author_id": a.get("author_id", ""),
|
||||
}
|
||||
for a in authors
|
||||
]
|
||||
resources = item.get("resources", [])
|
||||
if resources:
|
||||
result["pdf_link"] = resources[0].get("link", "")
|
||||
results.append(result)
|
||||
|
||||
return {
|
||||
"query": query,
|
||||
"total_results": (data.get("search_information", {}).get("total_results", 0)),
|
||||
"results": results,
|
||||
"count": len(results),
|
||||
}
|
||||
|
||||
except httpx.TimeoutException:
|
||||
return {"error": "Search request timed out"}
|
||||
except httpx.RequestError as e:
|
||||
return {"error": f"Network error: {e}"}
|
||||
except Exception as e:
|
||||
return {"error": f"Scholar search failed: {e}"}
|
||||
|
||||
@mcp.tool()
|
||||
def scholar_get_citations(result_id: str) -> dict:
|
||||
"""
|
||||
Get formatted citations for a Google Scholar paper.
|
||||
|
||||
Returns citation text in MLA, APA, Chicago, Harvard, and Vancouver
|
||||
formats, plus download links for BibTeX, EndNote, RefMan, RefWorks.
|
||||
|
||||
Args:
|
||||
result_id: The result_id from a scholar_search result
|
||||
|
||||
Returns:
|
||||
Dict with citations list and download links, or error dict
|
||||
"""
|
||||
if not result_id:
|
||||
return {"error": "result_id is required"}
|
||||
|
||||
client = _get_client()
|
||||
if isinstance(client, dict):
|
||||
return client
|
||||
|
||||
try:
|
||||
data = client.scholar_cite(result_id)
|
||||
if "error" in data:
|
||||
return data
|
||||
|
||||
return {
|
||||
"result_id": result_id,
|
||||
"citations": data.get("citations", []),
|
||||
"links": data.get("links", []),
|
||||
}
|
||||
|
||||
except httpx.TimeoutException:
|
||||
return {"error": "Citation request timed out"}
|
||||
except httpx.RequestError as e:
|
||||
return {"error": f"Network error: {e}"}
|
||||
except Exception as e:
|
||||
return {"error": f"Citation lookup failed: {e}"}
|
||||
|
||||
@mcp.tool()
|
||||
def scholar_get_author(
|
||||
author_id: str,
|
||||
num_articles: int = 20,
|
||||
start: int = 0,
|
||||
sort_by: str = "citedby",
|
||||
) -> dict:
|
||||
"""
|
||||
Get a Google Scholar author profile with h-index, citations, and articles.
|
||||
|
||||
Returns author name, affiliations, research interests, citation
|
||||
metrics (total citations, h-index, i10-index), and their articles.
|
||||
|
||||
Args:
|
||||
author_id: Google Scholar author ID (e.g. 'WLN3QrAAAAAJ')
|
||||
num_articles: Number of articles to return (1-100, default 20)
|
||||
start: Pagination offset for articles (default 0)
|
||||
sort_by: Sort articles by 'citedby' (default) or 'pubdate'
|
||||
|
||||
Returns:
|
||||
Dict with author profile, metrics, and articles, or error dict
|
||||
"""
|
||||
if not author_id:
|
||||
return {"error": "author_id is required"}
|
||||
|
||||
client = _get_client()
|
||||
if isinstance(client, dict):
|
||||
return client
|
||||
|
||||
try:
|
||||
data = client.scholar_author(
|
||||
author_id=author_id,
|
||||
start=start,
|
||||
num=num_articles,
|
||||
sort_by=sort_by,
|
||||
)
|
||||
if "error" in data:
|
||||
return data
|
||||
|
||||
author = data.get("author", {})
|
||||
cited_by = data.get("cited_by", {})
|
||||
|
||||
metrics = {}
|
||||
for entry in cited_by.get("table", []):
|
||||
for key, value in entry.items():
|
||||
metrics[key] = value
|
||||
|
||||
articles = []
|
||||
for article in data.get("articles", []):
|
||||
articles.append(
|
||||
{
|
||||
"title": article.get("title", ""),
|
||||
"authors": article.get("authors", ""),
|
||||
"publication": article.get("publication", ""),
|
||||
"year": article.get("year", ""),
|
||||
"cited_by_count": article.get("cited_by", {}).get("value", 0),
|
||||
"citation_id": article.get("citation_id", ""),
|
||||
}
|
||||
)
|
||||
|
||||
return {
|
||||
"author_id": author_id,
|
||||
"name": author.get("name", ""),
|
||||
"affiliations": author.get("affiliations", ""),
|
||||
"email": author.get("email", ""),
|
||||
"interests": [i.get("title", "") for i in author.get("interests", [])],
|
||||
"thumbnail": author.get("thumbnail", ""),
|
||||
"metrics": metrics,
|
||||
"articles": articles,
|
||||
"article_count": len(articles),
|
||||
}
|
||||
|
||||
except httpx.TimeoutException:
|
||||
return {"error": "Author lookup timed out"}
|
||||
except httpx.RequestError as e:
|
||||
return {"error": f"Network error: {e}"}
|
||||
except Exception as e:
|
||||
return {"error": f"Author lookup failed: {e}"}
|
||||
|
||||
@mcp.tool()
|
||||
def patents_search(
|
||||
query: str,
|
||||
page: int = 1,
|
||||
country: str | None = None,
|
||||
status: str | None = None,
|
||||
before_date: str | None = None,
|
||||
after_date: str | None = None,
|
||||
) -> dict:
|
||||
"""
|
||||
Search Google Patents for patents and patent applications.
|
||||
|
||||
Supports keyword search, inventor/assignee filtering via query operators,
|
||||
and date/country/status filters.
|
||||
|
||||
Query operators (use in query string):
|
||||
- inassignee:Google — filter by assignee
|
||||
- ininventor:"John Smith" — filter by inventor
|
||||
- inclaims:neural network — search within claims
|
||||
- intitle:machine learning — search within title
|
||||
|
||||
Args:
|
||||
query: Search query for patents (1-500 chars)
|
||||
page: Page number, 1-indexed (default 1)
|
||||
country: Filter by country code (e.g. 'US', 'EP', 'WO', 'CN')
|
||||
status: Patent status filter: 'GRANT' or 'APPLICATION'
|
||||
before_date: Patents filed before this date (YYYYMMDD)
|
||||
after_date: Patents filed after this date (YYYYMMDD)
|
||||
|
||||
Returns:
|
||||
Dict with patent results, or error dict
|
||||
"""
|
||||
if not query or len(query) > 500:
|
||||
return {"error": "Query must be 1-500 characters"}
|
||||
|
||||
client = _get_client()
|
||||
if isinstance(client, dict):
|
||||
return client
|
||||
|
||||
try:
|
||||
data = client.patents_search(
|
||||
query=query,
|
||||
page=page,
|
||||
country=country,
|
||||
status=status,
|
||||
before=before_date,
|
||||
after=after_date,
|
||||
)
|
||||
if "error" in data:
|
||||
return data
|
||||
|
||||
results = []
|
||||
for item in data.get("organic_results", []):
|
||||
results.append(
|
||||
{
|
||||
"title": item.get("title", ""),
|
||||
"snippet": item.get("snippet", ""),
|
||||
"link": item.get("link", ""),
|
||||
"patent_id": item.get("patent_id", ""),
|
||||
"publication_number": item.get("publication_number", ""),
|
||||
"inventor": item.get("inventor", ""),
|
||||
"assignee": item.get("assignee", ""),
|
||||
"filing_date": item.get("filing_date", ""),
|
||||
"grant_date": item.get("grant_date"),
|
||||
"publication_date": item.get("publication_date", ""),
|
||||
"priority_date": item.get("priority_date", ""),
|
||||
"pdf": item.get("pdf", ""),
|
||||
}
|
||||
)
|
||||
|
||||
return {
|
||||
"query": query,
|
||||
"total_results": (data.get("search_information", {}).get("total_results", 0)),
|
||||
"results": results,
|
||||
"count": len(results),
|
||||
"page": page,
|
||||
}
|
||||
|
||||
except httpx.TimeoutException:
|
||||
return {"error": "Patent search timed out"}
|
||||
except httpx.RequestError as e:
|
||||
return {"error": f"Network error: {e}"}
|
||||
except Exception as e:
|
||||
return {"error": f"Patent search failed: {e}"}
|
||||
|
||||
@mcp.tool()
|
||||
def patents_get_details(patent_id: str) -> dict:
|
||||
"""
|
||||
Get detailed information for a specific patent.
|
||||
|
||||
Fetches a single patent by its publication number (e.g. 'US20210012345A1')
|
||||
and returns full metadata including title, abstract, inventors, assignee,
|
||||
dates, classifications, and PDF link.
|
||||
|
||||
Args:
|
||||
patent_id: Patent publication number (e.g. 'US20210012345A1')
|
||||
|
||||
Returns:
|
||||
Dict with patent details, or error dict
|
||||
"""
|
||||
if not patent_id:
|
||||
return {"error": "patent_id is required"}
|
||||
|
||||
client = _get_client()
|
||||
if isinstance(client, dict):
|
||||
return client
|
||||
|
||||
try:
|
||||
data = client.patents_details(patent_id)
|
||||
if "error" in data:
|
||||
return data
|
||||
|
||||
results = data.get("organic_results", [])
|
||||
if not results:
|
||||
return {"error": f"No patent found for ID: {patent_id}"}
|
||||
|
||||
patent = results[0]
|
||||
return {
|
||||
"patent_id": patent_id,
|
||||
"title": patent.get("title", ""),
|
||||
"snippet": patent.get("snippet", ""),
|
||||
"link": patent.get("link", ""),
|
||||
"publication_number": patent.get("publication_number", ""),
|
||||
"inventor": patent.get("inventor", ""),
|
||||
"assignee": patent.get("assignee", ""),
|
||||
"filing_date": patent.get("filing_date", ""),
|
||||
"grant_date": patent.get("grant_date"),
|
||||
"publication_date": patent.get("publication_date", ""),
|
||||
"priority_date": patent.get("priority_date", ""),
|
||||
"pdf": patent.get("pdf", ""),
|
||||
"classifications": patent.get("classifications", {}),
|
||||
}
|
||||
|
||||
except httpx.TimeoutException:
|
||||
return {"error": "Patent detail request timed out"}
|
||||
except httpx.RequestError as e:
|
||||
return {"error": f"Network error: {e}"}
|
||||
except Exception as e:
|
||||
return {"error": f"Patent detail lookup failed: {e}"}
|
||||
@@ -0,0 +1,409 @@
|
||||
"""Tests for SerpAPI tools (Google Scholar & Patents) - FastMCP."""
|
||||
|
||||
from unittest.mock import patch
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
from fastmcp import FastMCP
|
||||
|
||||
from aden_tools.tools.serpapi_tool import register_tools
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def scholar_search_fn(mcp: FastMCP):
|
||||
"""Register and return the scholar_search tool function."""
|
||||
register_tools(mcp)
|
||||
return mcp._tool_manager._tools["scholar_search"].fn
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def scholar_cite_fn(mcp: FastMCP):
|
||||
"""Register and return the scholar_get_citations tool function."""
|
||||
register_tools(mcp)
|
||||
return mcp._tool_manager._tools["scholar_get_citations"].fn
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def scholar_author_fn(mcp: FastMCP):
|
||||
"""Register and return the scholar_get_author tool function."""
|
||||
register_tools(mcp)
|
||||
return mcp._tool_manager._tools["scholar_get_author"].fn
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def patents_search_fn(mcp: FastMCP):
|
||||
"""Register and return the patents_search tool function."""
|
||||
register_tools(mcp)
|
||||
return mcp._tool_manager._tools["patents_search"].fn
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def patents_details_fn(mcp: FastMCP):
|
||||
"""Register and return the patents_get_details tool function."""
|
||||
register_tools(mcp)
|
||||
return mcp._tool_manager._tools["patents_get_details"].fn
|
||||
|
||||
|
||||
# ---- Credential Tests ----
|
||||
|
||||
|
||||
class TestCredentials:
|
||||
"""Test credential handling for all SerpAPI tools."""
|
||||
|
||||
def test_scholar_search_no_creds(self, scholar_search_fn, monkeypatch):
|
||||
"""scholar_search without credentials returns helpful error."""
|
||||
monkeypatch.delenv("SERPAPI_API_KEY", raising=False)
|
||||
result = scholar_search_fn(query="machine learning")
|
||||
assert "error" in result
|
||||
assert "SerpAPI credentials not configured" in result["error"]
|
||||
assert "help" in result
|
||||
|
||||
def test_scholar_cite_no_creds(self, scholar_cite_fn, monkeypatch):
|
||||
"""scholar_get_citations without credentials returns error."""
|
||||
monkeypatch.delenv("SERPAPI_API_KEY", raising=False)
|
||||
result = scholar_cite_fn(result_id="abc123")
|
||||
assert "error" in result
|
||||
assert "SerpAPI credentials not configured" in result["error"]
|
||||
|
||||
def test_scholar_author_no_creds(self, scholar_author_fn, monkeypatch):
|
||||
"""scholar_get_author without credentials returns error."""
|
||||
monkeypatch.delenv("SERPAPI_API_KEY", raising=False)
|
||||
result = scholar_author_fn(author_id="WLN3QrAAAAAJ")
|
||||
assert "error" in result
|
||||
assert "SerpAPI credentials not configured" in result["error"]
|
||||
|
||||
def test_patents_search_no_creds(self, patents_search_fn, monkeypatch):
|
||||
"""patents_search without credentials returns error."""
|
||||
monkeypatch.delenv("SERPAPI_API_KEY", raising=False)
|
||||
result = patents_search_fn(query="neural network")
|
||||
assert "error" in result
|
||||
assert "SerpAPI credentials not configured" in result["error"]
|
||||
|
||||
def test_patents_details_no_creds(self, patents_details_fn, monkeypatch):
|
||||
"""patents_get_details without credentials returns error."""
|
||||
monkeypatch.delenv("SERPAPI_API_KEY", raising=False)
|
||||
result = patents_details_fn(patent_id="US20210012345A1")
|
||||
assert "error" in result
|
||||
assert "SerpAPI credentials not configured" in result["error"]
|
||||
|
||||
|
||||
# ---- Input Validation Tests ----
|
||||
|
||||
|
||||
class TestInputValidation:
|
||||
"""Test input validation for all tools."""
|
||||
|
||||
def test_scholar_empty_query(self, scholar_search_fn, monkeypatch):
|
||||
"""Empty query returns error."""
|
||||
monkeypatch.setenv("SERPAPI_API_KEY", "test-key")
|
||||
result = scholar_search_fn(query="")
|
||||
assert "error" in result
|
||||
assert "1-500" in result["error"]
|
||||
|
||||
def test_scholar_long_query(self, scholar_search_fn, monkeypatch):
|
||||
"""Query exceeding 500 chars returns error."""
|
||||
monkeypatch.setenv("SERPAPI_API_KEY", "test-key")
|
||||
result = scholar_search_fn(query="x" * 501)
|
||||
assert "error" in result
|
||||
assert "1-500" in result["error"]
|
||||
|
||||
def test_cite_empty_result_id(self, scholar_cite_fn, monkeypatch):
|
||||
"""Empty result_id returns error."""
|
||||
monkeypatch.setenv("SERPAPI_API_KEY", "test-key")
|
||||
result = scholar_cite_fn(result_id="")
|
||||
assert "error" in result
|
||||
assert "result_id" in result["error"]
|
||||
|
||||
def test_author_empty_id(self, scholar_author_fn, monkeypatch):
|
||||
"""Empty author_id returns error."""
|
||||
monkeypatch.setenv("SERPAPI_API_KEY", "test-key")
|
||||
result = scholar_author_fn(author_id="")
|
||||
assert "error" in result
|
||||
assert "author_id" in result["error"]
|
||||
|
||||
def test_patents_empty_query(self, patents_search_fn, monkeypatch):
|
||||
"""Empty patent query returns error."""
|
||||
monkeypatch.setenv("SERPAPI_API_KEY", "test-key")
|
||||
result = patents_search_fn(query="")
|
||||
assert "error" in result
|
||||
assert "1-500" in result["error"]
|
||||
|
||||
def test_patents_long_query(self, patents_search_fn, monkeypatch):
|
||||
"""Patent query exceeding 500 chars returns error."""
|
||||
monkeypatch.setenv("SERPAPI_API_KEY", "test-key")
|
||||
result = patents_search_fn(query="x" * 501)
|
||||
assert "error" in result
|
||||
|
||||
def test_patents_details_empty_id(self, patents_details_fn, monkeypatch):
|
||||
"""Empty patent_id returns error."""
|
||||
monkeypatch.setenv("SERPAPI_API_KEY", "test-key")
|
||||
result = patents_details_fn(patent_id="")
|
||||
assert "error" in result
|
||||
assert "patent_id" in result["error"]
|
||||
|
||||
|
||||
# ---- HTTP Error Handling Tests ----
|
||||
|
||||
|
||||
def _mock_response(status_code: int, json_data: dict | None = None, text: str = ""):
|
||||
"""Create a mock httpx.Response."""
|
||||
resp = httpx.Response(
|
||||
status_code=status_code,
|
||||
json=json_data,
|
||||
request=httpx.Request("GET", "https://serpapi.com/search.json"),
|
||||
)
|
||||
return resp
|
||||
|
||||
|
||||
class TestHTTPErrors:
|
||||
"""Test HTTP error handling."""
|
||||
|
||||
def test_401_returns_auth_error(self, scholar_search_fn, monkeypatch):
|
||||
"""HTTP 401 returns invalid API key error."""
|
||||
monkeypatch.setenv("SERPAPI_API_KEY", "bad-key")
|
||||
with patch("httpx.get", return_value=_mock_response(401, {"error": "Invalid API key"})):
|
||||
result = scholar_search_fn(query="test")
|
||||
assert "error" in result
|
||||
assert "Invalid SerpAPI API key" in result["error"]
|
||||
|
||||
def test_429_returns_rate_limit(self, scholar_search_fn, monkeypatch):
|
||||
"""HTTP 429 returns rate limit error."""
|
||||
monkeypatch.setenv("SERPAPI_API_KEY", "test-key")
|
||||
with patch("httpx.get", return_value=_mock_response(429)):
|
||||
result = scholar_search_fn(query="test")
|
||||
assert "error" in result
|
||||
assert "rate limit" in result["error"].lower()
|
||||
|
||||
def test_500_returns_server_error(self, patents_search_fn, monkeypatch):
|
||||
"""HTTP 500 returns server error."""
|
||||
monkeypatch.setenv("SERPAPI_API_KEY", "test-key")
|
||||
with patch("httpx.get", return_value=_mock_response(500, text="Internal Server Error")):
|
||||
result = patents_search_fn(query="test")
|
||||
assert "error" in result
|
||||
assert "500" in result["error"]
|
||||
|
||||
def test_timeout_returns_error(self, scholar_search_fn, monkeypatch):
|
||||
"""Timeout returns error dict."""
|
||||
monkeypatch.setenv("SERPAPI_API_KEY", "test-key")
|
||||
with patch("httpx.get", side_effect=httpx.TimeoutException("timed out")):
|
||||
result = scholar_search_fn(query="test")
|
||||
assert "error" in result
|
||||
assert "timed out" in result["error"].lower()
|
||||
|
||||
def test_network_error_returns_error(self, scholar_search_fn, monkeypatch):
|
||||
"""Network error returns error dict."""
|
||||
monkeypatch.setenv("SERPAPI_API_KEY", "test-key")
|
||||
with patch(
|
||||
"httpx.get",
|
||||
side_effect=httpx.ConnectError("Connection refused"),
|
||||
):
|
||||
result = scholar_search_fn(query="test")
|
||||
assert "error" in result
|
||||
assert "Network error" in result["error"] or "error" in result["error"].lower()
|
||||
|
||||
|
||||
# ---- Success Response Tests ----
|
||||
|
||||
|
||||
SCHOLAR_RESPONSE = {
|
||||
"search_information": {"total_results": 1000},
|
||||
"organic_results": [
|
||||
{
|
||||
"position": 0,
|
||||
"title": "Deep learning",
|
||||
"result_id": "vhbKQo7YFEEJ",
|
||||
"link": "https://www.nature.com/articles/nature14539",
|
||||
"snippet": "Deep learning allows computational models...",
|
||||
"publication_info": {
|
||||
"summary": "Y LeCun, Y Bengio, G Hinton - nature, 2015",
|
||||
"authors": [
|
||||
{"name": "Y LeCun", "author_id": "WLN3QrAAAAAJ"},
|
||||
{"name": "Y Bengio", "author_id": "kukA0LcAAAAJ"},
|
||||
],
|
||||
},
|
||||
"inline_links": {
|
||||
"cited_by": {
|
||||
"total": 75000,
|
||||
"cites_id": "17291221010185025511",
|
||||
},
|
||||
},
|
||||
"resources": [{"title": "PDF", "link": "https://example.com/paper.pdf"}],
|
||||
}
|
||||
],
|
||||
}
|
||||
|
||||
CITE_RESPONSE = {
|
||||
"citations": [
|
||||
{"title": "MLA", "snippet": "LeCun, Yann, et al..."},
|
||||
{"title": "APA", "snippet": "LeCun, Y., Bengio, Y..."},
|
||||
],
|
||||
"links": [
|
||||
{"name": "BibTeX", "link": "https://scholar.google.com/bibtex"},
|
||||
],
|
||||
}
|
||||
|
||||
AUTHOR_RESPONSE = {
|
||||
"author": {
|
||||
"name": "Yann LeCun",
|
||||
"affiliations": "NYU & Meta",
|
||||
"email": "Verified email at fb.com",
|
||||
"interests": [{"title": "machine learning"}, {"title": "deep learning"}],
|
||||
"thumbnail": "https://example.com/photo.jpg",
|
||||
},
|
||||
"articles": [
|
||||
{
|
||||
"title": "Gradient-based learning",
|
||||
"authors": "Y LeCun, L Bottou",
|
||||
"publication": "Proceedings of the IEEE, 1998",
|
||||
"year": "1998",
|
||||
"cited_by": {"value": 45000},
|
||||
"citation_id": "WLN3QrAAAAAJ:u5HHmVD_uO8C",
|
||||
}
|
||||
],
|
||||
"cited_by": {
|
||||
"table": [
|
||||
{"citations": {"all": 390000, "since_2019": 200000}},
|
||||
{"h_index": {"all": 165, "since_2019": 120}},
|
||||
{"i10_index": {"all": 420, "since_2019": 350}},
|
||||
],
|
||||
},
|
||||
}
|
||||
|
||||
PATENT_RESPONSE = {
|
||||
"search_information": {"total_results": 500},
|
||||
"organic_results": [
|
||||
{
|
||||
"title": "Machine learning model for prediction",
|
||||
"snippet": "A system and method...",
|
||||
"link": "https://patents.google.com/patent/US20210012345A1",
|
||||
"patent_id": "US20210012345A1",
|
||||
"publication_number": "US20210012345A1",
|
||||
"inventor": "John Smith",
|
||||
"assignee": "Google LLC",
|
||||
"filing_date": "2020-07-10",
|
||||
"grant_date": None,
|
||||
"publication_date": "2021-01-14",
|
||||
"priority_date": "2020-07-10",
|
||||
"pdf": "https://example.com/patent.pdf",
|
||||
}
|
||||
],
|
||||
}
|
||||
|
||||
|
||||
class TestScholarSearch:
|
||||
"""Tests for scholar_search with mock API responses."""
|
||||
|
||||
def test_successful_search(self, scholar_search_fn, monkeypatch):
|
||||
"""Successful scholar search returns structured results."""
|
||||
monkeypatch.setenv("SERPAPI_API_KEY", "test-key")
|
||||
with patch("httpx.get", return_value=_mock_response(200, SCHOLAR_RESPONSE)):
|
||||
result = scholar_search_fn(query="deep learning")
|
||||
|
||||
assert "error" not in result
|
||||
assert result["query"] == "deep learning"
|
||||
assert result["total_results"] == 1000
|
||||
assert result["count"] == 1
|
||||
assert len(result["results"]) == 1
|
||||
|
||||
paper = result["results"][0]
|
||||
assert paper["title"] == "Deep learning"
|
||||
assert paper["result_id"] == "vhbKQo7YFEEJ"
|
||||
assert paper["cited_by_count"] == 75000
|
||||
assert paper["cites_id"] == "17291221010185025511"
|
||||
assert paper["pdf_link"] == "https://example.com/paper.pdf"
|
||||
assert len(paper["authors"]) == 2
|
||||
assert paper["authors"][0]["name"] == "Y LeCun"
|
||||
|
||||
def test_search_with_year_filter(self, scholar_search_fn, monkeypatch):
|
||||
"""Search with year filters works."""
|
||||
monkeypatch.setenv("SERPAPI_API_KEY", "test-key")
|
||||
with patch("httpx.get", return_value=_mock_response(200, SCHOLAR_RESPONSE)) as mock:
|
||||
scholar_search_fn(query="AI", year_low=2020, year_high=2024)
|
||||
params = mock.call_args[1]["params"]
|
||||
assert params["as_ylo"] == 2020
|
||||
assert params["as_yhi"] == 2024
|
||||
|
||||
|
||||
class TestScholarCite:
|
||||
"""Tests for scholar_get_citations with mock API responses."""
|
||||
|
||||
def test_successful_cite(self, scholar_cite_fn, monkeypatch):
|
||||
"""Successful citation lookup returns formats."""
|
||||
monkeypatch.setenv("SERPAPI_API_KEY", "test-key")
|
||||
with patch("httpx.get", return_value=_mock_response(200, CITE_RESPONSE)):
|
||||
result = scholar_cite_fn(result_id="vhbKQo7YFEEJ")
|
||||
|
||||
assert "error" not in result
|
||||
assert result["result_id"] == "vhbKQo7YFEEJ"
|
||||
assert len(result["citations"]) == 2
|
||||
assert result["citations"][0]["title"] == "MLA"
|
||||
assert len(result["links"]) == 1
|
||||
|
||||
|
||||
class TestScholarAuthor:
|
||||
"""Tests for scholar_get_author with mock API responses."""
|
||||
|
||||
def test_successful_author(self, scholar_author_fn, monkeypatch):
|
||||
"""Successful author lookup returns profile and metrics."""
|
||||
monkeypatch.setenv("SERPAPI_API_KEY", "test-key")
|
||||
with patch("httpx.get", return_value=_mock_response(200, AUTHOR_RESPONSE)):
|
||||
result = scholar_author_fn(author_id="WLN3QrAAAAAJ")
|
||||
|
||||
assert "error" not in result
|
||||
assert result["name"] == "Yann LeCun"
|
||||
assert result["affiliations"] == "NYU & Meta"
|
||||
assert "machine learning" in result["interests"]
|
||||
assert result["metrics"]["h_index"]["all"] == 165
|
||||
assert result["article_count"] == 1
|
||||
assert result["articles"][0]["cited_by_count"] == 45000
|
||||
|
||||
|
||||
class TestPatentsSearch:
|
||||
"""Tests for patents_search with mock API responses."""
|
||||
|
||||
def test_successful_search(self, patents_search_fn, monkeypatch):
|
||||
"""Successful patent search returns structured results."""
|
||||
monkeypatch.setenv("SERPAPI_API_KEY", "test-key")
|
||||
with patch("httpx.get", return_value=_mock_response(200, PATENT_RESPONSE)):
|
||||
result = patents_search_fn(query="machine learning")
|
||||
|
||||
assert "error" not in result
|
||||
assert result["total_results"] == 500
|
||||
assert result["count"] == 1
|
||||
patent = result["results"][0]
|
||||
assert patent["patent_id"] == "US20210012345A1"
|
||||
assert patent["inventor"] == "John Smith"
|
||||
assert patent["assignee"] == "Google LLC"
|
||||
|
||||
def test_search_with_filters(self, patents_search_fn, monkeypatch):
|
||||
"""Search with country and status filters works."""
|
||||
monkeypatch.setenv("SERPAPI_API_KEY", "test-key")
|
||||
with patch("httpx.get", return_value=_mock_response(200, PATENT_RESPONSE)) as mock:
|
||||
patents_search_fn(query="AI", country="US", status="GRANT")
|
||||
params = mock.call_args[1]["params"]
|
||||
assert params["country"] == "US"
|
||||
assert params["status"] == "GRANT"
|
||||
|
||||
|
||||
class TestPatentsDetails:
|
||||
"""Tests for patents_get_details with mock API responses."""
|
||||
|
||||
def test_successful_details(self, patents_details_fn, monkeypatch):
|
||||
"""Successful patent detail lookup."""
|
||||
monkeypatch.setenv("SERPAPI_API_KEY", "test-key")
|
||||
with patch("httpx.get", return_value=_mock_response(200, PATENT_RESPONSE)):
|
||||
result = patents_details_fn(patent_id="US20210012345A1")
|
||||
|
||||
assert "error" not in result
|
||||
assert result["patent_id"] == "US20210012345A1"
|
||||
assert result["title"] == "Machine learning model for prediction"
|
||||
assert result["inventor"] == "John Smith"
|
||||
|
||||
def test_not_found(self, patents_details_fn, monkeypatch):
|
||||
"""Patent not found returns error."""
|
||||
monkeypatch.setenv("SERPAPI_API_KEY", "test-key")
|
||||
empty_response = {"organic_results": []}
|
||||
with patch("httpx.get", return_value=_mock_response(200, empty_response)):
|
||||
result = patents_details_fn(patent_id="INVALID123")
|
||||
assert "error" in result
|
||||
assert "No patent found" in result["error"]
|
||||
Reference in New Issue
Block a user