@@ -40,6 +40,8 @@ All notable changes to `mcp/sdk` will be documented in this file.
   * [BC Break] A handler calling `\Fiber::suspend()` itself with anything but a `NotificationSuspension` or `RequestSuspension` fails the request without a scheduler, also when no event loop is involved. It used to be resumed with `null`.
   * Fix a request failing with `503`, and a resumed request losing its session writes, while another request of the session holds the session lock across an await: both wait for the lock on the event loop.
   * Fix a handler still running after its request was answered, e.g. without a fiber scheduler: it no longer fires `ResponseEvent` or `ErrorEvent`, what it throws is logged, and calling the `ClientGateway` throws.
+* [BC Break] Bind handshake-era sessions to the principal that opened them: `AccessToken::getPrincipal()`, or the `StreamableHttpTransport::PRINCIPAL_ATTRIBUTE` request attribute when set. Requests of another principal, and to sessions opened before the upgrade, are answered with `404`, and to a session the store cannot read with `503`. `Protocol::processInput()`, the transports' `handleMessage()`, `handleSessionEnd()`, `handlePostRequest()` and `handleDeleteRequest()`, and the `onMessage()` and `onSessionEnd()` listeners take a `?string $principal` as last argument.
+* [BC Break] Answer malformed JSON and an `initialize` sent with a session id with HTTP `400`, and a failure before the session is checked with `500`, instead of `200`. Such an answer no longer carries, or removes, the messages queued for the session in `Mcp-Session-Id`.
 
 ### Client
 
 
@@ -313,11 +313,27 @@ challenge a client steps up from. Use `$token->hasScope()` for decisions that de
 arguments beyond the tool name. With a `ScopePolicy`, the token's scopes include those implied by
 its hierarchy.
 
+## Sessions and the Principal
+
+A handshake-era session only serves the principal that opened it, as the
+[security best practices](https://modelcontextprotocol.io/specification/2025-11-25/basic/security_best_practices#session-hijacking)
+recommend. The principal is the issuer, client and subject of the access token. They stay the same
+when the client refreshes its token, so the session survives the refresh.
+
+When another principal sends the session's `Mcp-Session-Id`, the server answers 404, as for an
+unknown session, and a `DELETE` does not end the session.
+
+A token without a `sub` claim only identifies its issuer and client, so all tokens of one client share
+their sessions. If your authorization server identifies users by another claim, set the
+`StreamableHttpTransport::PRINCIPAL_ATTRIBUTE` request attribute from it. A session opened without a
+token only serves requests without a token.
+
 ## Authenticating Outside the SDK
 
 If your application or framework already authenticates the request, skip `AuthorizationMiddleware`
 and hand the result to the transport as the PSR-7 request attribute `AccessToken::class`. Both HTTP
-transports read it from there, so handlers get it through `RequestContext::getAccessToken()` as usual:
+transports read it from there, so handlers get it through `RequestContext::getAccessToken()` as usual,
+and a session only serves the principal of the token that opened it:
 
 ```php
 use Mcp\Server\Authorization\AccessToken;
@@ -326,6 +342,16 @@ $request = $request->withAttribute(AccessToken::class, new AccessToken($scopes,
 $transport = new StreamableHttpTransport($request);
 ```
 
+If the claims do not identify the user, or there is no token, set the
+`StreamableHttpTransport::PRINCIPAL_ATTRIBUTE` attribute as well. It takes a non-empty string, so cast
+an integer user id. The server answers any other value with a 500:
+
+```php
+use Mcp\Server\Transport\StreamableHttpTransport;
+
+$request = $request->withAttribute(StreamableHttpTransport::PRINCIPAL_ATTRIBUTE, (string) $userId);
+```
+
 ## Custom Token Validators
 
 Implement `AuthorizationTokenValidatorInterface` for other token formats, e.g. opaque tokens
 
@@ -81,6 +81,16 @@ public function getClientId(): ?string
         return \is_string($clientId) ? $clientId : null;
     }
 
+    /**
+     * Who the token identifies: its issuer, client and subject, which outlive a token refresh.
+     */
+    public function getPrincipal(): string
+    {
+        $issuer = $this->getClaim('iss');
+
+        return serialize([\is_string($issuer) ? $issuer : null, $this->getClientId(), $this->getSubject()]);
+    }
+
     /**
      * Never serialized: the token must not outlive its request in a session store.
      */
 
@@ -74,6 +74,11 @@ class Protocol
 
     public const SESSION_LOGGING_LEVEL = '_mcp.logging_level';
 
