update_collection_version
Update an Ansible collection's version and documentation links in the IAC Memory MCP Server, ensuring accurate version control and resource management.
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_doc_url | No | New documentation URL | |
| new_source_url | No | New source URL | |
| new_version | Yes | New version |
Implementation Reference
- MCP tool handler function that processes arguments, calls the database update function, and returns success/error response.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 and validation for the update_collection_version tool."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)Registration of the tool handler in the ansible_tool_handlers dictionary mapping tool name to handler function.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 helper function that performs the SQL UPDATE on ansible_collections table to update version and URLs.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." )