diff --git a/connections/implementation/BUILD b/connections/implementation/BUILD
index 8078dba2..b99f0aaf 100644
--- a/connections/implementation/BUILD
+++ b/connections/implementation/BUILD
@@ -45,6 +45,7 @@ cc_library(
         "base_pcp_handler.cc",
         "ble_advertisement.cc",
         "ble_endpoint_channel.cc",
+        "ble_v2_endpoint_channel.cc",
         "bluetooth_bwu_handler.cc",
         "bluetooth_device_name.cc",
         "bluetooth_endpoint_channel.cc",
@@ -79,6 +80,7 @@ cc_library(
         "base_pcp_handler.h",
         "ble_advertisement.h",
         "ble_endpoint_channel.h",
+        "ble_v2_endpoint_channel.h",
         "bluetooth_bwu_handler.h",
         "bluetooth_device_name.h",
         "bluetooth_endpoint_channel.h",
diff --git a/connections/implementation/base_pcp_handler_test.cc b/connections/implementation/base_pcp_handler_test.cc
index 2fafa616..c924d4ad 100644
--- a/connections/implementation/base_pcp_handler_test.cc
+++ b/connections/implementation/base_pcp_handler_test.cc
@@ -54,6 +54,9 @@ constexpr std::array<char, 6> kFakeMacAddress = {'a', 'b', 'c', 'd', 'e', 'f'};
 
 constexpr BooleanMediumSelector kTestCases[] = {
     BooleanMediumSelector{},
+    BooleanMediumSelector{
+        .ble = true,
+    },
     BooleanMediumSelector{
         .bluetooth = true,
     },
@@ -62,6 +65,19 @@ constexpr BooleanMediumSelector kTestCases[] = {
     },
     BooleanMediumSelector{
         .bluetooth = true,
+        .ble = true,
+    },
+    BooleanMediumSelector{
+        .bluetooth = true,
+        .wifi_lan = true,
+    },
+    BooleanMediumSelector{
+        .ble = true,
+        .wifi_lan = true,
+    },
+    BooleanMediumSelector{
+        .bluetooth = true,
+        .ble = true,
         .wifi_lan = true,
     },
 };
diff --git a/connections/implementation/ble_v2_endpoint_channel.cc b/connections/implementation/ble_v2_endpoint_channel.cc
new file mode 100644
index 00000000..16215ae4
--- /dev/null
+++ b/connections/implementation/ble_v2_endpoint_channel.cc
@@ -0,0 +1,72 @@
+// Copyright 2022 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#include "connections/implementation/ble_v2_endpoint_channel.h"
+
+#include <string>
+#include <utility>
+
+#include "internal/platform/ble_v2.h"
+#include "internal/platform/logging.h"
+
+namespace location {
+namespace nearby {
+namespace connections {
+
+namespace {
+
+OutputStream* GetOutputStreamOrNull(BleV2Socket& socket) {
+  if (socket.GetRemotePeripheral().IsValid()) {
+    return &socket.GetOutputStream();
+  }
+  return nullptr;
+}
+
+InputStream* GetInputStreamOrNull(BleV2Socket& socket) {
+  if (socket.GetRemotePeripheral().IsValid()) {
+    return &socket.GetInputStream();
+  }
+  return nullptr;
+}
+
+}  // namespace
+
+BleV2EndpointChannel::BleV2EndpointChannel(const std::string& service_id,
+                                           const std::string& channel_name,
+                                           BleV2Socket socket)
+    : BaseEndpointChannel(service_id, channel_name,
+                          GetInputStreamOrNull(socket),
+                          GetOutputStreamOrNull(socket)),
+      ble_socket_(std::move(socket)) {}
+
+proto::connections::Medium BleV2EndpointChannel::GetMedium() const {
+  return proto::connections::Medium::BLE;
+}
+
+int BleV2EndpointChannel::GetMaxTransmitPacketSize() const {
+  return kDefaultBleMaxTransmitPacketSize;
+}
+
+void BleV2EndpointChannel::CloseImpl() {
+  Exception status = ble_socket_.Close();
+  if (!status.Ok()) {
+    NEARBY_LOGS(WARNING)
+        << "Failed to close underlying socket for BleEndpointChannel "
+        << GetName() << ": exception=" << status.value;
+  }
+}
+
+}  // namespace connections
+}  // namespace nearby
+}  // namespace location
diff --git a/connections/implementation/ble_v2_endpoint_channel.h b/connections/implementation/ble_v2_endpoint_channel.h
new file mode 100644
index 00000000..8d7a0533
--- /dev/null
+++ b/connections/implementation/ble_v2_endpoint_channel.h
@@ -0,0 +1,49 @@
+// Copyright 2022 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#ifndef CONNECTIONS_IMPLEMENTATION_BLE_V2_ENDPOINT_CHANNEL_H_
+#define CONNECTIONS_IMPLEMENTATION_BLE_V2_ENDPOINT_CHANNEL_H_
+
+#include <string>
+
+#include "connections/implementation/base_endpoint_channel.h"
+#include "internal/platform/ble_v2.h"
+
+namespace location {
+namespace nearby {
+namespace connections {
+
+class BleV2EndpointChannel final : public BaseEndpointChannel {
+ public:
+  // Creates both outgoing and incoming Ble channels.
+  BleV2EndpointChannel(const std::string& service_id,
+                       const std::string& channel_name, BleV2Socket socket);
+
+  proto::connections::Medium GetMedium() const override;
+
+  int GetMaxTransmitPacketSize() const override;
+
+ private:
+  static constexpr int kDefaultBleMaxTransmitPacketSize = 512;  // 512 bytes
+
+  void CloseImpl() override;
+
+  BleV2Socket ble_socket_;
+};
+
+}  // namespace connections
+}  // namespace nearby
+}  // namespace location
+
+#endif  // CONNECTIONS_IMPLEMENTATION_BLE_V2_ENDPOINT_CHANNEL_H_
diff --git a/connections/implementation/mediums/ble_v2.cc b/connections/implementation/mediums/ble_v2.cc
index c8de4a85..a3cd1600 100644
--- a/connections/implementation/mediums/ble_v2.cc
+++ b/connections/implementation/mediums/ble_v2.cc
@@ -64,9 +64,13 @@ BleV2::~BleV2() {
   while (!advertising_infos_.empty()) {
     StopAdvertising(advertising_infos_.begin()->first);
   }
+  while (!server_sockets_.empty()) {
+    StopAcceptingConnections(server_sockets_.begin()->first);
+  }
 
   serial_executor_.Shutdown();
   alarm_executor_.Shutdown();
+  accept_loops_runner_.Shutdown();
 }
 
 bool BleV2::IsAvailable() const {
@@ -165,12 +169,6 @@ bool BleV2::StopAdvertising(const std::string& service_id) {
   gatt_advertisements_.clear();
 
   // Restart the BLE advertisement if there is still an advertiser.
-  // TODO(b/213835576): Check the BLE Connections is off. We set the fake
-  // value for the time being till connections is implemented.
-  bool no_incoming_ble_sockets = true;
-  if (advertising_infos_.empty() && !no_incoming_ble_sockets) {
-    return true;
-  }
   if (!advertising_infos_.empty()) {
     if (!hosted_gatt_characteristics_.empty()) {
       // Set the value of characteristic to empty if there is still an
@@ -186,27 +184,26 @@ bool BleV2::StopAdvertising(const std::string& service_id) {
       }
       hosted_gatt_characteristics_.clear();
     }
-    // Get the next service_id to restart BLE advertisement.
-    const std::string& service_id = advertising_infos_.begin()->first;
-    if (!StartAdvertisingLocked(service_id)) {
+    const std::string& new_service_id = advertising_infos_.begin()->first;
+    if (!StartAdvertisingLocked(new_service_id)) {
       NEARBY_LOGS(ERROR)
           << "Failed to restart BLE advertisement after stopping "
-             "BLE advertisement for service_id="
-          << service_id;
-      advertising_infos_.erase(service_id);
+             "BLE advertisement for new service_id="
+          << new_service_id;
+      advertising_infos_.erase(new_service_id);
       return false;
     }
-    NEARBY_LOGS(INFO) << "Restart BLE advertising with service_id="
-                      << service_id;
-    return true;
+    NEARBY_LOGS(INFO) << "Restart BLE advertising with new service_id="
+                      << new_service_id;
+  } else if (incoming_sockets_.empty()) {
+    // Otherwise, if we aren't restarting the BLE advertisement, then shutdown
+    // the gatt server if it's not in use.
+    NEARBY_LOGS(VERBOSE) << "Aggressively stopping any pre-existing "
+                            "advertisement GATT servers "
+                            "because no incoming BLE sockets are connected.";
+    StopAdvertisementGattServerLocked();
   }
 
-  // If we aren't restarting the BLE advertisement, then shutdown
-  // the gatt server if it's not in use.
-  NEARBY_LOGS(VERBOSE) << "Aggressively stopping any pre-existing "
-                          "advertisement GATT servers "
-                          "because no incoming BLE sockets are connected.";
-  StopAdvertisementGattServerLocked();
   return true;
 }
 
@@ -348,6 +345,152 @@ bool BleV2::IsScanning(const std::string& service_id) const {
   return IsScanningLocked(service_id);
 }
 
+bool BleV2::StartAcceptingConnections(const std::string& service_id,
+                                      AcceptedConnectionCallback callback) {
+  MutexLock lock(&mutex_);
+
+  if (service_id.empty()) {
+    NEARBY_LOGS(INFO)
+        << "Refusing to start accepting BLE connections with empty service id.";
+    return false;
+  }
+
+  if (IsAcceptingConnectionsLocked(service_id)) {
+    NEARBY_LOGS(INFO)
+        << "Refusing to start accepting BLE connections for " << service_id
+        << " because another BLE peripheral socket is already in-progress.";
+    return false;
+  }
+
+  if (!radio_.IsEnabled()) {
+    NEARBY_LOGS(INFO) << "Can't start accepting BLE connections for "
+                      << service_id << " because Bluetooth isn't enabled.";
+    return false;
+  }
+
+  if (!IsAvailableLocked()) {
+    NEARBY_LOGS(INFO) << "Can't start accepting BLE connections for "
+                      << service_id << " because BLE isn't available.";
+    return false;
+  }
+
+  BleV2ServerSocket server_socket = medium_.OpenServerSocket(service_id);
+  if (!server_socket.IsValid()) {
+    NEARBY_LOGS(INFO)
+        << "Failed to start accepting Ble connections for service_id="
+        << service_id;
+    return false;
+  }
+
+  // Mark the fact that there's an in-progress Ble server accepting
+  // connections.
+  auto owned_server_socket =
+      server_sockets_.insert({service_id, std::move(server_socket)})
+          .first->second;
+
+  // Start the accept loop on a dedicated thread - this stays alive and
+  // listening for new incoming connections until StopAcceptingConnections() is
+  // invoked.
+  accept_loops_runner_.Execute(
+      "ble-accept", [this, &service_id, callback = std::move(callback),
+                     server_socket = std::move(owned_server_socket)]() mutable {
+        while (true) {
+          BleV2Socket client_socket = server_socket.Accept();
+          if (!client_socket.IsValid()) {
+            NEARBY_LOGS(WARNING) << "The client socket to accept is invalid.";
+            server_socket.Close();
+            break;
+          }
+          {
+            MutexLock lock(&mutex_);
+            client_socket.SetCloseNotifier([this, service_id]() {
+              MutexLock lock(&mutex_);
+              incoming_sockets_.erase(service_id);
+            });
+            incoming_sockets_.insert({service_id, client_socket});
+          }
+          callback.accepted_cb(std::move(client_socket), service_id);
+        }
+      });
+
+  return true;
+}
+
+bool BleV2::StopAcceptingConnections(const std::string& service_id) {
+  MutexLock lock(&mutex_);
+
+  const auto it = server_sockets_.find(service_id);
+  if (it == server_sockets_.end()) {
+    NEARBY_LOGS(INFO)
+        << "Can't stop accepting BLE connections because it was never started.";
+    return false;
+  }
+
+  // Closing the BleServerSocket will kick off the suicide of the thread
+  // in accept_loops_thread_pool_ that blocks on BleServerSocket.accept().
+  // That may take some time to complete, but there's no particular reason to
+  // wait around for it.
+  auto item = server_sockets_.extract(it);
+
+  // Store a handle to the BleServerSocket, so we can use it after
+  // removing the entry from server_sockets_; making it scoped
+  // is a bonus that takes care of deallocation before we leave this method.
+  BleV2ServerSocket& listening_socket = item.mapped();
+
+  // Regardless of whether or not we fail to close the existing
+  // BleServerSocket, remove it from server_sockets_ so that it
+  // frees up this service for another round.
+
+  // Finally, close the BleServerSocket.
+  if (!listening_socket.Close().Ok()) {
+    NEARBY_LOGS(INFO) << "Failed to close Ble server socket for service_id="
+                      << service_id;
+    return false;
+  }
+
+  return true;
+}
+
+bool BleV2::IsAcceptingConnections(const std::string& service_id) {
+  MutexLock lock(&mutex_);
+  return IsAcceptingConnectionsLocked(service_id);
+}
+
+BleV2Socket BleV2::Connect(const std::string& service_id,
+                           const BleV2Peripheral& peripheral,
+                           CancellationFlag* cancellation_flag) {
+  MutexLock lock(&mutex_);
+  // Socket to return. To allow for NRVO to work, it has to be a single object.
+  BleV2Socket socket;
+
+  if (service_id.empty()) {
+    NEARBY_LOGS(INFO) << "Refusing to create client Ble socket because "
+                         "service_id is empty.";
+    return socket;
+  }
+
+  if (!IsAvailableLocked()) {
+    NEARBY_LOGS(INFO) << "Can't create client Ble socket [service_id="
+                      << service_id << "]; Ble isn't available.";
+    return socket;
+  }
+
+  if (cancellation_flag->Cancelled()) {
+    NEARBY_LOGS(INFO) << "Can't create client Ble socket due to cancel.";
+    return socket;
+  }
+
+  socket = medium_.Connect(service_id,
+                           PowerLevelToTxPowerLevel(PowerLevel::kHighPower),
+                           peripheral, cancellation_flag);
+  if (!socket.IsValid()) {
+    NEARBY_LOGS(INFO) << "Failed to Connect via Ble [service_id=" << service_id
+                      << "]";
+  }
+
+  return socket;
+}
+
 bool BleV2::IsAvailableLocked() const { return medium_.IsValid(); }
 
 bool BleV2::IsAdvertisingLocked(const std::string& service_id) const {
@@ -358,6 +501,10 @@ bool BleV2::IsScanningLocked(const std::string& service_id) const {
   return scanned_service_ids_.contains(service_id);
 }
 
+bool BleV2::IsAcceptingConnectionsLocked(const std::string& service_id) {
+  return server_sockets_.contains(service_id);
+}
+
 bool BleV2::IsAdvertisementGattServerRunningLocked() {
   return gatt_server_ && gatt_server_->IsValid();
 }
@@ -398,6 +545,7 @@ bool BleV2::GenerateAdvertisementCharacteristic(
   std::vector<GattCharacteristic::Property> properties{
       GattCharacteristic::Property::kRead};
 
+  // NOLINTNEXTLINE(google3-legacy-absl-backports)
   absl::optional<GattCharacteristic> gatt_characteristic =
       gatt_server.CreateCharacteristic(
           std::string(mediums::bleutils::kCopresenceServiceUuid),
@@ -647,10 +795,7 @@ bool BleV2::StartGattAdvertisingLocked(
   // to a loss of GATT callbacks for that remote device. The only time a
   // remote device is indefinitely connected to this device's GATT server is
   // when it has a BLE socket connection.
-  // TODO(b/213835576): Check the BLE Connections is off. We set the fake
-  // value for the time being till connections is implemented.
-  bool no_incoming_ble_sockets = true;
-  if (no_incoming_ble_sockets) {
+  if (incoming_sockets_.empty()) {
     NEARBY_LOGS(VERBOSE)
         << "Aggressively stopping any pre-existing advertisement GATT "
            "servers because no incoming BLE sockets are connected";
diff --git a/connections/implementation/mediums/ble_v2.h b/connections/implementation/mediums/ble_v2.h
index ea1e40d6..f58e3105 100644
--- a/connections/implementation/mediums/ble_v2.h
+++ b/connections/implementation/mediums/ble_v2.h
@@ -16,6 +16,7 @@
 #define CORE_INTERNAL_MEDIUMS_BLE_V2_H_
 
 #include <cstdint>
+#include <functional>
 #include <memory>
 #include <string>
 #include <utility>
@@ -29,8 +30,10 @@
 #include "connections/implementation/mediums/bluetooth_radio.h"
 #include "connections/power_level.h"
 #include "internal/platform/ble_v2.h"
+#include "internal/platform/bluetooth_adapter.h"
 #include "internal/platform/byte_array.h"
 #include "internal/platform/cancelable_alarm.h"
+#include "internal/platform/multi_thread_executor.h"
 #include "internal/platform/mutex.h"
 #include "internal/platform/mutex_lock.h"
 #include "internal/platform/scheduled_executor.h"
@@ -46,6 +49,12 @@ class BleV2 final {
  public:
   using DiscoveredPeripheralCallback = mediums::DiscoveredPeripheralCallback;
 
+  // Callback that is invoked when a new connection is accepted.
+  struct AcceptedConnectionCallback {
+    std::function<void(BleV2Socket socket, const std::string& service_id)>
+        accepted_cb = DefaultCallback<BleV2Socket, const std::string&>();
+  };
+
   static constexpr absl::Duration kPeripheralLostTimeout = absl::Seconds(3);
 
   explicit BleV2(BluetoothRadio& bluetooth_radio);
@@ -95,6 +104,26 @@ class BleV2 final {
   bool IsScanning(const std::string& service_id) const
       ABSL_LOCKS_EXCLUDED(mutex_);
 
+  // Starts a worker thread, creates a Ble socket, associates it with a
+  // service id.
+  bool StartAcceptingConnections(const std::string& service_id,
+                                 AcceptedConnectionCallback callback)
+      ABSL_LOCKS_EXCLUDED(mutex_);
+
+  // Closes socket corresponding to a service id.
+  bool StopAcceptingConnections(const std::string& service_id)
+      ABSL_LOCKS_EXCLUDED(mutex_);
+
+  bool IsAcceptingConnections(const std::string& service_id)
+      ABSL_LOCKS_EXCLUDED(mutex_);
+
+  // Establishes connection to Ble peripheral.
+  // Returns socket instance. On success, BleSocket.IsValid() return true.
+  BleV2Socket Connect(const std::string& service_id,
+                      const BleV2Peripheral& peripheral,
+                      CancellationFlag* cancellation_flag)
+      ABSL_LOCKS_EXCLUDED(mutex_);
+
   // Returns true if this object owns a valid platform implementation.
   bool IsMediumValid() const ABSL_LOCKS_EXCLUDED(mutex_) {
     MutexLock lock(&mutex_);
@@ -119,6 +148,11 @@ class BleV2 final {
   bool IsScanningLocked(const std::string& service_id) const
       ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
 
+  // Same as IsListeningForIncomingConnections(), but must be called with
+  // `mutex_` held.
+  bool IsAcceptingConnectionsLocked(const std::string& service_id)
+      ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
+
   bool IsAdvertisementGattServerRunningLocked()
       ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
   bool StartAdvertisementGattServerLocked(const std::string& service_id,
@@ -159,6 +193,8 @@ class BleV2 final {
 
   void RunOnBleThread(Runnable runnable);
 
+  static constexpr int kMaxConcurrentAcceptLoops = 5;
+
   SingleThreadExecutor serial_executor_;
   ScheduledExecutor alarm_executor_;
 
@@ -177,6 +213,20 @@ class BleV2 final {
   std::unique_ptr<CancelableAlarm> lost_alarm_;
   mediums::DiscoveredPeripheralTracker discovered_peripheral_tracker_
       ABSL_GUARDED_BY(mutex_);
+
+  // A thread pool dedicated to running all the accept loops from
+  // StartAcceptingConnections().
+  MultiThreadExecutor accept_loops_runner_{kMaxConcurrentAcceptLoops};
+
+  // A map of service_id -> ServerSocket. If map is non-empty, we
+  // are currently listening for incoming connections.
+  absl::flat_hash_map<std::string, BleV2ServerSocket> server_sockets_
+      ABSL_GUARDED_BY(mutex_);
+
+  // Tracks currently connected incoming sockets. This lets the device know when
+  // it's okay to restart GATT server related operations.
+  absl::flat_hash_map<std::string, BleV2Socket> incoming_sockets_
+      ABSL_GUARDED_BY(mutex_);
 };
 
 }  // namespace connections
diff --git a/connections/implementation/mediums/ble_v2_test.cc b/connections/implementation/mediums/ble_v2_test.cc
index d87db2e7..c320ac3a 100644
--- a/connections/implementation/mediums/ble_v2_test.cc
+++ b/connections/implementation/mediums/ble_v2_test.cc
@@ -27,6 +27,17 @@ namespace nearby {
 namespace connections {
 namespace {
 
+using FeatureFlags = FeatureFlags::Flags;
+
+constexpr FeatureFlags kTestCases[] = {
+    FeatureFlags{
+        .enable_cancellation_flag = true,
+    },
+    FeatureFlags{
+        .enable_cancellation_flag = false,
+    },
+};
+
 constexpr absl::Duration kWaitDuration = absl::Milliseconds(1000);
 constexpr absl::string_view kServiceIDA =
     "com.google.location.nearby.apps.test.a";
@@ -34,13 +45,151 @@ constexpr absl::string_view kServiceIDB =
     "com.google.location.nearby.apps.test.b";
 constexpr absl::string_view kAdvertisementString = "\x0a\x0b\x0c\x0d";
 
-class BleV2Test : public testing::Test {
+class BleV2Test : public testing::TestWithParam<FeatureFlags> {
  protected:
   BleV2Test() { env_.Stop(); }
 
   MediumEnvironment& env_{MediumEnvironment::Instance()};
 };
 
+TEST_P(BleV2Test, CanConnect) {
+  FeatureFlags feature_flags = GetParam();
+  env_.SetFeatureFlags(feature_flags);
+  env_.Start();
+  BluetoothRadio radio_client;
+  BluetoothRadio radio_server;
+  BleV2 ble_client{radio_client};
+  BleV2 ble_server{radio_server};
+  radio_client.Enable();
+  radio_server.Enable();
+  std::string service_id(kServiceIDA);
+  ByteArray advertisement_bytes{std::string(kAdvertisementString)};
+  CountDownLatch discovered_latch(1);
+  CountDownLatch accept_latch(1);
+
+  BleV2Socket socket_for_server;
+  EXPECT_TRUE(ble_server.StartAcceptingConnections(
+      service_id, {
+                      .accepted_cb =
+                          [&socket_for_server, &accept_latch](
+                              BleV2Socket socket, const std::string&) {
+                            socket_for_server = std::move(socket);
+                            accept_latch.CountDown();
+                          },
+                  }));
+
+  ble_server.StartAdvertising(service_id, advertisement_bytes,
+                              PowerLevel::kHighPower,
+                              /*is_fast_advertisement=*/true);
+
+  BleV2Peripheral discovered_peripheral;
+  ble_client.StartScanning(
+      service_id, PowerLevel::kHighPower,
+      mediums::DiscoveredPeripheralCallback{
+          .peripheral_discovered_cb =
+              [&discovered_latch, &discovered_peripheral](
+                  BleV2Peripheral peripheral, const std::string& service_id,
+                  const ByteArray& advertisement_bytes,
+                  bool fast_advertisement) {
+                discovered_peripheral = peripheral;
+                NEARBY_LOG(
+                    INFO,
+                    "Discovered peripheral=%p [impl=%p], fast advertisement=%d",
+                    &peripheral, &peripheral.GetImpl(), fast_advertisement);
+                discovered_latch.CountDown();
+              },
+      });
+  discovered_latch.Await(kWaitDuration).result();
+  ASSERT_TRUE(discovered_peripheral.IsValid());
+
+  CancellationFlag flag;
+  BleV2Socket socket_for_client =
+      ble_client.Connect(service_id, discovered_peripheral, &flag);
+  EXPECT_TRUE(accept_latch.Await(kWaitDuration).result());
+  EXPECT_TRUE(ble_server.StopAcceptingConnections(service_id));
+  EXPECT_TRUE(ble_server.StopAdvertising(service_id));
+  EXPECT_TRUE(socket_for_server.IsValid());
+  EXPECT_TRUE(socket_for_client.IsValid());
+  EXPECT_TRUE(socket_for_server.GetRemotePeripheral().IsValid());
+  EXPECT_TRUE(socket_for_client.GetRemotePeripheral().IsValid());
+  env_.Stop();
+}
+
+TEST_P(BleV2Test, CanCancelConnect) {
+  FeatureFlags feature_flags = GetParam();
+  env_.SetFeatureFlags(feature_flags);
+  env_.Start();
+  BluetoothRadio radio_client;
+  BluetoothRadio radio_server;
+  BleV2 ble_client{radio_client};
+  BleV2 ble_server{radio_server};
+  radio_client.Enable();
+  radio_server.Enable();
+  std::string service_id(kServiceIDA);
+  ByteArray advertisement_bytes{std::string(kAdvertisementString)};
+  CountDownLatch discovered_latch(1);
+  CountDownLatch accept_latch(1);
+
+  BleV2Socket socket_for_server;
+  EXPECT_TRUE(ble_server.StartAcceptingConnections(
+      service_id, {
+                      .accepted_cb =
+                          [&socket_for_server, &accept_latch](
+                              BleV2Socket socket, const std::string&) {
+                            socket_for_server = std::move(socket);
+                            accept_latch.CountDown();
+                          },
+                  }));
+
+  ble_server.StartAdvertising(service_id, advertisement_bytes,
+                              PowerLevel::kHighPower,
+                              /*is_fast_advertisement=*/true);
+
+  BleV2Peripheral discovered_peripheral;
+  ble_client.StartScanning(
+      service_id, PowerLevel::kHighPower,
+      mediums::DiscoveredPeripheralCallback{
+          .peripheral_discovered_cb =
+              [&discovered_latch, &discovered_peripheral](
+                  BleV2Peripheral peripheral, const std::string& service_id,
+                  const ByteArray& advertisement_bytes,
+                  bool fast_advertisement) {
+                discovered_peripheral = peripheral;
+                NEARBY_LOG(
+                    INFO,
+                    "Discovered peripheral=%p [impl=%p], fast advertisement=%d",
+                    &peripheral, &peripheral.GetImpl(), fast_advertisement);
+                discovered_latch.CountDown();
+              },
+      });
+  EXPECT_TRUE(discovered_latch.Await(kWaitDuration).result());
+  ASSERT_TRUE(discovered_peripheral.IsValid());
+
+  CancellationFlag flag(true);
+  BleV2Socket socket_for_client =
+      ble_client.Connect(service_id, discovered_peripheral, &flag);
+  // If FeatureFlag is disabled, Cancelled is false as no-op.
+  if (!feature_flags.enable_cancellation_flag) {
+    EXPECT_TRUE(accept_latch.Await(kWaitDuration).result());
+    EXPECT_TRUE(ble_server.StopAcceptingConnections(service_id));
+    EXPECT_TRUE(ble_server.StopAdvertising(service_id));
+    EXPECT_TRUE(socket_for_server.IsValid());
+    EXPECT_TRUE(socket_for_client.IsValid());
+    EXPECT_TRUE(socket_for_server.GetRemotePeripheral().IsValid());
+    EXPECT_TRUE(socket_for_client.GetRemotePeripheral().IsValid());
+  } else {
+    EXPECT_FALSE(accept_latch.Await(kWaitDuration).result());
+    EXPECT_TRUE(ble_server.StopAcceptingConnections(service_id));
+    EXPECT_TRUE(ble_server.StopAdvertising(service_id));
+    EXPECT_FALSE(socket_for_server.IsValid());
+    EXPECT_FALSE(socket_for_client.IsValid());
+  }
+  env_.Stop();
+}
+
+INSTANTIATE_TEST_SUITE_P(ParametrisedBleTest, BleV2Test,
+                         ::testing::ValuesIn(kTestCases));
+
 TEST_F(BleV2Test, CanConstructValidObject) {
   env_.Start();
   BluetoothRadio radio_a;
diff --git a/connections/implementation/offline_service_controller_test.cc b/connections/implementation/offline_service_controller_test.cc
index 78ff4e6e..23d48310 100644
--- a/connections/implementation/offline_service_controller_test.cc
+++ b/connections/implementation/offline_service_controller_test.cc
@@ -44,6 +44,9 @@ constexpr absl::Duration kDefaultTimeout = absl::Milliseconds(1500);
 constexpr absl::Duration kDisconnectTimeout = absl::Milliseconds(15000);
 
 constexpr BooleanMediumSelector kTestCases[] = {
+    BooleanMediumSelector{
+        .ble = true,
+    },
     BooleanMediumSelector{
         .bluetooth = true,
     },
@@ -52,6 +55,19 @@ constexpr BooleanMediumSelector kTestCases[] = {
     },
     BooleanMediumSelector{
         .bluetooth = true,
+        .ble = true,
+    },
+    BooleanMediumSelector{
+        .bluetooth = true,
+        .wifi_lan = true,
+    },
+    BooleanMediumSelector{
+        .ble = true,
+        .wifi_lan = true,
+    },
+    BooleanMediumSelector{
+        .bluetooth = true,
+        .ble = true,
         .wifi_lan = true,
     },
 };
diff --git a/connections/implementation/p2p_cluster_pcp_handler.cc b/connections/implementation/p2p_cluster_pcp_handler.cc
index a97341bb..ddb6d827 100644
--- a/connections/implementation/p2p_cluster_pcp_handler.cc
+++ b/connections/implementation/p2p_cluster_pcp_handler.cc
@@ -23,6 +23,7 @@
 #include "connections/implementation/base_pcp_handler.h"
 #include "connections/implementation/ble_advertisement.h"
 #include "connections/implementation/ble_endpoint_channel.h"
+#include "connections/implementation/ble_v2_endpoint_channel.h"
 #include "connections/implementation/bluetooth_endpoint_channel.h"
 #include "connections/implementation/bwu_manager.h"
 #include "connections/implementation/mediums/utils.h"
@@ -187,7 +188,7 @@ Status P2pClusterPcpHandler::StopAdvertisingImpl(ClientProxy* client) {
 
   if (FeatureFlags::GetInstance().GetFlags().support_ble_v2) {
     ble_v2_medium_.StopAdvertising(client->GetAdvertisingServiceId());
-    // TODO(b/213689498): Implements Stop accept connenction.
+    ble_v2_medium_.StopAcceptingConnections(client->GetAdvertisingServiceId());
   } else {
     ble_medium_.StopAdvertising(client->GetAdvertisingServiceId());
     ble_medium_.StopAcceptingConnections(client->GetAdvertisingServiceId());
@@ -1485,7 +1486,53 @@ proto::connections::Medium P2pClusterPcpHandler::StartBleV2Advertising(
   // Bluetooth Classic.
   NEARBY_LOGS(INFO) << "P2pClusterPcpHandler::StartBleAdvertising: service_id="
                     << service_id << " : start";
-  // TODO(b/213689498): Implements accept connenction.
+  // TODO(edwinwu): Move the lambda to a named function.
+  if (!ble_v2_medium_.IsAcceptingConnections(service_id)) {
+    if (!bluetooth_radio_.Enable() ||
+        !ble_v2_medium_.StartAcceptingConnections(
+            service_id, {.accepted_cb = [this, client, local_endpoint_info](
+                                            BleV2Socket socket,
+                                            const std::string& service_id) {
+              if (!socket.IsValid()) {
+                NEARBY_LOGS(WARNING)
+                    << "Invalid socket in accept callback("
+                    << absl::BytesToHexString(local_endpoint_info.data())
+                    << "), client=" << client->GetClientId();
+                return;
+              }
+              RunOnPcpHandlerThread(
+                  "p2p-ble-on-incoming-connection",
+                  [this, client, local_endpoint_info, service_id,
+                   socket = std::move(socket)]()
+                      RUN_ON_PCP_HANDLER_THREAD() mutable {
+                        ByteArray remote_peripheral_info =
+                            socket.GetRemotePeripheral().GetId();
+                        auto channel = std::make_unique<BleV2EndpointChannel>(
+                            service_id, std::string(remote_peripheral_info),
+                            socket);
+
+                        OnIncomingConnection(client, remote_peripheral_info,
+                                             std::move(channel),
+                                             proto::connections::Medium::BLE);
+                      });
+            }})) {
+      NEARBY_LOGS(WARNING)
+          << "In StartBleAdvertising("
+          << absl::BytesToHexString(local_endpoint_info.data())
+          << "), client=" << client->GetClientId()
+          << " failed to start accepting for incoming BLE connections to "
+             "service_id="
+          << service_id;
+      return proto::connections::UNKNOWN_MEDIUM;
+    }
+    NEARBY_LOGS(INFO)
+        << "In StartBleAdvertising("
+        << absl::BytesToHexString(local_endpoint_info.data())
+        << "), client=" << client->GetClientId()
+        << " started accepting for incoming BLE connections to service_id="
+        << service_id;
+  }
+
   PowerLevel power_level = advertising_options.low_power
                                ? PowerLevel::kLowPower
                                : PowerLevel::kHighPower;
@@ -1530,7 +1577,7 @@ proto::connections::Medium P2pClusterPcpHandler::StartBleV2Advertising(
             << " failed to start accepting for incoming BLE connections to "
                "service_id="
             << service_id;
-        // TODO(b/213689498): Implements stop accepting connenction.
+        ble_v2_medium_.StopAcceptingConnections(service_id);
         return proto::connections::UNKNOWN_MEDIUM;
       }
       NEARBY_LOGS(INFO)
@@ -1575,7 +1622,7 @@ proto::connections::Medium P2pClusterPcpHandler::StartBleV2Advertising(
                          << absl::BytesToHexString(local_endpoint_info.data())
                          << "), client=" << client->GetClientId()
                          << " failed to create an advertisement.";
-    // TODO(b/213689498): Implements stop accepting connenction.
+    ble_v2_medium_.StopAcceptingConnections(service_id);
     return proto::connections::UNKNOWN_MEDIUM;
   }
 
@@ -1594,7 +1641,7 @@ proto::connections::Medium P2pClusterPcpHandler::StartBleV2Advertising(
         << "), client=" << client->GetClientId()
         << " couldn't start BLE Advertising with BleAdvertisement "
         << absl::BytesToHexString(advertisement_bytes.data());
-    // TODO(b/213689498): Implements stop accepting connenction.
+    ble_v2_medium_.StopAcceptingConnections(service_id);
     return proto::connections::UNKNOWN_MEDIUM;
   }
   NEARBY_LOGS(INFO) << "In startBleAdvertising("
@@ -1632,11 +1679,28 @@ BasePcpHandler::ConnectImplResult P2pClusterPcpHandler::BleV2ConnectImpl(
                        << " is attempting to connect to endpoint(id="
                        << endpoint->endpoint_id << ") over BLE.";
 
-  // TODO(b/213689498): Implements connection.
+  BleV2Peripheral& peripheral = endpoint->ble_peripheral;
+
+  BleV2Socket ble_socket = ble_v2_medium_.Connect(
+      endpoint->service_id, peripheral,
+      client->GetCancellationFlag(endpoint->endpoint_id));
+  if (!ble_socket.IsValid()) {
+    NEARBY_LOGS(ERROR)
+        << "In BleConnectImpl(), failed to connect to BLE device "
+        << absl::BytesToHexString(peripheral.GetId().data())
+        << " for endpoint(id=" << endpoint->endpoint_id << ").";
+    return BasePcpHandler::ConnectImplResult{
+        .status = {Status::kBleError},
+    };
+  }
+
+  auto channel = std::make_unique<BleV2EndpointChannel>(
+      endpoint->service_id, /*channel_name=*/endpoint->endpoint_id, ble_socket);
+
   return BasePcpHandler::ConnectImplResult{
       .medium = proto::connections::Medium::BLE,
-      .status = {Status::kBleError},
-      .endpoint_channel = nullptr,
+      .status = {Status::kSuccess},
+      .endpoint_channel = std::move(channel),
   };
 }
 
diff --git a/connections/implementation/p2p_cluster_pcp_handler_test.cc b/connections/implementation/p2p_cluster_pcp_handler_test.cc
index c4ba4738..b04601e3 100644
--- a/connections/implementation/p2p_cluster_pcp_handler_test.cc
+++ b/connections/implementation/p2p_cluster_pcp_handler_test.cc
@@ -15,6 +15,7 @@
 #include "connections/implementation/p2p_cluster_pcp_handler.h"
 
 #include <memory>
+#include <string>
 
 #include "gmock/gmock.h"
 #include "protobuf-matchers/protocol-buffer-matchers.h"
@@ -22,9 +23,9 @@
 #include "absl/time/time.h"
 #include "connections/implementation/bwu_manager.h"
 #include "connections/implementation/injected_bluetooth_device_store.h"
-#include "internal/platform/medium_environment.h"
 #include "internal/platform/count_down_latch.h"
 #include "internal/platform/logging.h"
+#include "internal/platform/medium_environment.h"
 
 namespace location {
 namespace nearby {
@@ -32,6 +33,9 @@ namespace connections {
 namespace {
 
 constexpr BooleanMediumSelector kTestCases[] = {
+    BooleanMediumSelector{
+        .ble = true,
+    },
     BooleanMediumSelector{
         .bluetooth = true,
     },
@@ -40,16 +44,36 @@ constexpr BooleanMediumSelector kTestCases[] = {
     },
     BooleanMediumSelector{
         .bluetooth = true,
+        .ble = true,
+    },
+    BooleanMediumSelector{
+        .bluetooth = true,
+        .wifi_lan = true,
+    },
+    BooleanMediumSelector{
+        .ble = true,
+        .wifi_lan = true,
+    },
+    BooleanMediumSelector{
+        .bluetooth = true,
+        .ble = true,
         .wifi_lan = true,
     },
 };
 
+// Combines the bool `support_ble_v2` as param testing but should revert it back
+// if ble_v2 is done and ble will be replaced by ble_v2.
 class P2pClusterPcpHandlerTest
-    : public ::testing::TestWithParam<BooleanMediumSelector> {
+    : public testing::TestWithParam<std::tuple<BooleanMediumSelector, bool>> {
  protected:
   void SetUp() override {
     NEARBY_LOG(INFO, "SetUp: begin");
     env_.Stop();
+    FeatureFlags::GetMutableFlagsForTesting().support_ble_v2 =
+        std::get<1>(GetParam());
+    if (advertising_options_.allowed.ble) {
+      NEARBY_LOG(INFO, "SetUp: BLE enabled");
+    }
     if (advertising_options_.allowed.bluetooth) {
       NEARBY_LOG(INFO, "SetUp: BT enabled");
     }
@@ -68,19 +92,19 @@ class P2pClusterPcpHandlerTest
   ConnectionOptions connection_options_{
       {
           Strategy::kP2pCluster,
-          GetParam(),
+          std::get<0>(GetParam()),
       },
   };
   AdvertisingOptions advertising_options_{
       {
           Strategy::kP2pCluster,
-          GetParam(),
+          std::get<0>(GetParam()),
       },
   };
   DiscoveryOptions discovery_options_{
       {
           Strategy::kP2pCluster,
-          GetParam(),
+          std::get<0>(GetParam()),
       },
   };
   MediumEnvironment& env_{MediumEnvironment::Instance()};
@@ -265,7 +289,8 @@ TEST_P(P2pClusterPcpHandlerTest, CanConnect) {
 }
 
 INSTANTIATE_TEST_SUITE_P(ParametrisedPcpHandlerTest, P2pClusterPcpHandlerTest,
-                         ::testing::ValuesIn(kTestCases));
+                         ::testing::Combine(::testing::ValuesIn(kTestCases),
+                                            ::testing::Bool()));
 
 }  // namespace
 }  // namespace connections
diff --git a/connections/implementation/payload_manager_test.cc b/connections/implementation/payload_manager_test.cc
index c7b6c3fb..6b5df1d4 100644
--- a/connections/implementation/payload_manager_test.cc
+++ b/connections/implementation/payload_manager_test.cc
@@ -36,6 +36,9 @@ constexpr absl::Duration kProgressTimeout = absl::Milliseconds(1000);
 constexpr absl::Duration kDefaultTimeout = absl::Milliseconds(1000);
 
 constexpr BooleanMediumSelector kTestCases[] = {
+    BooleanMediumSelector{
+        .ble = true,
+    },
     BooleanMediumSelector{
         .bluetooth = true,
     },
@@ -44,6 +47,19 @@ constexpr BooleanMediumSelector kTestCases[] = {
     },
     BooleanMediumSelector{
         .bluetooth = true,
+        .ble = true,
+    },
+    BooleanMediumSelector{
+        .bluetooth = true,
+        .wifi_lan = true,
+    },
+    BooleanMediumSelector{
+        .ble = true,
+        .wifi_lan = true,
+    },
+    BooleanMediumSelector{
+        .bluetooth = true,
+        .ble = true,
         .wifi_lan = true,
     },
 };
diff --git a/connections/implementation/pcp_manager_test.cc b/connections/implementation/pcp_manager_test.cc
index be023ec2..1e271a94 100644
--- a/connections/implementation/pcp_manager_test.cc
+++ b/connections/implementation/pcp_manager_test.cc
@@ -37,6 +37,9 @@ constexpr char kDeviceA[] = "device-A";
 constexpr char kDeviceB[] = "device-B";
 
 constexpr BooleanMediumSelector kTestCases[] = {
+    BooleanMediumSelector{
+        .ble = true,
+    },
     BooleanMediumSelector{
         .bluetooth = true,
     },
@@ -45,6 +48,19 @@ constexpr BooleanMediumSelector kTestCases[] = {
     },
     BooleanMediumSelector{
         .bluetooth = true,
+        .ble = true,
+    },
+    BooleanMediumSelector{
+        .bluetooth = true,
+        .wifi_lan = true,
+    },
+    BooleanMediumSelector{
+        .ble = true,
+        .wifi_lan = true,
+    },
+    BooleanMediumSelector{
+        .bluetooth = true,
+        .ble = true,
         .wifi_lan = true,
     },
 };
diff --git a/internal/platform/ble_v2.cc b/internal/platform/ble_v2.cc
index e9cb8696..f8f95eca 100644
--- a/internal/platform/ble_v2.cc
+++ b/internal/platform/ble_v2.cc
@@ -139,6 +139,19 @@ std::unique_ptr<GattClient> BleV2Medium::ConnectToGattServer(
   return std::make_unique<GattClient>(std::move(api_gatt_client));
 }
 
+BleV2ServerSocket BleV2Medium::OpenServerSocket(const std::string& service_id) {
+  return BleV2ServerSocket(impl_->OpenServerSocket(service_id));
+}
+
+BleV2Socket BleV2Medium::Connect(const std::string& service_id,
+                                 TxPowerLevel tx_power_level,
+                                 const BleV2Peripheral& peripheral,
+                                 CancellationFlag* cancellation_flag) {
+  return BleV2Socket(impl_->Connect(service_id, tx_power_level,
+                                    /*mutated=*/peripheral.GetImpl(),
+                                    cancellation_flag));
+}
+
 bool BleV2Medium::IsExtendedAdvertisementsAvailable() {
   return impl_->IsExtendedAdvertisementsAvailable();
 }
diff --git a/internal/platform/ble_v2.h b/internal/platform/ble_v2.h
index cd32c304..11846f36 100644
--- a/internal/platform/ble_v2.h
+++ b/internal/platform/ble_v2.h
@@ -25,11 +25,120 @@
 #include "internal/platform/byte_array.h"
 #include "internal/platform/implementation/ble_v2.h"
 #include "internal/platform/implementation/platform.h"
+#include "internal/platform/logging.h"
 #include "internal/platform/mutex.h"
 
 namespace location {
 namespace nearby {
 
+// Container of operations that can be performed over the BLE GATT client
+// socket.
+// This class is copyable but not moveable.
+class BleV2Socket final {
+ public:
+  BleV2Socket() = default;
+  explicit BleV2Socket(std::unique_ptr<api::ble_v2::BleSocket> socket)
+      : impl_(std::move(socket)) {}
+  BleV2Socket(const BleV2Socket&) = default;
+  BleV2Socket& operator=(const BleV2Socket&) = default;
+
+  // Returns the InputStream of the BleSocket.
+  // On error, returned stream will report Exception::kIo on any operation.
+  //
+  // The returned object is not owned by the caller, and can be invalidated once
+  // the BleSocket object is destroyed.
+  InputStream& GetInputStream() { return impl_->GetInputStream(); }
+
+  // Returns the OutputStream of the BleSocket.
+  // On error, returned stream will report Exception::kIo on any operation.
+  //
+  // The returned object is not owned by the caller, and can be invalidated once
+  // the BleSocket object is destroyed.
+  OutputStream& GetOutputStream() { return impl_->GetOutputStream(); }
+
+  // Sets the close notifier by cient side.
+  void SetCloseNotifier(std::function<void()> notifier) {
+    close_notifier_ = std::move(notifier);
+  }
+
+  // Returns Exception::kIo on error, Exception::kSuccess otherwise.
+  Exception Close() {
+    if (close_notifier_) {
+      auto notifier = std::move(close_notifier_);
+      notifier();
+    }
+    return impl_->Close();
+  }
+
+  // Returns BlePeripheral object which wraps a valid BlePeripheral pointer.
+  BleV2Peripheral GetRemotePeripheral() {
+    return BleV2Peripheral(impl_->GetRemotePeripheral());
+  }
+
+  // Returns true if a socket is usable. If this method returns false,
+  // it is not safe to call any other method.
+  // NOTE(socket validity):
+  // Socket created by a default public constructor is not valid, because
+  // it is missing platform implementation.
+  // The only way to obtain a valid socket is through connection, such as
+  // an object returned by BleMedium::Connect
+  // These methods may also return an invalid socket if connection failed for
+  // any reason.
+  bool IsValid() const { return impl_ != nullptr; }
+
+  // Returns reference to platform implementation.
+  // This is used to communicate with platform code, and for debugging purposes.
+  // Returned reference will remain valid for while BleSocket object is
+  // itself valid. Typically BleSocket lifetime matches duration of the
+  // connection, and is controlled by end user, since they hold the instance.
+  api::ble_v2::BleSocket& GetImpl() { return *impl_; }
+
+ private:
+  std::function<void()> close_notifier_;
+  std::shared_ptr<api::ble_v2::BleSocket> impl_;
+};
+
+// Container of operations that can be performed over the BLE GATT server
+// socket.
+// This class is copyable but not moveable.
+class BleV2ServerSocket final {
+ public:
+  explicit BleV2ServerSocket(
+      std::unique_ptr<api::ble_v2::BleServerSocket> socket)
+      : impl_(std::move(socket)) {}
+  BleV2ServerSocket(const BleV2ServerSocket&) = default;
+  BleV2ServerSocket& operator=(const BleV2ServerSocket&) = default;
+
+  // Blocks until either:
+  // - at least one incoming connection request is available, or
+  // - ServerSocket is closed.
+  // On success, returns connected socket, ready to exchange data.
+  // On error, "impl_" will be nullptr and the caller will check it by calling
+  // member function "IsValid()"
+  // Once error is reported, it is permanent, and
+  // ServerSocket has to be closed by caller.
+  BleV2Socket Accept() {
+    std::unique_ptr<api::ble_v2::BleSocket> socket = impl_->Accept();
+    if (!socket) {
+      NEARBY_LOGS(INFO) << "BleServerSocket Accept() failed on server socket: "
+                        << this;
+    }
+    return BleV2Socket(std::move(socket));
+  }
+
+  // Returns Exception::kIo on error, Exception::kSuccess otherwise.
+  Exception Close() {
+    NEARBY_LOGS(INFO) << "BleServerSocket Closing:: " << this;
+    return impl_->Close();
+  }
+
+  bool IsValid() const { return impl_ != nullptr; }
+  api::ble_v2::BleServerSocket& GetImpl() { return *impl_; }
+
+ private:
+  std::shared_ptr<api::ble_v2::BleServerSocket> impl_;
+};
+
 // Opaque wrapper over a GattServer.
 // Move only, disallow copy.
 //
@@ -173,6 +282,17 @@ class BleV2Medium final {
       BleV2Peripheral peripheral, api::ble_v2::TxPowerLevel tx_power_level,
       ClientGattConnectionCallback callback);
 
+  // Returns a new BleServerSocket.
+  // On Success, BleServerSocket::IsValid() returns true.
+  BleV2ServerSocket OpenServerSocket(const std::string& service_id);
+
+  // Returns a new BleLanSocket.
+  // On Success, BleLanSocket::IsValid() returns true.
+  BleV2Socket Connect(const std::string& service_id,
+                      api::ble_v2::TxPowerLevel tx_power_level,
+                      const BleV2Peripheral& peripheral,
+                      CancellationFlag* cancellation_flag);
+
   bool IsExtendedAdvertisementsAvailable();
 
   bool IsValid() const { return impl_ != nullptr; }
diff --git a/internal/platform/ble_v2_test.cc b/internal/platform/ble_v2_test.cc
index 6465c11b..a4a9100c 100644
--- a/internal/platform/ble_v2_test.cc
+++ b/internal/platform/ble_v2_test.cc
@@ -16,6 +16,7 @@
 
 #include <memory>
 #include <string>
+#include <utility>
 
 #include "gmock/gmock.h"
 #include "protobuf-matchers/protocol-buffer-matchers.h"
@@ -28,6 +29,17 @@ namespace location {
 namespace nearby {
 namespace {
 
+using FeatureFlags = FeatureFlags::Flags;
+
+constexpr FeatureFlags kTestCases[] = {
+    FeatureFlags{
+        .enable_cancellation_flag = true,
+    },
+    FeatureFlags{
+        .enable_cancellation_flag = false,
+    },
+};
+
 using ::location::nearby::api::ble_v2::BleAdvertisementData;
 using ::location::nearby::api::ble_v2::GattCharacteristic;
 using ::location::nearby::api::ble_v2::TxPowerLevel;
@@ -38,6 +50,8 @@ constexpr absl::string_view kAdvertisementString = "\x0a\x0b\x0c\x0d";
 constexpr absl::string_view kAdvertisementHeaderString = "\x0x\x0y\x0z";
 constexpr absl::string_view kCopresenceServiceUuid = "F3FE";
 constexpr TxPowerLevel kTxPowerLevel(TxPowerLevel::kHigh);
+constexpr absl::string_view kServiceIDA{
+    "com.google.location.nearby.apps.test.a"};
 
 // A stub BlePeripheral implementation.
 class BlePeripheralStub : public api::ble_v2::BlePeripheral {
@@ -52,13 +66,158 @@ class BlePeripheralStub : public api::ble_v2::BlePeripheral {
   std::string mac_address_;
 };
 
-class BleV2MediumTest : public testing::Test {
+class BleV2MediumTest : public ::testing::TestWithParam<FeatureFlags> {
  protected:
   BleV2MediumTest() { env_.Stop(); }
 
   MediumEnvironment& env_{MediumEnvironment::Instance()};
 };
 
+TEST_P(BleV2MediumTest, CanConnectToService) {
+  FeatureFlags feature_flags = GetParam();
+  env_.SetFeatureFlags(feature_flags);
+  env_.Start();
+  BluetoothAdapter adapter_a_;
+  BluetoothAdapter adapter_b_;
+  BleV2Medium ble_a(adapter_a_);
+  BleV2Medium ble_b(adapter_b_);
+  std::string service_id(kServiceIDA);
+  ByteArray advertisement_bytes{std::string(kAdvertisementString)};
+  CountDownLatch found_latch(1);
+  CountDownLatch lost_latch(1);
+
+  BleV2ServerSocket server_socket = ble_b.OpenServerSocket(service_id);
+  EXPECT_TRUE(server_socket.IsValid());
+
+  // Assemble regular advertisement data
+  BleAdvertisementData advertising_data;
+  advertising_data.is_extended_advertisement = false;
+  advertising_data.service_data = {
+      {std::string(kCopresenceServiceUuid), advertisement_bytes}};
+  (ble_b.StartAdvertising(advertising_data, {.tx_power_level = kTxPowerLevel,
+                                             .is_connectable = true}));
+
+  BleV2Peripheral discovered_peripheral;
+  ble_a.StartScanning(
+      {std::string(kCopresenceServiceUuid)}, kTxPowerLevel,
+      {
+          .advertisement_found_cb =
+              [&found_latch, &discovered_peripheral](
+                  BleV2Peripheral peripheral,
+                  const BleAdvertisementData& advertisement_data) {
+                discovered_peripheral = std::move(peripheral);
+                found_latch.CountDown();
+              },
+      });
+  EXPECT_TRUE(found_latch.Await(absl::Milliseconds(1000)).result());
+  BleV2Socket socket_a;
+  BleV2Socket socket_b;
+  EXPECT_FALSE(socket_a.IsValid());
+  EXPECT_FALSE(socket_b.IsValid());
+  {
+    CancellationFlag flag;
+    SingleThreadExecutor server_executor;
+    SingleThreadExecutor client_executor;
+    client_executor.Execute(
+        [&ble_a, &socket_a, &service_id,
+         discovered_peripheral = std::move(discovered_peripheral),
+         &server_socket, &flag]() {
+          socket_a = ble_a.Connect(service_id, kTxPowerLevel,
+                                   discovered_peripheral, &flag);
+          if (!socket_a.IsValid()) {
+            server_socket.Close();
+          }
+        });
+    server_executor.Execute([&socket_b, &server_socket]() {
+      socket_b = server_socket.Accept();
+      if (!socket_b.IsValid()) {
+        server_socket.Close();
+      }
+    });
+  }
+  EXPECT_TRUE(socket_a.IsValid());
+  EXPECT_TRUE(socket_b.IsValid());
+  server_socket.Close();
+  env_.Stop();
+}
+
+TEST_P(BleV2MediumTest, CanCancelConnect) {
+  FeatureFlags feature_flags = GetParam();
+  env_.SetFeatureFlags(feature_flags);
+  env_.Start();
+  BluetoothAdapter adapter_a_;
+  BluetoothAdapter adapter_b_;
+  BleV2Medium ble_a{adapter_a_};
+  BleV2Medium ble_b{adapter_b_};
+  std::string service_id(kServiceIDA);
+  ByteArray advertisement_bytes((std::string(kAdvertisementString)));
+  CountDownLatch found_latch(1);
+  CountDownLatch lost_latch(1);
+
+  BleV2ServerSocket server_socket = ble_b.OpenServerSocket(service_id);
+  EXPECT_TRUE(server_socket.IsValid());
+
+  // Assemble regular advertisement data.
+  BleAdvertisementData advertising_data;
+  advertising_data.is_extended_advertisement = false;
+  advertising_data.service_data = {
+      {std::string(kCopresenceServiceUuid), advertisement_bytes}};
+  (ble_b.StartAdvertising(advertising_data, {.tx_power_level = kTxPowerLevel,
+                                             .is_connectable = true}));
+
+  BleV2Peripheral discovered_peripheral;
+  ble_a.StartScanning(
+      {std::string(kCopresenceServiceUuid)}, kTxPowerLevel,
+      {
+          .advertisement_found_cb =
+              [&found_latch, &discovered_peripheral](
+                  BleV2Peripheral peripheral,
+                  const BleAdvertisementData& advertisement_data) {
+                discovered_peripheral = std::move(peripheral);
+                found_latch.CountDown();
+              },
+      });
+  EXPECT_TRUE(found_latch.Await(absl::Milliseconds(1000)).result());
+  BleV2Socket socket_a;
+  BleV2Socket socket_b;
+  EXPECT_FALSE(socket_a.IsValid());
+  EXPECT_FALSE(socket_b.IsValid());
+  {
+    CancellationFlag flag(true);
+    SingleThreadExecutor server_executor;
+    SingleThreadExecutor client_executor;
+    client_executor.Execute(
+        [&ble_a, &socket_a, &service_id,
+         discovered_peripheral = std::move(discovered_peripheral),
+         &server_socket, &flag]() {
+          socket_a = ble_a.Connect(service_id, kTxPowerLevel,
+                                   discovered_peripheral, &flag);
+          if (!socket_a.IsValid()) {
+            server_socket.Close();
+          }
+        });
+    server_executor.Execute([&socket_b, &server_socket]() {
+      socket_b = server_socket.Accept();
+      if (!socket_b.IsValid()) {
+        server_socket.Close();
+      }
+    });
+  }
+  // If FeatureFlag is disabled, Cancelled is false as no-op.
+  if (!feature_flags.enable_cancellation_flag) {
+    EXPECT_TRUE(socket_a.IsValid());
+    EXPECT_TRUE(socket_b.IsValid());
+  } else {
+    EXPECT_FALSE(socket_a.IsValid());
+    EXPECT_FALSE(socket_b.IsValid());
+  }
+  server_socket.Close();
+  env_.Stop();
+}
+
+INSTANTIATE_TEST_SUITE_P(ParametrisedBleMediumTest, BleV2MediumTest,
+                         ::testing::ValuesIn(kTestCases));
+
 TEST_F(BleV2MediumTest, ConstructorDestructorWorks) {
   env_.Start();
   BluetoothAdapter adapter_a;
diff --git a/internal/platform/bluetooth_adapter.h b/internal/platform/bluetooth_adapter.h
index cdd816fd..832f57fe 100644
--- a/internal/platform/bluetooth_adapter.h
+++ b/internal/platform/bluetooth_adapter.h
@@ -52,6 +52,7 @@ class BlePeripheral final {
 // particular BLE peripheral to connect to its GATT server.
 class BleV2Peripheral final {
  public:
+  BleV2Peripheral() = default;
   explicit BleV2Peripheral(api::ble_v2::BlePeripheral* peripheral)
       : impl_(peripheral) {}
   BleV2Peripheral(const BleV2Peripheral&) = default;
@@ -87,7 +88,7 @@ class BleV2Peripheral final {
 
   // Returns reference to platform implementation.
   // This is used to communicate with platform code, and for debugging purposes.
-  api::ble_v2::BlePeripheral& GetImpl() { return *impl_; }
+  api::ble_v2::BlePeripheral& GetImpl() const { return *impl_; }
   bool IsValid() const { return impl_ != nullptr; }
 
  private:
diff --git a/internal/platform/implementation/ble_v2.h b/internal/platform/implementation/ble_v2.h
index 70b1c9df..5d387554 100644
--- a/internal/platform/implementation/ble_v2.h
+++ b/internal/platform/implementation/ble_v2.h
@@ -26,6 +26,7 @@
 #include "absl/container/flat_hash_map.h"
 #include "absl/strings/string_view.h"
 #include "internal/platform/byte_array.h"
+#include "internal/platform/cancellation_flag.h"
 #include "internal/platform/exception.h"
 #include "internal/platform/input_stream.h"
 #include "internal/platform/listeners.h"
@@ -224,44 +225,48 @@ struct ServerGattConnectionCallback {
       characteristic_unsubscription_cb;
 };
 
+// A BLE GATT client socket for requesting GATT socket.
 class BleSocket {
  public:
-  virtual ~BleSocket() {}
+  virtual ~BleSocket() = default;
 
-  // Returns the remote BLE peripheral tied to this socket.
-  virtual BlePeripheral& GetRemotePeripheral() = 0;
-
-  // Writes a message on the socket and blocks until finished. Returns
-  // Exception::kIo upon error, and Exception::kSuccess otherwise.
-  virtual Exception Write(const ByteArray& message) = 0;
-
-  // Closes the socket and blocks until finished. Returns Exception::kIo upon
-  // error, and Exception::kSuccess otherwise.
-  virtual Exception Close() = 0;
+  // Returns the InputStream of the BleSocket.
+  // On error, returned stream will report Exception::kIo on any operation.
+  //
+  // The returned object is not owned by the caller, and can be invalidated once
+  // the BleSocket object is destroyed.
   virtual InputStream& GetInputStream() = 0;
+
+  // Returns the OutputStream of the BleSocket.
+  // On error, returned stream will report Exception::kIo on any operation.
+  //
+  // The returned object is not owned by the caller, and can be invalidated once
+  // the BleSocket object is destroyed.
   virtual OutputStream& GetOutputStream() = 0;
+
+  // Returns Exception::kIo on error, Exception::kSuccess otherwise.
+  virtual Exception Close() = 0;
+
+  // Returns valid BlePeripheral pointer if there is a connection, and
+  // nullptr otherwise.
+  virtual BlePeripheral* GetRemotePeripheral() = 0;
 };
 
-// Callback for asynchronous events on a BleSocket object.
-class BleSocketLifeCycleCallback {
+// A BLE GATT server socket for listening incoming GATT socket.
+class BleServerSocket {
  public:
-  virtual ~BleSocketLifeCycleCallback() = default;
+  virtual ~BleServerSocket() = default;
 
-  // Called when a message arrives on a socket.
-  virtual void OnMessageReceived(BleSocket* socket,
-                                 const ByteArray& message) = 0;
+  // Blocks until either:
+  // - at least one incoming connection request is available, or
+  // - ServerSocket is closed.
+  // On success, returns connected socket, ready to exchange data.
+  // Returns nullptr on error.
+  // Once error is reported, it is permanent, and ServerSocket has to be closed.
+  virtual std::unique_ptr<BleSocket> Accept() = 0;
 
-  // Called when a socket gets disconnected.
-  virtual void OnDisconnected(BleSocket* socket) = 0;
-};
-
-// Callback for asynchronous events on the server side of a BleSocket object.
-class ServerBleSocketLifeCycleCallback : public BleSocketLifeCycleCallback {
- public:
-  ~ServerBleSocketLifeCycleCallback() override {}
-
-  // Called when a new incoming socket has been established.
-  virtual void OnSocketEstablished(BleSocket* socket) = 0;
+  // Returns Exception::kIo on error, Exception::kSuccess otherwise.
+  virtual Exception Close() = 0;
 };
 
 // The main BLE medium used inside of Nearby. This serves as the entry point
@@ -327,13 +332,6 @@ class BleMedium {
   virtual std::unique_ptr<GattServer> StartGattServer(
       ServerGattConnectionCallback callback) = 0;
 
-  // Starts listening for incoming BLE sockets and returns false upon error.
-  virtual bool StartListeningForIncomingBleSockets(
-      const ServerBleSocketLifeCycleCallback& callback) = 0;
-
-  // Stops listening for incoming BLE sockets.
-  virtual void StopListeningForIncomingBleSockets() = 0;
-
   // https://developer.android.com/reference/android/bluetooth/BluetoothDevice.html#connectGatt(android.content.Context,%20boolean,%20android.bluetooth.BluetoothGattCallback)
   // https://developer.android.com/reference/android/bluetooth/BluetoothGatt.html#requestConnectionPriority(int)
   // https://developer.android.com/reference/android/bluetooth/BluetoothGatt.html#requestMtu(int)
@@ -350,11 +348,20 @@ class BleMedium {
       BlePeripheral& peripheral, TxPowerLevel tx_power_level,
       ClientGattConnectionCallback callback) = 0;
 
-  // Establishes a BLE socket to the specified remote peripheral. Returns
-  // nullptr on error.
-  virtual std::unique_ptr<BleSocket> EstablishBleSocket(
-      BlePeripheral* peripheral,
-      const BleSocketLifeCycleCallback& callback) = 0;
+  // Opens a BLE server socket based on service ID.
+  //
+  // On success, returns a new BleServerSocket.
+  // On error, returns nullptr.
+  virtual std::unique_ptr<BleServerSocket> OpenServerSocket(
+      const std::string& service_id) = 0;
+
+  // Connects to a BLE peripheral.
+  //
+  // On success, returns a new BleSocket.
+  // On error, returns nullptr.
+  virtual std::unique_ptr<BleSocket> Connect(
+      const std::string& service_id, TxPowerLevel tx_power_level,
+      BlePeripheral& peripheral, CancellationFlag* cancellation_flag) = 0;
 
   // Requests if support extended advertisement.
   virtual bool IsExtendedAdvertisementsAvailable() = 0;
diff --git a/internal/platform/implementation/g3/ble_v2.cc b/internal/platform/implementation/g3/ble_v2.cc
index 88d7f099..d48e7afa 100644
--- a/internal/platform/implementation/g3/ble_v2.cc
+++ b/internal/platform/implementation/g3/ble_v2.cc
@@ -14,6 +14,7 @@
 
 #include "internal/platform/implementation/g3/ble_v2.h"
 
+#include <functional>
 #include <iostream>
 #include <memory>
 #include <optional>
@@ -22,6 +23,7 @@
 
 #include "absl/strings/escaping.h"
 #include "absl/synchronization/mutex.h"
+#include "internal/platform/cancellation_flag_listener.h"
 #include "internal/platform/implementation/ble_v2.h"
 #include "internal/platform/logging.h"
 #include "internal/platform/medium_environment.h"
@@ -33,8 +35,6 @@ namespace g3 {
 namespace {
 
 using ::location::nearby::api::ble_v2::BleAdvertisementData;
-using ::location::nearby::api::ble_v2::BleSocket;
-using ::location::nearby::api::ble_v2::BleSocketLifeCycleCallback;
 using ::location::nearby::api::ble_v2::TxPowerLevel;
 
 std::string TxPowerLevelToName(TxPowerLevel power_mode) {
@@ -54,15 +54,158 @@ std::string TxPowerLevelToName(TxPowerLevel power_mode) {
 
 }  // namespace
 
+BleV2Socket::~BleV2Socket() {
+  absl::MutexLock lock(&mutex_);
+  DoClose();
+}
+
+void BleV2Socket::Connect(BleV2Socket& other) {
+  absl::MutexLock lock(&mutex_);
+  remote_socket_ = &other;
+  input_ = other.output_;
+}
+
+InputStream& BleV2Socket::GetInputStream() {
+  auto* remote_socket = GetRemoteSocket();
+  CHECK(remote_socket != nullptr);
+  return remote_socket->GetLocalInputStream();
+}
+
+OutputStream& BleV2Socket::GetOutputStream() { return GetLocalOutputStream(); }
+
+BleV2Socket* BleV2Socket::GetRemoteSocket() {
+  absl::MutexLock lock(&mutex_);
+  return remote_socket_;
+}
+
+bool BleV2Socket::IsConnected() const {
+  absl::MutexLock lock(&mutex_);
+  return IsConnectedLocked();
+}
+
+bool BleV2Socket::IsClosed() const {
+  absl::MutexLock lock(&mutex_);
+  return closed_;
+}
+
+Exception BleV2Socket::Close() {
+  absl::MutexLock lock(&mutex_);
+  DoClose();
+  return {Exception::kSuccess};
+}
+
+BleV2Peripheral* BleV2Socket::GetRemotePeripheral() {
+  BluetoothAdapter* remote_adapter = nullptr;
+  {
+    absl::MutexLock lock(&mutex_);
+    if (remote_socket_ == nullptr || remote_socket_->adapter_ == nullptr) {
+      return nullptr;
+    }
+    remote_adapter = remote_socket_->adapter_;
+  }
+  return remote_adapter ? &remote_adapter->GetPeripheralV2() : nullptr;
+}
+
+void BleV2Socket::DoClose() {
+  if (!closed_) {
+    remote_socket_ = nullptr;
+    output_->GetOutputStream().Close();
+    output_->GetInputStream().Close();
+    input_->GetOutputStream().Close();
+    input_->GetInputStream().Close();
+    closed_ = true;
+  }
+}
+
+bool BleV2Socket::IsConnectedLocked() const { return input_ != nullptr; }
+
+InputStream& BleV2Socket::GetLocalInputStream() {
+  absl::MutexLock lock(&mutex_);
+  return output_->GetInputStream();
+}
+
+OutputStream& BleV2Socket::GetLocalOutputStream() {
+  absl::MutexLock lock(&mutex_);
+  return output_->GetOutputStream();
+}
+
+std::unique_ptr<api::ble_v2::BleSocket> BleV2ServerSocket::Accept() {
+  absl::MutexLock lock(&mutex_);
+  while (!closed_ && pending_sockets_.empty()) {
+    cond_.Wait(&mutex_);
+  }
+  // whether or not we were running in the wait loop, return early if closed.
+  if (closed_) return {};
+  auto* remote_socket =
+      pending_sockets_.extract(pending_sockets_.begin()).value();
+  CHECK(remote_socket);
+
+  auto local_socket = std::make_unique<BleV2Socket>(adapter_);
+  local_socket->Connect(*remote_socket);
+  remote_socket->Connect(*local_socket);
+  cond_.SignalAll();
+  return local_socket;
+}
+
+bool BleV2ServerSocket::Connect(BleV2Socket& socket) {
+  absl::MutexLock lock(&mutex_);
+  if (closed_) return false;
+  if (socket.IsConnected()) {
+    NEARBY_LOGS(WARNING)
+        << "Failed to connect to Ble server socket: already connected";
+    return true;  // already connected.
+  }
+  // add client socket to the pending list
+  pending_sockets_.insert(&socket);
+  cond_.SignalAll();
+  while (!socket.IsConnected()) {
+    cond_.Wait(&mutex_);
+    if (closed_) return false;
+  }
+  return true;
+}
+
+void BleV2ServerSocket::SetCloseNotifier(std::function<void()> notifier) {
+  absl::MutexLock lock(&mutex_);
+  close_notifier_ = std::move(notifier);
+}
+
+BleV2ServerSocket::~BleV2ServerSocket() {
+  absl::MutexLock lock(&mutex_);
+  DoClose();
+}
+
+Exception BleV2ServerSocket::Close() {
+  absl::MutexLock lock(&mutex_);
+  return DoClose();
+}
+
+Exception BleV2ServerSocket::DoClose() {
+  bool should_notify = !closed_;
+  closed_ = true;
+  if (should_notify) {
+    cond_.SignalAll();
+    if (close_notifier_) {
+      auto notifier = std::move(close_notifier_);
+      mutex_.Unlock();
+      // Notifier may contain calls to public API, and may cause deadlock, if
+      // mutex_ is held during the call.
+      notifier();
+      mutex_.Lock();
+    }
+  }
+  return {Exception::kSuccess};
+}
+
 BleV2Medium::BleV2Medium(api::BluetoothAdapter& adapter)
     : adapter_(static_cast<BluetoothAdapter*>(&adapter)) {
-  auto& env = MediumEnvironment::Instance();
-  env.RegisterBleV2Medium(*this);
+  adapter_->SetBleV2Medium(this);
+  MediumEnvironment::Instance().RegisterBleV2Medium(*this);
 }
 
 BleV2Medium::~BleV2Medium() {
-  auto& env = MediumEnvironment::Instance();
-  env.UnregisterBleV2Medium(*this);
+  adapter_->SetBleV2Medium(nullptr);
+  MediumEnvironment::Instance().UnregisterBleV2Medium(*this);
 }
 
 bool BleV2Medium::StartAdvertising(
@@ -125,25 +268,12 @@ std::unique_ptr<api::ble_v2::GattServer> BleV2Medium::StartGattServer(
   return std::make_unique<GattServer>();
 }
 
-bool BleV2Medium::StartListeningForIncomingBleSockets(
-    const api::ble_v2::ServerBleSocketLifeCycleCallback& callback) {
-  return false;
-}
-
-void BleV2Medium::StopListeningForIncomingBleSockets() {}
-
 std::unique_ptr<api::ble_v2::GattClient> BleV2Medium::ConnectToGattServer(
     api::ble_v2::BlePeripheral& peripheral, TxPowerLevel tx_power_level,
     api::ble_v2::ClientGattConnectionCallback callback) {
   return std::make_unique<GattClient>();
 }
 
-std::unique_ptr<BleSocket> BleV2Medium::EstablishBleSocket(
-    api::ble_v2::BlePeripheral* peripheral,
-    const BleSocketLifeCycleCallback& callback) {
-  return nullptr;
-}
-
 bool BleV2Medium::IsExtendedAdvertisementsAvailable() {
   return is_support_extended_advertisement_;
 }
@@ -244,6 +374,83 @@ void BleV2Medium::GattClient::Disconnect() {
   is_connection_alive_ = false;
 }
 
+std::unique_ptr<api::ble_v2::BleServerSocket> BleV2Medium::OpenServerSocket(
+    const std::string& service_id) {
+  auto server_socket = std::make_unique<BleV2ServerSocket>(&GetAdapter());
+  server_socket->SetCloseNotifier([this, service_id]() {
+    absl::MutexLock lock(&mutex_);
+    server_sockets_.erase(service_id);
+  });
+  NEARBY_LOGS(INFO) << "G3 Ble Adding server socket: medium=" << this
+                    << ", service_id=" << service_id;
+  absl::MutexLock lock(&mutex_);
+  server_sockets_.insert({service_id, server_socket.get()});
+  return server_socket;
+}
+
+std::unique_ptr<api::ble_v2::BleSocket> BleV2Medium::Connect(
+    const std::string& service_id, TxPowerLevel tx_power_level,
+    api::ble_v2::BlePeripheral& remote_peripheral,
+    CancellationFlag* cancellation_flag) {
+  NEARBY_LOGS(INFO) << "G3 Ble Connect [self]: medium=" << this
+                    << ", adapter=" << &GetAdapter()
+                    << ", peripheral=" << &GetAdapter().GetPeripheralV2()
+                    << ", service_id=" << service_id;
+  // First, find an instance of remote medium, that exposed this peripheral.
+  auto& remote_adapter =
+      static_cast<BleV2Peripheral&>(remote_peripheral).GetAdapter();
+  auto* remote_medium =
+      static_cast<BleV2Medium*>(remote_adapter.GetBleV2Medium());
+  if (!remote_medium) {
+    return nullptr;
+  }
+
+  BleV2ServerSocket* remote_server_socket = nullptr;
+  NEARBY_LOGS(INFO) << "G3 Ble Connect [peer]: medium=" << remote_medium
+                    << ", adapter=" << &remote_adapter
+                    << ", peripheral=" << &remote_peripheral
+                    << ", service_id=" << service_id;
+  // Then, find our server socket context in this medium.
+  {
+    absl::MutexLock medium_lock(&remote_medium->mutex_);
+    auto item = remote_medium->server_sockets_.find(service_id);
+    remote_server_socket =
+        item != server_sockets_.end() ? item->second : nullptr;
+    if (remote_server_socket == nullptr) {
+      NEARBY_LOGS(ERROR)
+          << "G3 Ble Failed to find Ble Server socket: service_id="
+          << service_id;
+      return nullptr;
+    }
+  }
+
+  if (cancellation_flag->Cancelled()) {
+    NEARBY_LOGS(ERROR) << "G3 BLE Connect: Has been cancelled: "
+                          "service_id="
+                       << service_id;
+    return nullptr;
+  }
+
+  CancellationFlagListener listener(
+      cancellation_flag, [&remote_server_socket]() {
+        NEARBY_LOGS(INFO) << "G3 Ble Cancel Connect.";
+        if (remote_server_socket != nullptr) {
+          remote_server_socket->Close();
+        }
+      });
+
+  auto socket = std::make_unique<BleV2Socket>(&GetAdapter());
+  // Finally, Request to connect to this socket.
+  if (!remote_server_socket->Connect(*socket)) {
+    NEARBY_LOGS(ERROR) << "G3 Ble Failed to connect to existing Ble "
+                          "Server socket: service_id="
+                       << service_id;
+    return nullptr;
+  }
+  NEARBY_LOGS(INFO) << "G3 Ble Connect to socket=" << socket.get();
+  return socket;
+}
+
 }  // namespace g3
 }  // namespace nearby
 }  // namespace location
diff --git a/internal/platform/implementation/g3/ble_v2.h b/internal/platform/implementation/g3/ble_v2.h
index 142c51db..f7753814 100644
--- a/internal/platform/implementation/g3/ble_v2.h
+++ b/internal/platform/implementation/g3/ble_v2.h
@@ -15,6 +15,7 @@
 #ifndef PLATFORM_IMPL_G3_BLE_V2_H_
 #define PLATFORM_IMPL_G3_BLE_V2_H_
 
+#include <functional>
 #include <memory>
 #include <optional>
 #include <string>
@@ -23,12 +24,129 @@
 #include "internal/platform/byte_array.h"
 #include "internal/platform/implementation/ble_v2.h"
 #include "internal/platform/implementation/g3/bluetooth_adapter.h"
+#include "internal/platform/implementation/g3/pipe.h"
 
 namespace location {
 namespace nearby {
 namespace g3 {
 
-// TODO(b/213691253): Add g3 BleV2 medium tests after more functions are ready.
+class BleV2nMedium;
+
+class BleV2Socket : public api::ble_v2::BleSocket {
+ public:
+  explicit BleV2Socket(BluetoothAdapter* adapter) : adapter_(adapter) {}
+  BleV2Socket(const BleV2Socket&) = default;
+  BleV2Socket& operator=(const BleV2Socket&) = default;
+  BleV2Socket(BleV2Socket&&) = default;
+  BleV2Socket& operator=(BleV2Socket&&) = default;
+  ~BleV2Socket() override;
+
+  // Connect to another BleSocket, to form a functional low-level channel.
+  // from this point on, and until Close is called, connection exists.
+  void Connect(BleV2Socket& other) ABSL_LOCKS_EXCLUDED(mutex_);
+
+  // Returns the InputStream of this connected BleSocket.
+  InputStream& GetInputStream() override ABSL_LOCKS_EXCLUDED(mutex_);
+
+  // Returns the OutputStream of this connected BleSocket.
+  // This stream is for local side to write.
+  OutputStream& GetOutputStream() override ABSL_LOCKS_EXCLUDED(mutex_);
+
+  // Returns address of a remote BleSocket or nullptr.
+  BleV2Socket* GetRemoteSocket() ABSL_LOCKS_EXCLUDED(mutex_);
+
+  // Returns true if connection exists to the (possibly closed) remote socket.
+  bool IsConnected() const ABSL_LOCKS_EXCLUDED(mutex_);
+
+  // Returns true if socket is closed.
+  bool IsClosed() const ABSL_LOCKS_EXCLUDED(mutex_);
+
+  // Returns Exception::kIo on error, Exception::kSuccess otherwise.
+  Exception Close() override ABSL_LOCKS_EXCLUDED(mutex_);
+
+  // Returns valid BlePeripheral pointer if there is a connection, and
+  // nullptr otherwise.
+  BleV2Peripheral* GetRemotePeripheral() override ABSL_LOCKS_EXCLUDED(mutex_);
+
+ private:
+  void DoClose() ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
+
+  // Returns true if connection exists to the (possibly closed) remote socket.
+  bool IsConnectedLocked() const ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
+
+  // Returns InputStream of our side of a connection.
+  // This is what the remote side is supposed to read from.
+  // This is a helper for GetInputStream() method.
+  InputStream& GetLocalInputStream() ABSL_LOCKS_EXCLUDED(mutex_);
+
+  // Returns OutputStream of our side of a connection.
+  // This is what the local size is supposed to write to.
+  // This is a helper for GetOutputStream() method.
+  OutputStream& GetLocalOutputStream() ABSL_LOCKS_EXCLUDED(mutex_);
+
+  // Output pipe is initialized by constructor, it remains always valid, until
+  // it is closed. it represents output part of a local socket. Input part of a
+  // local socket comes from the peer socket, after connection.
+  std::shared_ptr<Pipe> output_{new Pipe};
+  std::shared_ptr<Pipe> input_;
+  mutable absl::Mutex mutex_;
+  BluetoothAdapter* adapter_ = nullptr;  // Our Adapter. Read only.
+  BleV2Socket* remote_socket_ ABSL_GUARDED_BY(mutex_) = nullptr;
+  bool closed_ ABSL_GUARDED_BY(mutex_) = false;
+};
+
+class BleV2ServerSocket : public api::ble_v2::BleServerSocket {
+ public:
+  explicit BleV2ServerSocket(BluetoothAdapter* adapter) : adapter_(adapter) {}
+  BleV2ServerSocket(const BleV2ServerSocket&) = default;
+  BleV2ServerSocket& operator=(const BleV2ServerSocket&) = default;
+  BleV2ServerSocket(BleV2ServerSocket&&) = default;
+  BleV2ServerSocket& operator=(BleV2ServerSocket&&) = default;
+  ~BleV2ServerSocket() override;
+
+  // Blocks until either:
+  // - at least one incoming connection request is available, or
+  // - ServerSocket is closed.
+  // On success, returns connected socket, ready to exchange data.
+  // Returns nullptr on error.
+  // Once error is reported, it is permanent, and ServerSocket has to be closed.
+  //
+  // Called by the server side of a connection.
+  // Returns BleSocket to the server side.
+  // If not null, returned socket is connected to its remote (client-side) peer.
+  std::unique_ptr<api::ble_v2::BleSocket> Accept() override
+      ABSL_LOCKS_EXCLUDED(mutex_);
+
+  // Blocks until either:
+  // - connection is available, or
+  // - server socket is closed, or
+  // - error happens.
+  //
+  // Called by the client side of a connection.
+  // Returns true, if socket is successfully connected.
+  bool Connect(BleV2Socket& socket) ABSL_LOCKS_EXCLUDED(mutex_);
+
+  // Called by the server side of a connection before passing ownership of
+  // BleServerSocker to user, to track validity of a pointer to this
+  // server socket.
+  void SetCloseNotifier(std::function<void()> notifier)
+      ABSL_LOCKS_EXCLUDED(mutex_);
+
+  // Returns Exception::kIo on error, Exception::kSuccess otherwise.
+  // Calls close_notifier if it was previously set, and marks socket as closed.
+  Exception Close() override ABSL_LOCKS_EXCLUDED(mutex_);
+
+ private:
+  Exception DoClose() ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
+
+  mutable absl::Mutex mutex_;
+  absl::CondVar cond_;
+  BluetoothAdapter* adapter_ = nullptr;  // Our Adapter. Read only.
+  absl::flat_hash_set<BleV2Socket*> pending_sockets_ ABSL_GUARDED_BY(mutex_);
+  std::function<void()> close_notifier_ ABSL_GUARDED_BY(mutex_);
+  bool closed_ ABSL_GUARDED_BY(mutex_) = false;
+};
+
 // Container of operations that can be performed over the BLE medium.
 class BleV2Medium : public api::ble_v2::BleMedium {
  public:
@@ -50,20 +168,27 @@ class BleV2Medium : public api::ble_v2::BleMedium {
   std::unique_ptr<api::ble_v2::GattServer> StartGattServer(
       api::ble_v2::ServerGattConnectionCallback callback) override
       ABSL_LOCKS_EXCLUDED(mutex_);
-  bool StartListeningForIncomingBleSockets(
-      const api::ble_v2::ServerBleSocketLifeCycleCallback& callback) override
-      ABSL_LOCKS_EXCLUDED(mutex_);
-  void StopListeningForIncomingBleSockets() override
-      ABSL_LOCKS_EXCLUDED(mutex_);
   std::unique_ptr<api::ble_v2::GattClient> ConnectToGattServer(
       api::ble_v2::BlePeripheral& peripheral,
       api::ble_v2::TxPowerLevel tx_power_level,
       api::ble_v2::ClientGattConnectionCallback callback) override
       ABSL_LOCKS_EXCLUDED(mutex_);
-  std::unique_ptr<api::ble_v2::BleSocket> EstablishBleSocket(
-      api::ble_v2::BlePeripheral* peripheral,
-      const api::ble_v2::BleSocketLifeCycleCallback& callback) override
-      ABSL_LOCKS_EXCLUDED(mutex_);
+
+  // Open server socket to listen for incoming connection.
+  //
+  // On success, returns a new BleServerSocket.
+  // On error, returns nullptr.
+  std::unique_ptr<api::ble_v2::BleServerSocket> OpenServerSocket(
+      const std::string& service_id) override ABSL_LOCKS_EXCLUDED(mutex_);
+
+  // Connects to existing remote Ble peripheral.
+  //
+  // On success, returns a new BleSocket.
+  // On error, returns nullptr.
+  std::unique_ptr<api::ble_v2::BleSocket> Connect(
+      const std::string& service_id, api::ble_v2::TxPowerLevel tx_power_level,
+      api::ble_v2::BlePeripheral& remote_peripheral,
+      CancellationFlag* cancellation_flag) override ABSL_LOCKS_EXCLUDED(mutex_);
 
   bool IsExtendedAdvertisementsAvailable() override;
 
@@ -116,6 +241,8 @@ class BleV2Medium : public api::ble_v2::BleMedium {
 
   absl::Mutex mutex_;
   BluetoothAdapter* adapter_;  // Our device adapter; read-only.
+  absl::flat_hash_map<std::string, BleV2ServerSocket*> server_sockets_
+      ABSL_GUARDED_BY(mutex_);
   // TODO(edwinwu): Adds extended advertisement for testing.
   bool is_support_extended_advertisement_ = false;
 };
diff --git a/internal/platform/implementation/g3/bluetooth_adapter.cc b/internal/platform/implementation/g3/bluetooth_adapter.cc
index 2f673f50..c9f7fb3a 100644
--- a/internal/platform/implementation/g3/bluetooth_adapter.cc
+++ b/internal/platform/implementation/g3/bluetooth_adapter.cc
@@ -15,6 +15,7 @@
 #include "internal/platform/implementation/g3/bluetooth_adapter.h"
 
 #include <string>
+#include <utility>
 
 #include "internal/platform/implementation/g3/bluetooth_classic.h"
 #include "internal/platform/medium_environment.h"
@@ -78,6 +79,10 @@ void BluetoothAdapter::SetBleMedium(api::BleMedium* medium) {
   ble_medium_ = medium;
 }
 
+void BluetoothAdapter::SetBleV2Medium(api::ble_v2::BleMedium* medium) {
+  ble_v2_medium_ = medium;
+}
+
 bool BluetoothAdapter::SetStatus(Status status) {
   BluetoothAdapter::ScanMode mode;
   bool enabled = status == Status::kEnabled;
diff --git a/internal/platform/implementation/g3/bluetooth_adapter.h b/internal/platform/implementation/g3/bluetooth_adapter.h
index 6af13745..3bc3832e 100644
--- a/internal/platform/implementation/g3/bluetooth_adapter.h
+++ b/internal/platform/implementation/g3/bluetooth_adapter.h
@@ -138,6 +138,9 @@ class BluetoothAdapter : public api::BluetoothAdapter {
   void SetBleMedium(api::BleMedium* medium);
   api::BleMedium* GetBleMedium() { return ble_medium_; }
 
+  void SetBleV2Medium(api::ble_v2::BleMedium* medium);
+  api::ble_v2::BleMedium* GetBleV2Medium() { return ble_v2_medium_; }
+
   void SetMacAddress(std::string& mac_address) { mac_address_ = mac_address; }
 
  private:
@@ -147,6 +150,7 @@ class BluetoothAdapter : public api::BluetoothAdapter {
   BleV2Peripheral peripheral_v2_{this};
   api::BluetoothClassicMedium* bluetooth_classic_medium_ = nullptr;
   api::BleMedium* ble_medium_ = nullptr;
+  api::ble_v2::BleMedium* ble_v2_medium_ = nullptr;
   std::string mac_address_;
   ScanMode mode_ ABSL_GUARDED_BY(mutex_) = ScanMode::kNone;
   std::string name_ ABSL_GUARDED_BY(mutex_) = "unknown G3 BT device";
diff --git a/internal/platform/implementation/windows/ble_v2.cc b/internal/platform/implementation/windows/ble_v2.cc
index b4ba597a..dd91a27f 100644
--- a/internal/platform/implementation/windows/ble_v2.cc
+++ b/internal/platform/implementation/windows/ble_v2.cc
@@ -33,8 +33,8 @@ namespace {
 
 using ::location::nearby::api::ble_v2::AdvertiseParameters;
 using ::location::nearby::api::ble_v2::BleAdvertisementData;
+using ::location::nearby::api::ble_v2::BleServerSocket;
 using ::location::nearby::api::ble_v2::BleSocket;
-using ::location::nearby::api::ble_v2::BleSocketLifeCycleCallback;
 using ::location::nearby::api::ble_v2::GattClient;
 using ::location::nearby::api::ble_v2::ServerGattConnectionCallback;
 using ::location::nearby::api::ble_v2::TxPowerLevel;
@@ -251,22 +251,21 @@ std::unique_ptr<api::ble_v2::GattServer> BleV2Medium::StartGattServer(
   return nullptr;
 }
 
-bool BleV2Medium::StartListeningForIncomingBleSockets(
-    const api::ble_v2::ServerBleSocketLifeCycleCallback& callback) {
-  return false;
-}
-
-void BleV2Medium::StopListeningForIncomingBleSockets() {}
-
 std::unique_ptr<GattClient> BleV2Medium::ConnectToGattServer(
     api::ble_v2::BlePeripheral& peripheral, TxPowerLevel tx_power_level,
     api::ble_v2::ClientGattConnectionCallback callback) {
   return nullptr;
 }
 
-std::unique_ptr<BleSocket> BleV2Medium::EstablishBleSocket(
-    api::ble_v2::BlePeripheral* peripheral,
-    const BleSocketLifeCycleCallback& callback) {
+std::unique_ptr<api::ble_v2::BleServerSocket> BleV2Medium::OpenServerSocket(
+    const std::string& service_id) {
+  return nullptr;
+}
+
+std::unique_ptr<api::ble_v2::BleSocket> BleV2Medium::Connect(
+    const std::string& service_id, TxPowerLevel tx_power_level,
+    api::ble_v2::BlePeripheral& remote_peripheral,
+    CancellationFlag* cancellation_flag) {
   return nullptr;
 }
 
diff --git a/internal/platform/implementation/windows/ble_v2.h b/internal/platform/implementation/windows/ble_v2.h
index c981113d..bd19d304 100644
--- a/internal/platform/implementation/windows/ble_v2.h
+++ b/internal/platform/implementation/windows/ble_v2.h
@@ -58,20 +58,17 @@ class BleV2Medium : public api::ble_v2::BleMedium {
   std::unique_ptr<api::ble_v2::GattServer> StartGattServer(
       api::ble_v2::ServerGattConnectionCallback callback) override
       ABSL_LOCKS_EXCLUDED(mutex_);
-  bool StartListeningForIncomingBleSockets(
-      const api::ble_v2::ServerBleSocketLifeCycleCallback& callback) override
-      ABSL_LOCKS_EXCLUDED(mutex_);
-  void StopListeningForIncomingBleSockets() override
-      ABSL_LOCKS_EXCLUDED(mutex_);
   std::unique_ptr<api::ble_v2::GattClient> ConnectToGattServer(
       api::ble_v2::BlePeripheral& peripheral,
       api::ble_v2::TxPowerLevel tx_power_level,
       api::ble_v2::ClientGattConnectionCallback callback) override
       ABSL_LOCKS_EXCLUDED(mutex_);
-  std::unique_ptr<api::ble_v2::BleSocket> EstablishBleSocket(
-      api::ble_v2::BlePeripheral* peripheral,
-      const api::ble_v2::BleSocketLifeCycleCallback& callback) override
-      ABSL_LOCKS_EXCLUDED(mutex_);
+  std::unique_ptr<api::ble_v2::BleServerSocket> OpenServerSocket(
+      const std::string& service_id) override ABSL_LOCKS_EXCLUDED(mutex_);
+  std::unique_ptr<api::ble_v2::BleSocket> Connect(
+      const std::string& service_id, api::ble_v2::TxPowerLevel tx_power_level,
+      api::ble_v2::BlePeripheral& remote_peripheral,
+      CancellationFlag* cancellation_flag) override ABSL_LOCKS_EXCLUDED(mutex_);
   bool IsExtendedAdvertisementsAvailable() override { return false; }
 
   BluetoothAdapter& GetAdapter() { return *adapter_; }