| """ |
| PrestaShop MCP Server |
| A Model Context Protocol server that provides tools for accessing PrestaShop product data. |
| """ |
|
|
| import asyncio |
| import json |
| import logging |
| from typing import Any, Dict, List, Optional |
| from urllib.parse import urlencode |
|
|
| import httpx |
| from mcp.server import Server |
| from mcp.types import ( |
| Resource, |
| Tool, |
| TextContent, |
| ImageContent, |
| EmbeddedResource, |
| CallToolRequest, |
| CallToolResult, |
| ListToolsRequest, |
| ListToolsResult, |
| ListResourcesRequest, |
| ListResourcesResult, |
| ReadResourceRequest, |
| ReadResourceResult, |
| ) |
|
|
| |
| logging.basicConfig(level=logging.INFO) |
| logger = logging.getLogger(__name__) |
|
|
| |
| DEFAULT_API_BASE = "https://shop.mcp.integration.ambris.com" |
| DEFAULT_TIMEOUT = 30 |
|
|
| class PrestaShopMCPServer: |
| """MCP Server for PrestaShop integration""" |
| |
| def __init__(self, api_base: str = DEFAULT_API_BASE): |
| self.api_base = api_base.rstrip('/') |
| self.server = Server("prestashop-mcp") |
| self.setup_handlers() |
| |
| def setup_handlers(self): |
| """Set up MCP server handlers""" |
| |
| @self.server.list_tools() |
| async def list_tools() -> List[Tool]: |
| """List available tools""" |
| return [ |
| Tool( |
| name="get_product_details", |
| description="Get detailed information about a PrestaShop product including name, description, price, and features", |
| inputSchema={ |
| "type": "object", |
| "properties": { |
| "product_id": { |
| "type": "integer", |
| "description": "The ID of the product to retrieve" |
| }, |
| "lang_id": { |
| "type": "integer", |
| "description": "Language ID for localized content", |
| "default": 1 |
| } |
| }, |
| "required": ["product_id"] |
| } |
| ), |
| Tool( |
| name="get_product_features", |
| description="Get product features and specifications from PrestaShop", |
| inputSchema={ |
| "type": "object", |
| "properties": { |
| "product_id": { |
| "type": "integer", |
| "description": "The ID of the product to get features for" |
| } |
| }, |
| "required": ["product_id"] |
| } |
| ), |
| Tool( |
| name="get_product_images", |
| description="Get product images and media from PrestaShop", |
| inputSchema={ |
| "type": "object", |
| "properties": { |
| "product_id": { |
| "type": "integer", |
| "description": "The ID of the product to get images for" |
| } |
| }, |
| "required": ["product_id"] |
| } |
| ), |
| Tool( |
| name="search_products", |
| description="Search for products in PrestaShop catalog", |
| inputSchema={ |
| "type": "object", |
| "properties": { |
| "query": { |
| "type": "string", |
| "description": "Search query term" |
| }, |
| "limit": { |
| "type": "integer", |
| "description": "Maximum number of results to return", |
| "default": 10 |
| }, |
| "category_id": { |
| "type": "integer", |
| "description": "Optional category ID to filter results" |
| } |
| }, |
| "required": ["query"] |
| } |
| ) |
| ] |
| |
| @self.server.call_tool() |
| async def call_tool(name: str, arguments: Dict[str, Any]) -> List[TextContent]: |
| """Handle tool calls""" |
| try: |
| if name == "get_product_details": |
| return await self._get_product_details(arguments) |
| elif name == "get_product_features": |
| return await self._get_product_features(arguments) |
| elif name == "get_product_images": |
| return await self._get_product_images(arguments) |
| elif name == "search_products": |
| return await self._search_products(arguments) |
| else: |
| return [TextContent( |
| type="text", |
| text=f"Unknown tool: {name}" |
| )] |
| except Exception as e: |
| logger.error(f"Error calling tool {name}: {e}") |
| return [TextContent( |
| type="text", |
| text=f"Error calling tool {name}: {str(e)}" |
| )] |
| |
| async def _make_api_request(self, endpoint: str, params: Dict[str, Any] = None) -> Dict[str, Any]: |
| """Make HTTP request to PrestaShop API""" |
| url = f"{self.api_base}/index.php" |
| |
| |
| default_params = { |
| "fc": "module", |
| "module": "ambaiagent", |
| "controller": "api" |
| } |
| |
| if params: |
| default_params.update(params) |
| |
| async with httpx.AsyncClient(timeout=DEFAULT_TIMEOUT) as client: |
| try: |
| response = await client.get(url, params=default_params) |
| response.raise_for_status() |
| |
| data = response.json() |
| if not data.get('success', False): |
| raise Exception(f"API error: {data.get('error', 'Unknown error')}") |
| |
| return data |
| |
| except httpx.TimeoutException: |
| raise Exception("Request timeout - PrestaShop API took too long to respond") |
| except httpx.HTTPStatusError as e: |
| raise Exception(f"HTTP error {e.response.status_code}: {e.response.text}") |
| except json.JSONDecodeError: |
| raise Exception("Invalid JSON response from PrestaShop API") |
| |
| async def _get_product_details(self, arguments: Dict[str, Any]) -> List[TextContent]: |
| """Get detailed product information""" |
| product_id = arguments.get("product_id") |
| lang_id = arguments.get("lang_id", 1) |
| |
| if not product_id: |
| return [TextContent( |
| type="text", |
| text="Error: product_id is required" |
| )] |
| |
| try: |
| data = await self._make_api_request("", { |
| "action": "get_product", |
| "product_id": product_id, |
| "lang_id": lang_id |
| }) |
| |
| product = data.get("product", {}) |
| |
| |
| result = f"""PRESTASHOP PRODUCT DETAILS (ID: {product_id}) |
| |
| Name: {product.get('name', 'N/A')} |
| Description: {product.get('description', 'N/A')} |
| Short Description: {product.get('description_short', 'N/A')} |
| Price: {product.get('price', 'N/A')} |
| Reference: {product.get('reference', 'N/A')} |
| EAN13: {product.get('ean13', 'N/A')} |
| Weight: {product.get('weight', 'N/A')} |
| Quantity: {product.get('quantity', 'N/A')} |
| Active: {product.get('active', 'N/A')} |
| Available for Order: {product.get('available_for_order', 'N/A')} |
| |
| Categories: {json.dumps(product.get('categories', []), indent=2)} |
| Features: {json.dumps(product.get('features', []), indent=2)} |
| Images: {json.dumps(product.get('images', []), indent=2)} |
| |
| Full Product Data: |
| {json.dumps(product, indent=2)}""" |
| |
| return [TextContent(type="text", text=result)] |
| |
| except Exception as e: |
| return [TextContent( |
| type="text", |
| text=f"Error retrieving product {product_id}: {str(e)}" |
| )] |
| |
| async def _get_product_features(self, arguments: Dict[str, Any]) -> List[TextContent]: |
| """Get product features""" |
| product_id = arguments.get("product_id") |
| |
| if not product_id: |
| return [TextContent( |
| type="text", |
| text="Error: product_id is required" |
| )] |
| |
| try: |
| data = await self._make_api_request("", { |
| "action": "get_product", |
| "product_id": product_id |
| }) |
| |
| product = data.get("product", {}) |
| features = product.get("features", []) |
| |
| if not features: |
| result = f"No features found for product {product_id}" |
| else: |
| result = f"PRESTASHOP PRODUCT FEATURES (ID: {product_id}):\n\n" |
| result += json.dumps(features, indent=2) |
| |
| return [TextContent(type="text", text=result)] |
| |
| except Exception as e: |
| return [TextContent( |
| type="text", |
| text=f"Error retrieving features for product {product_id}: {str(e)}" |
| )] |
| |
| async def _get_product_images(self, arguments: Dict[str, Any]) -> List[TextContent]: |
| """Get product images""" |
| product_id = arguments.get("product_id") |
| |
| if not product_id: |
| return [TextContent( |
| type="text", |
| text="Error: product_id is required" |
| )] |
| |
| try: |
| data = await self._make_api_request("", { |
| "action": "get_product", |
| "product_id": product_id |
| }) |
| |
| product = data.get("product", {}) |
| images = product.get("images", []) |
| |
| if not images: |
| result = f"No images found for product {product_id}" |
| else: |
| result = f"PRESTASHOP PRODUCT IMAGES (ID: {product_id}):\n\n" |
| result += json.dumps(images, indent=2) |
| |
| return [TextContent(type="text", text=result)] |
| |
| except Exception as e: |
| return [TextContent( |
| type="text", |
| text=f"Error retrieving images for product {product_id}: {str(e)}" |
| )] |
| |
| async def _search_products(self, arguments: Dict[str, Any]) -> List[TextContent]: |
| """Search for products""" |
| query = arguments.get("query") |
| limit = arguments.get("limit", 10) |
| category_id = arguments.get("category_id") |
| |
| if not query: |
| return [TextContent( |
| type="text", |
| text="Error: query is required" |
| )] |
| |
| try: |
| params = { |
| "action": "search_products", |
| "query": query, |
| "limit": limit |
| } |
| |
| if category_id: |
| params["category_id"] = category_id |
| |
| data = await self._make_api_request("", params) |
| |
| products = data.get("products", []) |
| |
| if not products: |
| result = f"No products found for query: '{query}'" |
| else: |
| result = f"PRESTASHOP PRODUCT SEARCH RESULTS for '{query}':\n\n" |
| result += json.dumps(products, indent=2) |
| |
| return [TextContent(type="text", text=result)] |
| |
| except Exception as e: |
| return [TextContent( |
| type="text", |
| text=f"Error searching for products with query '{query}': {str(e)}" |
| )] |
| |
| async def run(self, transport_type: str = "stdio"): |
| """Run the MCP server""" |
| if transport_type == "stdio": |
| from mcp.server.stdio import stdio_server |
| async with stdio_server() as (read_stream, write_stream): |
| await self.server.run( |
| read_stream, |
| write_stream, |
| self.server.create_initialization_options() |
| ) |
| else: |
| raise ValueError(f"Unsupported transport type: {transport_type}") |
|
|
| def main(): |
| """Main entry point""" |
| import argparse |
| |
| parser = argparse.ArgumentParser(description="PrestaShop MCP Server") |
| parser.add_argument( |
| "--api-base", |
| default=DEFAULT_API_BASE, |
| help="PrestaShop API base URL" |
| ) |
| parser.add_argument( |
| "--transport", |
| choices=["stdio"], |
| default="stdio", |
| help="Transport mechanism" |
| ) |
| |
| args = parser.parse_args() |
| |
| server = PrestaShopMCPServer(api_base=args.api_base) |
| |
| try: |
| asyncio.run(server.run(args.transport)) |
| except KeyboardInterrupt: |
| logger.info("Server stopped by user") |
| except Exception as e: |
| logger.error(f"Server error: {e}") |
| raise |
|
|
| if __name__ == "__main__": |
| main() |
|
|
