| name | amazon-sorftime-research-mcp-skill |
| description | Amazon product research, competitor analysis, category selection, keyword research, and review analysis using Sorftime MCP for cross-border e-commerce |
| triggers | ["analyze Amazon listing","research Amazon product category","find Amazon keywords","analyze product reviews","competitor research Amazon","Amazon category selection","product research workflow","Amazon market research"] |
Amazon Sorftime Research MCP Skill
Skill by ara.so — MCP Skills collection.
This skill enables AI agents to perform comprehensive Amazon product research, competitor analysis, category selection, keyword research, and review analysis using the Sorftime MCP service. It provides seven core skills for cross-border e-commerce decision-making: listing analysis, category selection, keyword research, review analysis, product research, Sif research, and Xiyou insight.
Overview
The project integrates three MCP services (Sorftime, Sif, Xiyou) and provides Python-based analysis tools for:
- Single Listing Analysis: Deep-dive into competitor products with keyword, review, and trend analysis
- Category Selection: Market analysis of Top 100 products with five-dimension scoring
- Keyword Research: 8-dimension intelligent classification of 1500+ keywords with ad strategy guidance
- Review Analysis: 6-dimension pain point analysis with service risk warnings
- Product Research: LLM-driven deep research workflow from data collection to decision
- Sif Research: Market validation, competitor growth paths, keyword layout, and traffic diagnostics
- Xiyou Insight: 7 scenario workflows for ad monitoring, traffic gap, competitor analysis, new product launch, ad budget transparency, and keyword database building
Installation
Prerequisites
- Claude Code CLI or similar AI coding agent
- Python 3.8+
- Bash shell environment
- Sorftime API Key from sorftime.com/zh-cn/mcp
Setup MCP Configuration
Create or update .mcp.json in your project root:
{
"mcpServers": {
"sorftime": {
"type": "streamableHttp",
"url": "https://mcp.sorftime.com?key=${SORFTIME_API_KEY}",
"name": "Sorftime MCP"
},
"sif": {
"type": "streamableHttp",
"url": "https://mcp.sif.com?key=${SIF_API_KEY}",
"name": "Sif MCP"
},
"xiyou": {
"type": "streamableHttp",
"url": "https://mcp.xiyou.com?key=${XIYOU_API_KEY}",
"name": "Xiyou MCP"
}
}
}
Set environment variables:
export SORFTIME_API_KEY="your_sorftime_api_key"
export SIF_API_KEY="your_sif_api_key"
export XIYOU_API_KEY="your_xiyou_api_key"
Install Python Dependencies
pip install -r requirements.txt
Core Skills and Commands
1. Listing Analysis (amazon-analyse)
Analyze a single ASIN with comprehensive data including product details, reviews, keywords, and trends.
/amazon-analyse B07PWTJ4H1 US
/amazon-analyse B08N5WRWNW DE
Python Implementation Pattern:
import json
import requests
from datetime import datetime
def analyze_listing(asin: str, site: str, mcp_url: str):
"""
Comprehensive listing analysis using Sorftime MCP
"""
product_detail = requests.post(
mcp_url,
json={
"method": "product_detail",
"params": {"asin": asin, "site": site}
}
).json()
traffic_keywords = requests.post(
mcp_url,
json={
"method": "product_traffic_terms",
"params": {"asin": asin, "site": site, "limit": 100}
}
).json()
reviews = requests.post(
mcp_url,
json={
"method": "product_reviews",
"params": {"asin": asin, "site": site, "limit": 100}
}
).json()
trends = requests.post(
mcp_url,
json={
"method": "product_trend",
"params": {
"asin": asin,
"site": site,
"days": 90
}
}
).json()
competitor_keywords = requests.post(
mcp_url,
json={
"method": "competitor_product_keywords",
"params": {"asin": asin, "site": site}
}
).json()
report_data = {
"asin": asin,
"site": site,
"product": product_detail,
"keywords": traffic_keywords,
"reviews": reviews,
"trends": trends,
"competitor_keywords": competitor_keywords,
"analysis_date": datetime.now().isoformat()
}
report_path = f"reports/analysis_{asin}_{site}_{datetime.now().strftime('%Y%m%d')}.md"
with open(report_path, 'w') as f:
f.write(generate_markdown_report(report_data))
return report_data
Output: reports/analysis_{ASIN}_{SITE}_{DATE}.md
2. Category Selection (category-selection)
Analyze category market with Top 100 products and five-dimension scoring model.
/category-selection "Sofas & Couches" US
/category-selection "Wireless Earbuds" US --limit 20
Python Implementation Pattern:
def analyze_category(category_name: str, site: str, limit: int = 100, mcp_url: str):
"""
Category selection analysis with five-dimension scoring
"""
category_search = requests.post(
mcp_url,
json={
"method": "category_name_search",
"params": {"query": category_name, "site": site}
}
).json()
node_id = category_search[0]['nodeId']
category_report = requests.post(
mcp_url,
json={
"method": "category_report",
"params": {
"nodeId": node_id,
"site": site,
"limit": limit
}
}
).json()
category_trends = requests.post(
mcp_url,
json={
"method": "category_trend",
"params": {"nodeId": node_id, "site": site}
}
).json()
scores = calculate_five_dimension_scores(category_report, category_trends)
report_dir = f"category-reports/{category_name}_{site}_{datetime.now().strftime('%Y%m%d')}"
os.makedirs(report_dir, exist_ok=True)
with open(f"{report_dir}/report.md", 'w') as f:
f.write(generate_category_markdown(category_report, scores))
generate_excel_report(f"{report_dir}/category_report.xlsx", category_report, scores)
generate_html_dashboard(f"{report_dir}/dashboard.html", category_report, scores)
return scores
def calculate_five_dimension_scores(report_data, trend_data):
"""
Five-dimension scoring model (100 points total)
"""
return {
"market_scale": calculate_market_scale_score(report_data),
"growth_potential": calculate_growth_score(trend_data),
"competition_level": calculate_competition_score(report_data),
"entry_barrier": calculate_barrier_score(report_data),
"profit_margin": calculate_profit_score(report_data)
}
Five-Dimension Scoring Model:
| Dimension | Points | Criteria |
|---|
| Market Scale | 30 | Total sales volume, search volume, top seller performance |
| Growth Potential | 20 | YoY growth rate, trend momentum, seasonality |
| Competition Level | 20 | Number of sellers, review count distribution, brand concentration |
| Entry Barrier | 15 | Average review count, established brand presence, capital requirements |
| Profit Margin | 15 | Price range, cost structure, margin opportunity |
Output:
category-reports/{CATEGORY}_{SITE}_{DATE}/report.mdcategory-reports/{CATEGORY}_{SITE}_{DATE}/dashboard.htmlcategory-reports/{CATEGORY}_{SITE}_{DATE}/category_report.xlsx
3. Keyword Research (keyword-research)
Deep keyword research with 8-dimension intelligent classification for 1500+ keywords.
/keyword-research B0D9ZTW7PS US
Python Implementation Pattern:
def research_keywords(asin: str, site: str, mcp_url: str):
"""
8-dimension keyword classification with ad strategy guidance
"""
traffic_keywords = requests.post(
mcp_url,
json={
"method": "product_traffic_terms",
"params": {"asin": asin, "site": site, "limit": 100}
}
).json()
all_keywords = []
for kw in traffic_keywords[:10]:
related = requests.post(
mcp_url,
json={
"method": "keyword_related_words",
"params": {"keyword": kw['keyword'], "site": site}
}
).json()
all_keywords.extend(related)
for kw in all_keywords:
detail = requests.post(
mcp_url,
json={
"method": "keyword_detail",
"params": {"keyword": kw['keyword'], "site": site}
}
).json()
kw.update(detail)
classified = classify_keywords_8_dimensions(all_keywords)
report_dir = f"keyword-reports/{asin}_{site}_{datetime.now().strftime('%Y%m%d')}"
os.makedirs(report_dir, exist_ok=True)
save_keywords_csv(f"{report_dir}/keywords.csv", classified)
with open(f"{report_dir}/negative_words.txt", 'w') as f:
f.write('\n'.join([kw['keyword'] for kw in classified['NEGATIVE']]))
for category, keywords in classified.items():
save_keywords_csv(f"{report_dir}/keywords_{category.lower()}.csv", keywords)
generate_keyword_dashboard(f"{report_dir}/dashboard.html", classified)
return classified
def classify_keywords_8_dimensions(keywords: list) -> dict:
"""
8-dimension intelligent classification
"""
categories = {
"NEGATIVE": [],
"BRAND": [],
"MATERIAL": [],
"SCENARIO": [],
"ATTRIBUTE": [],
"FUNCTION": [],
"CORE": [],
"OTHER": []
}
for kw in keywords:
category = classify_single_keyword(kw)
categories[category].append(kw)
return categories
8-Dimension Classification:
| Dimension | Use Case | Ad Strategy |
|---|
| NEGATIVE | Irrelevant/sensitive terms | Direct negation in campaigns |
| BRAND | Competitor brand names | Competitor targeting or negation |
| MATERIAL | Material descriptors (e.g., "stainless steel") | Exact match campaigns |
| SCENARIO | Use case keywords (e.g., "outdoor camping") | Scene-based ad groups |
| ATTRIBUTE | Attribute modifiers (e.g., "waterproof") | Long-tail exact match |
| FUNCTION | Functional keywords (e.g., "fast charging") | Broad match for discovery |
| CORE | Core product terms (e.g., "bluetooth speaker") | High bid, top-of-search placement |
| OTHER | Miscellaneous | Supplementary targeting |
Output:
keyword-reports/{ASIN}_{SITE}_{DATE}/report.mdkeyword-reports/{ASIN}_{SITE}_{DATE}/keywords.csvkeyword-reports/{ASIN}_{SITE}_{DATE}/negative_words.txtkeyword-reports/{ASIN}_{SITE}_{DATE}/keywords_{category}.csvkeyword-reports/{ASIN}_{SITE}_{DATE}/dashboard.html
4. Review Analysis (review-analysis)
6-dimension pain point analysis with service risk warnings.
/review-analysis B0DZCBYCNY US
Python Implementation Pattern:
def analyze_reviews(asin: str, site: str, mcp_url: str):
"""
6-dimension pain point analysis with service risk assessment
"""
reviews = requests.post(
mcp_url,
json={
"method": "product_reviews",
"params": {
"asin": asin,
"site": site,
"limit": 100,
"rating_filter": "1,2,3"
}
}
).json()
pain_points = analyze_six_dimension_pain_points(reviews)
service_risks = calculate_service_risks(reviews)
suggestions = generate_improvement_suggestions(pain_points, service_risks)
report_dir = f"review-analysis-reports/{asin}_{site}_{datetime.now().strftime('%Y%m%d')}"
os.makedirs(f"{report_dir}/data", exist_ok=True)
with open(f"{report_dir}/data/raw_reviews_sse.txt", 'w') as f:
f.write(json.dumps(reviews, indent=2))
analysis_data = {
"pain_points": pain_points,
"service_risks": service_risks,
"suggestions": suggestions
}
with open(f"{report_dir}/data/negative_reviews_analysis.json", 'w') as f:
f.write(json.dumps(analysis_data, indent=2, ensure_ascii=False))
with open(f"{report_dir}/report.md", 'w') as f:
f.write(generate_review_analysis_report(analysis_data))
return analysis_data
def analyze_six_dimension_pain_points(reviews: list) -> dict:
"""
6-dimension pain point framework
"""
dimensions = {
"electronic_failure": [],
"structural_issues": [],
"design_defects": [],
"appearance_material": [],
"description_mismatch": [],
"service_logistics": []
}
for review in reviews:
categorized_issues = categorize_review_issues(review)
for dimension, issues in categorized_issues.items():
dimensions[dimension].extend(issues)
return dimensions
def calculate_service_risks(reviews: list) -> dict:
"""
Service risk assessment with threshold warnings
"""
total_reviews = len(reviews)
risk_metrics = {
"used_defective_items": 0,
"missing_parts": 0,
"return_difficulties": 0,
"customer_service_issues": 0
}
for review in reviews:
if "used" in review['text'].lower() or "defective" in review['text'].lower():
risk_metrics["used_defective_items"] += 1
if "missing" in review['text'].lower():
risk_metrics["missing_parts"] += 1
if "return" in review['text'].lower() and "difficult" in review['text'].lower():
risk_metrics["return_difficulties"] += 1
if "customer service" in review['text'].lower() and review['rating'] <= 2:
risk_metrics["customer_service_issues"] += 1
risk_assessment = {}
for metric, count in risk_metrics.items():
percentage = (count / total_reviews) * 100
risk_level = determine_risk_level(metric, percentage)
risk_assessment[metric] = {
"count": count,
"percentage": percentage,
"risk_level": risk_level
}
return risk_assessment
6-Dimension Pain Point Framework:
| Dimension | Issues Identified | Severity Assessment |
|---|
| Electronic Failure | Battery, charging, connectivity, feature malfunction | High/Medium/Low |
| Structural Issues | Parts broken, seal failure, connector breaks | High/Medium/Low |
| Design Defects | Software UX, complexity, missing features | High/Medium/Low |
| Appearance/Material | Odor, allergies, color mismatch, scratches | High/Medium/Low |
| Description Mismatch | Feature gaps, size/color differences | High/Medium/Low |
| Service/Logistics | Used/defective items, missing parts, return difficulties | High/Medium/Low |
Service Risk Thresholds:
| Issue Type | Warning | Danger |
|---|
| Used/Defective Items | >2% | >5% |
| Missing Parts | >1% | >3% |
| Return Difficulties | >5% | >10% |
| Customer Service Issues | >3% | >7% |
Output:
review-analysis-reports/{ASIN}_{SITE}_{DATE}/report.mdreview-analysis-reports/{ASIN}_{SITE}_{DATE}/data/raw_reviews_sse.txtreview-analysis-reports/{ASIN}_{SITE}_{DATE}/data/negative_reviews_analysis.json
5. Product Research (product-research)
LLM-driven deep research workflow from data collection to decision.
/product-research "bluetooth speaker" US
/product-research "laptop backpack" GB
Python Implementation Pattern:
def product_research_workflow(keyword: str, site: str, mcp_url: str):
"""
LLM-driven product research workflow
Step 1: Information Collection
Step 2: Data Acquisition
Step 3: Attribute Tagging
Step 4: Cross Analysis
Step 5: Competitor & VOC
Step 6: Evaluation & Decision
Step 7: Report Generation
"""
products = requests.post(
mcp_url,
json={
"method": "product_search",
"params": {
"keyword": keyword,
"site": site,
"limit": 50
}
}
).json()
product_data = []
for product in products[:20]:
detail = get_product_complete_data(product['asin'], site, mcp_url)
product_data.append(detail)
analysis_results = {
"attribute_analysis": llm_attribute_tagging(product_data),
"cross_analysis": llm_cross_analysis(product_data),
"competitor_analysis": llm_competitor_analysis(product_data),
"voc_analysis": llm_voc_analysis(product_data),
"market_decision": llm_market_decision(product_data)
}
report_dir = f"product-research-reports/{keyword.replace(' ', '_')}_{site}_{datetime.now().strftime('%Y%m%d')}"
os.makedirs(report_dir, exist_ok=True)
with open(f"{report_dir}/data.json", 'w') as f:
f.write(json.dumps({
"products": product_data,
"analysis": analysis_results
}, indent=2, ensure_ascii=False))
report_markdown = llm_generate_report(product_data, analysis_results)
with open(f"{report_dir}/report.md", 'w') as f:
f.write(report_markdown)
generate_research_dashboard(f"{report_dir}/dashboard.html", product_data, analysis_results)
return analysis_results
def get_product_complete_data(asin: str, site: str, mcp_url: str) -> dict:
"""
Collect all data points for a product
"""
return {
"detail": get_product_detail(asin, site, mcp_url),
"trends": get_product_trends(asin, site, mcp_url),
"keywords": get_product_keywords(asin, site, mcp_url),
"reviews": get_product_reviews(asin, site, mcp_url),
"tiktok_videos": get_tiktok_videos(asin, site, mcp_url),
"supply_chain": get_1688_suppliers(asin, mcp_url)
}
Research Workflow Phases:
- Information Collection: Search and identify candidate products
- Data Acquisition: Collect comprehensive data (details, trends, keywords, reviews)
- Attribute Tagging: LLM extracts features, materials, use cases
- Cross Analysis: Compare products across dimensions
- Competitor & VOC: Analyze competitor strategies and customer voice
- Evaluation & Decision: LLM provides market entry recommendation
- Report Generation: Markdown report + structured data + dashboard
Output:
product-research-reports/{KEYWORD}_{SITE}_{DATE}/report.mdproduct-research-reports/{KEYWORD}_{SITE}_{DATE}/data.jsonproduct-research-reports/{KEYWORD}_{SITE}_{DATE}/dashboard.html
6. Sif Amazon Research (sif-amazon-research)
Market validation, competitor analysis, traffic diagnostics using Sif MCP.
/sif-amazon-research
Sif Research Decision Types:
| Decision Type | Purpose | Key Questions |
|---|
| Market Entry | Should we enter this market? | Is there demand? Can we compete? |
| Product Launch | Should we launch this product? | Is validation complete? Ready to scale? |
| Growth Strategy | Should we expand or optimize? | Double down or pause? |
| Root Cause Analysis | Why did performance change? | Traffic drop? Ad issue? Competition? |
Python Implementation Pattern:
def sif_research_workflow(decision_type: str, params: dict, sif_mcp_url: str):
"""
Sif MCP research with evidence-based decision framework
"""
context = {
"decision_type": decision_type,
"candidate_asin": params.get("asin"),
"keywords": params.get("keywords", []),
"site": params.get("site", "US")
}
evidence = collect_sif_evidence(context, sif_mcp_url)
validated_evidence = cross_validate_evidence(evidence)
decision = generate_business_decision(validated_evidence, context)
report_dir = f"sif-research-reports/{decision_type}_{datetime.now().strftime('%Y%m%d%H%M%S')}"
os.makedirs(report_dir, exist_ok=True)
with open(f"{report_dir}/decision_report.md", 'w') as f:
f.write(format_sif_decision_report(decision, validated_evidence))
return decision
def collect_sif_evidence(context: dict, sif_mcp_url: str) -> dict:
"""
Collect evidence from Sif MCP service
"""
evidence = {}
if context.get("keywords"):
evidence["market_demand"] = requests.post(
sif_mcp_url,
json={
"method": "keyword_market_research",
"params": {"keywords": context["keywords"], "site": context["site"]}
}
).json()
if context.get("candidate_asin"):
evidence["competition"] = requests.post(
sif_mcp_url,
json={
"method": "competitor_landscape",
"params": {"asin": context["candidate_asin"], "site": context["site"]}
}
).json()
if context.get("candidate_asin"):
evidence["traffic_structure"] = requests.post(
sif_mcp_url,
json={
"method": "traffic_source_analysis",
"params": {"asin": context["candidate_asin"], "site": context["site"]}
}
).json()
if context.get("candidate_asin"):
evidence["sales_trends"] = requests.post(
sif_mcp_url,
json={
"method": "sales_trend_analysis",
"params": {"asin": context["candidate_asin"], "site": context["site"]}
}
).json()
if context.get("candidate_asin"):
evidence["ad_performance"] = requests.post(
sif_mcp_url,
json={
"method": "ad_performance_analysis",
"params": {"asin": context["candidate_asin"], "site": context["site"]}
}
).json()
return evidence
def generate_business_decision(evidence: dict, context: dict) -> dict:
"""
Generate business decision with confidence level
"""
return {
"decision": "ENTER" | "DO_NOT_ENTER" | "LAUNCH" | "DO_NOT_LAUNCH" | "EXPAND" | "OPTIMIZE" | "STOP",
"confidence": 0.85,
"evidence_chain": [
{"finding": "Market demand validated", "confidence": 0.9, "source": "keyword_market_research"},
{"finding": "Competition moderate", "confidence": 0.8, "source": "competitor_landscape"}
],
"risks": [
{"risk": "Seasonal demand pattern", "severity": "MEDIUM", "mitigation": "Plan inventory accordingly"}
],
"missing_data": [
{"data": "Supplier MOQ requirements", "impact": "MEDIUM", "action": "Contact 1688 suppliers"}
],
"next_actions": [
{"action": "Validate supplier pricing", "priority": "HIGH", "deadline": "3 days"},
{"action": "Test ad creative variants", "priority": "MEDIUM", "deadline": "1 week"}
]
}
7. Xiyou Insight (xiyou-insight)
7-scenario analysis workflows for ad monitoring, traffic gaps, and competitor intelligence.
/xiyou-insight --scenario ad_monitoring --asin B07PWTJ4H1 --site US --keyword "bluetooth speaker"
/xiyou-insight --scenario traffic_gap --own_asin B07PWTJ4H1 --competitor_asins B08XYZ,B09ABC --site US
/xiyou-insight --scenario competitor_analysis --asin B08N5WRWNW --site US
/xiyou-insight --scenario new_product --asin B0D9ZTW7PS --site US
/xiyou-insight --scenario ad_budget --asin B07PWTJ4H1 --site US
/xiyou-insight --scenario keyword_database --site US --competitor_asins B07ABC,B08XYZ --core_keywords "wireless earbuds,bluetooth headphones"
7 Xiyou Scenarios:
| Scenario | Purpose | Key Tools |
|---|
| Ad Monitoring | Hourly rank tracking, ad placement effectiveness | get_asin_keyword_rank_hourly, get_asin_ad_change_trends |
| Traffic Gap | Find keywords where competitors rank but you don't | get_asin_keywords, cross-ASIN comparison |
| Competitor Analysis | Deconstruct competitor traffic and ad strategy | get_asin_traffic, get_asin_keywords, get_asin_ad_change_trends |
| New Product Launch | Track traffic stage transitions and keyword momentum | `get_a |
