amazon-sorftime-research-mcp-skill

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.

# Analyze US marketplace listing
/amazon-analyse B07PWTJ4H1 US

# Analyze German marketplace listing
/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
    """
    # 1. Get product details
    product_detail = requests.post(
        mcp_url,
        json={
            "method": "product_detail",
            "params": {"asin": asin, "site": site}
        }
    ).json()
    
    # 2. Get traffic keywords (top 100)
    traffic_keywords = requests.post(
        mcp_url,
        json={
            "method": "product_traffic_terms",
            "params": {"asin": asin, "site": site, "limit": 100}
        }
    ).json()
    
    # 3. Get product reviews (last 100)
    reviews = requests.post(
        mcp_url,
        json={
            "method": "product_reviews",
            "params": {"asin": asin, "site": site, "limit": 100}
        }
    ).json()
    
    # 4. Get product trends (90 days)
    trends = requests.post(
        mcp_url,
        json={
            "method": "product_trend",
            "params": {
                "asin": asin,
                "site": site,
                "days": 90
            }
        }
    ).json()
    
    # 5. Get competitor keyword layout
    competitor_keywords = requests.post(
        mcp_url,
        json={
            : ,
            : {: asin, : site}
        }
    ).json()
    
    
    report_data = {
        : asin,
        : site,
        : product_detail,
        : traffic_keywords,
        : reviews,
        : trends,
        : competitor_keywords,
        : datetime.now().isoformat()
    }
    
    
    report_path = 
     (report_path, )  f:
        f.write(generate_markdown_report(report_data))
    
     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.

# Analyze US Sofas & Couches category
/category-selection "Sofas & Couches" US

# Analyze specific number of products
/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
    """
    # 1. Search for category nodeId
    category_search = requests.post(
        mcp_url,
        json={
            "method": "category_name_search",
            "params": {"query": category_name, "site": site}
        }
    ).json()
    
    node_id = category_search[0]['nodeId']
    
    # 2. Get category report (Top products)
    category_report = requests.post(
        mcp_url,
        json={
            "method": "category_report",
            "params": {
                "nodeId": node_id,
                "site": site,
                "limit": limit
            }
        }
    ).json()
    
    # 3. Get category trends
    category_trends = requests.post(
        mcp_url,
        json={
            "method": "category_trend",
            "params": {"nodeId": node_id, "site": site}
        }
    ).json()
    
    # 4. Calculate five-dimension scores
    scores = calculate_five_dimension_scores(category_report, category_trends)
    
    # 5. Generate multi-format reports
    report_dir = f"category-reports/{category_name}_{site}_{datetime.now().strftime('%Y%m%d')}"
    os.makedirs(report_dir, exist_ok=True)
    
    # Save Markdown report
     (, )  f:
        f.write(generate_category_markdown(category_report, scores))
    
    
    generate_excel_report(, category_report, scores)
    
    
    generate_html_dashboard(, category_report, scores)
    
     scores

 ():
    
     {
        : calculate_market_scale_score(report_data),  
        : calculate_growth_score(trend_data),      
        : calculate_competition_score(report_data), 
        : calculate_barrier_score(report_data),       
        : 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.md
• category-reports/{CATEGORY}_{SITE}_{DATE}/dashboard.html
• category-reports/{CATEGORY}_{SITE}_{DATE}/category_report.xlsx

3. Keyword Research (keyword-research)

Deep keyword research with 8-dimension intelligent classification for 1500+ keywords.

# Research keywords for an ASIN
/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
    """
    # 1. Get traffic keywords
    traffic_keywords = requests.post(
        mcp_url,
        json={
            "method": "product_traffic_terms",
            "params": {"asin": asin, "site": site, "limit": 100}
        }
    ).json()
    
    # 2. Expand related keywords for each traffic term
    all_keywords = []
    for kw in traffic_keywords[:10]:  # Top 10 keywords
        related = requests.post(
            mcp_url,
            json={
                "method": "keyword_related_words",
                "params": {"keyword": kw['keyword'], "site": site}
            }
        ).json()
        all_keywords.extend(related)
    
    # 3. Get keyword details (search volume, CPC)
    for kw in all_keywords:
        detail = requests.post(
            mcp_url,
            json={
                "method": "keyword_detail",
                "params": {"keyword": kw['keyword'], "site": site}
            }
        ).json()
        kw.update(detail)
    
    # 4. Classify keywords into 8 dimensions
    classified = classify_keywords_8_dimensions(all_keywords)
    
    # 5. Generate reports
    report_dir = f"keyword-reports/{asin}_{site}_"
    os.makedirs(report_dir, exist_ok=)
    
    
    save_keywords_csv(, classified)
    
    
     (, )  f:
        f.write(.join([kw[]  kw  classified[]]))
    
    
     category, keywords  classified.items():
        save_keywords_csv(, keywords)
    
    
    generate_keyword_dashboard(, classified)
    
     classified

 () -> :
    
    categories = {
        : [],    
        : [],       
        : [],    
        : [],    
        : [],   
        : [],    
        : [],        
        : []        
    }
    
     kw  keywords:
        category = classify_single_keyword(kw)
        categories[category].append(kw)
    
     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.md
• keyword-reports/{ASIN}_{SITE}_{DATE}/keywords.csv
• keyword-reports/{ASIN}_{SITE}_{DATE}/negative_words.txt
• keyword-reports/{ASIN}_{SITE}_{DATE}/keywords_{category}.csv
• keyword-reports/{ASIN}_{SITE}_{DATE}/dashboard.html

4. Review Analysis (review-analysis)

6-dimension pain point analysis with service risk warnings.

# Analyze product reviews
/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
    """
    # 1. Get product reviews (negative reviews priority)
    reviews = requests.post(
        mcp_url,
        json={
            "method": "product_reviews",
            "params": {
                "asin": asin,
                "site": site,
                "limit": 100,
                "rating_filter": "1,2,3"  # Low ratings only
            }
        }
    ).json()
    
    # 2. Analyze pain points in 6 dimensions
    pain_points = analyze_six_dimension_pain_points(reviews)
    
    # 3. Calculate service risk metrics
    service_risks = calculate_service_risks(reviews)
    
    # 4. Generate improvement suggestions
    suggestions = generate_improvement_suggestions(pain_points, service_risks)
    
    # 5. Save reports
    report_dir = f"review-analysis-reports/{asin}_{site}_{datetime.now().strftime('%Y%m%d')}"
    os.makedirs(f"{report_dir}/data", exist_ok=True)
    
    # Save raw SSE response
    with open(f"{report_dir}/data/raw_reviews_sse.txt", 'w') as f:
        f.write(json.dumps(reviews, indent=2))
    
    # Save structured analysis
    analysis_data = {
        "pain_points": pain_points,
        : service_risks,
        : suggestions
    }
     (, )  f:
        f.write(json.dumps(analysis_data, indent=, ensure_ascii=))
    
    
     (, )  f:
        f.write(generate_review_analysis_report(analysis_data))
    
     analysis_data

 () -> :
    
    dimensions = {
        : [],      
        : [],       
        : [],          
        : [],     
        : [],    
        : []        
    }
    
     review  reviews:
        categorized_issues = categorize_review_issues(review)
         dimension, issues  categorized_issues.items():
            dimensions[dimension].extend(issues)
    
     dimensions

 () -> :
    
    total_reviews = (reviews)
    
    risk_metrics = {
        : ,      
        : ,             
        : ,       
        :     
    }
    
     review  reviews:
           review[].lower()    review[].lower():
            risk_metrics[] += 
           review[].lower():
            risk_metrics[] += 
           review[].lower()    review[].lower():
            risk_metrics[] += 
           review[].lower()  review[] <= :
            risk_metrics[] += 
    
    
    risk_assessment = {}
     metric, count  risk_metrics.items():
        percentage = (count / total_reviews) * 
        risk_level = determine_risk_level(metric, percentage)
        risk_assessment[metric] = {
            : count,
            : percentage,
            : risk_level
        }
    
     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.md
• review-analysis-reports/{ASIN}_{SITE}_{DATE}/data/raw_reviews_sse.txt
• review-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.

# Research bluetooth speaker market
/product-research "bluetooth speaker" US

# Research laptop backpack market
/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
    """
    # Step 1: Search products
    products = requests.post(
        mcp_url,
        json={
            "method": "product_search",
            "params": {
                "keyword": keyword,
                "site": site,
                "limit": 50
            }
        }
    ).json()
    
    # Step 2: Collect detailed data for top products
    product_data = []
    for product in products[:20]:
        detail = get_product_complete_data(product['asin'], site, mcp_url)
        product_data.append(detail)
    
    # Step 3-6: LLM analysis (pseudocode - actual LLM calls)
    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)
    }
    
    # Step 7: Generate reports
    report_dir = f"product-research-reports/{keyword.replace(' ', '_')}_{site}_{datetime.now().strftime('%Y%m%d')}"
    os.makedirs(report_dir, exist_ok=True)
    
    # Save structured data
     (, )  f:
        f.write(json.dumps({
            : product_data,
            : analysis_results
        }, indent=, ensure_ascii=))
    
    
    report_markdown = llm_generate_report(product_data, analysis_results)
     (, )  f:
        f.write(report_markdown)
    
    
    generate_research_dashboard(, product_data, analysis_results)
    
     analysis_results

 () -> :
    
     {
        : get_product_detail(asin, site, mcp_url),
        : get_product_trends(asin, site, mcp_url),
        : get_product_keywords(asin, site, mcp_url),
        : get_product_reviews(asin, site, mcp_url),
        : get_tiktok_videos(asin, site, mcp_url),
        : get_1688_suppliers(asin, mcp_url)
    }

Research Workflow Phases:

1. Information Collection: Search and identify candidate products
2. Data Acquisition: Collect comprehensive data (details, trends, keywords, reviews)
3. Attribute Tagging: LLM extracts features, materials, use cases
4. Cross Analysis: Compare products across dimensions
5. Competitor & VOC: Analyze competitor strategies and customer voice
6. Evaluation & Decision: LLM provides market entry recommendation
7. Report Generation: Markdown report + structured data + dashboard

Output:

• product-research-reports/{KEYWORD}_{SITE}_{DATE}/report.md
• product-research-reports/{KEYWORD}_{SITE}_{DATE}/data.json
• product-research-reports/{KEYWORD}_{SITE}_{DATE}/dashboard.html

6. Sif Amazon Research (sif-amazon-research)

Market validation, competitor analysis, traffic diagnostics using Sif MCP.

# Interactive research workflow
/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
    """
    # Step 1: Define decision context
    context = {
        "decision_type": decision_type,
        "candidate_asin": params.get("asin"),
        "keywords": params.get("keywords", []),
        "site": params.get("site", "US")
    }
    
    # Step 2: Collect Sif evidence
    evidence = collect_sif_evidence(context, sif_mcp_url)
    
    # Step 3: Cross-validate evidence
    validated_evidence = cross_validate_evidence(evidence)
    
    # Step 4: Generate business decision
    decision = generate_business_decision(validated_evidence, context)
    
    # Step 5: Output report
    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 = {}
    
    # Market demand evidence
    if context.get():
        evidence[] = requests.post(
            sif_mcp_url,
            json={
                : ,
                : {: context[], : context[]}
            }
        ).json()
    
    
     context.get():
        evidence[] = requests.post(
            sif_mcp_url,
            json={
                : ,
                : {: context[], : context[]}
            }
        ).json()
    
    
     context.get():
        evidence[] = requests.post(
            sif_mcp_url,
            json={
                : ,
                : {: context[], : context[]}
            }
        ).json()
    
    
     context.get():
        evidence[] = requests.post(
            sif_mcp_url,
            json={
                : ,
                : {: context[], : context[]}
            }
        ).json()
    
    
     context.get():
        evidence[] = requests.post(
            sif_mcp_url,
            json={
                : ,
                : {: context[], : context[]}
            }
        ).json()
    
     evidence

 () -> :
    
     {
        :  |  |  |  |  |  | ,
        : ,  
        : [
            {: , : , : },
            {: , : , : }
        ],
        : [
            {: , : , : }
        ],
        : [
            {: , : , : }
        ],
        : [
            {: , : , : },
            {: , : , : }
        ]
    }

7. Xiyou Insight (xiyou-insight)

7-scenario analysis workflows for ad monitoring, traffic gaps, and competitor intelligence.

# Ad monitoring workflow
/xiyou-insight --scenario ad_monitoring --asin B07PWTJ4H1 --site US --keyword "bluetooth speaker"

# Traffic gap analysis
/xiyou-insight --scenario traffic_gap --own_asin B07PWTJ4H1 --competitor_asins B08XYZ,B09ABC --site US

# Competitor analysis
/xiyou-insight --scenario competitor_analysis --asin B08N5WRWNW --site US

# New product launch tracking
/xiyou-insight --scenario new_product --asin B0D9ZTW7PS --site US

# Ad budget transparency
/xiyou-insight --scenario ad_budget --asin B07PWTJ4H1 --site US

# Keyword database building
/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