diff --git a/pw_rpc/client.cc b/pw_rpc/client.cc
index be2dd32..e579db8 100644
--- a/pw_rpc/client.cc
+++ b/pw_rpc/client.cc
@@ -47,7 +47,7 @@
     return Status::Unavailable();
   }
 
-  if (call == nullptr || call->id() != packet.call_id()) {
+  if (call == nullptr) {
     // The call for the packet does not exist. If the packet is a server stream
     // message, notify the server so that it can kill the stream. Otherwise,
     // silently drop the packet (as it would terminate the RPC anyway).
diff --git a/pw_rpc/endpoint.cc b/pw_rpc/endpoint.cc
index fccdf26..c9e40af 100644
--- a/pw_rpc/endpoint.cc
+++ b/pw_rpc/endpoint.cc
@@ -62,9 +62,9 @@
 
 void Endpoint::RegisterCall(Call& call) {
   Call* const existing_call = FindCallById(
-      call.channel_id_locked(), call.service_id(), call.method_id());
+      call.channel_id_locked(), call.service_id(), call.method_id(), call.id());
 
-  RegisterUniqueCall(call);
+  calls_.push_front(call);
 
   if (existing_call != nullptr) {
     // TODO(b/234876851): Ensure call object is locked when calling callback.
@@ -77,11 +77,20 @@
 
 Call* Endpoint::FindCallById(uint32_t channel_id,
                              uint32_t service_id,
-                             uint32_t method_id) {
+                             uint32_t method_id,
+                             uint32_t call_id) {
   for (Call& call : calls_) {
     if (channel_id == call.channel_id_locked() &&
         service_id == call.service_id() && method_id == call.method_id()) {
-      return &call;
+      if (call_id == call.id() || call_id == kOpenCallId) {
+        return &call;
+      }
+      if (call.id() == kOpenCallId) {
+        // Calls with ID of `kOpenCallId` were unrequested, and
+        // are updated to have the call ID of the first matching request.
+        call.set_id(call_id);
+        return &call;
+      }
     }
   }
   return nullptr;
diff --git a/pw_rpc/public/pw_rpc/internal/call.h b/pw_rpc/public/pw_rpc/internal/call.h
index 6e72257..4a26a61 100644
--- a/pw_rpc/public/pw_rpc/internal/call.h
+++ b/pw_rpc/public/pw_rpc/internal/call.h
@@ -15,6 +15,7 @@
 
 #include <cassert>
 #include <cstddef>
+#include <limits>
 #include <utility>
 
 #include "pw_containers/intrusive_list.h"
@@ -40,6 +41,11 @@
 class LockedEndpoint;
 class Packet;
 
+// Unrequested RPCs always use this call ID. When a subsequent request
+// or response is sent with a matching channel + service + method,
+// it will match a calls with this ID if one exists.
+constexpr uint32_t kOpenCallId = std::numeric_limits<uint32_t>::max();
+
 // Internal RPC Call class. The Call is used to respond to any type of RPC.
 // Public classes like ServerWriters inherit from it with private inheritance
 // and provide a public API for their use case. The Call's public API is used by
@@ -72,6 +78,8 @@
 
   uint32_t id() const PW_EXCLUSIVE_LOCKS_REQUIRED(rpc_lock()) { return id_; }
 
+  void set_id(uint32_t id) PW_EXCLUSIVE_LOCKS_REQUIRED(rpc_lock()) { id_ = id; }
+
   uint32_t channel_id() const PW_LOCKS_EXCLUDED(rpc_lock()) {
     LockGuard lock(rpc_lock());
     return channel_id_locked();
@@ -255,6 +263,7 @@
     const bool invoke = on_error_ != nullptr;
 
     // TODO(b/234876851): Ensure on_error_ is properly guarded.
+
     rpc_lock().unlock();
     if (invoke) {
       on_error_(error);
diff --git a/pw_rpc/public/pw_rpc/internal/client_call.h b/pw_rpc/public/pw_rpc/internal/client_call.h
index 2120933..5a23c1c 100644
--- a/pw_rpc/public/pw_rpc/internal/client_call.h
+++ b/pw_rpc/public/pw_rpc/internal/client_call.h
@@ -14,6 +14,7 @@
 #pragma once
 
 #include <cstdint>
+#include <mutex>
 
 #include "pw_bytes/span.h"
 #include "pw_function/function.h"
@@ -32,6 +33,11 @@
     rpc_lock().unlock();
   }
 
+  uint32_t id() PW_LOCKS_EXCLUDED(rpc_lock()) {
+    std::lock_guard lock(rpc_lock());
+    return Call::id();
+  }
+
  protected:
   constexpr ClientCall() = default;
 
diff --git a/pw_rpc/public/pw_rpc/internal/endpoint.h b/pw_rpc/public/pw_rpc/internal/endpoint.h
index c8caca5..48074c1 100644
--- a/pw_rpc/public/pw_rpc/internal/endpoint.h
+++ b/pw_rpc/public/pw_rpc/internal/endpoint.h
@@ -96,8 +96,10 @@
   // Finds a call object for an ongoing call associated with this packet, if
   // any. Returns nullptr if no matching call exists.
   Call* FindCall(const Packet& packet) PW_EXCLUSIVE_LOCKS_REQUIRED(rpc_lock()) {
-    return FindCallById(
-        packet.channel_id(), packet.service_id(), packet.method_id());
+    return FindCallById(packet.channel_id(),
+                        packet.service_id(),
+                        packet.method_id(),
+                        packet.call_id());
   }
 
   void AbortCallsForService(const Service& service)
@@ -141,8 +143,8 @@
 
   Call* FindCallById(uint32_t channel_id,
                      uint32_t service_id,
-                     uint32_t method_id)
-      PW_EXCLUSIVE_LOCKS_REQUIRED(rpc_lock());
+                     uint32_t method_id,
+                     uint32_t call_id) PW_EXCLUSIVE_LOCKS_REQUIRED(rpc_lock());
 
   ChannelList channels_ PW_GUARDED_BY(rpc_lock());
   IntrusiveList<Call> calls_ PW_GUARDED_BY(rpc_lock());
diff --git a/pw_rpc/public/pw_rpc/server.h b/pw_rpc/public/pw_rpc/server.h
index 9411564..9adfa9c 100644
--- a/pw_rpc/public/pw_rpc/server.h
+++ b/pw_rpc/public/pw_rpc/server.h
@@ -18,6 +18,7 @@
 
 #include "pw_containers/intrusive_list.h"
 #include "pw_rpc/channel.h"
+#include "pw_rpc/internal/call.h"
 #include "pw_rpc/internal/channel.h"
 #include "pw_rpc/internal/endpoint.h"
 #include "pw_rpc/internal/lock.h"
@@ -126,12 +127,8 @@
                     "streaming RPCs.");
     }
 
-    // Unrequested RPCs always use 0 as the call ID. When an actual request is
-    // sent, the call will be replaced with its real ID.
-    constexpr uint32_t kOpenCallId = 0;
-
     return internal::CallContext(
-        *this, channel_id, service, method, kOpenCallId);
+        *this, channel_id, service, method, internal::kOpenCallId);
   }
 
   std::tuple<Service*, const internal::Method*> FindMethod(
diff --git a/pw_rpc/raw/client_reader_writer_test.cc b/pw_rpc/raw/client_reader_writer_test.cc
index b3220a8..68ff35a 100644
--- a/pw_rpc/raw/client_reader_writer_test.cc
+++ b/pw_rpc/raw/client_reader_writer_test.cc
@@ -231,27 +231,6 @@
   EXPECT_TRUE(active_call_2.active());
 }
 
