diff --git a/src/EventHandler.cc b/src/EventHandler.cc
index c31a601713..1a26082a77 100644
--- a/src/EventHandler.cc
+++ b/src/EventHandler.cc
@@ -60,16 +60,12 @@ void EventHandler::Call(Args* vl, bool no_remote, double ts) {
     if ( ! no_remote ) {
         if ( ! auto_publish.empty() ) {
             // Send event in form [name, xs...] where xs represent the arguments.
-            broker::vector xs;
-            xs.reserve(vl->size());
+            BrokerListBuilder xs;
+            xs.Reserve(vl->size());
             bool valid_args = true;
 
-            for ( auto i = 0u; i < vl->size(); ++i ) {
-                auto opt_data = Broker::detail::val_to_data((*vl)[i].get());
-
-                if ( opt_data )
-                    xs.emplace_back(std::move(*opt_data));
-                else {
+            for ( size_t index = 0; index < vl->size(); ++index ) {
+                if ( ! xs.Add((*vl)[index]) ) {
                     valid_args = false;
                     auto_publish.clear();
                     reporter->Error("failed auto-remote event '%s', disabled", Name());
@@ -78,14 +74,16 @@ void EventHandler::Call(Args* vl, bool no_remote, double ts) {
             }
 
             if ( valid_args ) {
+                auto ev_args = std::move(xs).Build();
+
                 for ( auto it = auto_publish.begin();; ) {
                     const auto& topic = *it;
                     ++it;
 
                     if ( it != auto_publish.end() )
-                        broker_mgr->PublishEvent(topic, Name(), xs, ts);
+                        broker_mgr->PublishEvent(topic, Name(), ev_args, ts);
                     else {
-                        broker_mgr->PublishEvent(topic, Name(), std::move(xs), ts);
+                        broker_mgr->PublishEvent(topic, Name(), std::move(ev_args), ts);
                         break;
                     }
                 }
diff --git a/src/Frame.cc b/src/Frame.cc
index b0d09c1ca3..11b60a57f5 100644
--- a/src/Frame.cc
+++ b/src/Frame.cc
@@ -125,60 +125,58 @@ static bool val_is_func(const ValPtr& v, ScriptFunc* func) {
     return v->AsFunc() == func;
 }
 
-broker::expected<broker::data> Frame::Serialize() {
-    broker::vector body;
+std::optional<BrokerData> Frame::Serialize() {
+    BrokerListBuilder body;
 
     for ( int i = 0; i < size; ++i ) {
-        const auto& val = frame[i];
-        auto expected = Broker::detail::val_to_data(val.get());
-        if ( ! expected )
-            return broker::ec::invalid_data;
+        BrokerListBuilder val_tuple;
 
-        TypeTag tag = val->GetType()->Tag();
-        broker::vector val_tuple{std::move(*expected), static_cast<broker::integer>(tag)};
-        body.emplace_back(std::move(val_tuple));
+        const auto& val = frame[i];
+        if ( ! val_tuple.Add(val) )
+            return std::nullopt;
+
+        val_tuple.Add(static_cast<int64_t>(val->GetType()->Tag()));
+
+        body.Add(std::move(val_tuple));
     }
 
-    broker::vector rval;
-    rval.emplace_back(std::move(body));
+    BrokerListBuilder rval;
+    rval.Add(std::move(body));
 
-    return {std::move(rval)};
+    return std::move(rval).Build();
 }
 
-std::pair<bool, FramePtr> Frame::Unserialize(const broker::vector& data) {
-    if ( data.size() == 0 )
+std::pair<bool, FramePtr> Frame::Unserialize(BrokerListView data) {
+    if ( data.IsEmpty() )
         return std::make_pair(true, nullptr);
 
-    auto where = data.begin();
-    auto has_body = broker::get_if<broker::vector>(*where);
-    if ( ! has_body )
+    if ( ! data.Front().IsList() )
         return std::make_pair(false, nullptr);
 
-    broker::vector body = *has_body;
-    int frame_size = body.size();
-    auto rf = make_intrusive<Frame>(frame_size, nullptr, nullptr);
+    auto body = data.Front().ToList();
 
-    for ( int i = 0; i < frame_size; ++i ) {
-        auto has_vec = broker::get_if<broker::vector>(body[i]);
-        if ( ! has_vec )
+    auto frame_size = body.Size();
+    auto rf = make_intrusive<Frame>(static_cast<int>(frame_size), nullptr, nullptr);
+
+    for ( size_t index = 0; index < frame_size; ++index ) {
+        if ( ! body[index].IsList() )
             continue;
 
-        broker::vector val_tuple = *has_vec;
-        if ( val_tuple.size() != 2 )
+        auto val_tuple = body[index].ToList();
+
+        if ( val_tuple.Size() != 2 )
             return std::make_pair(false, nullptr);
 
-        auto has_type = broker::get_if<broker::integer>(val_tuple[1]);
-        if ( ! has_type )
+        auto type_int = val_tuple[1].ToInteger(-1);
+        if ( type_int < 0 || type_int >= NUM_TYPES )
             return std::make_pair(false, nullptr);
 
-        broker::integer g = *has_type;
-        Type t(static_cast<TypeTag>(g));
-
-        auto val = Broker::detail::data_to_val(std::move(val_tuple[0]), &t);
+        Type t{static_cast<TypeTag>(type_int)};
+        auto val = val_tuple[0].ToVal(&t);
         if ( ! val )
             return std::make_pair(false, nullptr);
 
-        rf->frame[i] = std::move(val);
+        rf->frame[index] = std::move(val);
     }
 
     return std::make_pair(true, std::move(rf));
diff --git a/src/Frame.h b/src/Frame.h
index 05407aaa97..306dab2453 100644
--- a/src/Frame.h
+++ b/src/Frame.h
@@ -2,8 +2,6 @@
 
 #pragma once
 
-#include <broker/data.hh>
-#include <broker/expected.hh>
 #include <memory>
 #include <optional>
 #include <string>
@@ -21,6 +19,9 @@ namespace zeek {
 
 using ValPtr = IntrusivePtr<Val>;
 
+class BrokerListView;
+class BrokerData;
+
 namespace detail {
 
 class CallExpr;
@@ -171,7 +172,7 @@ public:
      * sequence of two-element vectors, the first element reflecting
      * the frame value, the second its type.
      */
-    broker::expected<broker::data> Serialize();
+    std::optional<BrokerData> Serialize();
 
     /**
      * Instantiates a Frame from a serialized one.
@@ -180,7 +181,7 @@ public:
      * and the second is the unserialized frame with reference count +1, or
      * null if the serialization wasn't successful.
      */
-    static std::pair<bool, FramePtr> Unserialize(const broker::vector& data);
+    static std::pair<bool, FramePtr> Unserialize(BrokerListView data);
 
     // If the frame is run in the context of a trigger condition evaluation,
     // the trigger needs to be registered.
diff --git a/src/Func.cc b/src/Func.cc
index efd9b5a992..b6f6f33700 100644
--- a/src/Func.cc
+++ b/src/Func.cc
@@ -586,7 +586,7 @@ void ScriptFunc::ReplaceBody(const StmtPtr& old_body, StmtPtr new_body) {
     current_body = new_body;
 }
 
-bool ScriptFunc::DeserializeCaptures(const broker::vector& data) {
+bool ScriptFunc::DeserializeCaptures(BrokerListView data) {
     auto result = Frame::Unserialize(data);
 
     ASSERT(result.first);
@@ -649,7 +649,7 @@ FuncPtr ScriptFunc::DoClone() {
     return other;
 }
 
-broker::expected<broker::data> ScriptFunc::SerializeCaptures() const {
+std::optional<BrokerData> ScriptFunc::SerializeCaptures() const {
     if ( captures_vec ) {
         auto& cv = *captures_vec;
         auto& captures = *type->GetCaptures();
@@ -668,7 +668,7 @@ broker::expected<broker::data> ScriptFunc::SerializeCaptures() const {
         return captures_frame->Serialize();
 
     // No captures, return an empty vector.
-    return broker::vector{};
+    return BrokerListBuilder{}.Build();
 }
 
 void ScriptFunc::Describe(ODesc* d) const {
diff --git a/src/Func.h b/src/Func.h
index d10eedc291..1a247123e5 100644
--- a/src/Func.h
+++ b/src/Func.h
@@ -18,17 +18,12 @@
 #include "zeek/ZeekArgs.h"
 #include "zeek/ZeekList.h"
 
-namespace broker {
-class data;
-using vector = std::vector<data>;
-template<class>
-class expected;
-} // namespace broker
-
 namespace zeek {
 
 class Val;
 class FuncType;
+class BrokerData;
+class BrokerListView;
 
 namespace detail {
 
@@ -231,14 +226,14 @@ public:
      *
      * @return a serialized version of the function's capture frame.
      */
-    virtual broker::expected<broker::data> SerializeCaptures() const;
+    virtual std::optional<BrokerData> SerializeCaptures() const;
 
     /**
      * Sets the captures frame to one built from *data*.
      *
      * @param data a serialized frame
      */
-    bool DeserializeCaptures(const broker::vector& data);
+    bool DeserializeCaptures(BrokerListView data);
 
     using Func::AddBody;
 
diff --git a/src/OpaqueVal.cc b/src/OpaqueVal.cc
index 19165990f4..a4cbbc6b6f 100644
--- a/src/OpaqueVal.cc
+++ b/src/OpaqueVal.cc
@@ -23,6 +23,7 @@
 #include "zeek/Reporter.h"
 #include "zeek/Scope.h"
 #include "zeek/Var.h"
+#include "zeek/broker/Data.h"
 #include "zeek/digest.h"
 #include "zeek/probabilistic/BloomFilter.h"
 #include "zeek/probabilistic/CardinalityCounter.h"
@@ -37,18 +38,14 @@ inline void* EVP_MD_CTX_md_data(const EVP_MD_CTX* ctx) { return ctx->md_data; }
 
 namespace zeek {
 
-// Helper to retrieve a broker value out of a broker::vector at a specified
-// index, and casted to the expected destination type.
-template<typename S, typename V, typename D>
-inline bool get_vector_idx(const V& v, unsigned int i, D* dst) {
-    if ( i >= v.size() )
+// Helper to retrieve a broker count out of a list at a specified index, and
+// casted to the expected destination type.
+template<typename V, typename D>
+inline bool get_vector_idx(V& v, size_t i, D* dst) {
+    if ( i >= v.Size() || ! v[i].IsCount() )
         return false;
 
-    auto x = broker::get_if<S>(&v[i]);
-    if ( ! x )
-        return false;
-
-    *dst = static_cast<D>(*x);
+    *dst = static_cast<D>(v[i].ToCount());
     return true;
 }
 
@@ -73,65 +70,78 @@ OpaqueValPtr OpaqueMgr::Instantiate(const std::string& id) const {
     return x != _types.end() ? (*x->second)() : nullptr;
 }
 
-broker::expected<broker::data> OpaqueVal::Serialize() const {
+std::optional<BrokerData> OpaqueVal::Serialize() const {
     auto type = OpaqueMgr::mgr()->TypeID(this);
 
     auto d = DoSerialize();
     if ( ! d )
-        return d.error();
+        return std::nullopt;
 
-    return {broker::vector{std::move(type), std::move(*d)}};
+    BrokerListBuilder builder;
+    builder.Add(std::move(type));
+    builder.Add(std::move(*d));
+    return std::move(builder).Build();
 }
 
-OpaqueValPtr OpaqueVal::Unserialize(const broker::data& data) {
-    auto v = broker::get_if<broker::vector>(&data);
-
-    if ( ! (v && v->size() == 2) )
+OpaqueValPtr OpaqueVal::Unserialize(BrokerDataView data) {
+    if ( ! data.IsList() )
         return nullptr;
 
-    auto type = broker::get_if<std::string>(&(*v)[0]);
-    if ( ! type )
+    return Unserialize(data.ToList());
+}
+
+OpaqueValPtr OpaqueVal::Unserialize(BrokerListView v) {
+    if ( v.Size() != 2 || ! v[0].IsString() )
         return nullptr;
 
-    auto val = OpaqueMgr::mgr()->Instantiate(*type);
+    auto type = v[0].ToString();
+
+    auto val = OpaqueMgr::mgr()->Instantiate(std::string{type});
     if ( ! val )
         return nullptr;
 
-    if ( ! val->DoUnserialize((*v)[1]) )
+    if ( ! val->DoUnserialize(v[1]) )
         return nullptr;
 
     return val;
 }
 
-broker::expected<broker::data> OpaqueVal::SerializeType(const TypePtr& t) {
+std::optional<BrokerData> OpaqueVal::SerializeType(const TypePtr& t) {
     if ( t->InternalType() == TYPE_INTERNAL_ERROR )
-        return broker::ec::invalid_data;
+        return std::nullopt;
+
+    BrokerListBuilder builder;
 
     if ( t->InternalType() == TYPE_INTERNAL_OTHER ) {
         // Serialize by name.
-        assert(t->GetName().size());
-        return {broker::vector{true, t->GetName()}};
+        assert(! t->GetName().empty());
+        builder.Add(true);
+        builder.Add(t->GetName());
+    }
+    else {
+        // A base type.
+        builder.Add(false);
+        builder.Add(static_cast<uint64_t>(t->Tag()));
     }
 
-    // A base type.
-    return {broker::vector{false, static_cast<uint64_t>(t->Tag())}};
+    return {std::move(builder).Build()};
 }
 
-TypePtr OpaqueVal::UnserializeType(const broker::data& data) {
-    auto v = broker::get_if<broker::vector>(&data);
-    if ( ! (v && v->size() == 2) )
+TypePtr OpaqueVal::UnserializeType(BrokerDataView data) {
+    if ( ! data.IsList() )
         return nullptr;
 
-    auto by_name = broker::get_if<bool>(&(*v)[0]);
-    if ( ! by_name )
+    auto v = data.ToList();
+
+    if ( v.Size() != 2 || ! v[0].IsBool() )
         return nullptr;
 
-    if ( *by_name ) {
-        auto name = broker::get_if<std::string>(&(*v)[1]);
-        if ( ! name )
+    if ( v[0].ToBool() ) // Get by name?
+    {
+        if ( ! v[1].IsString() )
             return nullptr;
 
-        const auto& id = detail::global_scope()->Find(*name);
+        const auto& id = detail::global_scope()->Find(v[1].ToString());
         if ( ! id )
             return nullptr;
 
@@ -141,11 +151,10 @@ TypePtr OpaqueVal::UnserializeType(const broker::data& data) {
         return id->GetType();
     }
 
-    auto tag = broker::get_if<uint64_t>(&(*v)[1]);
-    if ( ! tag )
+    if ( ! v[1].IsCount() )
         return nullptr;
 
-    return base_type(static_cast<TypeTag>(*tag));
+    return base_type(static_cast<TypeTag>(v[1].ToCount()));
 }
 
 ValPtr OpaqueVal::DoClone(CloneState* state) {
@@ -153,7 +162,7 @@ ValPtr OpaqueVal::DoClone(CloneState* state) {
     if ( ! d )
         return nullptr;
 
-    auto rval = OpaqueVal::Unserialize(std::move(*d));
+    auto rval = OpaqueVal::Unserialize(d->AsView());
     return state->NewClone(this, std::move(rval));
 }
 
@@ -431,40 +440,43 @@ StringValPtr MD5Val::DoGet() {
 
 IMPLEMENT_OPAQUE_VALUE(MD5Val)
 
-broker::expected<broker::data> MD5Val::DoSerialize() const {
-    if ( ! IsValid() )
-        return {broker::vector{false}};
+std::optional<BrokerData> MD5Val::DoSerialize() const {
+    BrokerListBuilder builder;
 
-    broker::vector d = {true, do_serialize(ctx)};
-    return {std::move(d)};
+    if ( ! IsValid() ) {
+        builder.Add(false);
+        return std::move(builder).Build();
+    }
+
+    builder.Reserve(2);
+    builder.Add(true);
+    builder.Add(do_serialize(ctx));
+    return std::move(builder).Build();
 }
 
-bool MD5Val::DoUnserialize(const broker::data& data) {
-    auto d = broker::get_if<broker::vector>(&data);
-    if ( ! d )
+bool MD5Val::DoUnserialize(BrokerDataView data) {
+    if ( ! data.IsList() )
+        return false;
+    auto d = data.ToList();
+
+    if ( d.IsEmpty() || ! d[0].IsBool() )
         return false;
 
-    auto valid = broker::get_if<bool>(&(*d)[0]);
-    if ( ! valid )
-        return false;
-
-    if ( ! *valid ) {
+    if ( ! d[0].ToBool() ) {
         assert(! IsValid()); // default set by ctor
         return true;
     }
 
-    if ( (*d).size() != 2 )
+    if ( d.Size() != 2 || ! d[1].IsString() )
         return false;
 
-    auto s = broker::get_if<std::string>(&(*d)[1]);
-    if ( ! s )
-        return false;
+    auto s = d[1].ToString();
 
-    if ( s->size() != MD5VAL_STATE_SIZE )
+    if ( s.size() != MD5VAL_STATE_SIZE )
         return false;
 
     Init();
-    do_unserialize(ctx, s->data(), s->size());
+    do_unserialize(ctx, s.data(), s.size());
     return true;
 }
 
@@ -510,41 +522,44 @@ StringValPtr SHA1Val::DoGet() {
 
 IMPLEMENT_OPAQUE_VALUE(SHA1Val)
 
-broker::expected<broker::data> SHA1Val::DoSerialize() const {
-    if ( ! IsValid() )
-        return {broker::vector{false}};
+std::optional<BrokerData> SHA1Val::DoSerialize() const {
+    BrokerListBuilder builder;
 
-    broker::vector d = {true, do_serialize(ctx)};
+    if ( ! IsValid() ) {
+        builder.Add(false);
+        return std::move(builder).Build();
+    }
 
-    return {std::move(d)};
+    builder.Reserve(2);
+    builder.Add(true);
+    builder.Add(do_serialize(ctx));
+    return std::move(builder).Build();
 }
 
-bool SHA1Val::DoUnserialize(const broker::data& data) {
-    auto d = broker::get_if<broker::vector>(&data);
-    if ( ! d )
+bool SHA1Val::DoUnserialize(BrokerDataView data) {
+    if ( ! data.IsList() )
         return false;
 
-    auto valid = broker::get_if<bool>(&(*d)[0]);
-    if ( ! valid )
+    auto d = data.ToList();
+
+    if ( d.IsEmpty() || ! d[0].IsBool() )
         return false;
 
-    if ( ! *valid ) {
+    if ( ! d[0].ToBool() ) {
         assert(! IsValid()); // default set by ctor
         return true;
     }
 
-    if ( (*d).size() != 2 )
+    if ( d.Size() != 2 || ! d[1].IsString() )
         return false;
 
-    auto s = broker::get_if<std::string>(&(*d)[1]);
-    if ( ! s )
-        return false;
+    auto s = d[1].ToString();
 
-    if ( s->size() != SHA1VAL_STATE_SIZE )
+    if ( s.size() != SHA1VAL_STATE_SIZE )
         return false;
 
     Init();
-    do_unserialize(ctx, s->data(), s->size());
+    do_unserialize(ctx, s.data(), s.size());
     return true;
 }
 
@@ -593,41 +608,43 @@ StringValPtr SHA256Val::DoGet() {
 
 IMPLEMENT_OPAQUE_VALUE(SHA256Val)
 
-broker::expected<broker::data> SHA256Val::DoSerialize() const {
-    if ( ! IsValid() )
-        return {broker::vector{false}};
+std::optional<BrokerData> SHA256Val::DoSerialize() const {
+    BrokerListBuilder builder;
 
-    broker::vector d = {true, do_serialize(ctx)};
+    if ( ! IsValid() ) {
+        builder.Add(false);
+        return std::move(builder).Build();
+    }
 
-    return {std::move(d)};
+    builder.Add(true);
+    builder.Add(do_serialize(ctx));
+    return std::move(builder).Build();
 }
 
-bool SHA256Val::DoUnserialize(const broker::data& data) {
-    auto d = broker::get_if<broker::vector>(&data);
-    if ( ! d )
+bool SHA256Val::DoUnserialize(BrokerDataView data) {
+    if ( ! data.IsList() )
         return false;
 
-    auto valid = broker::get_if<bool>(&(*d)[0]);
-    if ( ! valid )
+    auto d = data.ToList();
+
+    if ( d.IsEmpty() || ! d[0].IsBool() )
         return false;
 
-    if ( ! *valid ) {
+    if ( ! d[0].ToBool() ) {
         assert(! IsValid()); // default set by ctor
         return true;
     }
 
-    if ( (*d).size() != 2 )
+    if ( d.Size() != 2 || ! d[1].IsString() )
         return false;
 
-    auto s = broker::get_if<std::string>(&(*d)[1]);
-    if ( ! s )
-        return false;
+    auto s = d[1].ToString();
 
-    if ( s->size() != SHA256VAL_STATE_SIZE )
+    if ( s.size() != SHA256VAL_STATE_SIZE )
         return false;
 
     Init();
-    do_unserialize(ctx, s->data(), s->size());
+    do_unserialize(ctx, s.data(), s.size());
     return true;
 }
 
@@ -645,69 +662,78 @@ bool EntropyVal::Get(double* r_ent, double* r_chisq, double* r_mean, double* r_m
 
 IMPLEMENT_OPAQUE_VALUE(EntropyVal)
 
-broker::expected<broker::data> EntropyVal::DoSerialize() const {
-    broker::vector d = {
-        static_cast<uint64_t>(state.totalc),   static_cast<uint64_t>(state.mp),
-        static_cast<uint64_t>(state.sccfirst), static_cast<uint64_t>(state.inmont),
-        static_cast<uint64_t>(state.mcount),   static_cast<uint64_t>(state.cexp),
-        static_cast<uint64_t>(state.montex),   static_cast<uint64_t>(state.montey),
-        static_cast<uint64_t>(state.montepi),  static_cast<uint64_t>(state.sccu0),
-        static_cast<uint64_t>(state.scclast),  static_cast<uint64_t>(state.scct1),
-        static_cast<uint64_t>(state.scct2),    static_cast<uint64_t>(state.scct3),
-    };
+std::optional<BrokerData> EntropyVal::DoSerialize() const {
+    BrokerListBuilder builder;
+    builder.Reserve(256 + 3 + RT_MONTEN + 11);
 
-    d.reserve(256 + 3 + RT_MONTEN + 11);
+    builder.Add(static_cast<uint64_t>(state.totalc));
+    builder.Add(static_cast<uint64_t>(state.mp));
+    builder.Add(static_cast<uint64_t>(state.sccfirst));
+    builder.Add(static_cast<uint64_t>(state.inmont));
+    builder.Add(static_cast<uint64_t>(state.mcount));
+    builder.Add(static_cast<uint64_t>(state.cexp));
+    builder.Add(static_cast<uint64_t>(state.montex));
+    builder.Add(static_cast<uint64_t>(state.montey));
+    builder.Add(static_cast<uint64_t>(state.montepi));
+    builder.Add(static_cast<uint64_t>(state.sccu0));
+    builder.Add(static_cast<uint64_t>(state.scclast));
+    builder.Add(static_cast<uint64_t>(state.scct1));
+    builder.Add(static_cast<uint64_t>(state.scct2));
+    builder.Add(static_cast<uint64_t>(state.scct3));
 
-    for ( int i = 0; i < 256; ++i )
-        d.emplace_back(static_cast<uint64_t>(state.ccount[i]));
+    for ( auto bin : state.ccount )
+        builder.Add(static_cast<uint64_t>(bin));
 
-    for ( int i = 0; i < RT_MONTEN; ++i )
-        d.emplace_back(static_cast<uint64_t>(state.monte[i]));
+    for ( auto val : state.monte )
+        builder.Add(static_cast<uint64_t>(val));
 
-    return {std::move(d)};
+    return std::move(builder).Build();
 }
 
-bool EntropyVal::DoUnserialize(const broker::data& data) {
-    auto d = broker::get_if<broker::vector>(&data);
-    if ( ! d )
+bool EntropyVal::DoUnserialize(BrokerDataView data) {
+    if ( ! data.IsList() )
         return false;
 
-    if ( ! get_vector_idx<uint64_t>(*d, 0, &state.totalc) )
+    auto d = data.ToList();
+
+    if ( ! get_vector_idx(d, 0, &state.totalc) )
         return false;
-    if ( ! get_vector_idx<uint64_t>(*d, 1, &state.mp) )
+    if ( ! get_vector_idx(d, 1, &state.mp) )
         return false;
-    if ( ! get_vector_idx<uint64_t>(*d, 2, &state.sccfirst) )
+    if ( ! get_vector_idx(d, 2, &state.sccfirst) )
         return false;
-    if ( ! get_vector_idx<uint64_t>(*d, 3, &state.inmont) )
+    if ( ! get_vector_idx(d, 3, &state.inmont) )
         return false;
-    if ( ! get_vector_idx<uint64_t>(*d, 4, &state.mcount) )
+    if ( ! get_vector_idx(d, 4, &state.mcount) )
         return false;
-    if ( ! get_vector_idx<uint64_t>(*d, 5, &state.cexp) )
+    if ( ! get_vector_idx(d, 5, &state.cexp) )
         return false;
-    if ( ! get_vector_idx<uint64_t>(*d, 6, &state.montex) )
+    if ( ! get_vector_idx(d, 6, &state.montex) )
         return false;
-    if ( ! get_vector_idx<uint64_t>(*d, 7, &state.montey) )
+    if ( ! get_vector_idx(d, 7, &state.montey) )
         return false;
-    if ( ! get_vector_idx<uint64_t>(*d, 8, &state.montepi) )
+    if ( ! get_vector_idx(d, 8, &state.montepi) )
         return false;
-    if ( ! get_vector_idx<uint64_t>(*d, 9, &state.sccu0) )
+    if ( ! get_vector_idx(d, 9, &state.sccu0) )
         return false;
-    if ( ! get_vector_idx<uint64_t>(*d, 10, &state.scclast) )
+    if ( ! get_vector_idx(d, 10, &state.scclast) )
         return false;
-    if ( ! get_vector_idx<uint64_t>(*d, 11, &state.scct1) )
+    if ( ! get_vector_idx(d, 11, &state.scct1) )
         return false;
-    if ( ! get_vector_idx<uint64_t>(*d, 12, &state.scct2) )
+    if ( ! get_vector_idx(d, 12, &state.scct2) )
         return false;
-    if ( ! get_vector_idx<uint64_t>(*d, 13, &state.scct3) )
+    if ( ! get_vector_idx(d, 13, &state.scct3) )
         return false;
 
-    for ( int i = 0; i < 256; ++i ) {
-        if ( ! get_vector_idx<uint64_t>(*d, 14 + i, &state.ccount[i]) )
+    auto index = size_t{14};
+
+    for ( auto& bin : state.ccount ) {
+        if ( ! get_vector_idx(d, index++, &bin) )
             return false;
     }
 
-    for ( int i = 0; i < RT_MONTEN; ++i ) {
-        if ( ! get_vector_idx<uint64_t>(*d, 14 + 256 + i, &state.monte[i]) )
+    for ( auto& val : state.monte ) {
+        if ( ! get_vector_idx(d, index++, &val) )
             return false;
     }
 
@@ -840,42 +866,44 @@ BloomFilterVal::~BloomFilterVal() {
 
 IMPLEMENT_OPAQUE_VALUE(BloomFilterVal)
 
-broker::expected<broker::data> BloomFilterVal::DoSerialize() const {
-    broker::vector d;
+std::optional<BrokerData> BloomFilterVal::DoSerialize() const {
+    BrokerListBuilder builder;
 
     if ( type ) {
         auto t = SerializeType(type);
         if ( ! t )
-            return broker::ec::invalid_data;
+            return std::nullopt;
 
-        d.emplace_back(std::move(*t));
+        builder.Add(std::move(*t));
     }
     else
-        d.emplace_back(broker::none());
+        builder.AddNil();
 
     auto bf = bloom_filter->Serialize();
     if ( ! bf )
-        return broker::ec::invalid_data; // Cannot serialize;
+        return std::nullopt;
 
-    d.emplace_back(*bf);
-    return {std::move(d)};
+    builder.Add(std::move(*bf));
+    return std::move(builder).Build();
 }
 
-bool BloomFilterVal::DoUnserialize(const broker::data& data) {
-    auto v = broker::get_if<broker::vector>(&data);
-
-    if ( ! (v && v->size() == 2) )
+bool BloomFilterVal::DoUnserialize(BrokerDataView data) {
+    if ( ! data.IsList() )
         return false;
 
-    auto no_type = broker::get_if<broker::none>(&(*v)[0]);
-    if ( ! no_type ) {
-        auto t = UnserializeType((*v)[0]);
+    auto v = data.ToList();
+
+    if ( v.Size() != 2 )
+        return false;
+
+    if ( ! v[0].IsNil() ) {
+        auto t = UnserializeType(v[0]);
 
         if ( ! (t && Typify(std::move(t))) )
             return false;
     }
 
-    auto bf = probabilistic::BloomFilter::Unserialize((*v)[1]);
+    auto bf = probabilistic::BloomFilter::Unserialize(v[1]);
     if ( ! bf )
         return false;
 
@@ -922,42 +950,45 @@ void CardinalityVal::Add(const Val* val) {
 
 IMPLEMENT_OPAQUE_VALUE(CardinalityVal)
 
-broker::expected<broker::data> CardinalityVal::DoSerialize() const {
-    broker::vector d;
+std::optional<BrokerData> CardinalityVal::DoSerialize() const {
+    BrokerListBuilder builder;
+    builder.Reserve(2);
 
     if ( type ) {
         auto t = SerializeType(type);
         if ( ! t )
-            return broker::ec::invalid_data;
+            return std::nullopt;
 
-        d.emplace_back(std::move(*t));
+        builder.Add(std::move(*t));
     }
     else
-        d.emplace_back(broker::none());
+        builder.AddNil();
 
     auto cs = c->Serialize();
     if ( ! cs )
-        return broker::ec::invalid_data;
+        return std::nullopt;
 
-    d.emplace_back(*cs);
-    return {std::move(d)};
+    builder.Add(std::move(*cs));
+    return std::move(builder).Build();
 }
 
-bool CardinalityVal::DoUnserialize(const broker::data& data) {
-    auto v = broker::get_if<broker::vector>(&data);
-
-    if ( ! (v && v->size() == 2) )
+bool CardinalityVal::DoUnserialize(BrokerDataView data) {
+    if ( ! data.IsList() )
         return false;
 
-    auto no_type = broker::get_if<broker::none>(&(*v)[0]);
-    if ( ! no_type ) {
-        auto t = UnserializeType((*v)[0]);
+    auto v = data.ToList();
+
+    if ( v.Size() != 2 )
+        return false;
+
+    if ( ! v[0].IsNil() ) {
+        auto t = UnserializeType(v[0]);
 
         if ( ! (t && Typify(std::move(t))) )
             return false;
     }
 
-    auto cu = probabilistic::detail::CardinalityCounter::Unserialize((*v)[1]);
+    auto cu = probabilistic::detail::CardinalityCounter::Unserialize(v[1]);
     if ( ! cu )
         return false;
 
@@ -986,24 +1017,26 @@ bool ParaglobVal::operator==(const ParaglobVal& other) const {
 
 IMPLEMENT_OPAQUE_VALUE(ParaglobVal)
 
-broker::expected<broker::data> ParaglobVal::DoSerialize() const {
-    broker::vector d;
+std::optional<BrokerData> ParaglobVal::DoSerialize() const {
     std::unique_ptr<std::vector<uint8_t>> iv = this->internal_paraglob->serialize();
-    for ( uint8_t a : *(iv.get()) )
-        d.emplace_back(static_cast<uint64_t>(a));
-    return {std::move(d)};
+    BrokerListBuilder builder;
+    builder.Reserve(iv->size());
+    for ( uint8_t a : *iv )
+        builder.AddCount(a);
+    return std::move(builder).Build();
 }
 
-bool ParaglobVal::DoUnserialize(const broker::data& data) {
-    auto d = broker::get_if<broker::vector>(&data);
-    if ( ! d )
+bool ParaglobVal::DoUnserialize(BrokerDataView data) {
+    if ( ! data.IsList() )
         return false;
 
-    std::unique_ptr<std::vector<uint8_t>> iv(new std::vector<uint8_t>);
-    iv->resize(d->size());
+    auto d = data.ToList();
 
-    for ( std::vector<broker::data>::size_type i = 0; i < d->size(); ++i ) {
-        if ( ! get_vector_idx<uint64_t>(*d, i, iv.get()->data() + i) )
+    auto iv = std::make_unique<std::vector<uint8_t>>();
+    iv->resize(d.Size());
+
+    for ( size_t index = 0; index < d.Size(); ++index ) {
+        if ( ! get_vector_idx(d, index, iv->data() + index) )
             return false;
     }
 
@@ -1032,11 +1065,9 @@ ValPtr ParaglobVal::DoClone(CloneState* state) {
     }
 }
 
-broker::expected<broker::data> TelemetryVal::DoSerialize() const {
-    return broker::make_error(broker::ec::invalid_data, "cannot serialize metric handles");
-}
+std::optional<BrokerData> TelemetryVal::DoSerialize() const { return std::nullopt; }
 
-bool TelemetryVal::DoUnserialize(const broker::data&) { return false; }
+bool TelemetryVal::DoUnserialize(BrokerDataView) { return false; }
 
 TelemetryVal::TelemetryVal(telemetry::IntCounter) : OpaqueVal(int_counter_metric_type) {}
 
diff --git a/src/OpaqueVal.h b/src/OpaqueVal.h
index e6560ac50c..9abcbebcd5 100644
--- a/src/OpaqueVal.h
+++ b/src/OpaqueVal.h
@@ -6,9 +6,9 @@
 #include <unistd.h>
 #endif
 
-#include <broker/expected.hh>
 #include <paraglob/paraglob.h>
 #include <sys/types.h> // for u_char
+#include <optional>
 
 #include "zeek/IntrusivePtr.h"
 #include "zeek/RandTest.h"
@@ -18,12 +18,12 @@
 #include "zeek/telemetry/Gauge.h"
 #include "zeek/telemetry/Histogram.h"
 
-namespace broker {
-class data;
-}
-
 namespace zeek {
 
+class BrokerData;
+class BrokerDataView;
+class BrokerListView;
+
 namespace probabilistic {
 class BloomFilter;
 }
@@ -90,8 +90,8 @@ private:
 #define DECLARE_OPAQUE_VALUE(T)                                                                                        \
     friend class zeek::OpaqueMgr::Register<T>;                                                                         \
     friend zeek::IntrusivePtr<T> zeek::make_intrusive<T>();                                                            \
-    broker::expected<broker::data> DoSerialize() const override;                                                       \
-    bool DoUnserialize(const broker::data& data) override;                                                             \
+    std::optional<BrokerData> DoSerialize() const override;                                                            \
+    bool DoUnserialize(BrokerDataView data) override;                                                                  \
     const char* OpaqueName() const override { return #T; }                                                             \
     static zeek::OpaqueValPtr OpaqueInstantiate() { return zeek::make_intrusive<T>(); }
 
@@ -117,7 +117,7 @@ public:
      * @return the broker representation, or an error if serialization
      * isn't supported or failed.
      */
-    broker::expected<broker::data> Serialize() const;
+    std::optional<BrokerData> Serialize() const;
 
     /**
      * Reinstantiates a value from its serialized Broker representation.
@@ -125,7 +125,12 @@ public:
      * @param data Broker representation as returned by *Serialize()*.
      * @return unserialized instances with reference count at +1
      */
-    static OpaqueValPtr Unserialize(const broker::data& data);
+    static OpaqueValPtr Unserialize(BrokerDataView data);
+
+    /**
+     * @copydoc Unserialize
+     */
+    static OpaqueValPtr Unserialize(BrokerListView data);
 
 protected:
     friend class Val;
@@ -138,7 +143,7 @@ protected:
      * @return the serialized data or an error if serialization
      * isn't supported or failed.
      */
-    virtual broker::expected<broker::data> DoSerialize() const = 0;
+    virtual std::optional<BrokerData> DoSerialize() const = 0;
 
     /**
      * Must be overridden to recreate the derived class' state from a
@@ -146,7 +151,7 @@ protected:
      *
      * @return true if successful.
      */
-    virtual bool DoUnserialize(const broker::data& data) = 0;
+    virtual bool DoUnserialize(BrokerDataView data) = 0;
 
     /**
      * Internal helper for the serialization machinery. Automatically
@@ -166,13 +171,13 @@ protected:
      * Helper function for derived class that need to record a type
      * during serialization.
      */
-    static broker::expected<broker::data> SerializeType(const TypePtr& t);
+    static std::optional<BrokerData> SerializeType(const TypePtr& t);
 
     /**
      * Helper function for derived class that need to restore a type
      * during unserialization. Returns the type at reference count +1.
      */
-    static TypePtr UnserializeType(const broker::data& data);
+    static TypePtr UnserializeType(BrokerDataView data);
 
     void ValDescribe(ODesc* d) const override;
     void ValDescribeReST(ODesc* d) const override;
@@ -414,8 +419,8 @@ protected:
     explicit TelemetryVal(telemetry::DblHistogram);
     explicit TelemetryVal(telemetry::DblHistogramFamily);
 
-    broker::expected<broker::data> DoSerialize() const override;
-    bool DoUnserialize(const broker::data& data) override;
+    std::optional<BrokerData> DoSerialize() const override;
+    bool DoUnserialize(BrokerDataView data) override;
 };
 
 template<class Handle>
diff --git a/src/Val.cc b/src/Val.cc
index 8d93426721..2b916402b8 100644
--- a/src/Val.cc
+++ b/src/Val.cc
@@ -2163,9 +2163,9 @@ void TableVal::SendToStore(const Val* index, const TableEntryVal* new_entry_val,
         else
             index_val = index;
 
-        auto broker_index = Broker::detail::val_to_data(index_val);
+        auto broker_index = BrokerData{};
 
-        if ( ! broker_index ) {
+        if ( ! broker_index.Convert(index_val) ) {
             emit_builtin_error("invalid Broker data conversation for table index");
             return;
         }
@@ -2195,26 +2195,25 @@ void TableVal::SendToStore(const Val* index, const TableEntryVal* new_entry_val,
                 }
 
                 if ( table_type->IsSet() )
-                    handle->store.put(std::move(*broker_index), broker::data(), expiry);
+                    handle->Put(std::move(broker_index), BrokerData{}, expiry);
                 else {
                     if ( ! new_entry_val ) {
                         emit_builtin_error("did not receive new value for Broker datastore send operation");
                         return;
                     }
 
-                    auto new_value = new_entry_val->GetVal().get();
-                    auto broker_val = Broker::detail::val_to_data(new_value);
-                    if ( ! broker_val ) {
+                    auto broker_val = BrokerData{};
+                    if ( ! broker_val.Convert(new_entry_val->GetVal()) ) {
                         emit_builtin_error("invalid Broker data conversation for table value");
                         return;
                     }
 
-                    handle->store.put(std::move(*broker_index), std::move(*broker_val), expiry);
+                    handle->Put(std::move(broker_index), std::move(broker_val), expiry);
                 }
                 break;
             }
 
-            case ELEMENT_REMOVED: handle->store.erase(std::move(*broker_index)); break;
+            case ELEMENT_REMOVED: handle->Erase(std::move(broker_index)); break;
 
             case ELEMENT_EXPIRED:
                 // we do nothing here. The Broker store does its own expiration - so the element
diff --git a/src/broker/Data.cc b/src/broker/Data.cc
index 6bf3c1500a..4440bc6f5d 100644
--- a/src/broker/Data.cc
+++ b/src/broker/Data.cc
@@ -38,6 +38,17 @@ TEST_CASE("converting Zeek to Broker protocol constants") {
     CHECK_EQ(to_broker_port_proto(TRANSPORT_UNKNOWN), broker::port::protocol::unknown);
 }
 
+namespace zeek {
+
+class BrokerDataAccess {
+public:
+    static broker::data& Unbox(BrokerData& data) { return data.value_; }
+
+    static broker::data&& Unbox(BrokerData&& data) { return std::move(data.value_); }
+};
+
+} // namespace zeek
+
 namespace zeek::Broker::detail {
 
 // Returns true if the given Zeek type is serialized as a broker::vector
@@ -210,7 +221,8 @@ struct val_converter {
             auto list_val = make_intrusive<ListVal>(TYPE_ANY);
 
             for ( size_t i = 0; i < indices->size(); ++i ) {
-                auto index_val = data_to_val(std::move((*indices)[i]), expected_index_types[i].get());
+                auto val = (*indices)[i]; // Must copy, because std::set has only immutable access.
+                auto index_val = data_to_val(val, expected_index_types[i].get());
 
                 if ( ! index_val )
                     return nullptr;
@@ -258,7 +270,8 @@ struct val_converter {
             auto list_val = make_intrusive<ListVal>(TYPE_ANY);
 
             for ( size_t i = 0; i < indices->size(); ++i ) {
-                auto index_val = data_to_val(std::move((*indices)[i]), expected_index_types[i].get());
+                auto val = (*indices)[i]; // Must copy, because the key is immutable.
+                auto index_val = data_to_val(val, expected_index_types[i].get());
 
                 if ( ! index_val )
                     return nullptr;
@@ -266,7 +279,7 @@ struct val_converter {
                 list_val->Append(std::move(index_val));
             }
 
-            auto value_val = data_to_val(std::move(item.second), tt->Yield().get());
+            auto value_val = data_to_val(item.second, tt->Yield().get());
 
             if ( ! value_val )
                 return nullptr;
@@ -283,7 +296,7 @@ struct val_converter {
             auto rval = make_intrusive<VectorVal>(IntrusivePtr{NewRef{}, vt});
 
             for ( auto& item : a ) {
-                auto item_val = data_to_val(std::move(item), vt->Yield().get());
+                auto item_val = data_to_val(item, vt->Yield().get());
 
                 if ( ! item_val )
                     return nullptr;
@@ -307,7 +320,7 @@ struct val_converter {
 
             unsigned int pos = 0;
             for ( auto& item : a ) {
-                auto item_val = data_to_val(std::move(item), pure ? lt->GetPureType().get() : types[pos].get());
+                auto item_val = data_to_val(item, pure ? lt->GetPureType().get() : types[pos].get());
                 pos++;
 
                 if ( ! item_val )
@@ -358,7 +371,7 @@ struct val_converter {
                     return nullptr;
 
                 auto* b = dynamic_cast<zeek::detail::ScriptFunc*>(rval->AsFunc());
-                if ( ! b || ! b->DeserializeCaptures(*frame) )
+                if ( ! b || ! b->DeserializeCaptures(BrokerListView{frame}) )
                     return nullptr;
             }
 
@@ -379,7 +392,7 @@ struct val_converter {
                     continue;
                 }
 
-                auto item_val = data_to_val(std::move(a[idx]), rt->GetFieldType(i).get());
+                auto item_val = data_to_val(a[idx], rt->GetFieldType(i).get());
 
                 if ( ! item_val )
                     return nullptr;
@@ -413,7 +426,7 @@ struct val_converter {
             return rval;
         }
         else if ( type->Tag() == TYPE_OPAQUE )
-            return OpaqueVal::Unserialize(a);
+            return OpaqueVal::Unserialize(BrokerListView{&a});
 
         return nullptr;
     }
@@ -688,7 +701,8 @@ struct type_checker {
         else if ( type->Tag() == TYPE_OPAQUE ) {
             // TODO: Could avoid doing the full unserialization here
             // and just check if the type is a correct match.
-            auto ov = OpaqueVal::Unserialize(a);
+            auto cpy = a;
+            auto ov = OpaqueVal::Unserialize(BrokerListView{&cpy});
             return ov != nullptr;
         }
 
@@ -703,14 +717,16 @@ static bool data_type_check(const broker::data& d, Type* t) {
     return visit(type_checker{t}, d);
 }
 
-ValPtr data_to_val(broker::data d, Type* type) {
-    if ( type->Tag() == TYPE_ANY )
-        return make_data_val(std::move(d));
+ValPtr data_to_val(broker::data& d, Type* type) {
+    if ( type->Tag() == TYPE_ANY ) {
+        BrokerData tmp{std::move(d)};
+        return std::move(tmp).ToRecordVal();
+    }
 
     return visit(val_converter{type}, d);
 }
 
-broker::expected<broker::data> val_to_data(const Val* v) {
+std::optional<broker::data> val_to_data(const Val* v) {
     switch ( v->GetType()->Tag() ) {
         case TYPE_BOOL: return {v->AsBool()};
         case TYPE_INT: return {v->AsInt()};
@@ -766,13 +782,13 @@ broker::expected<broker::data> val_to_data(const Val* v) {
                 if ( auto b = dynamic_cast<const zeek::detail::ScriptFunc*>(f) ) {
                     auto bc = b->SerializeCaptures();
                     if ( ! bc )
-                        return broker::ec::invalid_data;
+                        return std::nullopt;
 
-                    rval.emplace_back(std::move(*bc));
+                    rval.emplace_back(std::move(BrokerDataAccess::Unbox(*bc)));
                 }
                 else {
                     reporter->InternalWarning("Closure with non-ScriptFunc");
-                    return broker::ec::invalid_data;
+                    return std::nullopt;
                 }
             }
 
@@ -800,7 +816,7 @@ broker::expected<broker::data> val_to_data(const Val* v) {
                     auto key_part = val_to_data(vl->Idx(k).get());
 
                     if ( ! key_part )
-                        return broker::ec::invalid_data;
+                        return std::nullopt;
 
                     composite_key.emplace_back(std::move(*key_part));
                 }
@@ -818,7 +834,7 @@ broker::expected<broker::data> val_to_data(const Val* v) {
                     auto val = val_to_data(te.value->GetVal().get());
 
                     if ( ! val )
-                        return broker::ec::invalid_data;
+                        return std::nullopt;
 
                     get<broker::table>(rval).emplace(std::move(key), std::move(*val));
                 }
@@ -840,7 +856,7 @@ broker::expected<broker::data> val_to_data(const Val* v) {
                 auto item = val_to_data(item_val.get());
 
                 if ( ! item )
-                    return broker::ec::invalid_data;
+                    return std::nullopt;
 
                 rval.emplace_back(std::move(*item));
             }
@@ -863,7 +879,7 @@ broker::expected<broker::data> val_to_data(const Val* v) {
                 auto item = val_to_data(item_val.get());
 
                 if ( ! item )
-                    return broker::ec::invalid_data;
+                    return std::nullopt;
 
                 rval.emplace_back(std::move(*item));
             }
@@ -887,7 +903,7 @@ broker::expected<broker::data> val_to_data(const Val* v) {
                 auto item = val_to_data(item_val.get());
 
                 if ( ! item )
-                    return broker::ec::invalid_data;
+                    return std::nullopt;
 
                 rval.emplace_back(std::move(*item));
             }
@@ -906,12 +922,12 @@ broker::expected<broker::data> val_to_data(const Val* v) {
                 break;
             }
 
-            return {c};
+            return {BrokerDataAccess::Unbox(std::move(*c))};
         }
         default: reporter->Error("unsupported Broker::Data type: %s", type_name(v->GetType()->Tag())); break;
     }
 
-    return broker::ec::invalid_data;
+    return std::nullopt;
 }
 
 RecordValPtr make_data_val(Val* v) {
@@ -1021,19 +1037,21 @@ const TypePtr& DataVal::ScriptDataType() {
 
 IMPLEMENT_OPAQUE_VALUE(zeek::Broker::detail::DataVal)
 
-broker::expected<broker::data> DataVal::DoSerialize() const { return data; }
+std::optional<BrokerData> DataVal::DoSerialize() const { return BrokerData{data}; }
 
-bool DataVal::DoUnserialize(const broker::data& data_) {
-    data = data_;
+bool DataVal::DoUnserialize(BrokerDataView dv) {
+    data = std::move(*dv.value_);
     return true;
 }
 
 IMPLEMENT_OPAQUE_VALUE(zeek::Broker::detail::SetIterator)
 
-broker::expected<broker::data> SetIterator::DoSerialize() const { return broker::vector{dat, *it}; }
+std::optional<BrokerData> SetIterator::DoSerialize() const {
+    return BrokerData{broker::data{broker::vector{dat, *it}}};
+}
 
-bool SetIterator::DoUnserialize(const broker::data& data) {
-    auto v = get_if<broker::vector>(&data);
+bool SetIterator::DoUnserialize(BrokerDataView data) {
+    auto v = get_if<broker::vector>(data.value_);
     if ( ! (v && v->size() == 2) )
         return false;
 
@@ -1053,10 +1071,12 @@ bool SetIterator::DoUnserialize(const broker::data& data) {
 
 IMPLEMENT_OPAQUE_VALUE(zeek::Broker::detail::TableIterator)
 
-broker::expected<broker::data> TableIterator::DoSerialize() const { return broker::vector{dat, it->first}; }
+std::optional<BrokerData> TableIterator::DoSerialize() const {
+    return BrokerData{broker::data{broker::vector{dat, it->first}}};
+}
 
-bool TableIterator::DoUnserialize(const broker::data& data) {
-    auto v = get_if<broker::vector>(&data);
+bool TableIterator::DoUnserialize(BrokerDataView data) {
+    auto v = get_if<broker::vector>(data.value_);
     if ( ! (v && v->size() == 2) )
         return false;
 
@@ -1076,13 +1096,13 @@ bool TableIterator::DoUnserialize(const broker::data& data) {
 
 IMPLEMENT_OPAQUE_VALUE(zeek::Broker::detail::VectorIterator)
 
-broker::expected<broker::data> VectorIterator::DoSerialize() const {
+std::optional<BrokerData> VectorIterator::DoSerialize() const {
     broker::integer difference = it - dat.begin();
-    return broker::vector{dat, difference};
+    return BrokerData{broker::data{broker::vector{dat, difference}}};
 }
 
-bool VectorIterator::DoUnserialize(const broker::data& data) {
-    auto v = get_if<broker::vector>(&data);
+bool VectorIterator::DoUnserialize(BrokerDataView data) {
+    auto v = get_if<broker::vector>(data.value_);
     if ( ! (v && v->size() == 2) )
         return false;
 
@@ -1099,13 +1119,13 @@ bool VectorIterator::DoUnserialize(const broker::data& data) {
 
 IMPLEMENT_OPAQUE_VALUE(zeek::Broker::detail::RecordIterator)
 
-broker::expected<broker::data> RecordIterator::DoSerialize() const {
-    broker::integer difference = it - dat.begin();
-    return broker::vector{dat, difference};
+std::optional<BrokerData> RecordIterator::DoSerialize() const {
+    auto difference = static_cast<broker::integer>(it - dat.begin());
+    return BrokerData{broker::data{broker::vector{dat, difference}}};
 }
 
-bool RecordIterator::DoUnserialize(const broker::data& data) {
-    auto v = get_if<broker::vector>(&data);
+bool RecordIterator::DoUnserialize(BrokerDataView data) {
+    auto v = get_if<broker::vector>(data.value_);
     if ( ! (v && v->size() == 2) )
         return false;
 
@@ -1156,3 +1176,47 @@ threading::Field* data_to_threading_field(broker::data d) {
 }
 
 } // namespace zeek::Broker::detail
+
+namespace zeek {
+
+BrokerListView BrokerDataView::ToList() noexcept {
+    return BrokerListView{std::addressof(broker::get<broker::vector>(*value_))};
+}
+
+ValPtr BrokerDataView::ToVal(Type* type) { return zeek::Broker::detail::data_to_val(*value_, type); }
+
+bool BrokerData::Convert(const Val* value) {
+    if ( auto res = zeek::Broker::detail::val_to_data(value) ) {
+        value_ = std::move(*res);
+        return true;
+    }
+    return false;
+}
+
+RecordValPtr BrokerData::ToRecordVal() && {
+    auto rval = make_intrusive<RecordVal>(BifType::Record::Broker::Data);
+    rval->Assign(0, make_intrusive<zeek::Broker::detail::DataVal>(std::move(value_)));
+    return rval;
+}
+
+RecordValPtr BrokerData::ToRecordVal(const Val* v) {
+    auto rval = make_intrusive<RecordVal>(BifType::Record::Broker::Data);
+
+    BrokerData tmp;
+    if ( tmp.Convert(v) )
+        rval->Assign(0, make_intrusive<zeek::Broker::detail::DataVal>(std::move(tmp.value_)));
+    else
+        reporter->Warning("did not get a value from val_to_data");
+
+    return rval;
+}
+
+bool BrokerListBuilder::Add(const Val* value) {
+    if ( auto res = zeek::Broker::detail::val_to_data(value) ) {
+        values_.emplace_back(std::move(*res));
+        return true;
+    }
+    return false;
+}
+
+} // namespace zeek
diff --git a/src/broker/Data.h b/src/broker/Data.h
index 2e015d44ce..9105644e21 100644
--- a/src/broker/Data.h
+++ b/src/broker/Data.h
@@ -1,24 +1,42 @@
 #pragma once
 
+#include <cassert>
+#include <memory>
+#include <type_traits>
+
 #include "zeek/Expr.h"
 #include "zeek/Frame.h"
 #include "zeek/OpaqueVal.h"
 #include "zeek/Reporter.h"
 
+#include "broker/config.hh"
 #include "broker/data.hh"
 
 namespace zeek {
 
+/// A 64-bit timestamp with nanosecond precision.
+using BrokerTimespan = std::chrono::duration<int64_t, std::nano>;
+
 class ODesc;
 
-namespace threading {
+} // namespace zeek
+
+namespace zeek::Broker {
+
+class Manager;
+
+} // namespace zeek::Broker
+
+namespace zeek::threading {
 
 struct Value;
 struct Field;
 
-} // namespace threading
+} // namespace zeek::threading
 
-namespace Broker::detail {
+namespace zeek::Broker::detail {
+
+class StoreHandleVal;
 
 extern OpaqueTypePtr opaque_of_data_type;
 extern OpaqueTypePtr opaque_of_set_iterator;
@@ -59,7 +77,7 @@ EnumValPtr get_data_type(RecordVal* v, zeek::detail::Frame* frame);
  * @param v a Zeek value.
  * @return a Broker data value if the Zeek value could be converted to one.
  */
-broker::expected<broker::data> val_to_data(const Val* v);
+std::optional<broker::data> val_to_data(const Val* v);
 
 /**
  * Convert a Broker data value to a Zeek value.
@@ -68,7 +86,7 @@ broker::expected<broker::data> val_to_data(const Val* v);
  * @return a pointer to a new Zeek value or a nullptr if the conversion was not
  * possible.
  */
-ValPtr data_to_val(broker::data d, Type* type);
+ValPtr data_to_val(broker::data& d, Type* type);
 
 /**
  * Convert a zeek::threading::Field to a Broker data value.
@@ -255,5 +273,371 @@ protected:
     DECLARE_OPAQUE_VALUE(zeek::Broker::detail::RecordIterator)
 };
 
-} // namespace Broker::detail
+} // namespace zeek::Broker::detail
+
+namespace zeek {
+
+class BrokerListView;
+
+/**
+ * Non-owning reference (view) to a Broker data value.
+ */
+class BrokerDataView {
+public:
+    friend class zeek::Broker::detail::DataVal;
+    friend class zeek::Broker::detail::SetIterator;
+    friend class zeek::Broker::detail::TableIterator;
+    friend class zeek::Broker::detail::VectorIterator;
+    friend class zeek::Broker::detail::RecordIterator;
+
+    BrokerDataView() = delete;
+
+    BrokerDataView(const BrokerDataView&) noexcept = default;
+
+    explicit BrokerDataView(broker::data* value) noexcept : value_(value) { assert(value != nullptr); }
+
+    /**
+     * Checks whether the value represents the `nil` value.
+     */
+    [[nodiscard]] bool IsNil() const noexcept { return broker::is<broker::none>(*value_); }
+
+    /**
+     * Checks whether the value is a Boolean.
+     */
+    [[nodiscard]] bool IsBool() const noexcept { return broker::is<bool>(*value_); }
+
+    /**
+     * Converts the value to a Boolean.
+     */
+    [[nodiscard]] bool ToBool(bool fallback = 0) const noexcept {
+        if ( auto val = broker::get_if<bool>(value_); val ) {
+            return *val;
+        }
+        return fallback;
+    }
+
+    /**
+     * Checks whether the value is a string.
+     */
+    [[nodiscard]] bool IsString() const noexcept { return broker::is<std::string>(*value_); }
+
+    /**
+     * Converts the value to a string.
+     */
+    [[nodiscard]] std::string_view ToString() const noexcept {
+        if ( auto val = broker::get_if<std::string>(value_); val ) {
+            return *val;
+        }
+        return std::string_view{};
+    }
+
+    /**
+     * Checks whether the value is an integer.
+     */
+    [[nodiscard]] bool IsInteger() const noexcept { return broker::is<broker::integer>(*value_); }
+
+    /**
+     * Converts the value to an integer.
+     */
+    [[nodiscard]] int64_t ToInteger(int64_t fallback = 0) const noexcept {
+        if ( auto val = broker::get_if<broker::integer>(value_); val ) {
+            return *val;
+        }
+        return fallback;
+    }
+
+    /**
+     * Checks whether the value is a count.
+     */
+    [[nodiscard]] bool IsCount() const noexcept { return broker::is<broker::count>(*value_); }
+
+    /**
+     * Converts the value to a count.
+     */
+    [[nodiscard]] uint64_t ToCount(uint64_t fallback = 0) const noexcept {
+        if ( auto val = broker::get_if<broker::count>(value_); val ) {
+            return *val;
+        }
+        return fallback;
+    }
+
+    /**
+     * Checks whether the value is a real (double).
+     */
+    [[nodiscard]] bool IsReal() const noexcept { return broker::is<broker::real>(*value_); }
+
+    /**
+     * Converts the value to a real (double).
+     */
+    [[nodiscard]] double ToReal(double fallback = 0) const noexcept {
+        if ( auto val = broker::get_if<broker::real>(value_); val ) {
+            return *val;
+        }
+        return fallback;
+    }
+
+    /**
+     * Checks whether the value is a list.
+     */
+    [[nodiscard]] bool IsList() const noexcept { return broker::is<broker::vector>(*value_); }
+
+    /**
+     * Converts the value to a list.
+     * @pre IsList()
+     */
+    [[nodiscard]] BrokerListView ToList() noexcept;
+
+    /**
+     * Tries to convert this view to a Zeek value.
+     * @returns a Zeek value or nullptr if the conversion failed.
+     */
+    [[nodiscard]] ValPtr ToVal(Type* type);
+
+    /**
+     * Renders the value as a string.
+     */
+    friend std::string to_string(const BrokerDataView& data) { return broker::to_string(*data.value_); }
+
+private:
+    broker::data* value_;
+};
+
+/**
+ * Convenience function to check whether a list of Broker data values are all of type `count`.
+ */
+template<typename... Args>
+[[nodiscard]] inline bool IsCount(BrokerDataView arg, Args&&... args) {
+    return arg.IsCount() && (args.IsCount() && ...);
+}
+
+/**
+ * Convenience function to check whether a list of Broker data values are all of type `integer`.
+ */
+template<typename... Args>
+[[nodiscard]] inline auto ToCount(BrokerDataView arg, Args&&... args) {
+    return std::tuple{arg.ToCount(), args.ToCount()...};
+}
+
+/**
+ * Non-owning reference (view) to a Broker list value.
+ */
+class BrokerListView {
+public:
+    BrokerListView() = delete;
+
+    BrokerListView(const BrokerListView&) noexcept = default;
+
+    explicit BrokerListView(broker::vector* values) noexcept : values_(values) { assert(values != nullptr); }
+
+    /**
+     * Returns a view to the first element.
+     * @pre Size() > 0
+     */
+    [[nodiscard]] BrokerDataView Front() const { return BrokerDataView{std::addressof(values_->front())}; }
+
+    /**
+     * Returns a view to the last element.
+     * @pre Size() > 0
+     */
+    [[nodiscard]] BrokerDataView Back() const { return BrokerDataView{std::addressof(values_->back())}; }
+
+    /**
+     * Returns a view to the element at the given index.
+     * @pre index < Size()
+     */
+    [[nodiscard]] BrokerDataView operator[](size_t index) const {
+        return BrokerDataView{std::addressof((*values_)[index])};
+    }
+
+    /**
+     * Returns the number of elements in the list.
+     */
+    [[nodiscard]] size_t Size() const noexcept { return values_->size(); }
+
+    /**
+     * Checks whether the list is empty.
+     */
+    [[nodiscard]] size_t IsEmpty() const noexcept { return values_->empty(); }
+
+private:
+    broker::vector* values_;
+};
+
+class BrokerDataAccess;
+
+class BrokerListBuilder;
+
+/**
+ * Owning wrapper for a Broker data value.
+ */
+class BrokerData {
+public:
+    friend class BrokerDataAccess;
+    friend class BrokerListBuilder;
+    friend class zeek::Broker::Manager;
+    friend class zeek::Broker::detail::StoreHandleVal;
+
+    BrokerData() = default;
+
+    template<class DataType, class = std::enable_if_t<std::is_same_v<DataType, broker::data>>>
+    explicit BrokerData(DataType value) : value_(std::move(value)) {
+        // Note: we use enable_if here to avoid nasty implicit conversions of broker::data.
+    }
+
+    BrokerDataView AsView() noexcept { return BrokerDataView{std::addressof(value_)}; }
+
+    /**
+     * Attempts to parse a Zeek value into a Broker value. On success, the Broker
+     * value is stored in this object.
+     * @returns `true` if the conversion succeeded, `false` otherwise.
+     */
+    [[nodiscard]] bool Convert(const Val* value);
+
+    /**
+     * @copydoc Convert(const Val*)
+     */
+    [[nodiscard]] bool Convert(const ValPtr& value) { return Convert(value.get()); }
+
+    /**
+     * Converts this value to a Zeek record.
+     */
+    [[nodiscard]] RecordValPtr ToRecordVal() &&;
+
+    /**
+     * Convenience function for converting a Zeek value to a Broker value and then
+     * to a Zeek record.
+     */
+    [[nodiscard]] static RecordValPtr ToRecordVal(const Val* value);
+
+    /**
+     * @copydoc ToRecordVal(const Val*)
+     */
+    [[nodiscard]] static RecordValPtr ToRecordVal(const ValPtr& value) { return ToRecordVal(value.get()); }
+
+    /**
+     * Creates a Broker value from a string.
+     */
+    [[nodiscard]] static BrokerData FromString(const char* cstr, size_t len) {
+        return BrokerData{broker::data{std::string{cstr, len}}};
+    }
+
+    /**
+     * Renders the value as a string.
+     */
+    friend std::string to_string(const BrokerData& data) { return broker::to_string(data.value_); }
+
+private:
+    broker::data value_;
+};
+
+/**
+ * Utility class for building a BrokerData containing a list of values.
+ */
+class BrokerListBuilder {
+public:
+    friend class zeek::Broker::Manager;
+
+    /**
+     * Reserves space for up to `n` elements.
+     */
+    void Reserve(size_t n) { values_.reserve(n); }
+
+    /**
+     * Tries to convert a Zeek value into a Broker value and adds it to the list on success.
+     */
+    [[nodiscard]] bool Add(const Val* value);
+
+    /**
+     * @copydoc Add(const Val*)
+     */
+    [[nodiscard]] bool Add(const ValPtr& value) { return Add(value.get()); }
+
+    /**
+     * Adds `value` as a Broker `count` to the list, automatically converting it if necessary.
+     */
+    template<typename T>
+    void AddCount(T value) {
+        static_assert(std::is_integral_v<T> && ! std::is_same_v<bool, T>);
+        static_assert(std::is_unsigned_v<T>);
+        static_assert(sizeof(T) <= sizeof(broker::count));
+        values_.emplace_back(static_cast<broker::count>(value));
+    }
+
+    /**
+     * Adds `value` as a Broker `integer` to the list, automatically converting it if necessary.
+     */
+    template<typename T>
+    void AddInteger(T value) {
+        static_assert(std::is_integral_v<T> && ! std::is_same_v<bool, T>);
+        static_assert(std::is_signed_v<T>);
+        static_assert(sizeof(T) <= sizeof(broker::integer));
+        values_.emplace_back(static_cast<broker::integer>(value));
+    }
+
+    /**
+     * Appends `value` to the end of the list.
+     */
+    void Add(uint64_t value) { values_.emplace_back(static_cast<broker::count>(value)); }
+
+    /**
+     * Appends `value` to the end of the list.
+     */
+    void Add(int64_t value) { values_.emplace_back(static_cast<broker::integer>(value)); }
+
+    /**
+     * Appends `value` to the end of the list.
+     */
+    void Add(double value) { values_.emplace_back(value); }
+
+    /**
+     * Appends `value` to the end of the list.
+     */
+    void Add(bool value) { values_.emplace_back(value); }
+
+    /**
+     * Appends `value` to the end of the list.
+     */
+    void Add(std::string value) { values_.emplace_back(std::move(value)); }
+
+    /**
+     * Appends a string to the end of the list.
+     * @param cstr The characters to append.
+     * @param len The number of characters to append.
+     */
+    void AddString(const char* cstr, size_t len) { values_.emplace_back(std::string{cstr, len}); }
+
+    /**
+     * Appends `value` to the end of the list.
+     */
+    void Add(BrokerData value) { values_.emplace_back(std::move(value.value_)); }
+
+    /**
+     * Appends all elements from `builder` to the end of the list as a single element.
+     */
+    void Add(BrokerListBuilder&& builder) { values_.emplace_back(std::move(builder.values_)); }
+
+    /**
+     * Appends the `nil` value to the end of the list.
+     */
+    void AddNil() { values_.emplace_back(); }
+
+    /**
+     * Adds a list of values to the list (as a single element).
+     */
+    template<class... Ts>
+    void AddList(Ts&&... values) {
+        BrokerListBuilder sub;
+        (sub.Add(std::forward<Ts>(values)), ...);
+        values_.emplace_back(std::move(sub.values_));
+    }
+
+    /**
+     * Builds a `BrokerData` containing the list of values.
+     */
+    BrokerData Build() && { return BrokerData{broker::data{std::move(values_)}}; }
+
+private:
+    broker::vector values_;
+};
+
 } // namespace zeek
diff --git a/src/broker/Manager.cc b/src/broker/Manager.cc
index a7f64ede20..7d81420588 100644
--- a/src/broker/Manager.cc
+++ b/src/broker/Manager.cc
@@ -603,14 +603,14 @@ bool Manager::PublishIdentifier(std::string topic, std::string id) {
         // receiving side, but not sure what use that would be.
         return false;
 
-    auto data = detail::val_to_data(val.get());
+    auto data = BrokerData{};
 
-    if ( ! data ) {
+    if ( ! data.Convert(val) ) {
         Error("Failed to publish ID with unsupported type: %s (%s)", id.c_str(), type_name(val->GetType()->Tag()));
         return false;
     }
 
-    broker::zeek::IdentifierUpdate msg(std::move(id), std::move(*data));
+    broker::zeek::IdentifierUpdate msg(std::move(id), std::move(data.value_));
     DBG_LOG(DBG_BROKER, "Publishing id-update: %s", RenderMessage(topic, msg.as_data()).c_str());
     bstate->endpoint.publish(std::move(topic), msg.move_data());
     ++statistics.num_ids_outgoing;
@@ -888,7 +888,7 @@ RecordVal* Manager::MakeEvent(ValPList* args, zeek::detail::Frame* frame) {
         if ( same_type(got_type, detail::DataVal::ScriptDataType()) )
             data_val = {NewRef{}, (*args)[i]->AsRecordVal()};
         else
-            data_val = detail::make_data_val((*args)[i]);
+            data_val = BrokerData::ToRecordVal((*args)[i]);
 
         if ( ! data_val->HasField(0) ) {
             rval->Remove(0);
@@ -1026,10 +1026,11 @@ void Manager::ProcessStoreEventInsertUpdate(const TableValPtr& table, const std:
 
     const auto& its = table->GetType()->AsTableType()->GetIndexTypes();
     ValPtr zeek_key;
+    auto key_copy = key;
     if ( its.size() == 1 )
-        zeek_key = detail::data_to_val(key, its[0].get());
+        zeek_key = detail::data_to_val(key_copy, its[0].get());
     else
-        zeek_key = detail::data_to_val(key, table->GetType()->AsTableType()->GetIndices().get());
+        zeek_key = detail::data_to_val(key_copy, table->GetType()->AsTableType()->GetIndices().get());
 
     if ( ! zeek_key ) {
         reporter->Error(
@@ -1045,7 +1046,8 @@ void Manager::ProcessStoreEventInsertUpdate(const TableValPtr& table, const std:
     }
 
     // it is a table
-    auto zeek_value = detail::data_to_val(data, table->GetType()->Yield().get());
+    auto data_copy = data;
+    auto zeek_value = detail::data_to_val(data_copy, table->GetType()->Yield().get());
     if ( ! zeek_value ) {
         reporter->Error(
             "ProcessStoreEvent %s: could not convert value \"%s\" for key \"%s\" in "
@@ -1204,7 +1206,8 @@ void Manager::ProcessMessage(std::string_view topic, broker::zeek::Event& ev) {
     for ( size_t i = 0; i < args.size(); ++i ) {
         auto got_type = args[i].get_type_name();
         const auto& expected_type = arg_types[i];
-        auto val = detail::data_to_val(args[i], expected_type.get());
+        auto arg = args[i];
+        auto val = detail::data_to_val(arg, expected_type.get());
 
         if ( val )
             vl.emplace_back(std::move(val));
@@ -1254,13 +1257,15 @@ bool Manager::ProcessMessage(std::string_view, broker::zeek::LogCreate& lc) {
         return false;
     }
 
-    auto stream_id = detail::data_to_val(std::move(lc.stream_id()), log_id_type);
+    auto wrapped_stream_id = broker::data{lc.stream_id()};
+    auto stream_id = detail::data_to_val(wrapped_stream_id, log_id_type);
     if ( ! stream_id ) {
         reporter->Warning("failed to unpack remote log stream id");
         return false;
     }
 
-    auto writer_id = detail::data_to_val(std::move(lc.writer_id()), writer_id_type);
+    auto wrapped_writer_id = broker::data{lc.writer_id()};
+    auto writer_id = detail::data_to_val(wrapped_writer_id, writer_id_type);
     if ( ! writer_id ) {
         reporter->Warning("failed to unpack remote log writer id");
         return false;
@@ -1315,7 +1320,8 @@ bool Manager::ProcessMessage(std::string_view, broker::zeek::LogWrite& lw) {
     auto& stream_id_name = lw.stream_id().name;
 
     // Get stream ID.
-    auto stream_id = detail::data_to_val(std::move(lw.stream_id()), log_id_type);
+    auto wrapped_stream_id = broker::data{lw.stream_id()};
+    auto stream_id = detail::data_to_val(wrapped_stream_id, log_id_type);
 
     if ( ! stream_id ) {
         reporter->Warning("failed to unpack remote log stream id: %s", stream_id_name.data());
@@ -1323,7 +1329,8 @@ bool Manager::ProcessMessage(std::string_view, broker::zeek::LogWrite& lw) {
     }
 
     // Get writer ID.
-    auto writer_id = detail::data_to_val(std::move(lw.writer_id()), writer_id_type);
+    auto wrapped_writer_id = broker::data{lw.writer_id()};
+    auto writer_id = detail::data_to_val(wrapped_writer_id, writer_id_type);
     if ( ! writer_id ) {
         reporter->Warning("failed to unpack remote log writer id for stream: %s", stream_id_name.data());
         return false;
@@ -1383,7 +1390,7 @@ bool Manager::ProcessMessage(std::string_view, broker::zeek::IdentifierUpdate& i
         return false;
     }
 
-    auto val = detail::data_to_val(std::move(id_value), id->GetType().get());
+    auto val = detail::data_to_val(id_value, id->GetType().get());
 
     if ( ! val ) {
         reporter->Error("Failed to receive ID with unsupported type: %s (%s)", id_name.c_str(),
@@ -1516,8 +1523,10 @@ void Manager::ProcessStoreResponse(detail::StoreHandleVal* s, broker::store::res
         return;
     }
 
-    if ( response.answer )
-        request->second->Result(detail::query_result(detail::make_data_val(std::move(*response.answer))));
+    if ( response.answer ) {
+        BrokerData tmp{std::move(*response.answer)};
+        request->second->Result(detail::query_result(std::move(tmp).ToRecordVal()));
+    }
     else if ( response.answer.error() == broker::ec::request_timeout ) {
         // Fine, trigger's timeout takes care of things.
     }
@@ -1604,11 +1613,12 @@ void Manager::BrokerStoreToZeekTable(const std::string& name, const detail::Stor
     table->DisableChangeNotifications();
 
     for ( const auto& key : *set ) {
-        ValPtr zeek_key;
+        auto zeek_key = ValPtr{};
+        auto key_copy = key;
         if ( its.size() == 1 )
-            zeek_key = detail::data_to_val(key, its[0].get());
+            zeek_key = detail::data_to_val(key_copy, its[0].get());
         else
-            zeek_key = detail::data_to_val(key, table->GetType()->AsTableType()->GetIndices().get());
+            zeek_key = detail::data_to_val(key_copy, table->GetType()->AsTableType()->GetIndices().get());
 
         if ( ! zeek_key ) {
             reporter->Error(
diff --git a/src/broker/Manager.h b/src/broker/Manager.h
index 2b98aae520..7384fe6218 100644
--- a/src/broker/Manager.h
+++ b/src/broker/Manager.h
@@ -14,6 +14,7 @@
 #include <unordered_map>
 
 #include "zeek/IntrusivePtr.h"
+#include "zeek/broker/Data.h"
 #include "zeek/iosource/IOSource.h"
 #include "zeek/logging/WriterBackend.h"
 
@@ -179,6 +180,15 @@ public:
      */
     bool PublishEvent(std::string topic, std::string name, broker::vector args, double ts = run_state::network_time);
 
+    /**
+     * @copydoc PublishEvent(std::string, std::string, broker::vector, double)
+     */
+    bool PublishEvent(std::string topic, std::string name, BrokerData args, double ts = run_state::network_time) {
+        if ( ! args.AsView().IsList() )
+            return false;
+        return PublishEvent(std::move(topic), std::move(name), std::move(broker::get<broker::vector>(args.value_)), ts);
+    }
+
     /**
      * Send an event to any interested peers.
      * @param topic a topic string associated with the message.
diff --git a/src/broker/Store.cc b/src/broker/Store.cc
index 7ff0591092..ac8fb62559 100644
--- a/src/broker/Store.cc
+++ b/src/broker/Store.cc
@@ -23,6 +23,12 @@ EnumValPtr query_status(bool success) {
     return rval;
 }
 
+void StoreHandleVal::Put(BrokerData&& key, BrokerData&& value, std::optional<BrokerTimespan> expiry) {
+    store.put(std::move(key).value_, std::move(value).value_, expiry);
+}
+
+void StoreHandleVal::Erase(BrokerData&& key) { store.erase(std::move(key).value_); }
+
 void StoreHandleVal::ValDescribe(ODesc* d) const {
     d->Add("broker::store::");
 
@@ -34,12 +40,12 @@ void StoreHandleVal::ValDescribe(ODesc* d) const {
 
 IMPLEMENT_OPAQUE_VALUE(StoreHandleVal)
 
-broker::expected<broker::data> StoreHandleVal::DoSerialize() const {
+std::optional<BrokerData> StoreHandleVal::DoSerialize() const {
     // Cannot serialize.
-    return broker::ec::invalid_data;
+    return std::nullopt;
 }
 
-bool StoreHandleVal::DoUnserialize(const broker::data& data) {
+bool StoreHandleVal::DoUnserialize(BrokerDataView) {
     // Cannot unserialize.
     return false;
 }
diff --git a/src/broker/Store.h b/src/broker/Store.h
index 7cb2d1aa52..fd5f5c317a 100644
--- a/src/broker/Store.h
+++ b/src/broker/Store.h
@@ -8,6 +8,7 @@
 #include "zeek/Expr.h"
 #include "zeek/OpaqueVal.h"
 #include "zeek/Trigger.h"
+#include "zeek/broker/Data.h"
 #include "zeek/broker/data.bif.h"
 #include "zeek/broker/store.bif.h"
 
@@ -110,6 +111,10 @@ public:
           forward_to{},
           have_store{true} {}
 
+    void Put(BrokerData&& key, BrokerData&& value, std::optional<BrokerTimespan> expiry = std::nullopt);
+
+    void Erase(BrokerData&& key);
+
     void ValDescribe(ODesc* d) const override;
 
     broker::store store;
diff --git a/src/broker/data.bif b/src/broker/data.bif
index e25006cd65..8a4b914a73 100644
--- a/src/broker/data.bif
+++ b/src/broker/data.bif
@@ -52,7 +52,7 @@ function Broker::__opaque_clone_through_serialization%(d: any%): any
         	return zeek::val_mgr->False();
 		}
 
-	return OpaqueVal::Unserialize(std::move(*x));
+	return OpaqueVal::Unserialize(x->AsView());
 	%}
 
 function Broker::__set_create%(%): Broker::Data
diff --git a/src/file_analysis/analyzer/x509/X509.cc b/src/file_analysis/analyzer/x509/X509.cc
index af6835edb8..f79f3099f3 100644
--- a/src/file_analysis/analyzer/x509/X509.cc
+++ b/src/file_analysis/analyzer/x509/X509.cc
@@ -16,6 +16,7 @@
 #endif
 
 #include "zeek/Event.h"
+#include "zeek/broker/Data.h"
 #include "zeek/digest.h"
 #include "zeek/file_analysis/File.h"
 #include "zeek/file_analysis/Manager.h"
@@ -551,27 +552,28 @@ ValPtr X509Val::DoClone(CloneState* state) {
 
 IMPLEMENT_OPAQUE_VALUE(X509Val)
 
-broker::expected<broker::data> X509Val::DoSerialize() const {
+std::optional<BrokerData> X509Val::DoSerialize() const {
     unsigned char* buf = nullptr;
     int length = i2d_X509(certificate, &buf);
 
     if ( length < 0 )
-        return broker::ec::invalid_data;
+        return std::nullopt;
 
-    auto d = std::string(reinterpret_cast<const char*>(buf), length);
+    auto result = BrokerData::FromString(reinterpret_cast<const char*>(buf), static_cast<size_t>(length));
     OPENSSL_free(buf);
 
-    return {std::move(d)};
+    return std::move(result);
 }
 
-bool X509Val::DoUnserialize(const broker::data& data) {
-    auto s = broker::get_if<std::string>(&data);
-    if ( ! s )
+bool X509Val::DoUnserialize(BrokerDataView data) {
+    if ( ! data.IsString() )
         return false;
 
-    auto opensslbuf = reinterpret_cast<const unsigned char*>(s->data());
-    certificate = d2i_X509(NULL, &opensslbuf, s->size());
-    return (certificate != nullptr);
+    auto s = data.ToString();
+
+    auto opensslbuf = reinterpret_cast<const unsigned char*>(s.data());
+    certificate = d2i_X509(NULL, &opensslbuf, s.size());
+    return certificate != nullptr;
 }
 
 } // namespace zeek::file_analysis::detail
diff --git a/src/logging/Manager.cc b/src/logging/Manager.cc
index d4670e6979..d7fb723a70 100644
--- a/src/logging/Manager.cc
+++ b/src/logging/Manager.cc
@@ -67,11 +67,9 @@ ValPtr LogDelayTokenVal::DoClone(CloneState* state) {
 }
 
 // Delay tokens are only valid on the same worker.
-broker::expected<broker::data> LogDelayTokenVal::DoSerialize() const {
-    return broker::make_error(broker::ec::invalid_data, "cannot serialize delay tokens");
-}
+std::optional<BrokerData> LogDelayTokenVal::DoSerialize() const { return std::nullopt; }
 
-bool LogDelayTokenVal::DoUnserialize(const broker::data&) { return false; }
+bool LogDelayTokenVal::DoUnserialize(BrokerDataView) { return false; }
 
 IMPLEMENT_OPAQUE_VALUE(LogDelayTokenVal)
 
diff --git a/src/probabilistic/BitVector.cc b/src/probabilistic/BitVector.cc
index 3d0dafe655..8b7a9e428e 100644
--- a/src/probabilistic/BitVector.cc
+++ b/src/probabilistic/BitVector.cc
@@ -2,12 +2,12 @@
 
 #include "zeek/probabilistic/BitVector.h"
 
-#include <broker/data.hh>
 #include <openssl/sha.h>
 #include <cassert>
 #include <cstring>
 #include <limits>
 
+#include "zeek/broker/Data.h"
 #include "zeek/digest.h"
 
 namespace zeek::probabilistic::detail {
@@ -413,39 +413,43 @@ uint64_t BitVector::Hash() const {
     return digest;
 }
 
-broker::expected<broker::data> BitVector::Serialize() const {
-    broker::vector v = {static_cast<uint64_t>(num_bits), static_cast<uint64_t>(bits.size())};
-    v.reserve(2 + bits.size());
+std::optional<BrokerData> BitVector::Serialize() const {
+    BrokerListBuilder builder;
+    builder.Reserve(2 + bits.size());
 
-    for ( size_t i = 0; i < bits.size(); ++i )
-        v.emplace_back(static_cast<uint64_t>(bits[i]));
+    builder.AddCount(num_bits);
+    builder.AddCount(bits.size());
 
-    return {std::move(v)};
+    for ( auto bit : bits )
+        builder.AddCount(bit);
+
+    return std::move(builder).Build();
 }
 
-std::unique_ptr<BitVector> BitVector::Unserialize(const broker::data& data) {
-    auto v = broker::get_if<broker::vector>(&data);
-    if ( ! (v && v->size() >= 2) )
+std::unique_ptr<BitVector> BitVector::Unserialize(BrokerDataView data) {
+    if ( ! data.IsList() )
         return nullptr;
 
-    auto num_bits = broker::get_if<uint64_t>(&(*v)[0]);
-    auto size = broker::get_if<uint64_t>(&(*v)[1]);
+    auto v = data.ToList();
 
-    if ( ! (num_bits && size) )
+    if ( v.Size() < 2 || ! v[0].IsCount() || ! v[1].IsCount() )
         return nullptr;
 
-    if ( v->size() != 2 + *size )
+    auto num_bits = v[0].ToCount();
+    auto size = v[1].ToCount();
+
+    if ( v.Size() != 2 + size )
         return nullptr;
 
-    auto bv = std::unique_ptr<BitVector>(new BitVector());
-    bv->num_bits = *num_bits;
+    auto bv = std::make_unique<BitVector>();
+    bv->num_bits = num_bits;
 
-    for ( size_t i = 0; i < *size; ++i ) {
-        auto x = broker::get_if<uint64_t>(&(*v)[2 + i]);
-        if ( ! x )
+    for ( size_t i = 0; i < size; ++i ) {
+        auto x = v[2 + i];
+        if ( ! x.IsCount() )
             return nullptr;
 
-        bv->bits.push_back(*x);
+        bv->bits.push_back(x.ToCount());
     }
 
     return bv;
diff --git a/src/probabilistic/BitVector.h b/src/probabilistic/BitVector.h
index f7ce9235d4..5ddd582e63 100644
--- a/src/probabilistic/BitVector.h
+++ b/src/probabilistic/BitVector.h
@@ -2,14 +2,17 @@
 
 #pragma once
 
-#include <broker/expected.hh>
 #include <iterator>
 #include <memory>
+#include <optional>
 #include <vector>
 
-namespace broker {
-class data;
-}
+namespace zeek {
+
+class BrokerData;
+class BrokerDataView;
+
+} // namespace zeek
 
 namespace zeek::probabilistic::detail {
 
@@ -281,8 +284,8 @@ public:
      */
     uint64_t Hash() const;
 
-    broker::expected<broker::data> Serialize() const;
-    static std::unique_ptr<BitVector> Unserialize(const broker::data& data);
+    std::optional<BrokerData> Serialize() const;
+    static std::unique_ptr<BitVector> Unserialize(BrokerDataView data);
 
 private:
     /**
diff --git a/src/probabilistic/BloomFilter.cc b/src/probabilistic/BloomFilter.cc
index 67abc6f9eb..15908ba55c 100644
--- a/src/probabilistic/BloomFilter.cc
+++ b/src/probabilistic/BloomFilter.cc
@@ -2,12 +2,11 @@
 
 #include "zeek/probabilistic/BloomFilter.h"
 
-#include <broker/data.hh>
-#include <broker/error.hh>
 #include <cmath>
 #include <limits>
 
 #include "zeek/Reporter.h"
+#include "zeek/broker/Data.h"
 #include "zeek/probabilistic/CounterVector.h"
 #include "zeek/util.h"
 
@@ -19,48 +18,52 @@ BloomFilter::BloomFilter(const detail::Hasher* arg_hasher) { hasher = arg_hasher
 
 BloomFilter::~BloomFilter() { delete hasher; }
 
-broker::expected<broker::data> BloomFilter::Serialize() const {
+std::optional<BrokerData> BloomFilter::Serialize() const {
     auto h = hasher->Serialize();
 
     if ( ! h )
-        return broker::ec::invalid_data; // Cannot serialize
+        return std::nullopt; // Cannot serialize
 
     auto d = DoSerialize();
 
     if ( ! d )
-        return broker::ec::invalid_data; // Cannot serialize
+        return std::nullopt; // Cannot serialize
 
-    return {broker::vector{static_cast<uint64_t>(Type()), std::move(*h), std::move(*d)}};
+    BrokerListBuilder builder;
+    builder.Reserve(3);
+    builder.Add(static_cast<uint64_t>(Type()));
+    builder.Add(std::move(*h));
+    builder.Add(std::move(*d));
+    return std::move(builder).Build();
 }
 
-std::unique_ptr<BloomFilter> BloomFilter::Unserialize(const broker::data& data) {
-    auto v = broker::get_if<broker::vector>(&data);
-
-    if ( ! (v && v->size() == 3) )
+std::unique_ptr<BloomFilter> BloomFilter::Unserialize(BrokerDataView data) {
+    if ( ! data.IsList() )
         return nullptr;
 
-    auto type = broker::get_if<uint64_t>(&(*v)[0]);
-    if ( ! type )
-        return nullptr;
+    auto v = data.ToList();
 
-    auto hasher_ = detail::Hasher::Unserialize((*v)[1]);
-    if ( ! hasher_ )
+    if ( v.Size() != 3 || ! v[0].IsCount() )
         return nullptr;
 
     std::unique_ptr<BloomFilter> bf;
 
-    switch ( *type ) {
-        case Basic: bf = std::unique_ptr<BloomFilter>(new BasicBloomFilter()); break;
+    switch ( v[0].ToCount() ) {
+        case Basic: bf.reset(new BasicBloomFilter()); break;
 
-        case Counting: bf = std::unique_ptr<BloomFilter>(new CountingBloomFilter()); break;
+        case Counting: bf.reset(new CountingBloomFilter()); break;
 
         default: reporter->Error("found invalid bloom filter type"); return nullptr;
     }
 
-    if ( ! bf->DoUnserialize((*v)[2]) )
+    if ( ! bf->DoUnserialize(v[2]) )
+        return nullptr;
+
+    bf->hasher = detail::Hasher::Unserialize(v[1]).release();
+
+    if ( ! bf->hasher )
         return nullptr;
 
-    bf->hasher = hasher_.release();
     return bf;
 }
 
@@ -163,12 +166,9 @@ size_t BasicBloomFilter::Count(const zeek::detail::HashKey* key) const {
     return 1;
 }
 
-broker::expected<broker::data> BasicBloomFilter::DoSerialize() const {
-    auto b = bits->Serialize();
-    return b;
-}
+std::optional<BrokerData> BasicBloomFilter::DoSerialize() const { return bits->Serialize(); }
 
-bool BasicBloomFilter::DoUnserialize(const broker::data& data) {
+bool BasicBloomFilter::DoUnserialize(BrokerDataView data) {
     auto b = detail::BitVector::Unserialize(data);
     if ( ! b )
         return false;
@@ -280,12 +280,9 @@ size_t CountingBloomFilter::Count(const zeek::detail::HashKey* key) const {
     return min;
 }
 
-broker::expected<broker::data> CountingBloomFilter::DoSerialize() const {
-    auto c = cells->Serialize();
-    return c;
-}
+std::optional<BrokerData> CountingBloomFilter::DoSerialize() const { return cells->Serialize(); }
 
-bool CountingBloomFilter::DoUnserialize(const broker::data& data) {
+bool CountingBloomFilter::DoUnserialize(BrokerDataView data) {
     auto c = detail::CounterVector::Unserialize(data);
     if ( ! c )
         return false;
diff --git a/src/probabilistic/BloomFilter.h b/src/probabilistic/BloomFilter.h
index d1f52c3fab..8be1eb285a 100644
--- a/src/probabilistic/BloomFilter.h
+++ b/src/probabilistic/BloomFilter.h
@@ -4,7 +4,6 @@
 
 #include "zeek/zeek-config.h"
 
-#include <broker/expected.hh>
 #include <memory>
 #include <string>
 #include <vector>
@@ -12,11 +11,13 @@
 #include "zeek/probabilistic/BitVector.h"
 #include "zeek/probabilistic/Hasher.h"
 
-namespace broker {
-class data;
-}
+namespace zeek {
+class BrokerData;
+class BrokerDataView;
+} // namespace zeek
 
 namespace zeek::probabilistic {
+
 namespace detail {
 class CounterVector;
 }
@@ -104,8 +105,8 @@ public:
      */
     virtual std::string InternalState() const = 0;
 
-    broker::expected<broker::data> Serialize() const;
-    static std::unique_ptr<BloomFilter> Unserialize(const broker::data& data);
+    std::optional<BrokerData> Serialize() const;
+    static std::unique_ptr<BloomFilter> Unserialize(BrokerDataView data);
 
 protected:
     /**
@@ -120,8 +121,8 @@ protected:
      */
     explicit BloomFilter(const detail::Hasher* hasher);
 
-    virtual broker::expected<broker::data> DoSerialize() const = 0;
-    virtual bool DoUnserialize(const broker::data& data) = 0;
+    virtual std::optional<BrokerData> DoSerialize() const = 0;
+    virtual bool DoUnserialize(BrokerDataView data) = 0;
     virtual BloomFilterType Type() const = 0;
 
     const detail::Hasher* hasher;
@@ -200,8 +201,8 @@ protected:
     void Add(const zeek::detail::HashKey* key) override;
     bool Decrement(const zeek::detail::HashKey* key) override;
     size_t Count(const zeek::detail::HashKey* key) const override;
-    broker::expected<broker::data> DoSerialize() const override;
-    bool DoUnserialize(const broker::data& data) override;
+    std::optional<BrokerData> DoSerialize() const override;
+    bool DoUnserialize(BrokerDataView data) override;
     BloomFilterType Type() const override { return BloomFilterType::Basic; }
 
 private:
@@ -263,8 +264,8 @@ protected:
     void Add(const zeek::detail::HashKey* key) override;
     bool Decrement(const zeek::detail::HashKey* key) override;
     size_t Count(const zeek::detail::HashKey* key) const override;
-    broker::expected<broker::data> DoSerialize() const override;
-    bool DoUnserialize(const broker::data& data) override;
+    std::optional<BrokerData> DoSerialize() const override;
+    bool DoUnserialize(BrokerDataView data) override;
     BloomFilterType Type() const override { return BloomFilterType::Counting; }
 
 private:
diff --git a/src/probabilistic/CardinalityCounter.cc b/src/probabilistic/CardinalityCounter.cc
index d9c5b9097d..db8efdfda0 100644
--- a/src/probabilistic/CardinalityCounter.cc
+++ b/src/probabilistic/CardinalityCounter.cc
@@ -2,12 +2,12 @@
 
 #include "zeek/probabilistic/CardinalityCounter.h"
 
-#include <broker/data.hh>
 #include <cmath>
 #include <cstdint>
 #include <utility>
 
 #include "zeek/Reporter.h"
+#include "zeek/broker/Data.h"
 
 namespace zeek::probabilistic::detail {
 
@@ -173,42 +173,45 @@ const std::vector<uint8_t>& CardinalityCounter::GetBuckets() const { return buck
 
 uint64_t CardinalityCounter::GetM() const { return m; }
 
-broker::expected<broker::data> CardinalityCounter::Serialize() const {
-    broker::vector v = {m, V, alpha_m};
-    v.reserve(3 + m);
+std::optional<BrokerData> CardinalityCounter::Serialize() const {
+    BrokerListBuilder builder;
+    builder.Reserve(3 + m);
+    builder.Add(m);
+    builder.Add(V);
+    builder.Add(alpha_m);
 
     for ( size_t i = 0; i < m; ++i )
-        v.emplace_back(static_cast<uint64_t>(buckets[i]));
+        builder.AddCount(buckets[i]);
 
-    return {std::move(v)};
+    return std::move(builder).Build();
 }
 
-std::unique_ptr<CardinalityCounter> CardinalityCounter::Unserialize(const broker::data& data) {
-    auto v = broker::get_if<broker::vector>(&data);
-    if ( ! (v && v->size() >= 3) )
+std::unique_ptr<CardinalityCounter> CardinalityCounter::Unserialize(BrokerDataView data) {
+    if ( ! data.IsList() )
         return nullptr;
 
-    auto m = broker::get_if<uint64_t>(&(*v)[0]);
-    auto V = broker::get_if<uint64_t>(&(*v)[1]);
-    auto alpha_m = broker::get_if<double>(&(*v)[2]);
-
-    if ( ! (m && V && alpha_m) )
-        return nullptr;
-    if ( v->size() != 3 + *m )
+    auto v = data.ToList();
+    if ( v.Size() < 3 || ! IsCount(v[0], v[1]) || ! v[2].IsReal() )
         return nullptr;
 
-    auto cc = std::unique_ptr<CardinalityCounter>(new CardinalityCounter(*m, *V, *alpha_m));
-    if ( *m != cc->m )
-        return nullptr;
-    if ( cc->buckets.size() != *m )
+    auto [m, V] = ToCount(v[0], v[1]);
+    auto alpha_m = v[2].ToReal();
+
+    if ( v.Size() != 3 + m )
         return nullptr;
 
-    for ( size_t i = 0; i < *m; ++i ) {
-        auto x = broker::get_if<uint64_t>(&(*v)[3 + i]);
-        if ( ! x )
+    auto cc = std::unique_ptr<CardinalityCounter>(new CardinalityCounter(m, V, alpha_m));
+    if ( m != cc->m )
+        return nullptr;
+    if ( cc->buckets.size() != m )
+        return nullptr;
+
+    for ( size_t i = 0; i < m; ++i ) {
+        auto x = v[3 + i];
+        if ( ! x.IsCount() )
             return nullptr;
 
-        cc->buckets[i] = *x;
+        cc->buckets[i] = x.ToCount();
     }
 
     return cc;
diff --git a/src/probabilistic/CardinalityCounter.h b/src/probabilistic/CardinalityCounter.h
index 457d684538..68ac83cd46 100644
--- a/src/probabilistic/CardinalityCounter.h
+++ b/src/probabilistic/CardinalityCounter.h
@@ -2,14 +2,15 @@
 
 #pragma once
 
-#include <broker/expected.hh>
 #include <cstdint>
 #include <memory>
+#include <optional>
 #include <vector>
 
-namespace broker {
-class data;
-}
+namespace zeek {
+class BrokerData;
+class BrokerDataView;
+} // namespace zeek
 
 namespace zeek::probabilistic::detail {
 
@@ -89,8 +90,8 @@ public:
      */
     bool Merge(CardinalityCounter* c);
 
-    broker::expected<broker::data> Serialize() const;
-    static std::unique_ptr<CardinalityCounter> Unserialize(const broker::data& data);
+    std::optional<BrokerData> Serialize() const;
+    static std::unique_ptr<CardinalityCounter> Unserialize(BrokerDataView data);
 
 protected:
     /**
diff --git a/src/probabilistic/CounterVector.cc b/src/probabilistic/CounterVector.cc
index fb3bab6409..21239146ee 100644
--- a/src/probabilistic/CounterVector.cc
+++ b/src/probabilistic/CounterVector.cc
@@ -2,11 +2,11 @@
 
 #include "zeek/probabilistic/CounterVector.h"
 
-#include <broker/data.hh>
-#include <broker/error.hh>
 #include <cassert>
 #include <limits>
+#include <memory>
 
+#include "zeek/broker/Data.h"
 #include "zeek/probabilistic/BitVector.h"
 #include "zeek/util.h"
 
@@ -138,27 +138,34 @@ CounterVector operator|(const CounterVector& x, const CounterVector& y) {
 
 uint64_t CounterVector::Hash() const { return bits->Hash(); }
 
-broker::expected<broker::data> CounterVector::Serialize() const {
+std::optional<BrokerData> CounterVector::Serialize() const {
     auto b = bits->Serialize();
     if ( ! b )
-        return broker::ec::invalid_data; // Cannot serialize
+        return std::nullopt; // Cannot serialize
 
-    return {broker::vector{static_cast<uint64_t>(width), std::move(*b)}};
+    BrokerListBuilder builder;
+    builder.Reserve(2);
+    builder.AddCount(width);
+    builder.Add(std::move(*b));
+    return std::move(builder).Build();
 }
 
-std::unique_ptr<CounterVector> CounterVector::Unserialize(const broker::data& data) {
-    auto v = broker::get_if<broker::vector>(&data);
-    if ( ! (v && v->size() >= 2) )
+std::unique_ptr<CounterVector> CounterVector::Unserialize(BrokerDataView data) {
+    if ( ! data.IsList() )
         return nullptr;
 
-    auto width = broker::get_if<uint64_t>(&(*v)[0]);
-    auto bits = BitVector::Unserialize((*v)[1]);
-
-    if ( ! (width && bits) )
+    auto v = data.ToList();
+    if ( v.Size() < 2 || ! v[0].IsCount() )
         return nullptr;
 
-    auto cv = std::unique_ptr<CounterVector>(new CounterVector());
-    cv->width = *width;
+    auto width = v[0].ToCount();
+    auto bits = BitVector::Unserialize(v[1]);
+
+    if ( ! bits )
+        return nullptr;
+
+    auto cv = std::unique_ptr<CounterVector>{new CounterVector};
+    cv->width = width;
     cv->bits = bits.release();
     return cv;
 }
diff --git a/src/probabilistic/CounterVector.h b/src/probabilistic/CounterVector.h
index b8ca73de9d..33b4886332 100644
--- a/src/probabilistic/CounterVector.h
+++ b/src/probabilistic/CounterVector.h
@@ -4,14 +4,15 @@
 
 #include "zeek/zeek-config.h"
 
-#include <broker/expected.hh>
 #include <cstddef>
 #include <cstdint>
 #include <memory>
+#include <optional>
 
-namespace broker {
-class data;
-}
+namespace zeek {
+class BrokerData;
+class BrokerDataView;
+} // namespace zeek
 
 namespace zeek::probabilistic::detail {
 
@@ -149,8 +150,8 @@ public:
      */
     uint64_t Hash() const;
 
-    broker::expected<broker::data> Serialize() const;
-    static std::unique_ptr<CounterVector> Unserialize(const broker::data& data);
+    std::optional<BrokerData> Serialize() const;
+    static std::unique_ptr<CounterVector> Unserialize(BrokerDataView data);
 
 protected:
     friend CounterVector operator|(const CounterVector& x, const CounterVector& y);
diff --git a/src/probabilistic/Hasher.cc b/src/probabilistic/Hasher.cc
index 47fb74ab98..dcf159eecb 100644
--- a/src/probabilistic/Hasher.cc
+++ b/src/probabilistic/Hasher.cc
@@ -2,13 +2,13 @@
 
 #include "zeek/probabilistic/Hasher.h"
 
-#include <broker/data.hh>
 #include <highwayhash/sip_hash.h>
 #include <openssl/evp.h>
 #include <typeinfo>
 
 #include "zeek/NetVar.h"
 #include "zeek/Var.h"
+#include "zeek/broker/Data.h"
 #include "zeek/digest.h"
 
 namespace zeek::probabilistic::detail {
@@ -45,30 +45,33 @@ Hasher::Hasher(size_t arg_k, seed_t arg_seed) {
     seed = arg_seed;
 }
 
-broker::expected<broker::data> Hasher::Serialize() const {
-    return {broker::vector{static_cast<uint64_t>(Type()), static_cast<uint64_t>(k), seed.h[0], seed.h[1]}};
+std::optional<BrokerData> Hasher::Serialize() const {
+    BrokerListBuilder builder;
+    builder.Reserve(4);
+    builder.AddCount(static_cast<unsigned>(Type()));
+    builder.AddCount(k);
+    builder.AddCount(seed.h[0]);
+    builder.AddCount(seed.h[1]);
+    return std::move(builder).Build();
 }
 
-std::unique_ptr<Hasher> Hasher::Unserialize(const broker::data& data) {
-    auto v = broker::get_if<broker::vector>(&data);
-
-    if ( ! (v && v->size() == 4) )
+std::unique_ptr<Hasher> Hasher::Unserialize(BrokerDataView data) {
+    if ( ! data.IsList() )
         return nullptr;
 
-    auto type = broker::get_if<uint64_t>(&(*v)[0]);
-    auto k = broker::get_if<uint64_t>(&(*v)[1]);
-    auto h1 = broker::get_if<uint64_t>(&(*v)[2]);
-    auto h2 = broker::get_if<uint64_t>(&(*v)[3]);
+    auto v = data.ToList();
 
-    if ( ! (type && k && h1 && h2) )
+    if ( v.Size() != 4 || ! IsCount(v[0], v[1], v[2], v[3]) )
         return nullptr;
 
+    auto [type, k, h1, h2] = ToCount(v[0], v[1], v[2], v[3]);
+
     std::unique_ptr<Hasher> hasher;
 
-    switch ( *type ) {
-        case Default: hasher = std::unique_ptr<Hasher>(new DefaultHasher(*k, {*h1, *h2})); break;
+    switch ( type ) {
+        case Default: hasher.reset(new DefaultHasher(k, {h1, h2})); break;
 
-        case Double: hasher = std::unique_ptr<Hasher>(new DoubleHasher(*k, {*h1, *h2})); break;
+        case Double: hasher.reset(new DoubleHasher(k, {h1, h2})); break;
     }
 
     // Note that the derived classed don't hold any further state of
diff --git a/src/probabilistic/Hasher.h b/src/probabilistic/Hasher.h
index 2ddee369f1..53f46e75fa 100644
--- a/src/probabilistic/Hasher.h
+++ b/src/probabilistic/Hasher.h
@@ -7,9 +7,10 @@
 
 #include "zeek/Hash.h"
 
-namespace broker {
-class data;
-}
+namespace zeek {
+class BrokerData;
+class BrokerDataView;
+} // namespace zeek
 
 namespace zeek::probabilistic::detail {
 
@@ -106,8 +107,8 @@ public:
      */
     seed_t Seed() const { return seed; }
 
-    broker::expected<broker::data> Serialize() const;
-    static std::unique_ptr<Hasher> Unserialize(const broker::data& data);
+    std::optional<BrokerData> Serialize() const;
+    static std::unique_ptr<Hasher> Unserialize(BrokerDataView data);
 
 protected:
     Hasher() {}
@@ -179,8 +180,8 @@ public:
 
     friend bool operator!=(const UHF& x, const UHF& y) { return ! (x == y); }
 
-    broker::expected<broker::data> Serialize() const;
-    static UHF Unserialize(const broker::data& data);
+    std::optional<BrokerData> Serialize() const;
+    static UHF Unserialize(BrokerDataView data);
 
 private:
     static size_t compute_seed(Hasher::seed_t seed);
diff --git a/src/probabilistic/Topk.cc b/src/probabilistic/Topk.cc
index 69e923eddc..d62df0c6cc 100644
--- a/src/probabilistic/Topk.cc
+++ b/src/probabilistic/Topk.cc
@@ -360,69 +360,59 @@ void TopkVal::IncrementCounter(Element* e, unsigned int count) {
 
 IMPLEMENT_OPAQUE_VALUE(TopkVal)
 
-broker::expected<broker::data> TopkVal::DoSerialize() const {
-    broker::vector d = {size, numElements, pruned};
+std::optional<BrokerData> TopkVal::DoSerialize() const {
+    BrokerListBuilder builder;
+    builder.Reserve(8);
+
+    builder.Add(size);
+    builder.Add(numElements);
+    builder.Add(pruned);
 
     if ( type ) {
         auto t = SerializeType(type);
         if ( ! t )
-            return broker::ec::invalid_data;
+            return std::nullopt;
 
-        d.emplace_back(std::move(*t));
+        builder.Add(std::move(*t));
     }
     else
-        d.emplace_back(broker::none());
+        builder.AddNil();
 
     uint64_t i = 0;
-    std::list<Bucket*>::const_iterator it = buckets.begin();
-    while ( it != buckets.end() ) {
-        Bucket* b = *it;
-        uint32_t elements_count = b->elements.size();
+    for ( const auto* b : buckets ) {
+        builder.AddCount(b->elements.size());
+        builder.AddCount(b->count);
 
-        d.emplace_back(static_cast<uint64_t>(b->elements.size()));
-        d.emplace_back(b->count);
+        for ( const auto* element : b->elements ) {
+            builder.AddCount(element->epsilon);
+            BrokerData val;
+            if ( ! val.Convert(element->value) )
+                return std::nullopt;
 
-        std::list<Element*>::const_iterator eit = b->elements.begin();
-        while ( eit != b->elements.end() ) {
-            Element* element = *eit;
-            d.emplace_back(element->epsilon);
-            auto v = Broker::detail::val_to_data(element->value.get());
-            if ( ! v )
-                return broker::ec::invalid_data;
+            builder.Add(std::move(val));
 
-            d.emplace_back(*v);
-
-            eit++;
             i++;
         }
-
-        it++;
     }
 
     assert(i == numElements);
-    return {std::move(d)};
+    return std::move(builder).Build();
 }
 
-bool TopkVal::DoUnserialize(const broker::data& data) {
-    auto v = broker::get_if<broker::vector>(&data);
-
-    if ( ! (v && v->size() >= 4) )
+bool TopkVal::DoUnserialize(BrokerDataView data) {
+    if ( ! data.IsList() )
         return false;
 
-    auto size_ = broker::get_if<uint64_t>(&(*v)[0]);
-    auto numElements_ = broker::get_if<uint64_t>(&(*v)[1]);
-    auto pruned_ = broker::get_if<bool>(&(*v)[2]);
-
-    if ( ! (size_ && numElements_ && pruned_) )
+    auto v = data.ToList();
+    if ( v.Size() < 4 || ! v[0].IsCount() || ! v[1].IsCount() || ! v[2].IsBool() )
         return false;
 
-    size = *size_;
-    numElements = *numElements_;
-    pruned = *pruned_;
+    size = v[0].ToCount();
+    numElements = v[1].ToCount();
+    pruned = v[2].ToBool();
 
-    auto no_type = broker::get_if<broker::none>(&(*v)[3]);
-    if ( ! no_type ) {
-        auto t = UnserializeType((*v)[3]);
+    if ( ! v[3].IsNil() ) {
+        auto t = UnserializeType(v[3]);
 
         if ( ! t )
             return false;
@@ -430,29 +420,48 @@ bool TopkVal::DoUnserialize(const broker::data& data) {
         Typify(t);
     }
 
+    bool ok = true;
+    auto index = size_t{4}; // Index into v.
+    auto atEnd = [&v, &index] { return index >= v.Size(); };
+    auto nextCount = [&v, &ok, &index]() -> uint64_t {
+        if ( index >= v.Size() || ! v[index].IsCount() ) {
+            ok = false;
+            return 0;
+        }
+        auto res = v[index].ToCount();
+        ++index;
+        return res;
+    };
+
     uint64_t i = 0;
-    uint64_t idx = 4;
-
     while ( i < numElements ) {
-        auto elements_count = broker::get_if<uint64_t>(&(*v)[idx++]);
-        auto count = broker::get_if<uint64_t>(&(*v)[idx++]);
-
-        if ( ! (elements_count && count) )
+        auto elements_count = nextCount();
+        if ( ! ok )
             return false;
 
-        Bucket* b = new Bucket();
-        b->count = *count;
+        auto count = nextCount();
+        if ( ! ok )
+            return false;
+
+        auto* b = new Bucket();
+        b->count = count;
         b->bucketPos = buckets.insert(buckets.end(), b);
 
-        for ( uint64_t j = 0; j < *elements_count; j++ ) {
-            auto epsilon = broker::get_if<uint64_t>(&(*v)[idx++]);
-            auto val = Broker::detail::data_to_val((*v)[idx++], type.get());
+        for ( uint64_t j = 0; j < elements_count; j++ ) {
+            auto epsilon = nextCount();
+            if ( ! ok )
+                return false;
 
-            if ( ! (epsilon && val) )
+            if ( atEnd() )
+                return false;
+
+            auto val = v[index++].ToVal(type.get());
+
+            if ( ! val )
                 return false;
 
             Element* e = new Element();
-            e->epsilon = *epsilon;
+            e->epsilon = epsilon;
             e->value = std::move(val);
             e->parent = b;
 
@@ -463,12 +472,13 @@ bool TopkVal::DoUnserialize(const broker::data& data) {
 
             elementDict->Insert(key, e);
             delete key;
-
-            i++;
+            ++i;
         }
     }
 
-    assert(i == numElements);
+    if ( ! atEnd() )
+        return false;
+
     return true;
 }
 
diff --git a/src/script_opt/CPP/Func.cc b/src/script_opt/CPP/Func.cc
index 6e883e246f..1f9aad3229 100644
--- a/src/script_opt/CPP/Func.cc
+++ b/src/script_opt/CPP/Func.cc
@@ -2,7 +2,7 @@
 
 #include "zeek/script_opt/CPP/Func.h"
 
-#include <broker/error.hh>
+#include <string_view>
 
 #include "zeek/Desc.h"
 #include "zeek/broker/Data.h"
@@ -46,27 +46,29 @@ CPPLambdaFunc::CPPLambdaFunc(string _name, FuncTypePtr ft, CPPStmtPtr _l_body)
     l_body = std::move(_l_body);
 }
 
-broker::expected<broker::data> CPPLambdaFunc::SerializeCaptures() const {
+std::optional<BrokerData> CPPLambdaFunc::SerializeCaptures() const {
+    using namespace std::literals;
+
+    auto name = "CopyFrame"sv;
     auto vals = l_body->SerializeLambdaCaptures();
 
-    broker::vector rval;
-    rval.emplace_back(string("CopyFrame"));
-
-    broker::vector body;
+    BrokerListBuilder body;
+    body.Reserve(vals.size());
 
     for ( const auto& val : vals ) {
-        auto expected = Broker::detail::val_to_data(val.get());
-        if ( ! expected )
-            return broker::ec::invalid_data;
+        BrokerData tmp;
+        if ( ! tmp.Convert(val) )
+            return std::nullopt;
 
         TypeTag tag = val->GetType()->Tag();
-        broker::vector val_tuple{std::move(*expected), static_cast<broker::integer>(tag)};
-        body.emplace_back(std::move(val_tuple));
+        body.AddList(std::move(tmp), static_cast<int64_t>(tag));
     }
 
-    rval.emplace_back(std::move(body));
-
-    return {std::move(rval)};
+    BrokerListBuilder builder;
+    builder.Reserve(2);
+    builder.AddString(name.data(), name.size());
+    builder.Add(std::move(body));
+    return std::move(builder).Build();
 }
 
 void CPPLambdaFunc::SetCaptures(Frame* f) { l_body->SetLambdaCaptures(f); }
diff --git a/src/script_opt/CPP/Func.h b/src/script_opt/CPP/Func.h
index acf54eeb0f..5b37866783 100644
--- a/src/script_opt/CPP/Func.h
+++ b/src/script_opt/CPP/Func.h
@@ -83,7 +83,7 @@ public:
 
 protected:
     // Methods related to sending lambdas via Broker.
-    broker::expected<broker::data> SerializeCaptures() const override;
+    std::optional<BrokerData> SerializeCaptures() const override;
     void SetCaptures(Frame* f) override;
 
     FuncPtr DoClone() override;