chr-hertel · Oct 10, 2026 · Oct 10, 2026
@@ -5,6 +5,8 @@ All notable changes to `mcp/sdk` will be documented in this file.
 0.9.0
 -----
 
+* Fix concurrent requests of one session overwriting each other's session changes: `Session::save()` now writes only the keys it changed, onto what the store holds at that point.
+* Add `Builder::setSessionLock()` and `SymfonySessionLock`, an opt-in lock that serializes concurrent requests of one session, so they no longer lose each other's changes to the same key.
 * [BC Break] `SchemaValidator` takes an optional `Opis\JsonSchema\Validator` as its first constructor argument, moving `$logger` to second. Pass `logger:` by name.
 * [BC Break] `SchemaValidator::validateAgainstJsonSchema()` no longer validates an empty array as an object. Pass `new \stdClass()` for an empty object.
 * Add `Builder::setSchemaValidator()` to configure the validator used for `tools/call` input and output, e.g. with a resolver for external `$ref` schemas.

@@ -38,7 +38,8 @@
   "suggest": {
     "firebase/php-jwt": "Required for the JwtTokenValidator.",
     "psr/cache-implementation": "Required for JwtTokenValidator::fromIssuer() to cache metadata and keys.",
-    "symfony/finder": "Required for file-based discovery."
+    "symfony/finder": "Required for file-based discovery.",
+    "symfony/lock": "Required for the SymfonySessionLock."
   },
   "require-dev": {
     "ext-openssl": "*",
@@ -58,6 +59,7 @@
     "symfony/console": "^5.4 || ^6.4 || ^7.3 || ^8.0",
     "symfony/finder": "^5.4 || ^6.4 || ^7.3 || ^8.0",
     "symfony/http-client": "^5.4 || ^6.4 || ^7.3 || ^8.0",
+    "symfony/lock": "^5.4 || ^6.4 || ^7.3 || ^8.0",
     "symfony/process": "^5.4 || ^6.4 || ^7.3 || ^8.0"
   },
   "conflict": {

@@ -127,3 +127,38 @@ class RedisSessionStore implements SessionStoreInterface
     }
 }
 ```
+
+## Concurrent Requests
+
+A client can send several requests of one session at the same time, for example parallel tool calls. Each
+request loads the session, changes it, and saves the keys it changed. When two requests change different keys,
+both changes are kept. When they change the same key, for example a counter in a tool handler, the request that
+saves last wins and the other change is lost.
+
+To prevent this, configure a session lock. A request then holds the lock from loading its session to saving it,
+and the other requests of that session wait for it:
+
+```php
+use Mcp\Server\Session\SymfonySessionLock;
+use Symfony\Component\Lock\LockFactory;
+use Symfony\Component\Lock\Store\FlockStore;
+
+// composer require symfony/lock
+$server = Server::builder()
+    ->setSession(new FileSessionStore(__DIR__ . '/sessions'))
+    ->setSessionLock(new SymfonySessionLock(new LockFactory(new FlockStore())))
+    ->build();
+```
+
+All workers must share the lock store: `FlockStore` works for workers on one host, use e.g. a `RedisStore`
+when the workers run on several hosts.
+
+- `$timeout` (float): Seconds a request waits for the lock (default: `30.0`). After that, it fails with a
+  `TimeoutException` and the client gets an internal error. Use a value above the duration of your longest
+  handler.
+- `$ttl` (float): Seconds after which the store expires a lock that was never released, e.g. by a crashed
+  worker (default: `300.0`).
+
+A request that waits on the client, for example for an elicitation or a sampling result, releases the lock
+while it waits. The lock is off by default, and the stdio transport does not need it, because it serves one
+client in one process. Implement `SessionLockInterface` to use another locking mechanism.
@@ -54,6 +54,7 @@
 use Mcp\Server\Resource\SessionSubscriptionManager;
 use Mcp\Server\Resource\SubscriptionManagerInterface;
 use Mcp\Server\Session\InMemorySessionStore;
+use Mcp\Server\Session\SessionLockInterface;
 use Mcp\Server\Session\SessionManager;
 use Mcp\Server\Session\SessionManagerInterface;
 use Mcp\Server\Session\SessionStoreInterface;
@@ -116,6 +117,8 @@ final class Builder
 
     private ?SessionStoreInterface $sessionStore = null;
 
+    private ?SessionLockInterface $sessionLock = null;
+
     private int $gcProbability = 1;
 
     private int $gcDivisor = 100;
@@ -637,6 +640,17 @@ public function setSession(
         return $this;
     }
 
+    /**
+     * Locks a handshake-era session from loading to saving it, so concurrent requests of one session take turns
+     * instead of losing each other's changes to the same key. Off by default.
+     */
+    public function setSessionLock(?SessionLockInterface $sessionLock): self
+    {
+        $this->sessionLock = $sessionLock;
+
+        return $this;
+    }
+
     /**
      * @param string[] $scanDirs
      * @param string[] $excludeDirs
@@ -955,6 +969,7 @@ public function build(): Server
                 ? new InputRequiredShim($this->inputRequiredRounds, $this->inputRequiredTimeout, $parts['logger'])
                 : null,
             requestStateCodec: $this->requestStateCodec(),
+            sessionLock: $this->sessionLock,
         );
 
         $modernVersions = $this->modernVersions ?? ProtocolVersion::modernVersions();

@@ -28,6 +28,7 @@
 use Mcp\Server\Handler\Notification\NotificationHandlerInterface;
 use Mcp\Server\Handler\Request\RequestHandlerInterface;
 use Mcp\Server\Session\SessionInterface;
+use Mcp\Server\Session\SessionLockInterface;
 use Mcp\Server\Session\SessionManagerInterface;
 use Mcp\Server\Stateless\InputContext;
 use Mcp\Server\Stateless\RequestStateCodec;
@@ -94,6 +95,7 @@ public function __construct(
         private readonly ?EventDispatcherInterface $eventDispatcher = null,
         private readonly ?InputRequiredShim $inputRequiredShim = null,
         private readonly ?RequestStateCodec $requestStateCodec = null,
+        private readonly ?SessionLockInterface $sessionLock = null,
     ) {
         $this->awaitedRequestIds = new \WeakMap();
     }
@@ -221,6 +223,16 @@ private function doProcessInput(TransportInterface $transport, string $input, ?U
             return;
         }
 
+        // A request that suspends saves its session and returns, so the lock is free while it waits on the client.
+        $this->locked($session->getId(), fn () => $this->handleMessages($transport, $messages, $session, $accessToken));
+    }
+
+    /**
+     * @param TransportInterface<mixed> $transport
+     * @param array<int, mixed>         $messages
+     */
+    private function handleMessages(TransportInterface $transport, array $messages, SessionInterface $session, ?AccessToken $accessToken): void
+    {
         foreach ($messages as $message) {
             // Guarded per message so one faulty message cannot suppress the rest of a batch.
             try {
@@ -578,16 +590,23 @@ private function queueOutgoing(Request|Notification $message, array $context, Se
      */
     public function consumeOutgoingMessages(Uuid $sessionId): array
     {
-        $session = $this->sessionManager->createWithId($sessionId);
-        $queue = $session->get(self::SESSION_OUTGOING_QUEUE, []);
-
-        // Saving an unchanged session would only overwrite what a concurrent request saved in the meantime.
-        if ([] !== $queue) {
-            $session->set(self::SESSION_OUTGOING_QUEUE, []);
-            $session->save();
+        // The polling loop asks many times a second and mostly finds nothing: it only locks when there is something to take.
+        if (null !== $this->sessionLock && [] === $this->sessionManager->createWithId($sessionId)->get(self::SESSION_OUTGOING_QUEUE, [])) {
+            return [];
         }
 
-        return $queue;
+        return $this->locked($sessionId, function () use ($sessionId): array {
+            $session = $this->sessionManager->createWithId($sessionId);
+            $queue = $session->get(self::SESSION_OUTGOING_QUEUE, []);
+
+            // Saving an unchanged session would only overwrite what a concurrent request saved in the meantime.
+            if ([] !== $queue) {
+                $session->set(self::SESSION_OUTGOING_QUEUE, []);
+                $session->save();
+            }
+
+            return $queue;
+        });
     }
 
     /**
@@ -601,23 +620,16 @@ public function consumeOutgoingMessages(Uuid $sessionId): array
      */
     public function checkResponse(int $requestId, Uuid $sessionId): Response|Error|null
     {
-        $session = $this->sessionManager->createWithId($sessionId);
-        $responseData = $session->get(self::SESSION_RESPONSES.".{$requestId}");
-
-        if (null === $responseData) {
+        if (null !== $this->sessionLock && null === $this->sessionManager->createWithId($sessionId)->get(self::SESSION_RESPONSES.".{$requestId}")) {
             return null;
         }
 
-        $this->logger->debug('Found and consuming client response.', [
-            'request_id' => $requestId,
-            'session_id' => $sessionId->toRfc4122(),
-        ]);
+        // Taken under the lock, from a fresh read, before the transport resumes the fiber with it.
+        $responseData = $this->locked($sessionId, fn (): mixed => $this->consumeResponse($requestId, $sessionId));
 
-        $session->set(self::SESSION_RESPONSES.".{$requestId}", null);
-        $pending = $session->get(self::SESSION_PENDING_REQUESTS, []);
-        unset($pending[$requestId]);
-        $session->set(self::SESSION_PENDING_REQUESTS, $pending);
-        $session->save();
+        if (null === $responseData) {
+            return null;
+        }
 
         try {
             if (isset($responseData['error'])) {
@@ -636,6 +648,32 @@ public function checkResponse(int $requestId, Uuid $sessionId): Response|Error|n
         }
     }
 
+    /**
+     * Removes the client's response to a request from the session, along with the request pending on it.
+     */
+    private function consumeResponse(int $requestId, Uuid $sessionId): mixed
+    {
+        $session = $this->sessionManager->createWithId($sessionId);
+        $responseData = $session->get(self::SESSION_RESPONSES.".{$requestId}");
+
+        if (null === $responseData) {
+            return null;
+        }
+
+        $this->logger->debug('Found and consuming client response.', [
+            'request_id' => $requestId,
+            'session_id' => $sessionId->toRfc4122(),
+        ]);
+
+        $session->set(self::SESSION_RESPONSES.".{$requestId}", null);
+        $pending = $session->get(self::SESSION_PENDING_REQUESTS, []);
+        unset($pending[$requestId]);
+        $session->set(self::SESSION_PENDING_REQUESTS, $pending);
+        $session->save();
+
+        return $responseData;
+    }
+
     /**
      * Get pending requests for a session.
      *
@@ -672,26 +710,57 @@ public function handleFiberYield(mixed $yieldedValue, ?Uuid $sessionId): ?int
             return null;
         }
 
-        $session = $this->sessionManager->createWithId($sessionId);
-
         if ($yieldedValue->sessionId !== $sessionId->toRfc4122()) {
             $this->logger->warning('Fiber yielded payload with mismatched session ID.', [
                 'payload_session_id' => $yieldedValue->sessionId,
                 'expected_session_id' => $sessionId->toRfc4122(),
             ]);
         }
 
-        try {
-            if ($yieldedValue instanceof RequestSuspension) {
-                return $this->sendRequest($yieldedValue->request, $yieldedValue->timeout, $session);
+        return $this->locked($sessionId, function () use ($yieldedValue, $sessionId): ?int {
+            $session = $this->sessionManager->createWithId($sessionId);
+
+            try {
+                if ($yieldedValue instanceof RequestSuspension) {
+                    return $this->sendRequest($yieldedValue->request, $yieldedValue->timeout, $session);
+                }
+
+                $this->sendNotification($yieldedValue->notification, $session);
+            } finally {
+                $session->save();
             }
 
-            $this->sendNotification($yieldedValue->notification, $session);
-        } finally {
-            $session->save();
+            return null;
+        });
+    }
+
+    /**
+     * Runs $callback with the session locked, if a session lock is configured.
+     *
+     * @template T
+     *
+     * @param callable(): T $callback
+     *
+     * @return T
+     */
+    private function locked(Uuid $sessionId, callable $callback): mixed
+    {
+        if (null === $this->sessionLock) {
+            return $callback();
         }
 
-        return null;
+        $this->sessionLock->acquire($sessionId);
+
+        try {
+            return $callback();
+        } finally {
+            try {
+                $this->sessionLock->release($sessionId);
+            } catch (\Throwable $e) {
+                // The lock store expires the lock after its TTL; failing here would only discard what the callback did.
+                $this->logger->error(\sprintf('Failed to release session lock: %s', $e->getMessage()), ['exception' => $e]);
+            }
+        }
     }
 
     /**