-TEST(RawClientReaderWriter, NewCallCancelsPreviousAndCallsErrorCallback) {
-  RawClientTestContext ctx;
-
-  Status error;
-  RawClientReaderWriter active_call_1 = TestService::TestBidirectionalStreamRpc(
-      ctx.client(),
-      ctx.channel().id(),
-      FailIfOnNextCalled,
-      FailIfCalled,
-      [&error](Status status) { error = status; });
-
-  ASSERT_TRUE(active_call_1.active());
-
-  RawClientReaderWriter active_call_2 =
-      TestService::TestBidirectionalStreamRpc(ctx.client(), ctx.channel().id());
-
-  EXPECT_FALSE(active_call_1.active());
-  EXPECT_TRUE(active_call_2.active());
-  EXPECT_EQ(error, Status::Cancelled());
-}
-
 TEST(RawClientReader, NoClientStream_OutOfScope_SilentlyCloses) {
   RawClientTestContext ctx;
 
@@ -329,5 +308,49 @@
                kWriterData);
 }
 
+const char* span_as_cstr(ConstByteSpan span) {
+  return reinterpret_cast<const char*>(span.data());
+}
+
+TEST(RawClientReaderWriter,
+     MultipleCallsToSameMethodOkAndReceiveSeparateResponses) {
+  RawClientTestContext ctx;
+
+  ConstByteSpan data_1 = as_bytes(span("data_1_unset"));
+  ConstByteSpan data_2 = as_bytes(span("data_2_unset"));
+
+  Status error;
+  auto set_error = [&error](Status status) { error.Update(status); };
+  RawClientReaderWriter active_call_1 = TestService::TestBidirectionalStreamRpc(
+      ctx.client(),
+      ctx.channel().id(),
+      [&data_1](ConstByteSpan payload) { data_1 = payload; },
+      FailIfCalled,
+      set_error);
+
+  EXPECT_TRUE(active_call_1.active());
+
+  RawClientReaderWriter active_call_2 = TestService::TestBidirectionalStreamRpc(
+      ctx.client(),
+      ctx.channel().id(),
+      [&data_2](ConstByteSpan payload) { data_2 = payload; },
+      FailIfCalled,
+      set_error);
+
+  EXPECT_TRUE(active_call_1.active());
+  EXPECT_TRUE(active_call_2.active());
+  EXPECT_EQ(error, OkStatus());
+
+  ConstByteSpan message_1 = as_bytes(span("hello_1"));
+  ConstByteSpan message_2 = as_bytes(span("hello_2"));
+
+  ctx.server().SendServerStream<TestService::TestBidirectionalStreamRpc>(
+      message_2, active_call_2.id());
+  EXPECT_STREQ(span_as_cstr(data_2), span_as_cstr(message_2));
+  ctx.server().SendServerStream<TestService::TestBidirectionalStreamRpc>(
+      message_1, active_call_1.id());
+  EXPECT_STREQ(span_as_cstr(data_1), span_as_cstr(message_1));
+}
+
 }  // namespace
 }  // namespace pw::rpc
