MCP Dynamics 365 Commerce Server

""" Template and Guidelines for Implementing MCP Tools with Database Operations This file provides a template and examples for implementing MCP tools that interact with the mock database. Use this as a reference for implementing the remaining controllers. IMPLEMENTATION PATTERN: 1. Import the database manager 2. Initialize database in __init__ 3. Implement CRUD operations in handle_tool methods 4. Use try/catch for error handling 5. Return consistent response format with API endpoint info EXAMPLE IMPLEMENTATION: """ from typing import Any, Dict, List from datetime import datetime, timedelta import random import uuid from mcp.types import Tool from ..database import get_database from ..config import get_base_url class TemplateController: """Template controller showing implementation patterns""" def __init__(self): self.db = get_database() def get_tools(self) -> List[Tool]: """Return list of tools with proper schema definitions""" return [ Tool( name="example_create_entity", description="Create a new entity", inputSchema={ "type": "object", "properties": { "name": { "type": "string", "description": "Entity name" }, "description": { "type": "string", "description": "Entity description" }, "baseUrl": { "type": "string", "description": "Base URL of the Dynamics 365 Commerce site (uses DYNAMICS365_BASE_URL env var if not provided)" } }, "required": ["name"] } ), Tool( name="example_get_entity", description="Get entity by ID", inputSchema={ "type": "object", "properties": { "entityId": { "type": "string", "description": "Entity ID to retrieve" }, "baseUrl": { "type": "string", "description": "Base URL of the Dynamics 365 Commerce site (uses DYNAMICS365_BASE_URL env var if not provided)" } }, "required": ["entityId"] } ), Tool( name="example_update_entity", description="Update an existing entity", inputSchema={ "type": "object", "properties": { "entityId": { "type": "string", "description": "Entity ID to update" }, "name": { "type": "string", "description": "New name" }, "description": { "type": "string", "description": "New description" }, "baseUrl": { "type": "string", "description": "Base URL of the Dynamics 365 Commerce site (uses DYNAMICS365_BASE_URL env var if not provided)" } }, "required": ["entityId"] } ), Tool( name="example_delete_entity", description="Delete an entity", inputSchema={ "type": "object", "properties": { "entityId": { "type": "string", "description": "Entity ID to delete" }, "baseUrl": { "type": "string", "description": "Base URL of the Dynamics 365 Commerce site (uses DYNAMICS365_BASE_URL env var if not provided)" } }, "required": ["entityId"] } ), Tool( name="example_list_entities", description="List entities with optional filtering", inputSchema={ "type": "object", "properties": { "limit": { "type": "integer", "description": "Max results to return", "default": 25 }, "offset": { "type": "integer", "description": "Number to skip", "default": 0 }, "filter_field": { "type": "string", "description": "Field to filter on" }, "filter_value": { "type": "string", "description": "Value to filter by" }, "baseUrl": { "type": "string", "description": "Base URL of the Dynamics 365 Commerce site (uses DYNAMICS365_BASE_URL env var if not provided)" } }, "required": [] } ), Tool( name="example_search_entities", description="Search entities by query", inputSchema={ "type": "object", "properties": { "query": { "type": "string", "description": "Search query" }, "limit": { "type": "integer", "description": "Max results", "default": 20 }, "baseUrl": { "type": "string", "description": "Base URL of the Dynamics 365 Commerce site (uses DYNAMICS365_BASE_URL env var if not provided)" } }, "required": ["query"] } ) ] async def handle_tool(self, name: str, arguments: Dict[str, Any]) -> Dict[str, Any]: """Handle tool calls with database operations""" base_url = arguments.get("baseUrl", get_base_url()) collection = "entities" # Replace with actual collection name try: # CREATE operation if name == "example_create_entity": entity_data = { "name": arguments.get("name"), "description": arguments.get("description", ""), "status": "Active", "created_by": "API" # In real system, use authenticated user } entity_id = self.db.create(collection, entity_data) created_entity = self.db.read(collection, entity_id) return { "api": f"POST {base_url}/api/CommerceRuntime/Entities", "success": True, "entity": created_entity } # READ operation elif name == "example_get_entity": entity_id = arguments.get("entityId") entity = self.db.read(collection, entity_id) if not entity: return {"error": f"Entity {entity_id} not found"} return { "api": f"GET {base_url}/api/CommerceRuntime/Entities/{entity_id}", "entity": entity } # UPDATE operation elif name == "example_update_entity": entity_id = arguments.get("entityId") # Check if entity exists entity = self.db.read(collection, entity_id) if not entity: return {"error": f"Entity {entity_id} not found"} # Prepare updates updates = {} if "name" in arguments: updates["name"] = arguments["name"] if "description" in arguments: updates["description"] = arguments["description"] # Perform update success = self.db.update(collection, entity_id, updates) if success: updated_entity = self.db.read(collection, entity_id) return { "api": f"PUT {base_url}/api/CommerceRuntime/Entities/{entity_id}", "success": True, "entity": updated_entity } else: return {"error": "Failed to update entity"} # DELETE operation elif name == "example_delete_entity": entity_id = arguments.get("entityId") # Check if entity exists entity = self.db.read(collection, entity_id) if not entity: return {"error": f"Entity {entity_id} not found"} success = self.db.delete(collection, entity_id) return { "api": f"DELETE {base_url}/api/CommerceRuntime/Entities/{entity_id}", "success": success, "deletedEntity": entity if success else None } # LIST operation with filtering elif name == "example_list_entities": limit = arguments.get("limit", 25) offset = arguments.get("offset", 0) # Apply filters if provided filters = {} if arguments.get("filter_field") and arguments.get("filter_value"): filters[arguments["filter_field"]] = arguments["filter_value"] entities = self.db.list(collection, limit=limit, offset=offset, filters=filters) total_count = self.db.count(collection, filters=filters) return { "api": f"GET {base_url}/api/CommerceRuntime/Entities", "entities": entities, "pagination": { "limit": limit, "offset": offset, "total": total_count, "hasMore": offset + limit < total_count }, "filters": filters } # SEARCH operation elif name == "example_search_entities": query = arguments.get("query") limit = arguments.get("limit", 20) # Define searchable fields for this entity type search_fields = ["name", "description"] results = self.db.search(collection, query, fields=search_fields, limit=limit) return { "api": f"GET {base_url}/api/CommerceRuntime/Entities/Search?q={query}", "query": query, "results": results, "totalResults": len(results), "searchFields": search_fields } else: return {"error": f"Unknown tool: {name}"} except Exception as e: return {"error": f"Error in {name}: {str(e)}"} """ QUICK IMPLEMENTATION GUIDE FOR REMAINING CONTROLLERS: 1. For simple CRUD controllers (address, barcode, etc.): - Copy template above - Change collection name and field names - Update tool schemas with appropriate fields - Add any controller-specific logic 2. For calculation controllers (pricing, currency, etc.): - Keep database operations minimal - Focus on calculation logic - Use existing data from database for calculations - Return calculated results 3. For complex controllers (cart, sales_order): - Implement multi-entity operations - Handle relationships between entities - Update multiple collections as needed - Validate business rules 4. Common patterns: - Always check entity exists before update/delete - Use try/catch for error handling - Include API endpoint in response - Return consistent response format - Use database helper methods (create, read, update, delete, list, search) 5. Example field mappings by controller: - address: street, city, state, zip, country, type - barcode: code, product_id, format, symbology - currency: code, name, symbol, exchange_rate - delivery_options: name, cost, delivery_days, type - reason_codes: code, name, description, type, category - tender_types: type, name, payment_method, requires_signature 6. For controllers with existing complex schemas, simplify to 3-5 core tools: - Create/Add - Get/Retrieve - Update/Modify - Delete/Remove - List/Search This approach provides functional implementations while keeping complexity manageable. """