chnjinlei/hive
1
0
Fork 0
Code Issues Pull Requests Actions 41 Packages Projects Releases Wiki Activity

feat(tools): add SimilarWeb V5 API integration (#7066)

Browse Source 
Adds 29 MCP tools for SimilarWeb V5 covering traffic and engagement,
competitor intelligence, keywords/SERP, audience demographics, and
segment analysis. Includes credential spec, health checker, README,
and tests on ubuntu and windows.

Closes #7022
This commit is contained in:
SAURABH KUMAR
2026-04-26 18:07:44 +05:30
committed by GitHub
parent 445c9600ab
commit ea707438f2
9 changed files with 1389 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/src/aden_tools/credentials/__init__.py
+3
View File
@@ -128,6 +128,7 @@ from .shell_config import (
get_shell_source_command,
)
from .shopify import SHOPIFY_CREDENTIALS
from .similarweb import SIMILARWEB_CREDENTIALS
from .slack import SLACK_CREDENTIALS
from .snowflake import SNOWFLAKE_CREDENTIALS
from .store_adapter import CredentialStoreAdapter
@@ -209,6 +210,7 @@ CREDENTIAL_SPECS = {
**SAP_CREDENTIALS,
**SEARCH_CREDENTIALS,
**SERPAPI_CREDENTIALS,
**SIMILARWEB_CREDENTIALS,
**SHOPIFY_CREDENTIALS,
**SLACK_CREDENTIALS,
**SNOWFLAKE_CREDENTIALS,
@@ -306,6 +308,7 @@ __all__ = [
"SAP_CREDENTIALS",
"SEARCH_CREDENTIALS",
"SERPAPI_CREDENTIALS",
"SIMILARWEB_CREDENTIALS",
"SHOPIFY_CREDENTIALS",
"SLACK_CREDENTIALS",
"SNOWFLAKE_CREDENTIALS",
tools/src/aden_tools/credentials/health_check.py
+24
View File
@@ -1125,6 +1125,29 @@ class SerpApiHealthChecker(BaseHttpHealthChecker):
AUTH_QUERY_PARAM_NAME = "api_key"
class SimilarWebHealthChecker(BaseHttpHealthChecker):
"""Health checker for SimilarWeb API key."""
ENDPOINT = "https://api.similarweb.com/v5/website-analysis/websites/traffic-and-engagement/"
SERVICE_NAME = "SimilarWeb"
AUTH_TYPE = BaseHttpHealthChecker.AUTH_HEADER
AUTH_HEADER_NAME = "api-key"
AUTH_HEADER_TEMPLATE = "{token}"
def _build_params(self, credential_value: str) -> dict[str, str]:
params = super()._build_params(credential_value)
params.update(
{
"domain": "google.com",
"start_date": "2024-01",
"end_date": "2024-01",
"country": "world",
"granularity": "monthly",
}
)
return params
class ApolloHealthChecker(BaseHttpHealthChecker):
"""Health checker for Apollo.io API key."""
@@ -1386,6 +1409,7 @@ HEALTH_CHECKERS: dict[str, CredentialHealthChecker] = {
"prometheus": PrometheusHealthChecker(),
"resend": ResendHealthChecker(),
"serpapi": SerpApiHealthChecker(),
"similarweb": SimilarWebHealthChecker(),
"slack": SlackHealthChecker(),
"stripe": StripeHealthChecker(),
"telegram": TelegramHealthChecker(),
tools/src/aden_tools/credentials/similarweb.py
+53
View File
@@ -0,0 +1,53 @@
from __future__ import annotations
from aden_tools.credentials.base import CredentialSpec
SIMILARWEB_CREDENTIALS = {
"similarweb": CredentialSpec(
env_var="SIMILARWEB_API_KEY",
tools=[
"similarweb_v5_traffic_and_engagement",
"similarweb_v5_website_rank",
"similarweb_v5_traffic_sources",
"similarweb_v5_geography",
"similarweb_v5_demographics",
"similarweb_v5_company_info",
"similarweb_v5_top_sites_by_category",
"similarweb_v5_referrals",
"similarweb_v5_ppc_spend",
"similarweb_v5_geography_details",
"similarweb_v5_similar_sites",
"similarweb_v5_ad_networks",
"similarweb_v5_demographics_traffic",
"similarweb_v5_deduplicated_audience",
"similarweb_v5_audience_interests",
"similarweb_v5_audience_overlap",
"similarweb_v5_technologies",
"similarweb_v5_leading_folders",
"similarweb_v5_popular_pages",
"similarweb_v5_subdomains",
"similarweb_v5_keyword_competitors",
"similarweb_v5_keyword_opportunities",
"similarweb_v5_serp_features",
"similarweb_v5_organic_keywords",
"similarweb_v5_paid_keywords",
"similarweb_v5_serp_players",
"similarweb_v5_social_referrals",
"similarweb_v5_segments_list",
"similarweb_v5_segment_analysis",
],
required=True,
help_url="https://developer.similarweb.com/",
description="API key for SimilarWeb traffic and competitor insights.",
direct_api_key_supported=True,
api_key_instructions="""To get a SimilarWeb API key:
1. Go to the SimilarWeb Developer Portal (https://developer.similarweb.com/)
2. Or log into your SimilarWeb Pro account at pro.similarweb.com
3. Navigate to Account Settings > API (or Data Extraction / API section)
4. Click on "Generate API Key"
5. Copy the generated API key and securely store it in your .env file""",
credential_id="similarweb",
credential_key="api_key",
health_check_endpoint="https://api.similarweb.com/v5/website-analysis/websites/traffic-and-engagement/",
)
}
tools/src/aden_tools/tools/__init__.py
+2
View File
@@ -118,6 +118,7 @@ from .salesforce_tool import register_tools as register_salesforce
from .sap_tool import register_tools as register_sap
from .serpapi_tool import register_tools as register_serpapi
from .shopify_tool import register_tools as register_shopify
from .similarweb_tool import register_tools as register_similarweb
from .slack_tool import register_tools as register_slack
from .snowflake_tool import register_tools as register_snowflake
from .ssl_tls_scanner import register_tools as register_ssl_tls_scanner
@@ -320,6 +321,7 @@ def _register_unverified(
register_salesforce(mcp, credentials=credentials)
register_sap(mcp, credentials=credentials)
register_shopify(mcp, credentials=credentials)
register_similarweb(mcp, credentials=credentials)
register_snowflake(mcp, credentials=credentials)
register_supabase(mcp, credentials=credentials)
register_terraform(mcp, credentials=credentials)
tools/src/aden_tools/tools/similarweb_tool/README.md
+152
View File
@@ -0,0 +1,152 @@
# SimilarWeb Tool
Integration with SimilarWeb for deep website analytics, competitor intelligence, market research data, traffic sources, and audience demographics.
## Overview
This tool enables Hive agents to interact with SimilarWeb's data intelligence infrastructure for:
- Website traffic analysis and engagement metrics
- Competitor research and benchmarking
- SEO and keyword analysis
- Advertising strategy and PPC spend insights
- Audience demographics and geographic distribution
- Technical profile and company insights
## Available Tools
This integration provides the following MCP tools for comprehensive market intelligence operations:
**Website Overview & Traffic**
- `similarweb_v5_traffic_and_engagement` - Get traffic and engagement metrics (visits, duration, pages per visit, bounce rate)
- `similarweb_v5_traffic_sources` - Get marketing channels (traffic sources) breakdown
- `similarweb_v5_geography` - Get traffic distribution by geography
- `similarweb_v5_geography_details` - Get detailed traffic distribution by country
- `similarweb_v5_website_rank` - Get global, country, and category ranks for a website
**Competitor Intelligence**
- `similarweb_v5_similar_sites` - Get a list of websites similar to the given domain
- `similarweb_v5_top_sites_by_category` - Get top sites in a specific category (e.g., 'Games', 'Lifestyle')
- `similarweb_v5_company_info` - Get company information (HQ, industry, etc.) for a website domain
- `similarweb_v5_technologies` - Get technologies used on the website (CMS, Ads, Analytics, etc.)
**Marketing Channels & Referrals**
- `similarweb_v5_referrals` - Get detailed referral traffic sources for a domain
- `similarweb_v5_social_referrals` - Get traffic distribution from social networks
- `similarweb_v5_ppc_spend` - Get estimated PPC spend for a website domain
- `similarweb_v5_ad_networks` - Get performance data across different ad networks
**Keywords & Search**
- `similarweb_v5_keyword_competitors` - Get organic and paid keyword competitors
- `similarweb_v5_keyword_opportunities` - Get keyword gap analysis and opportunities
- `similarweb_v5_organic_keywords` - Get detailed organic keyword performance
- `similarweb_v5_paid_keywords` - Get detailed paid keyword performance
- `similarweb_v5_serp_features` - Get SERP features analysis
- `similarweb_v5_serp_players` - Get top websites driving search traffic for keywords
**Website Content & Structure**
- `similarweb_v5_leading_folders` - Get top sub-folders by traffic
- `similarweb_v5_popular_pages` - Get most visited pages
- `similarweb_v5_subdomains` - Get traffic breakdown by subdomain
**Audience & Segments**
- `similarweb_v5_demographics` - Get audience demographics (age and gender)
- `similarweb_v5_demographics_traffic` - Get traffic breakdown by audience demographic segments
- `similarweb_v5_deduplicated_audience` - Get unique visitor count across multiple domains
- `similarweb_v5_audience_interests` - Get interests and categories relevant to the website's audience
- `similarweb_v5_audience_overlap` - Get shared audience between the main domain and a comparison domain
- `similarweb_v5_segments_list` - List custom segments available for the domain
- `similarweb_v5_segment_analysis` - Get traffic and engagement for a specific segment ID
## Setup
### 1. Get SimilarWeb API Credentials
1. Go to the [SimilarWeb Developer Portal](https://developer.similarweb.com/)
2. Log into your SimilarWeb Pro account at `pro.similarweb.com`
3. Navigate to **Account Settings** -> **API** (or Data Extraction / API section)
4. Click on **Generate API Key**
5. Copy the generated API key.
### 2. Configure Environment Variables
```bash
export SIMILARWEB_API_KEY="your_api_key_here"
```
## Usage
Here are usage examples for the available MCP tools:
### Website Overview & Traffic
```python
similarweb_v5_traffic_and_engagement(domain="example.com", country="world", granularity="monthly")
similarweb_v5_traffic_sources(domain="example.com", country="world")
similarweb_v5_geography(domain="example.com")
similarweb_v5_geography_details(domain="example.com")
similarweb_v5_website_rank(domain="example.com")
```
### Competitor Intelligence
```python
similarweb_v5_similar_sites(domain="example.com")
similarweb_v5_top_sites_by_category(category="Games", country="world")
similarweb_v5_company_info(domain="example.com")
similarweb_v5_technologies(domain="example.com")
```
### Marketing Channels & Referrals
```python
similarweb_v5_referrals(domain="example.com", country="world")
similarweb_v5_social_referrals(domain="example.com", country="world")
similarweb_v5_ppc_spend(domain="example.com", country="world")
similarweb_v5_ad_networks(domain="example.com", country="world")
```
### Keywords & Search
```python
similarweb_v5_keyword_competitors(domain="example.com")
similarweb_v5_keyword_opportunities(domain="example.com")
similarweb_v5_organic_keywords(domain="example.com", country="world")
similarweb_v5_paid_keywords(domain="example.com", country="world")
similarweb_v5_serp_features(domain="example.com")
similarweb_v5_serp_players(domain="example.com")
```
### Website Content & Structure
```python
similarweb_v5_leading_folders(domain="example.com", country="world")
similarweb_v5_popular_pages(domain="example.com", country="world")
similarweb_v5_subdomains(domain="example.com", country="world")
```
### Audience & Segments
```python
similarweb_v5_demographics(domain="example.com")
similarweb_v5_demographics_traffic(domain="example.com")
similarweb_v5_deduplicated_audience(domains="example.com,competitor.com", country="world")
similarweb_v5_audience_interests(domain="example.com")
similarweb_v5_audience_overlap(domain="example.com", compare_to="competitor.com")
similarweb_v5_segments_list(domain="example.com")
similarweb_v5_segment_analysis(segment_id="12345", country="world")
```
## Authentication
The tool passes your `SIMILARWEB_API_KEY` to the API calls via the `api-key` HTTP header during communication with the endpoints hosted under `https://api.similarweb.com`. The framework's credential adapter intercepts the secret parameter injected into your workspace securely.
## Error Handling
The API responses gracefully return API errors inside regular Python dictionaries with a detailed message (e.g. `{"error": "HTTP error 403: ..."}`).
tools/src/aden_tools/tools/similarweb_tool/__init__.py
+3
View File
@@ -0,0 +1,3 @@
from .similarweb_tool import register_tools
__all__ = ["register_tools"]
tools/src/aden_tools/tools/similarweb_tool/similarweb_tool.py
+570
View File
@@ -0,0 +1,570 @@
"""
SimilarWeb Tool - Traffic and competitor insights for FastMCP.
Provides website analytics, demographic data, and competitor intelligence.
"""
from __future__ import annotations
from typing import TYPE_CHECKING, Any
import httpx
if TYPE_CHECKING:
from fastmcp import FastMCP
from aden_tools.credentials import CredentialStoreAdapter
def _get_api_key(credentials: CredentialStoreAdapter | None = None) -> str | dict[str, str]:
"""Get the SimilarWeb API key from credentials or environment."""
if credentials:
key = credentials.get("similarweb")
if key:
return key
import os
env_key = os.environ.get("SIMILARWEB_API_KEY")
if env_key:
return env_key
return {
"error": "SimilarWeb credentials not configured",
"help": (
"Set SIMILARWEB_API_KEY environment variable or configure "
"via credential store. Get a key at https://developer.similarweb.com/"
),
}
def _make_request(
endpoint: str,
api_key: str,
params: dict[str, Any] | None = None,
) -> dict[str, Any]:
"""Helper method to make requests to the SimilarWeb API V5."""
if params is None:
params = {}
# SimilarWeb API v5 uses api-key in the header
headers = {"api-key": api_key, "Accept": "application/json"}
url = f"https://api.similarweb.com/v5/{endpoint}"
try:
response = httpx.get(url, params=params, headers=headers, timeout=30.0)
response.raise_for_status()
return response.json()
except httpx.HTTPStatusError as e:
return {"error": f"HTTP error {e.response.status_code}: {e.response.text}"}
except Exception as e:
return {"error": f"Request failed: {str(e)}"}
def register_tools(mcp: FastMCP, credentials: CredentialStoreAdapter | None = None) -> None:
"""Register SimilarWeb V5 tools with the MCP server."""
@mcp.tool()
def similarweb_v5_traffic_and_engagement(
domain: str,
start_date: str | None = None,
end_date: str | None = None,
country: str = "world",
granularity: str = "monthly",
) -> dict[str, Any]:
"""
Get traffic and engagement metrics for a website using V5 API.
Args:
domain: The website domain (e.g., 'amazon.com')
start_date: Start date (YYYY-MM or YYYY-MM-DD)
end_date: End date (YYYY-MM or YYYY-MM-DD)
country: 2-letter country code or 'world'
granularity: 'daily', 'weekly', or 'monthly'
"""
api_key_res = _get_api_key(credentials)
if isinstance(api_key_res, dict):
return api_key_res
params = {
"metrics": "visits,bounce_rate,avg_visit_duration,pages_per_visit,total_page_views",
"country": country,
"granularity": granularity,
}
if start_date:
params["start_date"] = start_date
if end_date:
params["end_date"] = end_date
params["domain"] = domain
params["web_source"] = "desktop"
return _make_request("website-analysis/websites/traffic-and-engagement", api_key_res, params)
@mcp.tool()
def similarweb_v5_website_rank(
domain: str,
) -> dict[str, Any]:
"""Get global, country, and category ranks for a website."""
api_key_res = _get_api_key(credentials)
if isinstance(api_key_res, dict):
return api_key_res
return _make_request("website-analysis/websites/website-rank", api_key_res, {"domain": domain})
@mcp.tool()
def similarweb_v5_traffic_sources(
domain: str,
start_date: str | None = None,
end_date: str | None = None,
country: str = "world",
) -> dict[str, Any]:
"""Get marketing channels (traffic sources) breakdown for a website."""
api_key_res = _get_api_key(credentials)
if isinstance(api_key_res, dict):
return api_key_res
params = {"country": country}
if start_date:
params["start_date"] = start_date
if end_date:
params["end_date"] = end_date
params["domain"] = domain
return _make_request("website-analysis/websites/traffic-sources", api_key_res, params)
@mcp.tool()
def similarweb_v5_geography(
domain: str,
start_date: str | None = None,
end_date: str | None = None,
) -> dict[str, Any]:
"""Get traffic distribution by geography for a website."""
api_key_res = _get_api_key(credentials)
if isinstance(api_key_res, dict):
return api_key_res
params = {}
if start_date:
params["start_date"] = start_date
if end_date:
params["end_date"] = end_date
params["domain"] = domain
return _make_request("website-analysis/websites/traffic-geography", api_key_res, params)
@mcp.tool()
def similarweb_v5_demographics(
domain: str,
) -> dict[str, Any]:
"""Get audience demographics (age and gender) for a website."""
api_key_res = _get_api_key(credentials)
if isinstance(api_key_res, dict):
return api_key_res
return _make_request("website-analysis/websites/demographics/aggregated", api_key_res, {"domain": domain})
@mcp.tool()
def similarweb_v5_company_info(
domain: str,
) -> dict[str, Any]:
"""Get company information (HQ, industry, etc.) for a website domain."""
api_key_res = _get_api_key(credentials)
if isinstance(api_key_res, dict):
return api_key_res
return _make_request("website-analysis/websites/company-info/company-info", api_key_res, {"domain": domain})
@mcp.tool()
def similarweb_v5_top_sites_by_category(
category: str,
country: str = "world",
) -> dict[str, Any]:
"""Get top sites in a specific category (e.g., 'Games', 'Lifestyle')."""
api_key_res = _get_api_key(credentials)
if isinstance(api_key_res, dict):
return api_key_res
params = {"category": category, "country": country}
return _make_request("website-analysis/websites/top-sites-by-category/aggregated", api_key_res, params)
@mcp.tool()
def similarweb_v5_referrals(
domain: str,
start_date: str | None = None,
end_date: str | None = None,
country: str = "world",
) -> dict[str, Any]:
"""Get detailed referral traffic sources for a domain."""
api_key_res = _get_api_key(credentials)
if isinstance(api_key_res, dict):
return api_key_res
params = {"country": country}
if start_date:
params["start_date"] = start_date
if end_date:
params["end_date"] = end_date
params["domain"] = domain
return _make_request("website-analysis/websites/referrals/aggregated", api_key_res, params)
@mcp.tool()
def similarweb_v5_ppc_spend(
domain: str,
start_date: str | None = None,
end_date: str | None = None,
country: str = "world",
) -> dict[str, Any]:
"""Get estimated PPC spend for a website domain."""
api_key_res = _get_api_key(credentials)
if isinstance(api_key_res, dict):
return api_key_res
params = {"country": country}
if start_date:
params["start_date"] = start_date
if end_date:
params["end_date"] = end_date
params["domain"] = domain
return _make_request("website-analysis/websites/ppc-spend", api_key_res, params)
@mcp.tool()
def similarweb_v5_geography_details(
domain: str,
start_date: str | None = None,
end_date: str | None = None,
) -> dict[str, Any]:
"""Get detailed traffic distribution by country (aggregated)."""
api_key_res = _get_api_key(credentials)
if isinstance(api_key_res, dict):
return api_key_res
params = {}
if start_date:
params["start_date"] = start_date
if end_date:
params["end_date"] = end_date
params["domain"] = domain
return _make_request("website-analysis/websites/geography/aggregated", api_key_res, params)
@mcp.tool()
def similarweb_v5_similar_sites(
domain: str,
) -> dict[str, Any]:
"""Get a list of websites similar to the given domain."""
api_key_res = _get_api_key(credentials)
if isinstance(api_key_res, dict):
return api_key_res
return _make_request("website-analysis/websites/similar-sites/aggregated", api_key_res, {"domain": domain})
@mcp.tool()
def similarweb_v5_ad_networks(
domain: str,
start_date: str | None = None,
end_date: str | None = None,
country: str = "world",
) -> dict[str, Any]:
"""Get performance data across different ad networks for a domain."""
api_key_res = _get_api_key(credentials)
if isinstance(api_key_res, dict):
return api_key_res
params = {"country": country}
if start_date:
params["start_date"] = start_date
if end_date:
params["end_date"] = end_date
params["domain"] = domain
return _make_request("website-analysis/ad-networks/aggregated", api_key_res, params)
@mcp.tool()
def similarweb_v5_demographics_traffic(
domain: str,
) -> dict[str, Any]:
"""Get traffic breakdown by audience demographic segments."""
api_key_res = _get_api_key(credentials)
if isinstance(api_key_res, dict):
return api_key_res
return _make_request(
"website-analysis/websites/traffic-by-demographics/aggregated", api_key_res, {"domain": domain}
)
@mcp.tool()
def similarweb_v5_deduplicated_audience(
domains: str,
start_date: str | None = None,
end_date: str | None = None,
country: str = "world",
) -> dict[str, Any]:
"""
Get unique visitor count across multiple domains (comma-separated).
Args:
domains: Comma-separated domains (e.g. 'amazon.com,ebay.com')
"""
api_key_res = _get_api_key(credentials)
if isinstance(api_key_res, dict):
return api_key_res
params = {"domains": domains, "country": country}
if start_date:
params["start_date"] = start_date
if end_date:
params["end_date"] = end_date
return _make_request("website-analysis/websites/deduplicated-audience", api_key_res, params)
@mcp.tool()
def similarweb_v5_audience_interests(
domain: str,
) -> dict[str, Any]:
"""Get interests and categories relevant to the website's audience."""
api_key_res = _get_api_key(credentials)
if isinstance(api_key_res, dict):
return api_key_res
return _make_request("website-analysis/websites/audience-interests/aggregated", api_key_res, {"domain": domain})
@mcp.tool()
def similarweb_v5_audience_overlap(
domain: str,
compare_to: str,
) -> dict[str, Any]:
"""
Get shared audience between the main domain and a comparison domain.
Args:
domain: The main domain
compare_to: Domain to compare overlap with
"""
api_key_res = _get_api_key(credentials)
if isinstance(api_key_res, dict):
return api_key_res
params = {"domain": domain, "compare_to": compare_to}
return _make_request("website-analysis/websites/audience-overlap/aggregated", api_key_res, params)
@mcp.tool()
def similarweb_v5_technologies(
domain: str,
) -> dict[str, Any]:
"""Get technologies used on the website (CMS, Ads, Analytics, etc.)."""
api_key_res = _get_api_key(credentials)
if isinstance(api_key_res, dict):
return api_key_res
return _make_request("website-analysis/websites/technologies/aggregated", api_key_res, {"domain": domain})
@mcp.tool()
def similarweb_v5_leading_folders(
domain: str,
start_date: str | None = None,
end_date: str | None = None,
country: str = "world",
) -> dict[str, Any]:
"""Get top sub-folders by traffic for a domain."""
api_key_res = _get_api_key(credentials)
if isinstance(api_key_res, dict):
return api_key_res
params = {"country": country}
if start_date:
params["start_date"] = start_date
if end_date:
params["end_date"] = end_date
params["domain"] = domain
return _make_request("website-analysis/websites/pages/leading-folders/aggregated", api_key_res, params)
@mcp.tool()
def similarweb_v5_popular_pages(
domain: str,
start_date: str | None = None,
end_date: str | None = None,
country: str = "world",
) -> dict[str, Any]:
"""Get most visited pages on the given domain."""
api_key_res = _get_api_key(credentials)
if isinstance(api_key_res, dict):
return api_key_res
params = {"country": country}
if start_date:
params["start_date"] = start_date
if end_date:
params["end_date"] = end_date
params["domain"] = domain
return _make_request("website-content/pages/popular-pages/aggregated", api_key_res, params)
@mcp.tool()
def similarweb_v5_subdomains(
domain: str,
start_date: str | None = None,
end_date: str | None = None,
country: str = "world",
) -> dict[str, Any]:
"""Get traffic breakdown by subdomain for a domain."""
api_key_res = _get_api_key(credentials)
if isinstance(api_key_res, dict):
return api_key_res
params = {"country": country}
if start_date:
params["start_date"] = start_date
if end_date:
params["end_date"] = end_date
params["domain"] = domain
return _make_request("website-content/subdomains/aggregated", api_key_res, params)
@mcp.tool()
def similarweb_v5_keyword_competitors(
domain: str,
) -> dict[str, Any]:
"""Get organic and paid keyword competitors for a domain."""
api_key_res = _get_api_key(credentials)
if isinstance(api_key_res, dict):
return api_key_res
return _make_request(
"website-analysis/websites/keywords-competitors/aggregated", api_key_res, {"domain": domain}
)
@mcp.tool()
def similarweb_v5_keyword_opportunities(
domain: str,
) -> dict[str, Any]:
"""Get keyword gap analysis and opportunities for a domain."""
api_key_res = _get_api_key(credentials)
if isinstance(api_key_res, dict):
return api_key_res
return _make_request(
"website-analysis/websites/keywords-opportunities/aggregated", api_key_res, {"domain": domain}
)
@mcp.tool()
def similarweb_v5_serp_features(
domain: str,
) -> dict[str, Any]:
"""Get SERP features analysis for the domain."""
api_key_res = _get_api_key(credentials)
if isinstance(api_key_res, dict):
return api_key_res
return _make_request("website-analysis/websites/keywords/serp-features", api_key_res, {"domain": domain})
@mcp.tool()
def similarweb_v5_organic_keywords(
domain: str,
start_date: str | None = None,
end_date: str | None = None,
country: str = "world",
) -> dict[str, Any]:
"""Get detailed organic keyword performance for a domain."""
api_key_res = _get_api_key(credentials)
if isinstance(api_key_res, dict):
return api_key_res
params = {"country": country}
if start_date:
params["start_date"] = start_date
if end_date:
params["end_date"] = end_date
params["domain"] = domain
return _make_request("website-analysis/websites/keywords/organic/aggregated", api_key_res, params)
@mcp.tool()
def similarweb_v5_paid_keywords(
domain: str,
start_date: str | None = None,
end_date: str | None = None,
country: str = "world",
) -> dict[str, Any]:
"""Get detailed paid keyword performance for a domain."""
api_key_res = _get_api_key(credentials)
if isinstance(api_key_res, dict):
return api_key_res
params = {"country": country}
if start_date:
params["start_date"] = start_date
if end_date:
params["end_date"] = end_date
params["domain"] = domain
return _make_request("website-analysis/websites/keywords/paid/aggregated", api_key_res, params)
@mcp.tool()
def similarweb_v5_serp_players(
domain: str,
) -> dict[str, Any]:
"""Get top websites driving search traffic for keywords (SERP players)."""
api_key_res = _get_api_key(credentials)
if isinstance(api_key_res, dict):
return api_key_res
return _make_request(
"website-analysis/websites/keywords/serp-players/aggregated", api_key_res, {"domain": domain}
)
@mcp.tool()
def similarweb_v5_social_referrals(
domain: str,
start_date: str | None = None,
end_date: str | None = None,
country: str = "world",
) -> dict[str, Any]:
"""Get traffic distribution from social networks for a domain."""
api_key_res = _get_api_key(credentials)
if isinstance(api_key_res, dict):
return api_key_res
params = {"country": country}
if start_date:
params["start_date"] = start_date
if end_date:
params["end_date"] = end_date
params["domain"] = domain
return _make_request("website-analysis/websites/social-referrals/aggregated", api_key_res, params)
@mcp.tool()
def similarweb_v5_segments_list(
domain: str,
) -> dict[str, Any]:
"""List custom segments available for the domain."""
api_key_res = _get_api_key(credentials)
if isinstance(api_key_res, dict):
return api_key_res
return _make_request("segment-analysis/segments/describe", api_key_res, {"domain": domain})
@mcp.tool()
def similarweb_v5_segment_analysis(
segment_id: str,
start_date: str | None = None,
end_date: str | None = None,
country: str = "world",
) -> dict[str, Any]:
"""Get traffic and engagement for a specific segment ID."""
api_key_res = _get_api_key(credentials)
if isinstance(api_key_res, dict):
return api_key_res
params = {"country": country}
if start_date:
params["start_date"] = start_date
if end_date:
params["end_date"] = end_date
params["segment"] = segment_id
return _make_request("segment-analysis/segments/traffic-and-engagement", api_key_res, params)
tools/tests/test_health_checks.py
+1
View File
@@ -91,6 +91,7 @@ class TestHealthCheckerRegistry:
"prometheus",
"resend",
"serpapi",
"similarweb",
"slack",
"stripe",
"telegram",
tools/tests/tools/test_similarweb_tool.py
+581
View File
@@ -0,0 +1,581 @@
"""Tests for similarweb_tool - Website traffic and competitor analytics (V5 API)."""
from __future__ import annotations
from unittest.mock import MagicMock, patch
import httpx
import pytest
from fastmcp import FastMCP
from aden_tools.tools.similarweb_tool.similarweb_tool import register_tools
class MockCredentials:
def get(self, key: str) -> str | None:
if key == "similarweb":
return "test_api_key_123"
return None
@pytest.fixture
def credentials() -> MockCredentials:
return MockCredentials()
@pytest.fixture
def mcp_with_tools(credentials: MockCredentials) -> FastMCP:
mcp = FastMCP("SimilarWebTest")
register_tools(mcp, credentials=credentials)
return mcp
class TestSimilarWebToolV5:
def _mock_response(self, mock_get: MagicMock, json_data: dict) -> None:
mock_response = MagicMock()
mock_response.json.return_value = json_data
mock_response.raise_for_status.return_value = None
mock_get.return_value = mock_response
def _assert_v5_request(
self,
mock_get: MagicMock,
expected_full_endpoint: str,
expected_params: dict | None = None,
) -> None:
mock_get.assert_called_once()
actual_url = mock_get.call_args[0][0]
expected_url = f"https://api.similarweb.com/v5/{expected_full_endpoint}"
assert actual_url == expected_url
call_kwargs = mock_get.call_args[1]
assert call_kwargs["headers"]["api-key"] == "test_api_key_123"
assert call_kwargs["headers"]["Accept"] == "application/json"
if expected_params:
for k, v in expected_params.items():
assert call_kwargs["params"][k] == v
@patch("aden_tools.tools.similarweb_tool.similarweb_tool.httpx.get")
def test_similarweb_v5_traffic_and_engagement_success(
self,
mock_get: MagicMock,
mcp_with_tools: FastMCP,
) -> None:
response_data = {
"meta": {"request": {"domain": "amazon.com"}},
"visits": [{"date": "2023-01-01", "visits": 1000}],
}
self._mock_response(mock_get, response_data)
tool = mcp_with_tools._tool_manager._tools["similarweb_v5_traffic_and_engagement"]
result = tool.fn(domain="amazon.com", country="us", granularity="daily")
self._assert_v5_request(
mock_get,
"website-analysis/websites/traffic-and-engagement",
{
"domain": "amazon.com",
"country": "us",
"granularity": "daily",
"metrics": "visits,bounce_rate,avg_visit_duration,pages_per_visit,total_page_views",
"web_source": "desktop",
},
)
assert result == response_data
@patch("aden_tools.tools.similarweb_tool.similarweb_tool.httpx.get")
def test_similarweb_v5_website_rank_success(
self,
mock_get: MagicMock,
mcp_with_tools: FastMCP,
) -> None:
response_data = {"global_rank": 10, "country_rank": 5}
self._mock_response(mock_get, response_data)
tool = mcp_with_tools._tool_manager._tools["similarweb_v5_website_rank"]
result = tool.fn(domain="amazon.com")
self._assert_v5_request(mock_get, "website-analysis/websites/website-rank", {"domain": "amazon.com"})
assert result == response_data
@patch("aden_tools.tools.similarweb_tool.similarweb_tool.httpx.get")
def test_similarweb_v5_traffic_sources_success(
self,
mock_get: MagicMock,
mcp_with_tools: FastMCP,
) -> None:
response_data = {"search": 0.4, "direct": 0.3}
self._mock_response(mock_get, response_data)
tool = mcp_with_tools._tool_manager._tools["similarweb_v5_traffic_sources"]
result = tool.fn(domain="amazon.com", country="world")
self._assert_v5_request(
mock_get, "website-analysis/websites/traffic-sources", {"domain": "amazon.com", "country": "world"}
)
assert result == response_data
@patch("aden_tools.tools.similarweb_tool.similarweb_tool.httpx.get")
def test_similarweb_v5_geography_success(
self,
mock_get: MagicMock,
mcp_with_tools: FastMCP,
) -> None:
response_data = {"top_countries": [{"country": "US", "share": 0.5}]}
self._mock_response(mock_get, response_data)
tool = mcp_with_tools._tool_manager._tools["similarweb_v5_geography"]
result = tool.fn(domain="amazon.com")
self._assert_v5_request(mock_get, "website-analysis/websites/traffic-geography", {"domain": "amazon.com"})
assert result == response_data
@patch("aden_tools.tools.similarweb_tool.similarweb_tool.httpx.get")
def test_similarweb_v5_demographics_success(
self,
mock_get: MagicMock,
mcp_with_tools: FastMCP,
) -> None:
response_data = {"age_distribution": {"18-24": 0.2}}
self._mock_response(mock_get, response_data)
tool = mcp_with_tools._tool_manager._tools["similarweb_v5_demographics"]
result = tool.fn(domain="amazon.com")
self._assert_v5_request(mock_get, "website-analysis/websites/demographics/aggregated", {"domain": "amazon.com"})
assert result == response_data
@patch("aden_tools.tools.similarweb_tool.similarweb_tool.httpx.get")
def test_similarweb_v5_company_info_success(
self,
mock_get: MagicMock,
mcp_with_tools: FastMCP,
) -> None:
response_data = {"company_name": "Amazon", "headquarters": "Seattle, WA"}
self._mock_response(mock_get, response_data)
tool = mcp_with_tools._tool_manager._tools["similarweb_v5_company_info"]
result = tool.fn(domain="amazon.com")
self._assert_v5_request(
mock_get, "website-analysis/websites/company-info/company-info", {"domain": "amazon.com"}
)
assert result == response_data
@patch("aden_tools.tools.similarweb_tool.similarweb_tool.httpx.get")
def test_similarweb_v5_top_sites_by_category_success(
self,
mock_get: MagicMock,
mcp_with_tools: FastMCP,
) -> None:
response_data = {"top_sites": [{"domain": "google.com", "rank": 1}]}
self._mock_response(mock_get, response_data)
tool = mcp_with_tools._tool_manager._tools["similarweb_v5_top_sites_by_category"]
result = tool.fn(category="Search Engines")
self._assert_v5_request(
mock_get,
"website-analysis/websites/top-sites-by-category/aggregated",
{"category": "Search Engines", "country": "world"},
)
assert result == response_data
@patch("aden_tools.tools.similarweb_tool.similarweb_tool.httpx.get")
def test_similarweb_v5_keyword_competitors_success(
self,
mock_get: MagicMock,
mcp_with_tools: FastMCP,
) -> None:
response_data = {"competitors": [{"domain": "competitor.com", "overlap": 0.8}]}
self._mock_response(mock_get, response_data)
tool = mcp_with_tools._tool_manager._tools["similarweb_v5_keyword_competitors"]
result = tool.fn(domain="amazon.com")
self._assert_v5_request(
mock_get, "website-analysis/websites/keywords-competitors/aggregated", {"domain": "amazon.com"}
)
assert result == response_data
@patch("aden_tools.tools.similarweb_tool.similarweb_tool.httpx.get")
def test_similarweb_v5_technologies_success(
self,
mock_get: MagicMock,
mcp_with_tools: FastMCP,
) -> None:
response_data = {"technologies": [{"name": "React", "category": "Frontend Framework"}]}
self._mock_response(mock_get, response_data)
tool = mcp_with_tools._tool_manager._tools["similarweb_v5_technologies"]
result = tool.fn(domain="amazon.com")
self._assert_v5_request(mock_get, "website-analysis/websites/technologies/aggregated", {"domain": "amazon.com"})
assert result == response_data
@patch("aden_tools.tools.similarweb_tool.similarweb_tool.httpx.get")
def test_similarweb_v5_deduplicated_audience_success(
self,
mock_get: MagicMock,
mcp_with_tools: FastMCP,
) -> None:
response_data = {"total_unique_visitors": 1000000}
self._mock_response(mock_get, response_data)
tool = mcp_with_tools._tool_manager._tools["similarweb_v5_deduplicated_audience"]
result = tool.fn(domains="amazon.com,ebay.com")
self._assert_v5_request(
mock_get,
"website-analysis/websites/deduplicated-audience",
{"domains": "amazon.com,ebay.com", "country": "world"},
)
assert result == response_data
@patch("aden_tools.tools.similarweb_tool.similarweb_tool.httpx.get")
def test_similarweb_v5_referrals_success(
self,
mock_get: MagicMock,
mcp_with_tools: FastMCP,
) -> None:
response_data = {"referrals": [{"domain": "google.com", "share": 0.5}]}
self._mock_response(mock_get, response_data)
tool = mcp_with_tools._tool_manager._tools["similarweb_v5_referrals"]
result = tool.fn(domain="amazon.com")
self._assert_v5_request(
mock_get, "website-analysis/websites/referrals/aggregated", {"domain": "amazon.com", "country": "world"}
)
assert result == response_data
@patch("aden_tools.tools.similarweb_tool.similarweb_tool.httpx.get")
def test_similarweb_v5_ppc_spend_success(
self,
mock_get: MagicMock,
mcp_with_tools: FastMCP,
) -> None:
response_data = {"ppc_spend": 5000}
self._mock_response(mock_get, response_data)
tool = mcp_with_tools._tool_manager._tools["similarweb_v5_ppc_spend"]
result = tool.fn(domain="amazon.com")
self._assert_v5_request(
mock_get, "website-analysis/websites/ppc-spend", {"domain": "amazon.com", "country": "world"}
)
assert result == response_data
@patch("aden_tools.tools.similarweb_tool.similarweb_tool.httpx.get")
def test_similarweb_v5_geography_details_success(
self,
mock_get: MagicMock,
mcp_with_tools: FastMCP,
) -> None:
response_data = {"top_countries": [{"country": "US", "share": 0.5}]}
self._mock_response(mock_get, response_data)
tool = mcp_with_tools._tool_manager._tools["similarweb_v5_geography_details"]
result = tool.fn(domain="amazon.com")
self._assert_v5_request(mock_get, "website-analysis/websites/geography/aggregated", {"domain": "amazon.com"})
assert result == response_data
@patch("aden_tools.tools.similarweb_tool.similarweb_tool.httpx.get")
def test_similarweb_v5_similar_sites_success(
self,
mock_get: MagicMock,
mcp_with_tools: FastMCP,
) -> None:
response_data = {"similar_sites": [{"domain": "ebay.com"}]}
self._mock_response(mock_get, response_data)
tool = mcp_with_tools._tool_manager._tools["similarweb_v5_similar_sites"]
result = tool.fn(domain="amazon.com")
self._assert_v5_request(
mock_get, "website-analysis/websites/similar-sites/aggregated", {"domain": "amazon.com"}
)
assert result == response_data
@patch("aden_tools.tools.similarweb_tool.similarweb_tool.httpx.get")
def test_similarweb_v5_ad_networks_success(
self,
mock_get: MagicMock,
mcp_with_tools: FastMCP,
) -> None:
response_data = {"ad_networks": [{"name": "Google Ads", "share": 0.5}]}
self._mock_response(mock_get, response_data)
tool = mcp_with_tools._tool_manager._tools["similarweb_v5_ad_networks"]
result = tool.fn(domain="amazon.com")
self._assert_v5_request(
mock_get, "website-analysis/ad-networks/aggregated", {"domain": "amazon.com", "country": "world"}
)
assert result == response_data
@patch("aden_tools.tools.similarweb_tool.similarweb_tool.httpx.get")
def test_similarweb_v5_demographics_traffic_success(
self,
mock_get: MagicMock,
mcp_with_tools: FastMCP,
) -> None:
response_data = {"demographics": {"male": 0.5, "female": 0.5}}
self._mock_response(mock_get, response_data)
tool = mcp_with_tools._tool_manager._tools["similarweb_v5_demographics_traffic"]
result = tool.fn(domain="amazon.com")
self._assert_v5_request(
mock_get, "website-analysis/websites/traffic-by-demographics/aggregated", {"domain": "amazon.com"}
)
assert result == response_data
@patch("aden_tools.tools.similarweb_tool.similarweb_tool.httpx.get")
def test_similarweb_v5_audience_interests_success(
self,
mock_get: MagicMock,
mcp_with_tools: FastMCP,
) -> None:
response_data = {"interests": ["shopping", "tech"]}
self._mock_response(mock_get, response_data)
tool = mcp_with_tools._tool_manager._tools["similarweb_v5_audience_interests"]
result = tool.fn(domain="amazon.com")
self._assert_v5_request(
mock_get, "website-analysis/websites/audience-interests/aggregated", {"domain": "amazon.com"}
)
assert result == response_data
@patch("aden_tools.tools.similarweb_tool.similarweb_tool.httpx.get")
def test_similarweb_v5_audience_overlap_success(
self,
mock_get: MagicMock,
mcp_with_tools: FastMCP,
) -> None:
response_data = {"overlap": 0.3}
self._mock_response(mock_get, response_data)
tool = mcp_with_tools._tool_manager._tools["similarweb_v5_audience_overlap"]
result = tool.fn(domain="amazon.com", compare_to="ebay.com")
self._assert_v5_request(
mock_get,
"website-analysis/websites/audience-overlap/aggregated",
{"domain": "amazon.com", "compare_to": "ebay.com"},
)
assert result == response_data
@patch("aden_tools.tools.similarweb_tool.similarweb_tool.httpx.get")
def test_similarweb_v5_leading_folders_success(
self,
mock_get: MagicMock,
mcp_with_tools: FastMCP,
) -> None:
response_data = {"folders": [{"name": "/products/", "share": 0.4}]}
self._mock_response(mock_get, response_data)
tool = mcp_with_tools._tool_manager._tools["similarweb_v5_leading_folders"]
result = tool.fn(domain="amazon.com")
self._assert_v5_request(
mock_get,
"website-analysis/websites/pages/leading-folders/aggregated",
{"domain": "amazon.com", "country": "world"},
)
assert result == response_data
@patch("aden_tools.tools.similarweb_tool.similarweb_tool.httpx.get")
def test_similarweb_v5_popular_pages_success(
self,
mock_get: MagicMock,
mcp_with_tools: FastMCP,
) -> None:
response_data = {"pages": [{"url": "amazon.com/best-sellers", "share": 0.1}]}
self._mock_response(mock_get, response_data)
tool = mcp_with_tools._tool_manager._tools["similarweb_v5_popular_pages"]
result = tool.fn(domain="amazon.com")
self._assert_v5_request(
mock_get, "website-content/pages/popular-pages/aggregated", {"domain": "amazon.com", "country": "world"}
)
assert result == response_data
@patch("aden_tools.tools.similarweb_tool.similarweb_tool.httpx.get")
def test_similarweb_v5_subdomains_success(
self,
mock_get: MagicMock,
mcp_with_tools: FastMCP,
) -> None:
response_data = {"subdomains": [{"name": "aws.amazon.com", "share": 0.2}]}
self._mock_response(mock_get, response_data)
tool = mcp_with_tools._tool_manager._tools["similarweb_v5_subdomains"]
result = tool.fn(domain="amazon.com")
self._assert_v5_request(
mock_get, "website-content/subdomains/aggregated", {"domain": "amazon.com", "country": "world"}
)
assert result == response_data
@patch("aden_tools.tools.similarweb_tool.similarweb_tool.httpx.get")
def test_similarweb_v5_keyword_opportunities_success(
self,
mock_get: MagicMock,
mcp_with_tools: FastMCP,
) -> None:
response_data = {"opportunities": [{"keyword": "buy electronics", "score": 90}]}
self._mock_response(mock_get, response_data)
tool = mcp_with_tools._tool_manager._tools["similarweb_v5_keyword_opportunities"]
result = tool.fn(domain="amazon.com")
self._assert_v5_request(
mock_get, "website-analysis/websites/keywords-opportunities/aggregated", {"domain": "amazon.com"}
)
assert result == response_data
@patch("aden_tools.tools.similarweb_tool.similarweb_tool.httpx.get")
def test_similarweb_v5_serp_features_success(
self,
mock_get: MagicMock,
mcp_with_tools: FastMCP,
) -> None:
response_data = {"serp_features": {"featured_snippets": 10}}
self._mock_response(mock_get, response_data)
tool = mcp_with_tools._tool_manager._tools["similarweb_v5_serp_features"]
result = tool.fn(domain="amazon.com")
self._assert_v5_request(mock_get, "website-analysis/websites/keywords/serp-features", {"domain": "amazon.com"})
assert result == response_data
@patch("aden_tools.tools.similarweb_tool.similarweb_tool.httpx.get")
def test_similarweb_v5_organic_keywords_success(
self,
mock_get: MagicMock,
mcp_with_tools: FastMCP,
) -> None:
response_data = {"keywords": [{"phrase": "shopping", "visits": 1000}]}
self._mock_response(mock_get, response_data)
tool = mcp_with_tools._tool_manager._tools["similarweb_v5_organic_keywords"]
result = tool.fn(domain="amazon.com")
self._assert_v5_request(
mock_get,
"website-analysis/websites/keywords/organic/aggregated",
{"domain": "amazon.com", "country": "world"},
)
assert result == response_data
@patch("aden_tools.tools.similarweb_tool.similarweb_tool.httpx.get")
def test_similarweb_v5_paid_keywords_success(
self,
mock_get: MagicMock,
mcp_with_tools: FastMCP,
) -> None:
response_data = {"keywords": [{"phrase": "buy books", "visits": 500}]}
self._mock_response(mock_get, response_data)
tool = mcp_with_tools._tool_manager._tools["similarweb_v5_paid_keywords"]
result = tool.fn(domain="amazon.com")
self._assert_v5_request(
mock_get, "website-analysis/websites/keywords/paid/aggregated", {"domain": "amazon.com", "country": "world"}
)
assert result == response_data
@patch("aden_tools.tools.similarweb_tool.similarweb_tool.httpx.get")
def test_similarweb_v5_serp_players_success(
self,
mock_get: MagicMock,
mcp_with_tools: FastMCP,
) -> None:
response_data = {"players": [{"domain": "walmart.com", "share": 0.1}]}
self._mock_response(mock_get, response_data)
tool = mcp_with_tools._tool_manager._tools["similarweb_v5_serp_players"]
result = tool.fn(domain="amazon.com")
self._assert_v5_request(
mock_get, "website-analysis/websites/keywords/serp-players/aggregated", {"domain": "amazon.com"}
)
assert result == response_data
@patch("aden_tools.tools.similarweb_tool.similarweb_tool.httpx.get")
def test_similarweb_v5_social_referrals_success(
self,
mock_get: MagicMock,
mcp_with_tools: FastMCP,
) -> None:
response_data = {"social": [{"name": "Facebook", "share": 0.5}]}
self._mock_response(mock_get, response_data)
tool = mcp_with_tools._tool_manager._tools["similarweb_v5_social_referrals"]
result = tool.fn(domain="amazon.com")
self._assert_v5_request(
mock_get,
"website-analysis/websites/social-referrals/aggregated",
{"domain": "amazon.com", "country": "world"},
)
assert result == response_data
@patch("aden_tools.tools.similarweb_tool.similarweb_tool.httpx.get")
def test_similarweb_v5_segments_list_success(
self,
mock_get: MagicMock,
mcp_with_tools: FastMCP,
) -> None:
response_data = {"segments": [{"id": "seg1", "name": "Segment 1"}]}
self._mock_response(mock_get, response_data)
tool = mcp_with_tools._tool_manager._tools["similarweb_v5_segments_list"]
result = tool.fn(domain="amazon.com")
self._assert_v5_request(mock_get, "segment-analysis/segments/describe", {"domain": "amazon.com"})
assert result == response_data
@patch("aden_tools.tools.similarweb_tool.similarweb_tool.httpx.get")
def test_similarweb_v5_segment_analysis_success(
self,
mock_get: MagicMock,
mcp_with_tools: FastMCP,
) -> None:
response_data = {"metrics": {"visits": 1000}}
self._mock_response(mock_get, response_data)
tool = mcp_with_tools._tool_manager._tools["similarweb_v5_segment_analysis"]
result = tool.fn(segment_id="seg1")
self._assert_v5_request(
mock_get, "segment-analysis/segments/traffic-and-engagement", {"segment": "seg1", "country": "world"}
)
assert result == response_data
@patch("aden_tools.tools.similarweb_tool.similarweb_tool.httpx.get")
def test_make_request_error_handling(
self,
mock_get: MagicMock,
mcp_with_tools: FastMCP,
) -> None:
# Mock a 403 error
mock_response = MagicMock()
mock_response.status_code = 403
mock_response.text = "Forbidden"
mock_response.raise_for_status.side_effect = httpx.HTTPStatusError(
"403 Forbidden", request=MagicMock(), response=mock_response
)
mock_get.return_value = mock_response
tool = mcp_with_tools._tool_manager._tools["similarweb_v5_website_rank"]
result = tool.fn(domain="forbidden.com")
assert "error" in result
assert "403" in result["error"]
assert "Forbidden" in result["error"]