get_module_version_compatibility
Check if a specific Ansible module works with a target collection version to prevent compatibility issues in IaC deployments.
Instructions
Check module compatibility across collection versions
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| collection_name | Yes | Name of the Ansible collection | |
| module_name | Yes | Name of the module to check | |
| version | Yes | Target collection version to check compatibility against |
Implementation Reference
- Main handler function executing the tool: logs input, fetches compatibility data from DB helper, formats detailed output including issues, breaking changes, and version history as TextContent.
async def handle_get_module_version_compatibility( db: Any, arguments: Dict[str, Any], operation_id: str ) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]: """Handle get_module_version_compatibility tool.""" try: logger.info( "Getting module version compatibility", extra={ "collection_name": arguments["collection_name"], "module_name": arguments["module_name"], "version": arguments["version"], "operation_id": operation_id, }, ) try: # Get compatibility info result = get_module_compatibility( db, arguments["collection_name"], arguments["module_name"], arguments["version"], ) except Exception as db_error: error_msg = str(db_error) logger.error(error_msg, extra={"operation_id": operation_id}) return [TextContent(type="text", text=error_msg)] # Format output output = [ "Module Compatibility Check:", f"Module: {result['module_name']}", f"Collection: {result['collection_name']}", f"Target Version: {result['target_version']}", f"Current Version: {result['current_version']}", f"Is Compatible: {'Yes' if result['is_compatible'] else 'No'}", "\nCompatibility Issues:", ] if not result["compatibility_issues"]: output.append("- No issues found") else: for issue in result["compatibility_issues"]: output.append(f"- {issue}") if result["breaking_changes"]: output.extend( [ "\nBreaking Changes:", *[f"- {change}" for change in result["breaking_changes"]], ] ) output.extend( [ "\nVersion History:", *[ f"- {v['collection_version']} (Module v{v['module_version']}) - {v['released']}" for v in result["version_history"] ], ] ) return [TextContent(type="text", text="\n".join(output))] except Exception as e: error_msg = f"Failed to check module version compatibility: {str(e)}" logger.error(error_msg, extra={"operation_id": operation_id}) raise McpError( types.ErrorData( code=types.INTERNAL_ERROR, message=error_msg, data={ "tool": "get_module_version_compatibility", "operation_id": operation_id, }, ) ) - JSON schema defining input parameters: collection_name, module_name, version (all required strings). Used for validation.
"get_module_version_compatibility": { "type": "object", "description": "Check module compatibility across collection versions", "required": ["collection_name", "module_name", "version"], "properties": { "collection_name": { "type": "string", "description": "Name of the Ansible collection", }, "module_name": { "type": "string", "description": "Name of the module to check", }, "version": { "type": "string", "description": "Target collection version to check compatibility against", }, }, }, - src/iac_memory_mcp_server/tools/ansible.py:559-570 (registration)Registration dictionary mapping tool name 'get_module_version_compatibility' to its handler function, alongside other Ansible tools.
ansible_tool_handlers = { "get_ansible_collection_info": handle_get_ansible_collection_info, "list_ansible_collections": handle_list_ansible_collections, "get_collection_version_history": handle_get_collection_version_history, "get_ansible_module_info": handle_get_ansible_module_info, "list_collection_modules": handle_list_collection_modules, "get_module_version_compatibility": handle_get_module_version_compatibility, "add_ansible_collection": handle_add_ansible_collection, "add_ansible_module": handle_add_ansible_module, "update_collection_version": handle_update_collection_version, "update_module_version": handle_update_module_version, } - Database helper function that performs SQL queries to retrieve module data and history, compares schemas for compatibility (e.g., new required fields, removed properties), and computes results used by the handler.
def get_module_compatibility( db: DatabaseManager, collection_name: str, module_name: str, version: str ) -> Dict: """Check module compatibility across collection versions. Args: db: Database manager instance collection_name: Name of the collection module_name: Name of the module version: Target version to check compatibility against Returns: Dict containing compatibility status and any potential issues """ logger.info( "Getting module compatibility info", extra={ "collection_name": collection_name, "module_name": module_name, "version": version, "operation": "get_module_compatibility", }, ) try: with db.get_connection() as conn: conn.execute("PRAGMA busy_timeout = 5000") # 5 second timeout # Get latest version of the module current = conn.execute( """ SELECT m.*, c.name as collection_name, c.version as collection_version FROM ansible_modules m JOIN ansible_collections c ON m.collection_id = c.id WHERE c.name = ? AND m.name = ? ORDER BY m.version DESC LIMIT 1 """, (collection_name, module_name), ).fetchone() if not current: raise DatabaseError( f"Module '{module_name}' not found in collection '{collection_name}'" ) # Get version history history = conn.execute( """ SELECT m.version as module_version, c.version as collection_version, m.schema, c.updated_at as released FROM ansible_modules m JOIN ansible_collections c ON m.collection_id = c.id WHERE c.name = ? AND m.name = ? ORDER BY c.updated_at DESC """, (collection_name, module_name), ).fetchall() # Check compatibility result = { "module_name": module_name, "collection_name": collection_name, "target_version": version, "current_version": current["version"], "is_compatible": True, "compatibility_issues": [], "breaking_changes": [], "version_history": [], } # Add version history for h in history: result["version_history"].append( { "module_version": h["module_version"], "collection_version": h["collection_version"], "released": h["released"], } ) # Compare schemas to detect breaking changes target = None for h in history: if h["collection_version"] == version: target = h break if not target: result["is_compatible"] = False result["compatibility_issues"].append( f"Target version {version} not found in module history" ) return result # Basic schema comparison try: import json current_schema = json.loads(current["schema"]) target_schema = json.loads(target["schema"]) # Check for required fields in current that weren't required in target if "required" in target_schema or "required" in current_schema: current_required = set(current_schema.get("required", [])) target_required = set(target_schema.get("required", [])) new_required = current_required - target_required if new_required: result["is_compatible"] = False result["breaking_changes"].append( f"Required fields added: {', '.join(new_required)}" ) # Check for removed properties if "properties" in current_schema and "properties" in target_schema: removed_props = set(current_schema["properties"].keys()) - set( target_schema["properties"].keys() ) if removed_props: result["breaking_changes"].append( f"Properties removed: {', '.join(removed_props)}" ) # Add compatibility notes if result["breaking_changes"]: result["compatibility_issues"].extend(result["breaking_changes"]) else: result["compatibility_issues"].append( "No breaking changes detected" ) except json.JSONDecodeError: result["compatibility_issues"].append( "Unable to compare schemas - invalid JSON format" ) return result