diff --git a/pw_rpc/raw/client_testing.cc b/pw_rpc/raw/client_testing.cc
index 3057dce..310d730 100644
--- a/pw_rpc/raw/client_testing.cc
+++ b/pw_rpc/raw/client_testing.cc
@@ -28,10 +28,11 @@
 void FakeServer::CheckProcessPacket(internal::PacketType type,
                                     uint32_t service_id,
                                     uint32_t method_id,
+                                    std::optional<uint32_t> call_id,
                                     ConstByteSpan payload,
                                     Status status) const {
   if (Status process_packet_status =
-          ProcessPacket(type, service_id, method_id, payload, status);
+          ProcessPacket(type, service_id, method_id, call_id, payload, status);
       !process_packet_status.ok()) {
     PW_LOG_CRITICAL("Failed to process packet in pw::rpc::FakeServer");
     PW_LOG_CRITICAL(
@@ -50,10 +51,10 @@
 Status FakeServer::ProcessPacket(internal::PacketType type,
                                  uint32_t service_id,
                                  uint32_t method_id,
+                                 std::optional<uint32_t> call_id,
                                  ConstByteSpan payload,
                                  Status status) const {
-  uint32_t call_id = 0;
-  {
+  if (!call_id.has_value()) {
     internal::LockGuard lock(output_.mutex_);
     auto view = internal::test::PacketsView(
         output_.packets(),
@@ -71,7 +72,7 @@
 
   auto packet_encoding_result =
       internal::Packet(
-          type, channel_id_, service_id, method_id, call_id, payload, status)
+          type, channel_id_, service_id, method_id, *call_id, payload, status)
           .Encode(packet_buffer_);
   PW_CHECK_OK(packet_encoding_result.status());
   return client_.ProcessPacket(*packet_encoding_result);
diff --git a/pw_rpc/raw/public/pw_rpc/raw/client_reader_writer.h b/pw_rpc/raw/public/pw_rpc/raw/client_reader_writer.h
index 1dbae7d..1239989 100644
--- a/pw_rpc/raw/public/pw_rpc/raw/client_reader_writer.h
+++ b/pw_rpc/raw/public/pw_rpc/raw/client_reader_writer.h
@@ -1,4 +1,4 @@
-// Copyright 2021 The Pigweed Authors
+// Copyright 2022 The Pigweed Authors
 //
 // 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
@@ -38,6 +38,8 @@
   using internal::Call::active;
   using internal::Call::channel_id;
 
+  using internal::ClientCall::id;
+
   // Functions for setting the callbacks.
   using internal::StreamResponseClientCall::set_on_completed;
   using internal::StreamResponseClientCall::set_on_error;
diff --git a/pw_rpc/raw/public/pw_rpc/raw/client_testing.h b/pw_rpc/raw/public/pw_rpc/raw/client_testing.h
index 692d1fd..89cefe0 100644
--- a/pw_rpc/raw/public/pw_rpc/raw/client_testing.h
+++ b/pw_rpc/raw/public/pw_rpc/raw/client_testing.h
@@ -45,8 +45,9 @@
   template <auto kMethod,
             typename = std::enable_if_t<
                 HasServerStream(internal::MethodInfo<kMethod>::kType)>>
-  void SendResponse(Status status) const {
-    SendPacket<kMethod>(internal::PacketType::RESPONSE, {}, status);
+  void SendResponse(Status status,
+                    std::optional<uint32_t> call_id = std::nullopt) const {
+    SendPacket<kMethod>(internal::PacketType::RESPONSE, {}, status, call_id);
   }
 
   // Sends a response packet for a unary or client streaming streaming RPC to
@@ -54,45 +55,54 @@
   template <auto kMethod,
             typename = std::enable_if_t<
                 !HasServerStream(internal::MethodInfo<kMethod>::kType)>>
-  void SendResponse(ConstByteSpan payload, Status status) const {
-    SendPacket<kMethod>(internal::PacketType::RESPONSE, payload, status);
+  void SendResponse(ConstByteSpan payload,
+                    Status status,
+                    std::optional<uint32_t> call_id = std::nullopt) const {
+    SendPacket<kMethod>(
+        internal::PacketType::RESPONSE, payload, status, call_id);
   }
 
   // Sends a stream packet for a server or bidirectional streaming RPC to the
   // client.
   template <auto kMethod>
-  void SendServerStream(ConstByteSpan payload) const {
+  void SendServerStream(ConstByteSpan payload,
+                        std::optional<uint32_t> call_id = std::nullopt) const {
     static_assert(HasServerStream(internal::MethodInfo<kMethod>::kType),
                   "Only server and bidirectional streaming methods can receive "
                   "server stream packets");
-    SendPacket<kMethod>(internal::PacketType::SERVER_STREAM, payload);
+    SendPacket<kMethod>(
+        internal::PacketType::SERVER_STREAM, payload, OkStatus(), call_id);
   }
 
   // Sends a server error packet to the client.
   template <auto kMethod>
-  void SendServerError(Status error) const {
-    SendPacket<kMethod>(internal::PacketType::SERVER_ERROR, {}, error);
+  void SendServerError(Status error,
+                       std::optional<uint32_t> call_id = std::nullopt) const {
+    SendPacket<kMethod>(internal::PacketType::SERVER_ERROR, {}, error, call_id);
   }
 
  private:
   template <auto kMethod>
   void SendPacket(internal::PacketType type,
-                  ConstByteSpan payload = {},
-                  Status status = OkStatus()) const {
+                  ConstByteSpan payload,
+                  Status status,
+                  std::optional<uint32_t> call_id) const {
     using Info = internal::MethodInfo<kMethod>;
     CheckProcessPacket(
-        type, Info::kServiceId, Info::kMethodId, payload, status);
+        type, Info::kServiceId, Info::kMethodId, call_id, payload, status);
   }
 
   void CheckProcessPacket(internal::PacketType type,
                           uint32_t service_id,
                           uint32_t method_id,
+                          std::optional<uint32_t> call_id,
                           ConstByteSpan payload,
                           Status status) const;
 
   Status ProcessPacket(internal::PacketType type,
                        uint32_t service_id,
                        uint32_t method_id,
+                       std::optional<uint32_t> call_id,
                        ConstByteSpan payload,
                        Status status) const;
 
diff --git a/pw_rpc/server.cc b/pw_rpc/server.cc
index 9ef2c29..270674a 100644
--- a/pw_rpc/server.cc
+++ b/pw_rpc/server.cc
@@ -38,8 +38,6 @@
                 Endpoint::ProcessPacket(packet_data, Packet::kServer));
 
   internal::rpc_lock().lock();
-  internal::ServerCall* const call =
-      static_cast<internal::ServerCall*>(FindCall(packet));
 
   // Verbose log for debugging.
   // PW_LOG_DEBUG("RPC server received packet type %u for %u:%08x/%08x",
@@ -72,10 +70,11 @@
     return OkStatus();  // OK since the packet was handled.
   }
 
+  internal::ServerCall* const call =
+      static_cast<internal::ServerCall*>(FindCall(packet));
+
   switch (packet.type()) {
     case PacketType::REQUEST: {
-      // If the REQUEST is for an ongoing RPC, the existing call will be
-      // cancelled when the new call object is created.
       const internal::CallContext context(
           *this, channel->id(), *service, *method, packet.call_id());
       method->Invoke(context, packet);
@@ -86,7 +85,7 @@
       break;
     case PacketType::CLIENT_ERROR:
     case PacketType::DEPRECATED_CANCEL:
-      if (call != nullptr && call->id() == packet.call_id()) {
+      if (call != nullptr) {
         call->HandleError(packet.status());
       } else {
         internal::rpc_lock().unlock();
@@ -125,7 +124,7 @@
 void Server::HandleClientStreamPacket(const internal::Packet& packet,
                                       internal::Channel& channel,
                                       internal::ServerCall* call) const {
-  if (call == nullptr || call->id() != packet.call_id()) {
+  if (call == nullptr) {
     channel.Send(Packet::ServerError(packet, Status::FailedPrecondition()))
         .IgnoreError();  // Errors are logged in Channel::Send.
     internal::rpc_lock().unlock();
diff --git a/pw_rpc/server_test.cc b/pw_rpc/server_test.cc
index 2814d40..e82bd2b 100644
--- a/pw_rpc/server_test.cc
+++ b/pw_rpc/server_test.cc
@@ -19,6 +19,7 @@
 
 #include "gtest/gtest.h"
 #include "pw_assert/check.h"
+#include "pw_rpc/internal/call.h"
 #include "pw_rpc/internal/method.h"
 #include "pw_rpc/internal/packet.h"
 #include "pw_rpc/internal/test_method.h"
@@ -67,6 +68,8 @@
   static constexpr std::array<TestMethodUnion, 0> methods_ = {};
 };
 
+uint32_t kDefaultCallId = 24601;
+
 class BasicServer : public ::testing::Test {
  protected:
   static constexpr byte kDefaultPayload[] = {
@@ -88,31 +91,36 @@
                                 uint32_t channel_id,
                                 uint32_t service_id,
                                 uint32_t method_id,
-                                span<const byte> payload = kDefaultPayload,
-                                Status status = OkStatus()) {
-    auto result =
-        Packet(type, channel_id, service_id, method_id, 0, payload, status)
-            .Encode(request_buffer_);
-    EXPECT_EQ(OkStatus(), result.status());
-    return result.value_or(ConstByteSpan());
+                                uint32_t call_id = kDefaultCallId) {
+    return EncodePacketWithBody(type,
+                                channel_id,
+                                service_id,
+                                method_id,
+                                call_id,
+                                kDefaultPayload,
+                                OkStatus());
   }
 
   span<const byte> EncodeCancel(uint32_t channel_id = 1,
                                 uint32_t service_id = 42,
-                                uint32_t method_id = 100) {
-    return EncodePacket(PacketType::CLIENT_ERROR,
-                        channel_id,
-                        service_id,
-                        method_id,
-                        {},
-                        Status::Cancelled());
+                                uint32_t method_id = 100,
+                                uint32_t call_id = kDefaultCallId) {
+    return EncodePacketWithBody(PacketType::CLIENT_ERROR,
+                                channel_id,
+                                service_id,
+                                method_id,
+                                call_id,
+                                {},
+                                Status::Cancelled());
   }
 
   template <typename T = ConstByteSpan>
   ConstByteSpan PacketForRpc(PacketType type,
                              Status status = OkStatus(),
-                             T&& payload = {}) {
-    return EncodePacket(type, 1, 42, 100, as_bytes(span(payload)), status);
+                             T&& payload = {},
+                             uint32_t call_id = kDefaultCallId) {
+    return EncodePacketWithBody(
+        type, 1, 42, 100, call_id, as_bytes(span(payload)), status);
   }
 
   RawFakeChannelOutput<2> output_;
@@ -124,6 +132,21 @@
 
  private:
   byte request_buffer_[64];
+
+  span<const byte> EncodePacketWithBody(PacketType type,
+                                        uint32_t channel_id,
+                                        uint32_t service_id,
+                                        uint32_t method_id,
+                                        uint32_t call_id,
+                                        span<const byte> payload,
+                                        Status status) {
+    auto result =
+        Packet(
+            type, channel_id, service_id, method_id, call_id, payload, status)
+            .Encode(request_buffer_);
+    EXPECT_EQ(OkStatus(), result.status());
+    return result.value_or(ConstByteSpan());
+  }
 };
 
 TEST_F(BasicServer, ProcessPacket_ValidMethodInService1_InvokesMethod) {
@@ -155,11 +178,12 @@
 }
 
 TEST_F(BasicServer, UnregisterService_CannotCallMethod) {
+  const uint32_t kCallId = 8675309;
   server_.UnregisterService(service_1_, service_42_);
 
-  EXPECT_EQ(
-      OkStatus(),
-      server_.ProcessPacket(EncodePacket(PacketType::REQUEST, 1, 1, 100)));
+  EXPECT_EQ(OkStatus(),
+            server_.ProcessPacket(
+                EncodePacket(PacketType::REQUEST, 1, 1, 100, kCallId)));
 
   const Packet& packet =
       static_cast<internal::test::FakeChannelOutput&>(output_).last_packet();
@@ -167,6 +191,7 @@
   EXPECT_EQ(packet.channel_id(), 1u);
   EXPECT_EQ(packet.service_id(), 1u);
   EXPECT_EQ(packet.method_id(), 100u);
+  EXPECT_EQ(packet.call_id(), kCallId);
   EXPECT_EQ(packet.status(), Status::NotFound());
 }
 
@@ -369,8 +394,11 @@
  protected:
   BidiMethod() {
     internal::rpc_lock().lock();
-    internal::CallContext context(
-        server_, channels_[0].id(), service_42_, service_42_.method(100), 0);
+    internal::CallContext context(server_,
+                                  channels_[0].id(),
+                                  service_42_,
+                                  service_42_.method(100),
+                                  kDefaultCallId);
     // A local temporary is required since the constructor requires a lock,
     // but the *move* constructor takes out the lock.
     internal::test::FakeServerReaderWriter responder_temp(
@@ -383,7 +411,7 @@
   internal::test::FakeServerReaderWriter responder_;
 };
 
-TEST_F(BidiMethod, DuplicateCall_CancelsExistingThenCallsAgain) {
+TEST_F(BidiMethod, DuplicateCallId_CancelsExistingThenCallsAgain) {
   int cancelled = 0;
   responder_.set_on_error([&cancelled](Status error) {
     if (error.IsCancelled()) {
@@ -401,6 +429,65 @@
   EXPECT_EQ(method.invocations(), 1u);
 }
 
+TEST_F(BidiMethod, DuplicateMethodDifferentCallId_NotCancelled) {
+  int cancelled = 0;
+  responder_.set_on_error([&cancelled](Status error) {
+    if (error.IsCancelled()) {
+      cancelled += 1;
+    }
+  });
+
+  const uint32_t kSecondCallId = 1625;
+  EXPECT_EQ(OkStatus(),
+            server_.ProcessPacket(PacketForRpc(
+                PacketType::REQUEST, OkStatus(), {}, kSecondCallId)));
+
+  EXPECT_EQ(cancelled, 0);
+}
+
+const char* span_as_cstr(ConstByteSpan span) {
+  return reinterpret_cast<const char*>(span.data());
+}
+
+TEST_F(BidiMethod, DuplicateMethodDifferentCallIdEachCallGetsSeparateResponse) {
+  const uint32_t kSecondCallId = 1625;
+
+  internal::rpc_lock().lock();
+  internal::test::FakeServerReaderWriter responder_2(
+      internal::CallContext(server_,
+                            channels_[0].id(),
+                            service_42_,
+                            service_42_.method(100),
+                            kSecondCallId)
+          .ClaimLocked());
+  internal::rpc_lock().unlock();
+
+  ConstByteSpan data_1 = as_bytes(span("data_1_unset"));
+  responder_.set_on_next(
+      [&data_1](ConstByteSpan payload) { data_1 = payload; });
+
+  ConstByteSpan data_2 = as_bytes(span("data_2_unset"));
+  responder_2.set_on_next(
+      [&data_2](ConstByteSpan payload) { data_2 = payload; });
+
+  const char* kMessage1 = "hello_1";
+  const char* kMessage2 = "hello_2";
+
+  EXPECT_EQ(
+      OkStatus(),
+      server_.ProcessPacket(PacketForRpc(
+          PacketType::CLIENT_STREAM, OkStatus(), "hello_2", kSecondCallId)));
+
+  EXPECT_STREQ(span_as_cstr(data_2), kMessage2);
+
+  EXPECT_EQ(
+      OkStatus(),
+      server_.ProcessPacket(PacketForRpc(
+          PacketType::CLIENT_STREAM, OkStatus(), "hello_1", kDefaultCallId)));
+
+  EXPECT_STREQ(span_as_cstr(data_1), kMessage1);
+}
+
 TEST_F(BidiMethod, Cancel_ClosesServerWriter) {
   EXPECT_EQ(OkStatus(), server_.ProcessPacket(EncodeCancel()));
 
@@ -468,7 +555,47 @@
                 PacketForRpc(PacketType::CLIENT_STREAM, {}, "hello")));
 
   EXPECT_EQ(output_.total_packets(), 0u);
-  EXPECT_STREQ(reinterpret_cast<const char*>(data.data()), "hello");
+  EXPECT_STREQ(span_as_cstr(data), "hello");
+}
+
+TEST_F(BidiMethod, ClientStream_CallsCallbackOnCallWithOpenId) {
+  ConstByteSpan data = as_bytes(span("?"));
+  responder_.set_on_next([&data](ConstByteSpan payload) { data = payload; });
+
+  ASSERT_EQ(
+      OkStatus(),
+      server_.ProcessPacket(PacketForRpc(
+          PacketType::CLIENT_STREAM, {}, "hello", internal::kOpenCallId)));
+
+  EXPECT_EQ(output_.total_packets(), 0u);
+  EXPECT_STREQ(span_as_cstr(data), "hello");
+}
+
+TEST_F(BidiMethod, ClientStream_CallsOpenIdOnCallWithDifferentId) {
+  const uint32_t kSecondCallId = 1625;
+  internal::CallContext context(server_,
+                                channels_[0].id(),
+                                service_42_,
+                                service_42_.method(100),
+                                internal::kOpenCallId);
+  internal::rpc_lock().lock();
+  auto temp_responder =
+      internal::test::FakeServerReaderWriter(context.ClaimLocked());
+  internal::rpc_lock().unlock();
+  responder_ = std::move(temp_responder);
+
+  ConstByteSpan data = as_bytes(span("?"));
+  responder_.set_on_next([&data](ConstByteSpan payload) { data = payload; });
+
+  ASSERT_EQ(OkStatus(),
+            server_.ProcessPacket(PacketForRpc(
+                PacketType::CLIENT_STREAM, {}, "hello", kSecondCallId)));
+
+  EXPECT_EQ(output_.total_packets(), 0u);
+  EXPECT_STREQ(span_as_cstr(data), "hello");
+
+  std::lock_guard lock(internal::rpc_lock());
+  EXPECT_EQ(responder_.as_server_call().id(), kSecondCallId);
 }
 
 TEST_F(BidiMethod, UnregsiterService_AbortsActiveCalls) {
@@ -518,8 +645,11 @@
 class ServerStreamingMethod : public BasicServer {
  protected:
   ServerStreamingMethod() {
-    internal::CallContext context(
-        server_, channels_[0].id(), service_42_, service_42_.method(100), 0);
+    internal::CallContext context(server_,
+                                  channels_[0].id(),
+                                  service_42_,
+                                  service_42_.method(100),
+                                  kDefaultCallId);
     internal::rpc_lock().lock();
     internal::test::FakeServerWriter responder_temp(context.ClaimLocked());
     internal::rpc_lock().unlock();