get_module_version_compatibility
Check Ansible module compatibility across collection versions to ensure modules work with specific target versions.
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
- The async handler function that processes tool arguments, calls the database helper get_module_compatibility, formats the compatibility results into a readable text response, and handles errors.
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, }, ) ) - Database helper function implementing the core logic: queries current module version, fetches version history, parses schemas, detects breaking changes like new required fields or removed properties, and computes compatibility status.
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 except sqlite3.Error as e: error_msg = f"Failed to check module compatibility: {str(e)}" logger.error(error_msg) raise DatabaseError(error_msg) - JSON schema defining the tool's input parameters and validation rules.
"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:473-482 (registration)Local registration mapping the tool name 'get_module_version_compatibility' to its handler function within Ansible-specific tool handlers.
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, } - src/iac_memory_mcp_server/tools/__init__.py:43-55 (registration)Aggregate registration of all tools including Ansible handlers to the MCP server, via tool_handlers dict that unpacks ansible_tool_handlers and the call_tool decorator.
def register_tools(server: Server) -> None: """Register all tool handlers with the server.""" @server.call_tool() async def call_tool( name: str, arguments: Dict[str, Any], ctx: RequestContext | None = None ): return await handle_call_tool(name, arguments, ctx) @server.list_tools() async def list_tools(ctx: RequestContext = None): return await handle_list_tools(ctx)