+    private const SESSION_NOT_FOUND_MESSAGE = 'Session not found or has expired.';
+
+    /** Session key for the fingerprint of the principal that opened the session, null when it was opened without one */
+    private const SESSION_PRINCIPAL = '_mcp.principal';
+
     /**
      * Deliberately generic: unexpected throwables carry internal details such as file paths, class
      * names and argument types, which must not be handed to the peer. The full exception is logged.
@@ -147,7 +152,7 @@ public function connect(TransportInterface $transport): void
     {
         $transport->onMessage($this->processInput(...));
 
-        $transport->onSessionEnd($this->destroySession(...));
+        $transport->onSessionEnd($this->endSession(...));
 
         // The transport keeps these callbacks, so they reference it weakly to not keep it alive.
         $transportRef = \WeakReference::create($transport);
@@ -175,13 +180,14 @@ public function connect(TransportInterface $transport): void
      *
      * @param TransportInterface<mixed> $transport
      * @param AccessToken|null          $accessToken the token the input was authorized with, if the transport authorizes
+     * @param string|null               $principal   who the input was authenticated as, if the transport authenticates
      */
-    public function processInput(TransportInterface $transport, string $input, ?Uuid $sessionId, ?AccessToken $accessToken = null): void
+    public function processInput(TransportInterface $transport, string $input, ?Uuid $sessionId, ?AccessToken $accessToken = null, ?string $principal = null): void
     {
         // Last line of defense: a malformed message must never escape as a PHP error and take the
         // server process down.
         try {
-            $this->doProcessInput($transport, $input, $sessionId, $accessToken);
+            $this->doProcessInput($transport, $input, $sessionId, $accessToken, $principal);
         } catch (\Throwable $e) {
             $this->logger->error(\sprintf('Uncaught exception while processing input: %s', $e->getMessage()), ['exception' => $e]);
 
@@ -192,7 +198,7 @@ public function processInput(TransportInterface $transport, string $input, ?Uuid
             }
 
             try {
-                $this->sendResponse($transport, Error::forInternalError(self::INTERNAL_ERROR_MESSAGE, $id), null);
+                $this->sendResponse($transport, Error::forInternalError(self::INTERNAL_ERROR_MESSAGE, $id), null, ['status_code' => 500]);
             } catch (\Throwable $e) {
                 $this->logger->error(\sprintf('Failed to send internal error response: %s', $e->getMessage()), ['exception' => $e]);
             }
@@ -226,7 +232,7 @@ private static function findResponseId(string $input): string|int|null
     /**
      * @param TransportInterface<mixed> $transport
      */
-    private function doProcessInput(TransportInterface $transport, string $input, ?Uuid $sessionId, ?AccessToken $accessToken): void
+    private function doProcessInput(TransportInterface $transport, string $input, ?Uuid $sessionId, ?AccessToken $accessToken, ?string $principal): void
     {
         $this->logger->info('Received message to process.');
         $this->logger->debug('Received message payload.', ['message' => $input]);
@@ -238,7 +244,8 @@ private function doProcessInput(TransportInterface $transport, string $input, ?U
         } catch (\JsonException $e) {
             $this->logger->warning('Failed to decode json message.', ['exception' => $e]);
             $error = Error::forParseError($e->getMessage());
-            $this->sendResponse($transport, $error, null);
+            // The status keeps a refusal from carrying the queue of a session nobody checked yet, here and below.
+            $this->sendResponse($transport, $error, null, ['status_code' => 400]);
 
             return;
         } catch (InvalidInputMessageException $e) {
@@ -247,13 +254,29 @@ private function doProcessInput(TransportInterface $transport, string $input, ?U
         }
 
         // Taken before the session is read and held until it is saved, so no other request of
-        // the session reads a state this one is about to replace. A session created here for
-        // an `initialize` is not known to anyone else yet and needs no lock, and an unknown
-        // session id is refused by resolveSession() without one; the session is looked up
-        // again under the lock, as it may have ended while this request waited.
-        $takesLock = null !== $sessionId && null !== $this->sessionLock && $this->sessionManager->exists($sessionId);
+        // the session reads a state this one is about to replace. An `initialize` never reads an
+        // existing session and an unknown session id is refused by resolveSession(), so neither
+        // needs the lock. The session is looked up again under the lock, as it may have ended
+        // while this request waited.
+        $takesLock = null !== $sessionId && null !== $this->sessionLock && !$message instanceof InitializeRequest && $this->sessionManager->exists($sessionId);
 
         if ($takesLock) {
+            // Who opened the session is checked before waiting for the lock too, so that another
+            // principal can neither tell the session is busy nor keep it busy.
+            try {
+                $serves = $this->servesPrincipal($this->sessionManager->createWithId($sessionId), $principal);
+            } catch (SessionStoreException $e) {
+                $this->refuseUnavailableStore($transport, $sessionId, self::findResponseId($input), $e);
+
+                return;
+            }
+
+            if (!$serves) {
+                $this->refuseOtherPrincipal($transport, $sessionId);
+
+                return;
+            }
+
             try {
                 $refusal = $this->acquireSessionLock($sessionId) ? null : Error::forServerError(self::SESSION_BUSY_MESSAGE, self::findResponseId($input));
             } catch (SessionLockException) {
@@ -268,7 +291,7 @@ private function doProcessInput(TransportInterface $transport, string $input, ?U
         }
 
         try {
-            $this->handleMessage($transport, $sessionId, $message, $accessToken);
+            $this->handleMessage($transport, $sessionId, $message, $accessToken, $principal);
         } finally {
             if ($takesLock) {
                 $this->releaseSessionLock($sessionId);
@@ -279,9 +302,9 @@ private function doProcessInput(TransportInterface $transport, string $input, ?U
     /**
      * @param TransportInterface<mixed> $transport
      */
-    private function handleMessage(TransportInterface $transport, ?Uuid $sessionId, MessageInterface|InvalidInputMessageException $message, ?AccessToken $accessToken): void
+    private function handleMessage(TransportInterface $transport, ?Uuid $sessionId, MessageInterface|InvalidInputMessageException $message, ?AccessToken $accessToken, ?string $principal): void
     {
-        $session = $this->resolveSession($transport, $sessionId, $message);
+        $session = $this->resolveSession($transport, $sessionId, $message, $principal);
         if (null === $session) {
             return;
         }
@@ -1107,19 +1130,22 @@ private function errorForThrowable(\Throwable $e, Request $request, SessionInter
      *
      * @param TransportInterface<mixed> $transport
      * @param Uuid|null                 $sessionId The session ID from the transport
+     * @param string|null               $principal Who the messages were authenticated as
      */
-    private function resolveSession(TransportInterface $transport, ?Uuid $sessionId, MessageInterface|InvalidInputMessageException $message): ?SessionInterface
+    private function resolveSession(TransportInterface $transport, ?Uuid $sessionId, MessageInterface|InvalidInputMessageException $message, ?string $principal): ?SessionInterface
     {
         if ($message instanceof InitializeRequest) {
             // Spec: An initialize request must not have a session ID.
             if ($sessionId) {
                 $error = Error::forInvalidRequest('A session ID MUST NOT be sent with an "initialize" request.');
-                $this->sendResponse($transport, $error, null);
+                $this->sendResponse($transport, $error, null, ['status_code' => 400]);
 
                 return null;
             }
 
             $session = $this->sessionManager->create();
+            $session->set(self::SESSION_PRINCIPAL, self::fingerprint($principal));
+
             $this->logger->debug('Created new session for initialize', [
                 'session_id' => $session->getId()->toRfc4122(),
             ]);
@@ -1140,25 +1166,88 @@ private function resolveSession(TransportInterface $transport, ?Uuid $sessionId,
         try {
             $exists = $this->sessionManager->exists($sessionId);
         } catch (SessionStoreException $e) {
-            // Not a 404: the session may well exist, so the client should retry instead of re-initializing.
-            $this->logger->error('Session store is unavailable.', [
-                'session_id' => $sessionId->toRfc4122(),
-                'exception' => $e,
-            ]);
-            $error = Error::forInternalError('Session store is unavailable.', self::findRequestId($message));
-            $this->sendResponse($transport, $error, null, ['status_code' => 503]);
+            $this->refuseUnavailableStore($transport, $sessionId, self::findRequestId($message), $e);
 
             return null;
         }
 
         if (!$exists) {
-            $error = Error::forInvalidRequest('Session not found or has expired.');
+            $error = Error::forInvalidRequest(self::SESSION_NOT_FOUND_MESSAGE);
             $this->sendResponse($transport, $error, null, ['status_code' => 404]);
 
             return null;
         }
 
-        return $this->sessionManager->createWithId($sessionId);
+        $session = $this->sessionManager->createWithId($sessionId);
+
+        try {
+            $serves = $this->servesPrincipal($session, $principal);
+        } catch (SessionStoreException $e) {
+            $this->refuseUnavailableStore($transport, $sessionId, self::findRequestId($message), $e);
+
+            return null;
+        }
+
+        if (!$serves) {
+            $this->refuseOtherPrincipal($transport, $sessionId);
+
+            return null;
+        }
+
+        return $session;
+    }
+
+    /**
+     * Whether the session serves the principal: the one that opened it, or no principal for a session opened without one.
+     *
+     * A session that does not record who opened it serves no one.
+     *
+     * @throws SessionStoreException when the session exists but its data cannot be read
+     */
+    private function servesPrincipal(SessionInterface $session, ?string $principal): bool
+    {
+        if ($session->get(self::SESSION_PRINCIPAL, false) === self::fingerprint($principal)) {
+            return true;
+        }
+
+        if ([] === $session->all()) {
+            throw new SessionStoreException(\sprintf('Session "%s" could not be read.', $session->getId()->toRfc4122()));
+        }
+
+        return false;
+    }
+
+    /**
+     * Not a 404: the session may well exist, so the client should retry instead of re-initializing.
+     *
+     * @param TransportInterface<mixed> $transport
+     */
+    private function refuseUnavailableStore(TransportInterface $transport, Uuid $sessionId, string|int|null $requestId, SessionStoreException $exception): void
+    {
+        $this->logger->error('Session store is unavailable.', [
+            'session_id' => $sessionId->toRfc4122(),
+            'exception' => $exception,
+        ]);
+        $this->sendResponse($transport, Error::forInternalError('Session store is unavailable.', $requestId), null, ['status_code' => 503]);
+    }
+
+    /**
+     * @param TransportInterface<mixed> $transport
+     */
+    private function refuseOtherPrincipal(TransportInterface $transport, Uuid $sessionId): void
+    {
+        $this->logger->warning('Refused a request of another principal than the one that opened the session.', [
+            'session_id' => $sessionId->toRfc4122(),
+        ]);
+
+        // Answered like an unknown session, so that another principal cannot tell the session exists.
+        $error = Error::forInvalidRequest(self::SESSION_NOT_FOUND_MESSAGE);
+        $this->sendResponse($transport, $error, null, ['status_code' => 404]);
+    }
+
+    private static function fingerprint(?string $principal): ?string
+    {
+        return null === $principal ? null : hash('sha256', $principal);
     }
 
     /**
@@ -1173,6 +1262,26 @@ private static function findRequestId(MessageInterface|InvalidInputMessageExcept
         };
     }
 
+    /**
+     * Ends a session at the request of a client, which must be the principal that opened it.
+     *
+     * @return bool false when the session does not exist, or is another principal's
+     *
+     * @throws SessionStoreException when who opened the session cannot be established; the session is not ended
+     */
+    private function endSession(Uuid $sessionId, ?string $principal = null): bool
+    {
+        if ($this->sessionManager->exists($sessionId) && !$this->servesPrincipal($this->sessionManager->createWithId($sessionId), $principal)) {
+            $this->logger->warning('Refused to end a session for another principal than the one that opened it.', [
+                'session_id' => $sessionId->toRfc4122(),
+            ]);
+
+            return false;
+        }
+
+        return $this->destroySession($sessionId);
+    }
+
     /**
      * Destroy a specific session.
      *
 
@@ -11,6 +11,7 @@
 
 namespace Mcp\Server\Transport;
 
+use Mcp\Exception\SessionStoreException;
 use Mcp\Schema\JsonRpc\Error;
 use Mcp\Schema\JsonRpc\Response;
 use Mcp\Server\Authorization\AccessToken;
@@ -166,20 +167,34 @@ protected function resumeHandlerFiber(\Fiber $fiber, ?Uuid $sessionId, Response|
         }
     }
 
-    protected function handleMessage(string $payload, ?Uuid $sessionId, ?AccessToken $accessToken = null): void
+    protected function handleMessage(string $payload, ?Uuid $sessionId, ?AccessToken $accessToken = null, ?string $principal = null): void
     {
         if (\is_callable($this->messageListener)) {
-            ($this->messageListener)($this, $payload, $sessionId, $accessToken);
+            ($this->messageListener)($this, $payload, $sessionId, $accessToken, $principal);
+        }
+    }
+
+    /**
+     * Ends the session of a transport shutting down, which has no client to report a store failure to.
+     */
+    protected function endSessionOnShutdown(?Uuid $sessionId): void
+    {
+        try {
+            $this->handleSessionEnd($sessionId);
+        } catch (SessionStoreException $e) {
+            $this->logger->error('Session store is unavailable, the session was not ended.', ['exception' => $e]);
         }
     }
 
     /**
      * @return bool false when the listener reported that the session does not exist
+     *
+     * @throws SessionStoreException when the listener cannot tell who opened the session
      */
-    protected function handleSessionEnd(?Uuid $sessionId): bool
+    protected function handleSessionEnd(?Uuid $sessionId, ?string $principal = null): bool
     {
         if ($sessionId && \is_callable($this->sessionEndListener)) {
-            return false !== ($this->sessionEndListener)($sessionId);
+            return false !== ($this->sessionEndListener)($sessionId, $principal);
         }
 
         return true;