| """ |
| HTTP Transport wrapper for the PrestaShop MCP Server |
| This allows the MCP server to be called via HTTP requests (like the old PHP version) |
| """ |
|
|
| import asyncio |
| import json |
| import logging |
| from typing import Any, Dict |
|
|
| from fastapi import FastAPI, HTTPException, Request |
| from fastapi.responses import JSONResponse |
| from pydantic import BaseModel |
| import uvicorn |
|
|
| from .server import PrestaShopMCPServer |
|
|
| logger = logging.getLogger(__name__) |
|
|
| class MCPRequest(BaseModel): |
| jsonrpc: str = "2.0" |
| method: str |
| params: Dict[str, Any] = {} |
| id: str |
|
|
| class MCPHTTPWrapper: |
| """HTTP wrapper for MCP server to maintain compatibility with existing integrations""" |
| |
| def __init__(self, api_base: str = None): |
| self.mcp_server = PrestaShopMCPServer(api_base) if api_base else PrestaShopMCPServer() |
| self.app = FastAPI(title="PrestaShop MCP HTTP Server") |
| self.setup_routes() |
| |
| def setup_routes(self): |
| """Setup HTTP routes""" |
| |
| @self.app.post("/") |
| async def handle_mcp_request(request: MCPRequest): |
| """Handle MCP requests over HTTP""" |
| try: |
| if request.method == "tools/list": |
| |
| tools = await self._get_tools() |
| return { |
| "jsonrpc": "2.0", |
| "result": {"tools": tools}, |
| "id": request.id |
| } |
| |
| elif request.method == "tools/call": |
| |
| tool_name = request.params.get("name") |
| arguments = request.params.get("arguments", {}) |
| |
| result = await self._call_tool(tool_name, arguments) |
| |
| return { |
| "jsonrpc": "2.0", |
| "result": { |
| "content": [ |
| { |
| "type": "text", |
| "text": result |
| } |
| ] |
| }, |
| "id": request.id |
| } |
| |
| else: |
| raise HTTPException( |
| status_code=400, |
| detail=f"Unknown method: {request.method}" |
| ) |
| |
| except Exception as e: |
| logger.error(f"Error handling request: {e}") |
| return { |
| "jsonrpc": "2.0", |
| "error": { |
| "code": -32603, |
| "message": "Internal error", |
| "data": str(e) |
| }, |
| "id": request.id |
| } |
| |
| @self.app.get("/health") |
| async def health_check(): |
| """Health check endpoint""" |
| return {"status": "healthy", "service": "prestashop-mcp-server"} |
| |
| async def _get_tools(self): |
| """Get list of available tools""" |
| |
| tools = [ |
| { |
| "name": "get_product_details", |
| "description": "Get detailed information about a PrestaShop product", |
| "inputSchema": { |
| "type": "object", |
| "properties": { |
| "product_id": {"type": "integer", "description": "Product ID"}, |
| "lang_id": {"type": "integer", "description": "Language ID", "default": 1} |
| }, |
| "required": ["product_id"] |
| } |
| }, |
| { |
| "name": "get_product_features", |
| "description": "Get product features from PrestaShop", |
| "inputSchema": { |
| "type": "object", |
| "properties": { |
| "product_id": {"type": "integer", "description": "Product ID"} |
| }, |
| "required": ["product_id"] |
| } |
| }, |
| { |
| "name": "get_product_images", |
| "description": "Get product images from PrestaShop", |
| "inputSchema": { |
| "type": "object", |
| "properties": { |
| "product_id": {"type": "integer", "description": "Product ID"} |
| }, |
| "required": ["product_id"] |
| } |
| }, |
| { |
| "name": "search_products", |
| "description": "Search for products in PrestaShop catalog", |
| "inputSchema": { |
| "type": "object", |
| "properties": { |
| "query": {"type": "string", "description": "Search query"}, |
| "limit": {"type": "integer", "description": "Max results", "default": 10} |
| }, |
| "required": ["query"] |
| } |
| } |
| ] |
| return tools |
| |
| async def _call_tool(self, tool_name: str, arguments: Dict[str, Any]) -> str: |
| """Call a tool and return the result as text""" |
| try: |
| if tool_name == "get_product_details": |
| result = await self.mcp_server._get_product_details(arguments) |
| elif tool_name == "get_product_features": |
| result = await self.mcp_server._get_product_features(arguments) |
| elif tool_name == "get_product_images": |
| result = await self.mcp_server._get_product_images(arguments) |
| elif tool_name == "search_products": |
| result = await self.mcp_server._search_products(arguments) |
| else: |
| return f"Unknown tool: {tool_name}" |
| |
| |
| if result and len(result) > 0: |
| return result[0].text |
| else: |
| return "No result returned" |
| |
| except Exception as e: |
| return f"Error calling tool {tool_name}: {str(e)}" |
|
|
| def create_app(api_base: str = None) -> FastAPI: |
| """Create FastAPI app instance""" |
| wrapper = MCPHTTPWrapper(api_base) |
| return wrapper.app |
|
|
| def start_server(host="0.0.0.0", port=8081): |
| """Start the HTTP server for MCP""" |
| print(f"Starting MCP HTTP server on {host}:{port}") |
| app = create_app() |
| uvicorn.run(app, host=host, port=port) |
|
|
| def main(): |
| """Main entry point for the server""" |
| start_server() |
|
|
| if __name__ == "__main__": |
| main() |
|
|
