diff --git a/src/a2a/contrib/tasks/vertex_task_store.py b/src/a2a/contrib/tasks/vertex_task_store.py
index 0457694e4..602d5c6fd 100644
--- a/src/a2a/contrib/tasks/vertex_task_store.py
+++ b/src/a2a/contrib/tasks/vertex_task_store.py
@@ -116,7 +116,12 @@ def _get_metadata_change_event(
         task: CompatTask,
         event_sequence_number: int,
     ) -> vertexai_types.TaskEvent | None:
-        if task.metadata != previous_task.metadata:
+        # We generate metadata change events if the metadata was changed.
+        # We don't generate events if the metadata was changed from
+        # one empty value to another, e.g. {} to None.
+        if task.metadata != previous_task.metadata and (
+            task.metadata or previous_task.metadata
+        ):
             return vertexai_types.TaskEvent(
                 event_data=vertexai_types.TaskEventData(
                     metadata_change=vertexai_types.TaskMetadataChange(
diff --git a/tests/contrib/tasks/test_vertex_task_store.py b/tests/contrib/tasks/test_vertex_task_store.py
index 4be8cd4e6..c77493022 100644
--- a/tests/contrib/tasks/test_vertex_task_store.py
+++ b/tests/contrib/tasks/test_vertex_task_store.py
@@ -534,6 +534,58 @@ async def test_metadata_field_mapping(
     assert retrieved_none.metadata == {}
 
 
+@pytest.mark.asyncio
+async def test_metadata_empty_transitions(
+    vertex_store: VertexTaskStore,
+) -> None:
+    """Test that updating metadata between {} and None does not generate events."""
+    task_id = 'task-metadata-empty-test'
+
+    # Step 1: Create task with metadata={}
+    task = Task(
+        id=task_id,
+        context_id='session-meta-empty',
+        status=TaskStatus(state=TaskState.TASK_STATE_SUBMITTED),
+        metadata={},
+    )
+    await vertex_store.save(task, ServerCallContext())
+
+    full_name = f'{vertex_store._agent_engine_resource_id}/a2aTasks/{task_id}'
+
+    # Get initial event sequence number
+    stored_task_before = (
+        await vertex_store._client.aio.agent_engines.a2a_tasks.get(
+            name=full_name
+        )
+    )
+    initial_seq = stored_task_before.next_event_sequence_number
+
+    # Step 2: Update metadata to None
+    updated_task = Task()
+    updated_task.CopyFrom(task)
+    updated_task.metadata.Clear()
+    await vertex_store.save(updated_task, ServerCallContext())
+
+    # Step 3: Update back to {}
+    task_back = Task()
+    task_back.CopyFrom(updated_task)
+    task_back.metadata = {}
+    await vertex_store.save(task_back, ServerCallContext())
+
+    # Verify that retrieved task still has {} (due to mapping)
+    retrieved = await vertex_store.get(task_id, ServerCallContext())
+    assert retrieved is not None
+    assert retrieved.metadata == {}
+
+    # Verify that next_event_sequence_number did NOT increase (no events generated)
+    stored_task_after = (
+        await vertex_store._client.aio.agent_engines.a2a_tasks.get(
+            name=full_name
+        )
+    )
+    assert stored_task_after.next_event_sequence_number == initial_seq
+
+
 @pytest.mark.asyncio
 async def test_update_task_status_details(
     vertex_store: VertexTaskStore,