@@ -1798,6 +1798,31 @@ transport = MCP::Server::Transports::StreamableHTTPTransport.new(server, session
 
 Stateless mode (`stateless: true`) retains no sessions, so neither limit applies to it.
 
+#### Session Ownership
+
+`StreamableHTTPTransport` issues a random `SecureRandom.uuid` session ID and validates incoming requests by session
+existence and idle timeout only. It does not bind a session to a user, because the transport never receives
+an authenticated identity on its own. A caller that obtains a valid session ID could therefore act on that session,
+so binding a session to a user is the deploying application's responsibility (the MCP spec frames this as a SHOULD).
+
+The primary control is the `session_request_validator`. It is called as `->(request, session_id) { true | false }`
+on every non-`initialize` POST, GET, and DELETE against an existing session (including notification and response POSTs,
+so a stolen session ID cannot, for example, POST `notifications/cancelled` against a victim's request). A falsy return
+rejects the request with HTTP 403. Use it to compare the request's authenticated principal against the one recorded
+when the session was created:
+
+```ruby
+transport = MCP::Server::Transports::StreamableHTTPTransport.new(
+  server,
+  session_request_validator: ->(request, session_id) { owns_session?(request, session_id) },
+)
+```
+
+Without a validator the transport does not enforce ownership. As a limited defense in depth (not authentication),
+it also records the `Origin` header at `initialize` and rejects a later request whose `Origin` differs, but only
+when both are present - a non-browser client that omits `Origin` (e.g. `curl` or a script) is not stopped by this check.
+Enforcing ownership against a determined attacker requires supplying the validator with an authenticated principal.
+
 ### Pagination
 
 The MCP Ruby SDK supports [pagination](https://modelcontextprotocol.io/specification/2025-11-25/server/utilities/pagination)
 
@@ -58,6 +58,14 @@ class InvalidJsonError < StandardError; end
         # @param dns_rebinding_protection [Boolean] when `true` (default), validates the `Host` and
         #   `Origin` headers to prevent DNS rebinding; pass `false` when an upstream proxy already
         #   validates them.
+        # @param session_request_validator [#call, nil] An optional
+        #   `->(request, session_id) { true | false }` invoked on every non-`initialize` POST, GET, and DELETE
+        #   against an existing session (regular requests, notifications, and client responses alike).
+        #   Returning a falsy value rejects the request with HTTP 403. The SDK issues a random `SecureRandom.uuid`
+        #   session ID and otherwise only checks existence/idle-timeout, so binding a session to a user is
+        #   the deploying application's responsibility (the transport never receives the authenticated identity
+        #   on its own); this is the seam to enforce ownership and mitigate session poisoning. Without a validator,
+        #   ownership is not enforced.
         def initialize(
           server,
           stateless: false,
@@ -66,15 +74,17 @@ def initialize(
           max_sessions: DEFAULT_MAX_SESSIONS,
           allowed_origins: nil,
           allowed_hosts: nil,
-          dns_rebinding_protection: true
+          dns_rebinding_protection: true,
+          session_request_validator: nil
         )
           super(server)
-          # Maps `session_id` to `{ get_sse_stream: stream_object, server_session: ServerSession, last_active_at: float_from_monotonic_clock }`.
+          # Maps `session_id` to `{ get_sse_stream: stream_object, server_session: ServerSession, last_active_at: float_from_monotonic_clock, origin: origin_header }`.
           @sessions = {}
           @mutex = Mutex.new
 
           @stateless = stateless
           @enable_json_response = enable_json_response
+          @session_request_validator = session_request_validator
           @dns_rebinding_protection = dns_rebinding_protection
 
           # Host names are case-insensitive, so the allow lists are compared down-cased.
@@ -447,16 +457,25 @@ def handle_post(request)
               return session_not_found_response
             end
 
-            handle_initialization(body_string, body)
-          elsif notification?(body)
-            dispatch_notification(body_string, session_id)
-            handle_accepted
-          elsif response?(body)
-            return session_not_found_response if !@stateless && !session_exists?(session_id)
-
-            handle_response(body, session_id: session_id)
+            handle_initialization(request, body_string, body)
           else
-            handle_regular_request(body_string, session_id, related_request_id: body[:id])
+            # Ownership gate for every request against an existing session, applied uniformly to notifications, client responses,
+            # and regular requests. This covers write paths beyond tool calls - notably `notifications/cancelled`, which would
+            # otherwise let a stolen session ID cancel a victim's in-flight request. `initialize` is exempt (it establishes the session).
+            if !@stateless && session_id && !validate_session_request(request, session_id)
+              return forbidden_response
+            end
+
+            if notification?(body)
+              dispatch_notification(body_string, session_id)
+              handle_accepted
+            elsif response?(body)
+              return session_not_found_response if !@stateless && !session_exists?(session_id)
+
+              handle_response(body, session_id: session_id)
+            else
+              handle_regular_request(body_string, session_id, related_request_id: body[:id])
+            end
           end
         rescue StandardError => e
           MCP.configuration.exception_reporter.call(e, { request: body_string })
@@ -481,6 +500,7 @@ def handle_get(request)
 
           error_response = validate_and_touch_session(session_id)
           return error_response if error_response
+          return forbidden_response unless validate_session_request(request, session_id)
 
           protocol_version_error = validate_protocol_version_header(request)
           return protocol_version_error if protocol_version_error
@@ -503,6 +523,7 @@ def handle_delete(request)
 
           return missing_session_id_response unless (session_id = extract_session_id(request))
           return session_not_found_response unless session_exists?(session_id)
+          return forbidden_response unless validate_session_request(request, session_id)
 
           protocol_version_error = validate_protocol_version_header(request)
           return protocol_version_error if protocol_version_error
@@ -564,6 +585,27 @@ def extract_session_id(request)
           request.env["HTTP_MCP_SESSION_ID"]
         end
 
+        # Session-ownership gate for requests against an existing session (the spec's session-binding guidance).
+        # The session ID alone is unguessable but not proof of ownership, so a stolen ID must not silently grant access.
+        # Two layers, both returning `false` to trigger a 403:
+        #
+        # - Built-in Origin consistency (defense in depth, not authentication): if the session recorded an `Origin`
+        #   at `initialize` and this request carries a different one, reject. Both must be present to compare,
+        #   so non-browser clients that send no `Origin` are unaffected.
+        # - The application-supplied `session_request_validator`, which can enforce true ownership when it has
+        #   an authenticated principal.
+        def validate_session_request(request, session_id)
+          session = @mutex.synchronize { @sessions[session_id] }
+          return true unless session
+
+          session_origin = session[:origin]
+          request_origin = request.env["HTTP_ORIGIN"]
+          return false if session_origin && request_origin && session_origin != request_origin
+          return @session_request_validator.call(request, session_id) if @session_request_validator
+
+          true
+        end
+
         def validate_accept_header(request, required_types)
           accept_header = request.env["HTTP_ACCEPT"]
           return not_acceptable_response(required_types) unless accept_header
@@ -682,7 +724,7 @@ def handle_response(body, session_id:)
           handle_accepted
         end
 
-        def handle_initialization(body_string, body)
+        def handle_initialization(request, body_string, body)
           session_id = nil
 
           if @stateless
@@ -709,6 +751,9 @@ def handle_initialization(body_string, body)
                 get_sse_stream: nil,
                 server_session: server_session,
                 last_active_at: Process.clock_gettime(Process::CLOCK_MONOTONIC),
+                # Captured for the built-in Origin-consistency defense in `validate_session_request`.
+                # Not authentication.
+                origin: request.env["HTTP_ORIGIN"],
               }
               true
             end
@@ -962,7 +1007,7 @@ def same_origin?(origin, request)
           authority.delete_suffix(default_port) == host.downcase.delete_suffix(default_port)
         end
 
-        def forbidden_response(message)
+        def forbidden_response(message = "Forbidden: session request validation failed")
           json_rpc_error_response(
             status: 403,
             code: JsonRpcHandler::ErrorCode::INVALID_REQUEST,
 
@@ -4741,6 +4741,196 @@ def string
           transport.close
         end
 
+        test "rejects a POST to a session whose Origin differs from initialize with 403" do
+          # Both origins are allow-listed so they pass DNS rebinding protection; the rejection
+          # below comes from the per-session Origin-consistency gate, not from the rebinding check.
+          transport = StreamableHTTPTransport.new(
+            @server,
+            enable_json_response: true,
+            allowed_origins: ["https://app.example.com", "https://evil.example.com"],
+          )
+          init = create_rack_request(
+            "POST",
+            "/",
+            { "CONTENT_TYPE" => "application/json", "HTTP_ORIGIN" => "https://app.example.com" },
+            { jsonrpc: "2.0", method: "initialize", id: "init" }.to_json,
+          )
+          session_id = transport.handle_request(init)[1]["Mcp-Session-Id"]
+
+          attacker = create_rack_request(
+            "POST",
+            "/",
+            {
+              "CONTENT_TYPE" => "application/json",
+              "HTTP_MCP_SESSION_ID" => session_id,
+              "HTTP_ORIGIN" => "https://evil.example.com",
+            },
+            { jsonrpc: "2.0", method: "ping", id: "1" }.to_json,
+          )
+          response = transport.handle_request(attacker)
+
+          assert_equal(403, response[0])
+        ensure
+          transport.close
+        end
+
+        test "allows a POST with a matching Origin and one with no Origin" do
+          transport = StreamableHTTPTransport.new(
+            @server,
+            enable_json_response: true,
+            allowed_origins: ["https://app.example.com"],
+          )
+          init = create_rack_request(
+            "POST",
+            "/",
+            { "CONTENT_TYPE" => "application/json", "HTTP_ORIGIN" => "https://app.example.com" },
+            { jsonrpc: "2.0", method: "initialize", id: "init" }.to_json,
+          )
+          session_id = transport.handle_request(init)[1]["Mcp-Session-Id"]
+
+          same_origin = create_rack_request(
+            "POST",
+            "/",
+            {
+              "CONTENT_TYPE" => "application/json",
+              "HTTP_MCP_SESSION_ID" => session_id,
+              "HTTP_ORIGIN" => "https://app.example.com",
+            },
+            { jsonrpc: "2.0", method: "ping", id: "1" }.to_json,
+          )
+
+          # A non-browser client that sends no Origin is not rejected by the built-in check.
+          no_origin = create_rack_request(
+            "POST",
+            "/",
+            { "CONTENT_TYPE" => "application/json", "HTTP_MCP_SESSION_ID" => session_id },
+            { jsonrpc: "2.0", method: "ping", id: "2" }.to_json,
+          )
+
+          assert_equal(200, transport.handle_request(same_origin)[0])
+          assert_equal(200, transport.handle_request(no_origin)[0])
+        ensure
+          transport.close
+        end
+
+        test "a custom session_request_validator rejects and allows requests" do
+          allowed = { value: true }
+          validator = ->(_request, _session_id) { allowed[:value] }
+          transport = StreamableHTTPTransport.new(@server, enable_json_response: true, session_request_validator: validator)
+          init = create_rack_request(
+            "POST",
+            "/",
+            { "CONTENT_TYPE" => "application/json" },
+            { jsonrpc: "2.0", method: "initialize", id: "init" }.to_json,
+          )
+          session_id = transport.handle_request(init)[1]["Mcp-Session-Id"]
+
+          ping = lambda do
+            create_rack_request(
+              "POST",
+              "/",
+              { "CONTENT_TYPE" => "application/json", "HTTP_MCP_SESSION_ID" => session_id },
+              { jsonrpc: "2.0", method: "ping", id: "1" }.to_json,
+            )
+          end
+
+          assert_equal(200, transport.handle_request(ping.call)[0])
+          allowed[:value] = false
+          assert_equal(403, transport.handle_request(ping.call)[0])
+        ensure
+          transport.close
+        end
+
+        test "the validator also gates notification and response POSTs, not just regular requests" do
+          # A stolen session ID must not be able to POST `notifications/cancelled` (which would cancel a victim's in-flight request)
+          # or a client response by bypassing the ownership gate. `initialize` establishes the session and is exempt,
+          # so a single transport whose validator always rejects still initializes.
+          transport = StreamableHTTPTransport.new(
+            @server,
+            enable_json_response: true,
+            session_request_validator: ->(_request, _session_id) { false },
+          )
+          init = create_rack_request(
+            "POST",
+            "/",
+            { "CONTENT_TYPE" => "application/json" },
+            { jsonrpc: "2.0", method: "initialize", id: "init" }.to_json,
+          )
+          session_id = transport.handle_request(init)[1]["Mcp-Session-Id"]
+          refute_nil(session_id)
+
+          headers = { "CONTENT_TYPE" => "application/json", "HTTP_MCP_SESSION_ID" => session_id }
+          cancelled = create_rack_request(
+            "POST",
+            "/",
+            headers,
+            { jsonrpc: "2.0", method: "notifications/cancelled", params: { requestId: "1" } }.to_json,
+          )
+          client_response = create_rack_request(
+            "POST",
+            "/",
+            headers,
+            { jsonrpc: "2.0", id: "1", result: {} }.to_json,
+          )
+
+          assert_equal(403, transport.handle_request(cancelled)[0])
+          assert_equal(403, transport.handle_request(client_response)[0])
+        ensure
+          transport.close
+        end
+
+        test "session_request_validator receives the request and session id" do
+          seen = {}
+          validator = ->(request, session_id) {
+            seen[:origin] = request.env["HTTP_ORIGIN"]
+            seen[:session_id] = session_id
+            true
+          }
+          transport = StreamableHTTPTransport.new(
+            @server,
+            enable_json_response: true,
+            allowed_origins: ["https://app.example.com"],
+            session_request_validator: validator,
+          )
+          init = create_rack_request(
+            "POST",
+            "/",
+            { "CONTENT_TYPE" => "application/json" },
+            { jsonrpc: "2.0", method: "initialize", id: "init" }.to_json,
+          )
+          session_id = transport.handle_request(init)[1]["Mcp-Session-Id"]
+
+          transport.handle_request(create_rack_request(
+            "POST",
+            "/",
+            { "CONTENT_TYPE" => "application/json", "HTTP_MCP_SESSION_ID" => session_id, "HTTP_ORIGIN" => "https://app.example.com" },
+            { jsonrpc: "2.0", method: "ping", id: "1" }.to_json,
+          ))
+
+          assert_equal(session_id, seen[:session_id])
+          assert_equal("https://app.example.com", seen[:origin])
+        ensure
+          transport.close
+        end
+
+        test "session validation does not apply in stateless mode" do
+          rejecting = ->(_request, _session_id) { false }
+          transport = StreamableHTTPTransport.new(@server, stateless: true, enable_json_response: true, session_request_validator: rejecting)
+
+          request = create_rack_request(
+            "POST",
+            "/",
+            { "CONTENT_TYPE" => "application/json" },
+            { jsonrpc: "2.0", method: "ping", id: "1" }.to_json,
+          )
+          response = transport.handle_request(request)
+
+          # Stateless mode keeps no sessions, so the validator is never consulted.
+          assert_equal(200, response[0])
+        ensure
+          transport.close
+        end
+
         private
 
         def create_rack_request(method, path, headers, body = nil)