update_collection_version
Modify version details and documentation links for Ansible collections stored in the Infrastructure-as-Code cache to maintain accurate IaC information.
Instructions
Update an existing Ansible collection's version information and documentation links
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| collection_id | Yes | Collection ID | |
| new_version | Yes | New version | |
| new_source_url | No | New source URL | |
| new_doc_url | No | New documentation URL |
Implementation Reference
- MCP tool handler function that processes arguments, calls the database update function, handles errors, and returns success message.async def handle_update_collection_version( db: Any, arguments: Dict[str, Any], operation_id: str ) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]: """Handle update_collection_version tool.""" try: logger.info( "Updating collection version", extra={ "collection_id": arguments["collection_id"], "new_version": arguments["new_version"], "operation_id": operation_id, }, ) # Update collection version result = update_collection_version( db, arguments["collection_id"], arguments["new_version"], arguments.get("new_source_url"), arguments.get("new_doc_url"), ) return [TextContent( type="text", text=f"Updated collection version: {result}" )] except Exception as e: error_msg = f"Failed to update collection version: {str(e)}" logger.error(error_msg, extra={"operation_id": operation_id}) raise McpError( types.ErrorData( code=types.INTERNAL_ERROR, message=error_msg, data={ "tool": "update_collection_version", "operation_id": operation_id, }, ) )
- JSON schema defining input parameters for the update_collection_version tool, including required fields and descriptions."update_collection_version": { "type": "object", "description": "Update an existing Ansible collection's version information and documentation links", "required": ["collection_id", "new_version"], "properties": { "collection_id": {"type": "string", "description": "Collection ID"}, "new_version": {"type": "string", "description": "New version"}, "new_source_url": {"type": "string", "description": "New source URL"}, "new_doc_url": {"type": "string", "description": "New documentation URL"}, }, },
- src/iac_memory_mcp_server/tools/ansible.py:559-570 (registration)Dictionary mapping tool names to their handler functions, registering update_collection_version with its handler.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, }
- Core database function that performs the SQL UPDATE on ansible_collections table to update version and optional URLs, with transaction handling and error management.def update_collection_version( db: DatabaseManager, collection_id: str, new_version: str, new_source_url: Optional[str] = None, new_doc_url: Optional[str] = None, ) -> bool: """Update an Ansible collection's version and optional URLs.""" try: updates = ["version = ?"] params = [new_version] if new_source_url: updates.append("source_url = ?") params.append(new_source_url) if new_doc_url: updates.append("doc_url = ?") params.append(new_doc_url) updates.append("updated_at = CURRENT_TIMESTAMP") params.append(collection_id) with db.get_connection() as conn: # Set busy timeout before any operations conn.execute( "PRAGMA busy_timeout = 5000" ) # 5 second timeout per testing rules conn.execute("BEGIN IMMEDIATE") # Start transaction try: cursor = conn.execute( f"""UPDATE ansible_collections SET {', '.join(updates)} WHERE id = ?""", tuple(params), ) success = cursor.rowcount > 0 conn.commit() return success except Exception: conn.rollback() raise except sqlite3.Error as e: raise DatabaseError( f"Failed to update collection version: {str(e)}. " f"Operation timed out after 5 seconds." )