Digikala MCP Server

import httpx from typing import Annotated, Optional from pydantic import Field from mcp.server.fastmcp import FastMCP from mcp.types import ToolAnnotations, Icon # Configure HTTP client with timeout HTTP_TIMEOUT = 30.0 # Currency conversion constants RIAL_TO_TOOMAN = 10 # 1 Tooman = 10 Rials def tooman_to_rial(tooman: int) -> int: """Convert Tooman to Rial (multiply by 10)""" return tooman * RIAL_TO_TOOMAN def rial_to_tooman(rial: int) -> int: """Convert Rial to Tooman (divide by 10)""" return rial // RIAL_TO_TOOMAN mcp = FastMCP( name="Digikala Discovery MCP Server", instructions=""" # Digikala Product Discovery Assistant This MCP server enables intelligent product discovery on Digikala through iterative search and refinement. ## Discovery Workflow ### Step 1: Initial Query Optimization For each user request, consider 2-4 initial query variations and call `get_optimized_keywords_and_categories` for each to discover: - Optimized search keywords - Most relevant category IDs to narrow search scope - **IMPORTANT**: Always try BOTH Persian (Farsi) and English versions of queries, as the API responds differently and Farsi queries typically yield better results ### Step 2: Execute Searches Use `search_products` with keyword-category pairs from Step 1. The API requires both a category_id and keyword. Be strategic about: - **Use category_id**: ALWAYS provide the category_id obtained from Step 1 - the search API requires it in the URL path - **Keyword-category pairing**: Use the keyword and category_id combinations from get_optimized_keywords_and_categories results - **Bilingual search**: Execute searches with BOTH Persian and English keywords - they produce different results, with Farsi often being more comprehensive - **Sort options**: Choose appropriate sort (1=Relevance, 2=Price Low-High, 3=Price High-Low, 4=Newest, 5=Best Selling, 6=Most Viewed, 7=Highest Rated) - **Price filtering**: Use `price_min_tooman` and `price_max_tooman` to filter out irrelevant products, even if not explicitly requested by user - **Pagination**: Results are auto-paginated with 20 items per page ### Step 3: Visual/Descriptive Search (Optional) For clothing, accessories, wearables, and shoes, use `search_text_lenz` with 2-3 word descriptions of the product's visual appearance for exceptional results. ### Step 4: Iterative Refinement Carefully evaluate search results and determine next actions: - **Same query, next page**: Increment page number if current results are promising - **Same query, different sort**: Change sort option to explore different angles - **New search**: Try different query/category combinations - **Keep iterating**: Continue through multiple search calls to navigate closer to desired results **CRITICAL - Exhaustive Exploration**: NEVER stop after finding initial good candidates. Always: - Try multiple different query variations (at least 3-5 different approaches) - Explore different category paths and combinations - Use different sort orders to discover hidden gems - Check recommendations from promising products to find alternatives - Compare results across different traversal chains before concluding - Only stop when you're confident you've explored the major search paths and found the best available options ### Step 5: Deep Dive into Candidates When you find potentially matching products, use `get_product_details` to examine: - `review.description`: Expert/editorial review - `attributes[]`: Technical specifications (.title, .values) - `suggestion`: Buy recommendation stats (.count, .percent) - `comments_overview`: Customer sentiment (.overview, .advantages[], .disadvantages[]) **Quality Criteria**: Always prioritize products with: - Rating of 4.0 stars or higher (out of 5) - Sufficient rating count (at least 10 reviews for reliability) - Positive customer feedback (high recommended_percent in customer_feedback) - Good buy suggestion percentage ### Step 6: Expand Search with Recommendations If a product is close but not exact, use `get_product_recommendations` to discover similar products for further exploration. ## Presenting Results to Users **ALWAYS provide clickable product links**: When explaining or recommending products to users, format links as: - `[Product Title](https://www.digikala.com{url})` where `{url}` is the product's URI from the API response - Example: `[گوشی موبایل سامسونگ مدل Galaxy S23](https://www.digikala.com/product/dkp-12345678/)` - Include the link alongside product name in every mention or recommendation - This enables users to directly access and view the products you're recommending ## Key Principles - Start broad, narrow down iteratively - Always optimize queries first - **Search in both Persian (Farsi) and English** - API responds differently, Farsi often yields better results - Use multiple search strategies in parallel - **Prioritize quality products**: Focus on products with 4+ star ratings and positive customer feedback - **Never settle early**: Explore multiple query variations and traversal paths before concluding - **Always provide clickable links**: Format product recommendations as `[Title](https://www.digikala.com{url})` - Examine product details before making recommendations - Leverage recommendations to expand search space - Consider price ranges intelligently to filter noise """, website_url="https://www.digikala.com", icons=[Icon(src="https://www.digikala.com/favicon.ico", mimeType="image/x-icon")], ) @mcp.tool( name="get_optimized_keywords_and_categories", title="Get optimized keywords and categories", description="Get optimized search keywords and relevant categories. IMPORTANT: Always call this with BOTH Persian (Farsi) and English queries separately, as the API responds differently to each language and Farsi typically yields better, more comprehensive results.", annotations=ToolAnnotations( readOnlyHint=True, destructiveHint=False, idempotentHint=False, openWorldHint=True ), ) def get_optimized_keywords_and_categories(query: Annotated[str, Field(description="Single word query to get optimized keywords and categories for. Must be only one word (e.g., 'laptop', 'shoe', 'phone')")]) -> str: """ Get search autocomplete suggestions and related categories for a query. Args: query: Search query string to get suggestions for Returns: Formatted markdown with optimized queries and categories """ response = httpx.get( "https://api.digikala.com/v1/autocomplete/", params={"q": query}, timeout=HTTP_TIMEOUT ) response.raise_for_status() data = response.json() # Extract optimized keywords optimized_queries = [] for item in data.get("data", {}).get("auto_complete", []): keyword = item.get("keyword", "") if keyword and keyword not in optimized_queries: optimized_queries.append(keyword) # Extract categories with their queries categories = [] seen_categories = set() for item in data.get("data", {}).get("categories", []): category_data = item.get("category", {}) category_id = category_data.get("id") if category_id and category_id not in seen_categories: seen_categories.add(category_id) categories.append({ "id": category_id, "title_fa": category_data.get("title_fa", ""), "title_en": category_data.get("title_en", ""), "code": category_data.get("code", ""), "query": item.get("keyword", "") }) is_text_lenz_eligible = data.get("data", {}).get("is_text_lenz_eligible", False) # Format as markdown markdown = f"# Optimized Keywords & Categories for: {query}\n\n" # Optimized keywords section markdown += "## Optimized Keywords\n\n" if optimized_queries: for i, kw in enumerate(optimized_queries, 1): markdown += f"{i}. `{kw}`\n" else: markdown += "No optimized keywords found.\n" markdown += "\n" # Categories section markdown += "## Keyword-Category Pairs\n\n" if categories: markdown += "Use these pairs with `search_products(category_id, keyword)`:\n\n" for cat in categories: markdown += f"### {cat['title_fa']} ({cat['title_en']})\n" markdown += f"- **Category ID**: `{cat['id']}`\n" markdown += f"- **Keyword**: `{cat['query']}`\n" markdown += f"- **Code**: `{cat['code']}`\n\n" else: markdown += "No categories found.\n\n" # Text-Lenz eligibility if is_text_lenz_eligible: markdown += "---\n\n" markdown += "**Text-Lenz Eligible**: This query is eligible for AI-powered semantic search using `search_text_lenz`.\n" return markdown @mcp.tool( name="search_products", title="Search Products", description="Search for products within a specific category with filtering and sorting. Returns 20 products per page. CRITICAL WORKFLOW: ALWAYS call get_optimized_keywords_and_categories FIRST to get the category_id for your search - this API requires it. Use the keyword-category pairs from that response. Search with BOTH Persian (Farsi) and English keywords separately - they produce different results, with Farsi often being more comprehensive. All prices in TOOMAN (1 Tooman = 10 Rials).", annotations=ToolAnnotations( readOnlyHint=True, destructiveHint=False, idempotentHint=True, openWorldHint=True ), ) def search_products( category_id: Annotated[int, Field(description="Category ID to search within (REQUIRED - get this from get_optimized_keywords_and_categories)")], keyword: Annotated[str, Field(description="Search keyword string for products, retrieved from the get_optimized_keywords_and_categories tool results")], page: Annotated[int, Field(description="Page number (starts from 1)")] = 1, sort: Annotated[int, Field(description="Sort order: 1=Relevance, 2=Price Low-High, 3=Price High-Low, 4=Newest, 5=Best Selling, 6=Most Viewed, 7=Highest Rated")] = 1, price_min_tooman: Annotated[Optional[int], Field(description="Minimum price in Toomans (e.g., 1000 Toomans = 10,000 Rials)")] = None, price_max_tooman: Annotated[Optional[int], Field(description="Maximum price in Toomans (e.g., 5000 Toomans = 50,000 Rials)")] = None, discount_min: Annotated[Optional[int], Field(description="Minimum discount percentage")] = None, discount_max: Annotated[Optional[int], Field(description="Maximum discount percentage")] = None, colors: Annotated[Optional[list[int]], Field(description="List of color IDs to filter by")] = None ) -> str: """ Search for products on Digikala within a specific category with advanced filtering and sorting. IMPORTANT: This uses Search API V2 which requires a category_id in the URL path. Always call get_optimized_keywords_and_categories first to get relevant category IDs. NOTE: All price inputs/outputs use TOOMAN currency (1 Tooman = 10 Rials). The tool automatically converts to Rials for API calls and returns simplified product summaries. Args: category_id: Category ID to search within (get from get_optimized_keywords_and_categories) keyword: Search keyword string for products, retrieved from the get_optimized_keywords_and_categories tool results page: Page number (starts from 1) sort: Sort order (1=Relevance, 2=Price Low-High, 3=Price High-Low, 4=Newest, 5=Best Selling, 6=Most Viewed, 7=Highest Rated) price_min_tooman: Minimum price in Toomans (e.g., 1000 Toomans = 10,000 Rials) price_max_tooman: Maximum price in Toomans (e.g., 5000 Toomans = 50,000 Rials) discount_min: Minimum discount percentage discount_max: Maximum discount percentage colors: List of color IDs to filter by Returns: Formatted markdown with product listings and pagination info """ params = { "q": keyword, "page": page, "sort": sort } # Add color filter if colors: params["colors[]"] = colors # Add price range filter (convert Tooman to Rial for API) if price_min_tooman is not None: params["price[min]"] = tooman_to_rial(price_min_tooman) if price_max_tooman is not None: params["price[max]"] = tooman_to_rial(price_max_tooman) # Add discount range filter if discount_min is not None: params["discount[min]"] = discount_min if discount_max is not None: params["discount[max]"] = discount_max response = httpx.get( f"https://api.digikala.com/v2/category/{category_id}/", params=params, timeout=HTTP_TIMEOUT ) response.raise_for_status() search_result = response.json() # Extract and simplify products products = [] for widget in search_result.get('data', {}).get('widgets', []): if widget.get('type') == 'vertical_product_listing': for product_widget in widget.get('data', {}).get('widgets', []): if product_widget.get('type') == 'product': product_data = product_widget['data'] # Skip non-marketable products if product_data.get('status') != 'marketable': continue default_variant = product_data.get('default_variant', {}) price = default_variant.get('price', {}) rating = product_data.get('rating', {}) data_layer = product_data.get('data_layer', {}) digiplus = product_data.get('digiplus', {}) url_info = product_data.get('url', {}) selling_price_rial = price.get('selling_price', 0) rrp_price_rial = price.get('rrp_price', 0) # Convert rating to stars (divide by 20), only if count >= 10 rating_count = rating.get('count', 0) rating_rate = rating.get('rate', 0) rating_stars = round(rating_rate / 20, 1) if rating_count >= 10 else None products.append({ "id": product_data.get('id', 0), "title_fa": product_data.get('title_fa', ''), "title_en": product_data.get('title_en', ''), "url": url_info.get('uri', ''), "selling_price_rial": selling_price_rial, "selling_price_tooman": rial_to_tooman(selling_price_rial), "rrp_price_rial": rrp_price_rial, "rrp_price_tooman": rial_to_tooman(rrp_price_rial), "discount_percent": price.get('discount_percent', 0), "rating_stars": rating_stars, "rating_count": rating_count, "brand": data_layer.get('brand', ''), "category": data_layer.get('category', ''), "is_incredible": price.get('is_incredible', False), "is_digiplus_jet_eligible": digiplus.get('is_jet_eligible', False) }) # Determine if there are more pages has_more = len(products) == 20 # Format as markdown sort_names = ["Relevance", "Price Low-High", "Price High-Low", "Newest", "Best Selling", "Most Viewed", "Highest Rated"] sort_name = sort_names[sort - 1] if 1 <= sort <= 7 else "Unknown" markdown = f"# Search Results: {keyword}\n\n" markdown += f"**Category ID**: `{category_id}` | **Page**: {page} | **Sort**: {sort_name}\n\n" if price_min_tooman or price_max_tooman: price_filter = "**Price Filter**: " if price_min_tooman: price_filter += f"{price_min_tooman:,} Tooman" if price_min_tooman and price_max_tooman: price_filter += " - " if price_max_tooman: price_filter += f"{price_max_tooman:,} Tooman" markdown += price_filter + "\n\n" markdown += f"**Found**: {len(products)} products on this page\n\n" markdown += "---\n\n" if not products: markdown += "No products found matching the criteria.\n" return markdown # List products for i, product in enumerate(products, 1): markdown += f"## {i}. {product['title_fa']}\n\n" markdown += f"**ID**: `{product['id']}`\n\n" # Price markdown += f"**Price**: {product['selling_price_tooman']:,} Tooman ({product['selling_price_rial']:,} Rial)\n\n" if product['discount_percent'] > 0: markdown += f"**Discount**: {product['discount_percent']}% off (was {product['rrp_price_tooman']:,} Tooman)\n\n" # Rating if product['rating_stars']: stars = "⭐" * int(product['rating_stars']) markdown += f"**Rating**: {stars} {product['rating_stars']}/5 ({product['rating_count']} reviews)\n\n" else: markdown += f"**Rating**: Not enough reviews ({product['rating_count']} reviews)\n\n" # Brand and category if product['brand']: markdown += f"**Brand**: {product['brand']}\n\n" if product['category']: markdown += f"**Category**: {product['category']}\n\n" # Badges badges = [] if product['is_incredible']: badges.append("🔥 Incredible Offer") if product['is_digiplus_jet_eligible']: badges.append("⚡ DigiPlus Jet") if badges: markdown += f"**Badges**: {' | '.join(badges)}\n\n" # URL markdown += f"**Link**: [View Product](https://www.digikala.com{product['url']})\n\n" markdown += "---\n\n" # Pagination info if has_more: markdown += f"📄 **More products available** - Use `page={page + 1}` to see next page\n" else: markdown += "📄 **End of results** - No more pages available\n" return markdown @mcp.tool( name="get_product_details", title="Get Product Details", description="Get comprehensive product information including specifications, expert reviews, customer feedback, and buy suggestions. Use this when you need detailed information about a specific product.", annotations=ToolAnnotations( readOnlyHint=True, destructiveHint=False, idempotentHint=True, openWorldHint=True ), ) def get_product_details( product_id: Annotated[int, Field(description="Unique product identifier")] ) -> str: """ Get detailed product information including specifications, variants, and reviews. Args: product_id: Unique product identifier Returns: Formatted markdown with product details, specifications, reviews, and customer feedback """ response = httpx.get( f"https://api.digikala.com/v2/product/{product_id}/", timeout=HTTP_TIMEOUT ) response.raise_for_status() data = response.json() product = data.get('data', {}).get('product', {}) # Skip if not marketable if product.get('status') != 'marketable': return f"# Product Not Available\n\n**Product ID**: `{product_id}`\n\n**Status**: {product.get('status', 'unknown')}\n\nThis product is not available for purchase." # Extract basic info title_fa = product.get('title_fa', '') title_en = product.get('title_en', '') url = product.get('url', {}).get('uri', '') category_title = product.get('category', {}).get('title_fa', '') brand_title = product.get('brand', {}).get('title_fa', '') # Extract rating rating = product.get('rating', {}) rating_count = rating.get('count', 0) rating_rate = rating.get('rate', 0) rating_stars = round(rating_rate / 20, 1) if rating_count >= 10 else None # Extract pricing default_variant = product.get('default_variant', {}) price = default_variant.get('price', {}) selling_price_rial = price.get('selling_price', 0) rrp_price_rial = price.get('rrp_price', 0) selling_price_tooman = rial_to_tooman(selling_price_rial) rrp_price_tooman = rial_to_tooman(rrp_price_rial) discount_percent = price.get('discount_percent', 0) is_incredible = price.get('is_incredible', False) # Start building markdown markdown = f"# {title_fa}\n\n" if title_en: markdown += f"*{title_en}*\n\n" markdown += f"**Product ID**: `{product_id}`\n\n" markdown += f"[View on Digikala](https://www.digikala.com{url})\n\n" markdown += "---\n\n" # Price section markdown += "## Price Information\n\n" markdown += f"**Current Price**: {selling_price_tooman:,} Tooman ({selling_price_rial:,} Rial)\n\n" if discount_percent > 0: markdown += f"**Original Price**: ~~{rrp_price_tooman:,} Tooman~~\n\n" markdown += f"**Discount**: {discount_percent}% OFF 🔥\n\n" if is_incredible: markdown += f"**🔥 Incredible Offer**\n\n" # Rating section markdown += "## Rating & Reviews\n\n" if rating_stars: stars = "⭐" * int(rating_stars) markdown += f"**Rating**: {stars} **{rating_stars}/5**\n\n" markdown += f"**Total Reviews**: {rating_count:,}\n\n" else: markdown += f"**Rating**: Not enough reviews yet ({rating_count} reviews)\n\n" # Category and brand markdown += "## Product Information\n\n" if brand_title: markdown += f"**Brand**: {brand_title}\n\n" if category_title: markdown += f"**Category**: {category_title}\n\n" # Expert review review = product.get('review') if review and review.get('description'): markdown += "---\n\n" markdown += "## Expert Review\n\n" markdown += f"{review.get('description')}\n\n" # Specifications markdown += "---\n\n" markdown += "## Specifications\n\n" spec_count = 0 for spec_group in product.get('specifications', []): for attr in spec_group.get('attributes', []): title = attr.get('title', '') values = attr.get('values', []) if title and values: spec_count += 1 markdown += f"**{title}**: {', '.join(str(v) for v in values)}\n\n" if spec_count == 0: markdown += "No specifications available.\n\n" # Buy suggestion suggestion = product.get('suggestion') if suggestion: count = suggestion.get('count', 0) percent = suggestion.get('percent', 0) if count > 0: markdown += "---\n\n" markdown += "## Buy Recommendation\n\n" markdown += f"**{percent}%** of {count:,} customers recommend buying this product\n\n" # Customer feedback comments_overview = product.get('comments_overview', {}) if comments_overview: total_count = comments_overview.get('total_count', 0) recommended_percent = comments_overview.get('recommended_percent', 0) not_recommended_percent = comments_overview.get('not_recommended_percent', 0) if total_count > 0: markdown += "---\n\n" markdown += "## Customer Feedback\n\n" markdown += f"**Total Comments**: {total_count:,}\n\n" markdown += f"**Recommended**: {recommended_percent}% 👍\n\n" markdown += f"**Not Recommended**: {not_recommended_percent}% 👎\n\n" # Extract advantages and disadvantages advantages = [] disadvantages = [] for comment in product.get('last_comments', [])[:5]: advantages.extend(comment.get('advantages', [])) disadvantages.extend(comment.get('disadvantages', [])) advantages = list(set(advantages))[:10] disadvantages = list(set(disadvantages))[:10] if advantages: markdown += "### Top Advantages\n\n" for adv in advantages: markdown += f"- ✅ {adv}\n" markdown += "\n" if disadvantages: markdown += "### Top Disadvantages\n\n" for dis in disadvantages: markdown += f"- ❌ {dis}\n" markdown += "\n" # Variant information variants = [] for variant in product.get('variants', []): if variant.get('status') == 'marketable': v_price = variant.get('price', {}) v_selling_rial = v_price.get('selling_price', 0) variants.append({ "color": variant.get('color', {}).get('title', '') if variant.get('color') else None, "selling_price_tooman": rial_to_tooman(v_selling_rial), "discount_percent": v_price.get('discount_percent', 0) }) if len(variants) > 1: markdown += "---\n\n" markdown += f"## Available Variants ({len(variants)})\n\n" prices = [v["selling_price_tooman"] for v in variants] markdown += f"**Price Range**: {min(prices):,} - {max(prices):,} Tooman\n\n" for variant in variants: if variant["color"]: markdown += f"- **{variant['color']}**: {variant['selling_price_tooman']:,} Tooman" if variant['discount_percent'] > 0: markdown += f" ({variant['discount_percent']}% off)" markdown += "\n" markdown += "\n" return markdown @mcp.tool( name="get_product_recommendations", title="Get Product Recommendations", description="Get similar products and related category recommendations. Use this when a product is close but not exactly what you need, to explore similar alternatives.", annotations=ToolAnnotations( readOnlyHint=True, destructiveHint=False, idempotentHint=True, openWorldHint=True ), ) def get_product_recommendations( product_id: Annotated[int, Field(description="Product ID for which recommendations are needed")], offset: Annotated[Optional[int], Field(description="Optional recommendation tab/offset to retrieve specific recommendation type")] = None ) -> str: """ Get tabular product recommendations (similar products, related categories). Args: product_id: Product ID for which recommendations are needed offset: Optional recommendation tab/offset to retrieve specific recommendation type Returns: Formatted markdown with recommended products """ params = {} if offset is not None: params["offset"] = offset response = httpx.get( f"https://api.digikala.com/v1/product/{product_id}/tabular-recommendation/", params=params, timeout=HTTP_TIMEOUT ) response.raise_for_status() data = response.json() recommendation_type = data.get('data', {}).get('type', '') title = data.get('data', {}).get('data', {}).get('title', '') # Extract available recommendation tabs available_tabs = [] meta = data.get('data', {}).get('meta', {}) for offset_info in meta.get('offsets', []): available_tabs.append({ "offset": offset_info.get('offset'), "type": offset_info.get('type', ''), "title": offset_info.get('title', '') }) # Extract and simplify recommended products products = [] for product in data.get('data', {}).get('data', {}).get('products', []): # Skip non-marketable products if product.get('status') != 'marketable': continue default_variant = product.get('default_variant', {}) price = default_variant.get('price', {}) rating = product.get('rating', {}) data_layer = product.get('data_layer', {}) digiplus = product.get('digiplus', {}) url_info = product.get('url', {}) selling_price_rial = price.get('selling_price', 0) rrp_price_rial = price.get('rrp_price', 0) # Convert rating to stars rating_count = rating.get('count', 0) rating_rate = rating.get('rate', 0) rating_stars = round(rating_rate / 20, 1) if rating_count >= 10 else None products.append({ "id": product.get('id', 0), "title_fa": product.get('title_fa', ''), "title_en": product.get('title_en', ''), "url": url_info.get('uri', ''), "selling_price_rial": selling_price_rial, "selling_price_tooman": rial_to_tooman(selling_price_rial), "rrp_price_rial": rrp_price_rial, "rrp_price_tooman": rial_to_tooman(rrp_price_rial), "discount_percent": price.get('discount_percent', 0), "rating_stars": rating_stars, "rating_count": rating_count, "brand": data_layer.get('brand', ''), "category": data_layer.get('category', ''), "is_incredible": price.get('is_incredible', False), "is_digiplus_jet_eligible": digiplus.get('is_jet_eligible', False) }) # Format as markdown markdown = f"# Product Recommendations\n\n" markdown += f"**Product ID**: `{product_id}`\n\n" if title: markdown += f"**Recommendation Type**: {title}\n\n" # Available tabs if available_tabs: markdown += "## Available Recommendation Tabs\n\n" for tab in available_tabs: markdown += f"- **{tab['title']}** (offset: `{tab['offset']}`, type: `{tab['type']}`)\n" markdown += "\n" markdown += f"## Recommended Products ({len(products)})\n\n" markdown += "---\n\n" if not products: markdown += "No recommendations found.\n" return markdown # List products for i, product in enumerate(products, 1): markdown += f"### {i}. {product['title_fa']}\n\n" markdown += f"**ID**: `{product['id']}`\n\n" # Price markdown += f"**Price**: {product['selling_price_tooman']:,} Tooman ({product['selling_price_rial']:,} Rial)\n\n" if product['discount_percent'] > 0: markdown += f"**Discount**: {product['discount_percent']}% off\n\n" # Rating if product['rating_stars']: stars = "⭐" * int(product['rating_stars']) markdown += f"**Rating**: {stars} {product['rating_stars']}/5 ({product['rating_count']} reviews)\n\n" # Brand if product['brand']: markdown += f"**Brand**: {product['brand']}\n\n" # URL markdown += f"[View Product](https://www.digikala.com{product['url']})\n\n" markdown += "---\n\n" return markdown @mcp.tool( name="search_text_lenz", title="Search with Text-Lenz AI", description="AI-powered semantic search using Text-Lenz. Exceptional for clothing, accessories, wearables, and shoes. Use 2-3 word visual descriptions (e.g., 'red summer dress', 'black running shoes'). Understands natural language and context.", annotations=ToolAnnotations( readOnlyHint=True, destructiveHint=False, idempotentHint=True, openWorldHint=True ), ) def search_text_lenz( query: Annotated[str, Field(description="Visual/descriptive query (2-3 words work best, e.g., 'blue cotton shirt')")], page: Annotated[int, Field(description="Page number for pagination (default: 1)")] = 1 ) -> str: """ Perform AI-powered semantic search using Text-Lenz technology. Text-Lenz enhances traditional search by understanding natural language queries, context, and visual descriptions for more accurate results. Args: query: Visual/descriptive search query (works best with 2-3 word descriptions) page: Page number for pagination (default: 1) Returns: Formatted markdown with products and related suggestions """ params = { "q": query, "page": page } response = httpx.get( "https://api.digikala.com/v1/search/text-lenz/", params=params, timeout=HTTP_TIMEOUT ) response.raise_for_status() data = response.json() search_data = data.get('data', {}) is_text_lenz_eligible = search_data.get('is_text_lenz_eligible', False) related_searches = search_data.get('related_search_words', []) # Pagination info pager = search_data.get('pager', {}) current_page = pager.get('current_page', 1) total_pages = pager.get('total_pages', 1) total_items = pager.get('total_items', 0) has_more = current_page < total_pages # Extract and simplify products products = [] for product in search_data.get('products', []): # Skip non-marketable products if product.get('status') != 'marketable': continue default_variant = product.get('default_variant', {}) price = default_variant.get('price', {}) rating = product.get('rating', {}) data_layer = product.get('data_layer', {}) digiplus = product.get('digiplus', {}) url_info = product.get('url', {}) selling_price_rial = price.get('selling_price', 0) rrp_price_rial = price.get('rrp_price', 0) # Convert rating to stars rating_count = rating.get('count', 0) rating_rate = rating.get('rate', 0) rating_stars = round(rating_rate / 20, 1) if rating_count >= 10 else None products.append({ "id": product.get('id', 0), "title_fa": product.get('title_fa', ''), "title_en": product.get('title_en', ''), "url": url_info.get('uri', ''), "selling_price_rial": selling_price_rial, "selling_price_tooman": rial_to_tooman(selling_price_rial), "rrp_price_rial": rrp_price_rial, "rrp_price_tooman": rial_to_tooman(rrp_price_rial), "discount_percent": price.get('discount_percent', 0), "rating_stars": rating_stars, "rating_count": rating_count, "brand": data_layer.get('brand', ''), "category": data_layer.get('category', ''), "is_incredible": price.get('is_incredible', False), "is_digiplus_jet_eligible": digiplus.get('is_jet_eligible', False) }) # Format as markdown markdown = f"# Text-Lenz Search: {query}\n\n" markdown += f"🔍 **AI-Powered Semantic Search** | Page {current_page}/{total_pages}\n\n" if is_text_lenz_eligible: markdown += "✅ **Text-Lenz Active** - Using AI-enhanced search\n\n" markdown += f"**Total Results**: {total_items:,} | **Showing**: {len(products)} products\n\n" # Related searches if related_searches: markdown += "## Related Searches\n\n" for rs in related_searches: markdown += f"- {rs}\n" markdown += "\n" markdown += "---\n\n" if not products: markdown += "No products found matching the criteria.\n" return markdown # List products for i, product in enumerate(products, 1): markdown += f"## {i}. {product['title_fa']}\n\n" markdown += f"**ID**: `{product['id']}`\n\n" # Price markdown += f"**Price**: {product['selling_price_tooman']:,} Tooman ({product['selling_price_rial']:,} Rial)\n\n" if product['discount_percent'] > 0: markdown += f"**Discount**: {product['discount_percent']}% off (was {product['rrp_price_tooman']:,} Tooman)\n\n" # Rating if product['rating_stars']: stars = "⭐" * int(product['rating_stars']) markdown += f"**Rating**: {stars} {product['rating_stars']}/5 ({product['rating_count']} reviews)\n\n" else: markdown += f"**Rating**: Not enough reviews ({product['rating_count']} reviews)\n\n" # Brand and category if product['brand']: markdown += f"**Brand**: {product['brand']}\n\n" if product['category']: markdown += f"**Category**: {product['category']}\n\n" # Badges badges = [] if product['is_incredible']: badges.append("🔥 Incredible Offer") if product['is_digiplus_jet_eligible']: badges.append("⚡ DigiPlus Jet") if badges: markdown += f"**Badges**: {' | '.join(badges)}\n\n" # URL markdown += f"**Link**: [View Product](https://www.digikala.com{product['url']})\n\n" markdown += "---\n\n" # Pagination info if has_more: markdown += f"📄 **More products available** - Use `page={current_page + 1}` to see next page\n" else: markdown += "📄 **End of results** - No more pages available\n" return markdown if __name__ == "__main__": mcp.run()