@@ -1758,14 +1758,23 @@ Session-scoped standalone notifications (`resources/updated`, `elicitation/compl
 broadcast notifications (`tools/list_changed`, etc.) still flow to clients connected to the GET SSE stream.
 This mode is suitable for simple tool servers that do not need server-initiated requests.
 
-By default, sessions do not expire. To mitigate session hijacking risks, you can set a `session_idle_timeout` (in seconds).
-When configured, sessions that receive no HTTP requests for this duration are automatically expired and cleaned up:
+By default, stateful sessions are bounded so an `initialize` flood cannot retain sessions until memory is exhausted:
+they expire after `session_idle_timeout` seconds of inactivity (default 1800, i.e. 30 minutes) and the concurrent
+session count is capped at `max_sessions` (default 10000). A session's idle timer is reset by activity that touches it
+(a GET, or a regular-request POST), and expired sessions are collected by a background reaper roughly once a minute,
+so cleanup lags inactivity by up to that interval. At the cap, the transport first reclaims any already-expired slots
+and then, if still full, rejects a new `initialize` with HTTP 503 (it does not evict an existing session).
 
 ```ruby
-# Session timeout of 30 minutes
-transport = MCP::Server::Transports::StreamableHTTPTransport.new(server, session_idle_timeout: 1800)
+# Tune the limits
+transport = MCP::Server::Transports::StreamableHTTPTransport.new(server, session_idle_timeout: 900, max_sessions: 5000)
+
+# Opt out of expiry and/or the cap (not recommended on internet-facing deployments)
+transport = MCP::Server::Transports::StreamableHTTPTransport.new(server, session_idle_timeout: nil, max_sessions: nil)
 ```
 
+Stateless mode (`stateless: true`) retains no sessions, so neither limit applies to it.
+
 ### Pagination
 
 The MCP Ruby SDK supports [pagination](https://modelcontextprotocol.io/specification/2025-11-25/server/utilities/pagination)
 
@@ -24,17 +24,40 @@ class InvalidJsonError < StandardError; end
           "Connection" => "keep-alive",
         }.freeze
 
-        def initialize(server, stateless: false, enable_json_response: false, session_idle_timeout: nil)
+        # Secure defaults for stateful mode. Without a finite idle timeout, sessions live until an explicit client DELETE,
+        # so an unauthenticated `initialize` flood retains unbounded `ServerSession` objects until memory is exhausted.
+        # These defaults expire idle sessions and cap the concurrent count, like the C# SDK (the only reference SDK that
+        # hardens this by default, with a 2h idle timeout and a 10k idle-session count). One difference: at the cap this transport
+        # rejects a new `initialize` with 503 (after reclaiming any already-expired slots), whereas the C# SDK evicts
+        # the oldest idle session. Rejecting keeps established sessions stable and avoids evicting a legitimate idle session on
+        # an attacker's behalf, at the cost of refusing new sessions while genuinely full. Pass `session_idle_timeout: nil` to
+        # opt out of expiry and `max_sessions: nil` to opt out of the cap.
+        DEFAULT_SESSION_IDLE_TIMEOUT = 1800
+        DEFAULT_MAX_SESSIONS = 10_000
+
+        # Distinguishes "argument omitted, apply the secure default" from an explicit `nil` (opt out of expiry).
+        UNSET_IDLE_TIMEOUT = Object.new.freeze
+        private_constant :UNSET_IDLE_TIMEOUT
+
+        def initialize(server, stateless: false, enable_json_response: false,
+          session_idle_timeout: UNSET_IDLE_TIMEOUT, max_sessions: DEFAULT_MAX_SESSIONS)
           super(server)
           # Maps `session_id` to `{ get_sse_stream: stream_object, server_session: ServerSession, last_active_at: float_from_monotonic_clock }`.
           @sessions = {}
           @mutex = Mutex.new
 
           @stateless = stateless
           @enable_json_response = enable_json_response
-          @session_idle_timeout = session_idle_timeout
           @pending_responses = {}
 
+          # Resolve the idle timeout: an explicit value (including `nil` to opt out) wins; otherwise apply the secure default,
+          # which does not apply to stateless mode since it retains no sessions.
+          @session_idle_timeout = if session_idle_timeout.equal?(UNSET_IDLE_TIMEOUT)
+            stateless ? nil : DEFAULT_SESSION_IDLE_TIMEOUT
+          else
+            session_idle_timeout
+          end
+
           if @session_idle_timeout
             if @stateless
               raise ArgumentError, "session_idle_timeout is not supported in stateless mode."
@@ -43,6 +66,13 @@ def initialize(server, stateless: false, enable_json_response: false, session_id
             end
           end
 
+          unless max_sessions.nil? || (max_sessions.is_a?(Integer) && max_sessions > 0)
+            raise ArgumentError, "max_sessions must be a positive Integer or nil"
+          end
+
+          # The cap guards the stateful session store; stateless mode keeps none.
+          @max_sessions = stateless ? nil : max_sessions
+
           start_reaper_thread if @session_idle_timeout
         end
 
@@ -622,13 +652,30 @@ def handle_initialization(body_string, body)
             session_id = SecureRandom.uuid
             server_session = ServerSession.new(server: @server, transport: self, session_id: session_id)
 
-            @mutex.synchronize do
+            # Cap the concurrent session count so an `initialize` flood cannot retain unbounded sessions until memory is exhausted.
+            # The check and insert share the mutex so concurrent initializes cannot race past the limit.
+            reclaimed = []
+            inserted = @mutex.synchronize do
+              # When at the cap, first reclaim slots held by already-expired sessions the 60s reaper has not yet collected,
+              # so the cap rejects only when genuinely full rather than up to a reaper interval after sessions expired.
+              if @max_sessions && @sessions.size >= @max_sessions
+                @sessions.each_key.select { |id| session_expired?(@sessions[id]) }.each do |id|
+                  cleanup_and_collect_stream(id, reclaimed)
+                end
+              end
+
+              next false if @max_sessions && @sessions.size >= @max_sessions
+
               @sessions[session_id] = {
                 get_sse_stream: nil,
                 server_session: server_session,
                 last_active_at: Process.clock_gettime(Process::CLOCK_MONOTONIC),
               }
+              true
             end
+
+            reclaimed.each { |stream| close_stream_safely(stream) }
+            return too_many_sessions_response unless inserted
           end
 
           response = server_session.handle_json(body_string)
@@ -655,6 +702,14 @@ def handle_accepted
           [202, {}, []]
         end
 
+        def too_many_sessions_response
+          json_rpc_error_response(
+            status: 503,
+            code: JsonRpcHandler::ErrorCode::INTERNAL_ERROR,
+            message: "Service unavailable: maximum concurrent sessions (#{@max_sessions}) reached",
+          )
+        end
+
         def handle_regular_request(body_string, session_id, related_request_id: nil)
           server_session = nil
 
 
@@ -3808,35 +3808,116 @@ def string
           transport.close
         end
 
-        test "default session_idle_timeout is nil and sessions do not expire" do
+        test "applies a finite session_idle_timeout by default and starts the reaper" do
+          # Secure default: stateful mode expires idle sessions out of the box.
           transport = StreamableHTTPTransport.new(@server)
-          assert_nil(transport.instance_variable_get(:@reaper_thread))
+          assert_equal(
+            StreamableHTTPTransport::DEFAULT_SESSION_IDLE_TIMEOUT,
+            transport.instance_variable_get(:@session_idle_timeout),
+          )
+          assert(transport.instance_variable_get(:@reaper_thread).alive?)
 
           init_request = create_rack_request(
             "POST",
             "/",
             { "CONTENT_TYPE" => "application/json" },
             { jsonrpc: "2.0", method: "initialize", id: "init" }.to_json,
           )
-          init_response = transport.handle_request(init_request)
-          session_id = init_response[1]["Mcp-Session-Id"]
-
+          # A freshly created session is well within the timeout, so normal use is unaffected.
+          session_id = transport.handle_request(init_request)[1]["Mcp-Session-Id"]
           request = create_rack_request(
             "POST",
             "/",
-            {
-              "CONTENT_TYPE" => "application/json",
-              "HTTP_MCP_SESSION_ID" => session_id,
-            },
+            { "CONTENT_TYPE" => "application/json", "HTTP_MCP_SESSION_ID" => session_id },
             { jsonrpc: "2.0", method: "ping", id: "456" }.to_json,
           )
 
-          response = transport.handle_request(request)
-          assert_equal(200, response[0])
+          assert_equal(200, transport.handle_request(request)[0])
+        ensure
+          transport.close
+        end
+
+        test "session_idle_timeout: nil opts out of expiry" do
+          transport = StreamableHTTPTransport.new(@server, session_idle_timeout: nil)
+          assert_nil(transport.instance_variable_get(:@session_idle_timeout))
+          assert_nil(transport.instance_variable_get(:@reaper_thread))
         ensure
           transport.close
         end
 
+        test "stateless mode keeps no idle timeout despite the secure default" do
+          transport = StreamableHTTPTransport.new(@server, stateless: true)
+          assert_nil(transport.instance_variable_get(:@session_idle_timeout))
+          assert_nil(transport.instance_variable_get(:@reaper_thread))
+        ensure
+          transport.close
+        end
+
+        test "rejects new initialize with 503 once max_sessions is reached" do
+          transport = StreamableHTTPTransport.new(@server, max_sessions: 2)
+
+          2.times do
+            request = create_rack_request(
+              "POST",
+              "/",
+              { "CONTENT_TYPE" => "application/json" },
+              { jsonrpc: "2.0", method: "initialize", id: "init" }.to_json,
+            )
+            assert_equal(200, transport.handle_request(request)[0])
+          end
+
+          overflow = create_rack_request(
+            "POST",
+            "/",
+            { "CONTENT_TYPE" => "application/json" },
+            { jsonrpc: "2.0", method: "initialize", id: "init" }.to_json,
+          )
+          response = transport.handle_request(overflow)
+
+          assert_equal(503, response[0])
+          assert_equal(2, transport.instance_variable_get(:@sessions).size)
+        ensure
+          transport.close
+        end
+
+        test "at the cap, reclaims an already-expired session before rejecting" do
+          # A very short idle timeout so the first session is expired by the time the cap would otherwise reject
+          # a new initialize; the expired slot is reclaimed.
+          transport = StreamableHTTPTransport.new(@server, max_sessions: 1, session_idle_timeout: 0.01)
+          init = lambda do
+            transport.handle_request(create_rack_request(
+              "POST",
+              "/",
+              { "CONTENT_TYPE" => "application/json" },
+              { jsonrpc: "2.0", method: "initialize", id: "init" }.to_json,
+            ))
+          end
+
+          assert_equal(200, init.call[0])
+          sleep(0.05) # let the first session's idle timer expire
+          second = init.call
+
+          # The expired session was reclaimed in-line, so the second initialize succeeds (200) rather than
+          # being rejected with 503, and the store stays at the cap.
+          assert_equal(200, second[0])
+          assert_equal(1, transport.instance_variable_get(:@sessions).size)
+        ensure
+          transport.close
+        end
+
+        test "max_sessions: nil opts out of the session cap" do
+          transport = StreamableHTTPTransport.new(@server, max_sessions: nil)
+          assert_nil(transport.instance_variable_get(:@max_sessions))
+        ensure
+          transport.close
+        end
+
+        test "raises ArgumentError for an invalid max_sessions" do
+          [0, -1, 1.5, "10"].each do |invalid|
+            assert_raises(ArgumentError) { StreamableHTTPTransport.new(@server, max_sessions: invalid) }
+          end
+        end
+
         test "raises ArgumentError when session_idle_timeout is zero" do
           error = assert_raises(ArgumentError) do
             StreamableHTTPTransport.new(@server, session_idle_timeout: 0)