[Server] Send every response through TransportInterface::send()

CHANGELOG.md

@@ -35,7 +35,8 @@ All notable changes to `mcp/sdk` will be documented in this file.
 * [BC Break] Add `ScopePolicy` as third argument of `AuthorizationMiddleware`, answering `403 insufficient_scope` per method and tool, with scope hierarchies; the `resource_metadata` challenge URL comes from the configured resource instead of the `Host` header.
 * Expose `WWW-Authenticate` in the default `CorsMiddleware`.
 * [BC Break] Fix concurrent Streamable HTTP streams on one session resuming each other's fibers: each stream now polls only the client request its own fiber sent, so an elicitation answer reaches the tool call that asked for it. `Protocol::handleFiberYield()` returns the ID of the request it sent.
-* Fix lost responses on concurrent requests of one session over Streamable HTTP: a POST is answered with its own responses instead of taking them from the session's outgoing queue. Adds `InlineResponseTransportInterface` for transports that answer each request on the exchange that carried it.
+* Fix lost responses on concurrent requests of one session over Streamable HTTP: a POST is answered with its own responses instead of taking them from the session's outgoing queue.
+* [BC Break] `TransportInterface::send()` receives every response of a handler that does not suspend, not only errors raised before a session exists; the session's outgoing queue only carries server-initiated requests and notifications. A custom transport must deliver each message it gets, as a batch gets several.
 * Serve both protocol eras over stdio: `StdioTransport` settles the era on the client's first request and serves `2026-07-28` requests, `subscriptions/listen` and `notifications/cancelled` on the one channel.
 * [BC Break] `StatelessAwareTransportInterface` declares `setHandshakeVersions()`, so a server without the modern era names only the revisions it negotiates when refusing a `2026-07-28` request, e.g. the one set with `Builder::setProtocolVersion()`.
 * Answer a bare `initialize` on a `2026-07-28`-only endpoint with `-32022` naming the served revisions, and a request without a session on the handshake leg with its id.

src/Server/Protocol.php

@@ -16,6 +16,7 @@
 use Mcp\Event\RequestEvent;
 use Mcp\Event\ResponseEvent;
 use Mcp\Exception\InvalidInputMessageException;
+use Mcp\Exception\RuntimeException;
 use Mcp\JsonRpc\MessageFactory;
 use Mcp\Schema\JsonRpc\Error;
 use Mcp\Schema\JsonRpc\Notification;
@@ -32,7 +33,6 @@
 use Mcp\Server\Stateless\RequestStateCodec;
 use Mcp\Server\Suspension\NotificationSuspension;
 use Mcp\Server\Suspension\RequestSuspension;
-use Mcp\Server\Transport\InlineResponseTransportInterface;
 use Mcp\Server\Transport\TransportInterface;
 use Psr\EventDispatcher\EventDispatcherInterface;
 use Psr\Log\LoggerInterface;
@@ -244,7 +244,12 @@ private function doProcessInput(TransportInterface $transport, string $input, ?U
             }
         }
 
-        $session->save();
+        try {
+            $session->save();
+        } catch (\Throwable $e) {
+            // The responses already reached the transport: an error for the same requests would contradict them.
+            $this->logger->error(\sprintf('Failed to save session: %s', $e->getMessage()), ['exception' => $e]);
+        }
     }
 
     /**
@@ -343,13 +348,32 @@ private function handleRequest(TransportInterface $transport, Request $request,
                 $result = $fiber->start();
 
                 if ($fiber->isSuspended()) {
+                    $beforeSuspension = $session->all();
+
                     $awaitedRequestId = null;
                     if ($result instanceof NotificationSuspension) {
                         $this->sendNotification($result->notification, $session);
                     } elseif ($result instanceof RequestSuspension) {
                         $awaitedRequestId = $this->sendRequest($result->request, $result->timeout, $session);
                     }
 
+                    // The transport resumes the fiber from what the session holds: it must
+                    // not get the fiber if the request the fiber awaits was not stored. A
+                    // later save must not store that request either, nobody would resume it.
+                    try {
+                        $saved = $session->save();
+                    } catch (\Throwable $e) {
+                        $session->hydrate($beforeSuspension);
+
+                        throw $e;
+                    }
+
+                    if (!$saved) {
+                        $session->hydrate($beforeSuspension);
+
+                        throw new RuntimeException('Failed to save the session of a suspended request.');
+                    }
+
                     $this->trackAwaitedRequest($transport, $awaitedRequestId);
                     $transport->attachFiberToSession($fiber, $session->getId());
 
@@ -481,61 +505,53 @@ public function sendNotification(Notification $notification, SessionInterface $s
     }
 
     /**
-     * Sends a response either immediately or queued for later delivery.
+     * Sends a response through the transport, on the exchange that carried its request.
+     *
+     * Responses never go through the session's outgoing queue: the session is shared by
+     * the concurrent requests of a client and written back whole, so a queued response
+     * could be overwritten by another request, or taken by it.
      *
      * @param TransportInterface<mixed>                            $transport
      * @param Response<ResultInterface|array<string, mixed>>|Error $response
      * @param array<string, mixed>                                 $context
      */
     private function sendResponse(TransportInterface $transport, Response|Error $response, ?SessionInterface $session, array $context = []): void
     {
-        // Queued in the session, a response can be overwritten or taken by a concurrent
-        // request of the same session: a transport that can answer on the request's
-        // own exchange gets it directly.
-        if (null === $session || $transport instanceof InlineResponseTransportInterface) {
-            $this->logger->debug('Sending immediate response', [
-                'response_id' => $response->getId(),
-            ]);
+        $this->logger->debug('Sending response', [
+            'response_id' => $response->getId(),
+        ]);
 
-            try {
-                $encoded = json_encode($response, \JSON_THROW_ON_ERROR);
-            } catch (\JsonException $e) {
-                $this->logger->error('Failed to encode response to JSON.', [
-                    'message_id' => $response->getId(),
-                    'exception' => $e,
-                ]);
-
-                $fallbackError = new Error(
-                    id: $response->getId(),
-                    code: Error::INTERNAL_ERROR,
-                    message: 'Response could not be encoded to JSON'
-                );
-
-                $encoded = json_encode($fallbackError, \JSON_THROW_ON_ERROR);
-            }
+        try {
+            $encoded = json_encode($response, \JSON_THROW_ON_ERROR);
+        } catch (\JsonException $e) {
+            $this->logger->error('Failed to encode response to JSON.', [
+                'message_id' => $response->getId(),
+                'exception' => $e,
+            ]);
 
-            $context['type'] = 'response';
-            if (null !== $session) {
-                $context['session_id'] = $session->getId();
-            }
+            $fallbackError = new Error(
+                id: $response->getId(),
+                code: Error::INTERNAL_ERROR,
+                message: 'Response could not be encoded to JSON'
+            );
 
-            $transport->send($encoded, $context);
-        } else {
-            $this->logger->info('Queueing server response', [
-                'response_id' => $response->getId(),
-            ]);
+            $encoded = json_encode($fallbackError, \JSON_THROW_ON_ERROR);
+        }
 
-            $this->queueOutgoing($response, ['type' => 'response'], $session);
+        $context['type'] = 'response';
+        if (null !== $session) {
+            $context['session_id'] = $session->getId();
         }
+
+        $transport->send($encoded, $context);
     }
 
     /**
      * Helper to queue outgoing messages in session.
      *
-     * @param Request|Notification|Response<ResultInterface|array<string, mixed>>|Error $message
-     * @param array<string, mixed>                                                      $context
+     * @param array<string, mixed> $context
      */
-    private function queueOutgoing(Request|Notification|Response|Error $message, array $context, SessionInterface $session): void
+    private function queueOutgoing(Request|Notification $message, array $context, SessionInterface $session): void
     {
         try {
             $encoded = json_encode($message, \JSON_THROW_ON_ERROR);

src/Server/Transport/StreamableHttpTransport.php

@@ -54,7 +54,7 @@
  *
  * @author Kyrian Obikwelu <koshnawaza@gmail.com>
  */
-class StreamableHttpTransport extends BaseTransport implements StatelessAwareTransportInterface, InlineResponseTransportInterface
+class StreamableHttpTransport extends BaseTransport implements StatelessAwareTransportInterface
 {
     use ReadsBoundedBody;
 
@@ -80,8 +80,8 @@ class StreamableHttpTransport extends BaseTransport implements StatelessAwareTra
     private ?string $immediateResponse = null;
     private ?int $immediateStatusCode = null;
 
-    /** @var list<string> responses to the requests of the current POST, see {@see InlineResponseTransportInterface} */
-    private array $inlineResponses = [];
+    /** @var list<string> responses to the requests of the current POST */
+    private array $responses = [];
 
     /** @var list<MiddlewareInterface>|null null until {@see self::listen()} resolves the defaults */
     private ?array $middleware;
@@ -181,14 +181,15 @@ public function setHandshakeVersions(array $versions): void
 
     public function send(string $data, array $context): void
     {
-        if (isset($context['session_id'])) {
-            $this->inlineResponses[] = $data;
+        // An error with its own HTTP status rejects the whole POST and is its only answer.
+        if (isset($context['status_code'])) {
+            $this->immediateResponse = $data;
+            $this->immediateStatusCode = $context['status_code'];
 
             return;
         }
 
-        $this->immediateResponse = $data;
-        $this->immediateStatusCode = $context['status_code'] ?? 200;
+        $this->responses[] = $data;
     }
 
     public function listen(): ResponseInterface
@@ -222,7 +223,7 @@ protected function handlePostRequest(string $body, ?AccessToken $accessToken = n
         $this->immediateStatusCode = null;
 
         if (null !== $immediateResponse) {
-            $this->inlineResponses = [];
+            $this->responses = [];
 
             return $this->responseFactory->createResponse($immediateStatusCode ?? 200)
                 ->withHeader('Content-Type', 'application/json')
@@ -251,8 +252,8 @@ protected function handleDeleteRequest(): ResponseInterface
 
     protected function createJsonResponse(): ResponseInterface
     {
-        $messages = [...array_column($this->getOutgoingMessages($this->sessionId), 'message'), ...$this->inlineResponses];
-        $this->inlineResponses = [];
+        $messages = [...array_column($this->getOutgoingMessages($this->sessionId), 'message'), ...$this->responses];
+        $this->responses = [];
 
         if ([] === $messages) {
             return $this->responseFactory->createResponse(202)
@@ -277,18 +278,18 @@ protected function createStreamedResponse(): ResponseInterface
         $fiber = $this->sessionFiber;
 
         // The other requests of a batch whose handler did not suspend.
-        $inlineResponses = $this->inlineResponses;
-        $this->inlineResponses = [];
+        $responses = $this->responses;
+        $this->responses = [];
 
-        $callback = function () use ($fiber, $inlineResponses): void {
+        $callback = function () use ($fiber, $responses): void {
             if (null === $fiber) {
                 return;
             }
 
             try {
                 $this->logger->info('SSE: Starting request processing loop');
 
-                foreach ($inlineResponses as $message) {
+                foreach ($responses as $message) {
                     echo "event: message\n";
                     echo "data: {$message}\n\n";
                     @ob_flush();

src/Server/Transport/TransportInterface.php

@@ -48,9 +48,12 @@ public function listen(): mixed;
     /**
      * Send a message to the client immediately (bypassing session queue).
      *
-     * Used for session resolution errors when no session is available, and for
-     * every response on a {@see InlineResponseTransportInterface}.
-     * The transport decides HOW to send based on context.
+     * Used for every response of a handler that returns without suspending, on
+     * the exchange that carried its request: a transport must deliver each one,
+     * a batch gets several. A handler that suspends returns its result from the
+     * fiber the transport resumes instead. An error with a `status_code` was
+     * raised before a session was resolved and rejects the whole exchange. The
+     * transport decides HOW to send based on context.
      *
      * @param array<string, mixed> $context Context about this message:
      *                                      - 'session_id': Uuid|null

tests/Unit/Fixtures/RecordingTransport.php

@@ -0,0 +1,28 @@
+<?php
+
+/*
+ * This file is part of the official PHP MCP SDK.
+ *
+ * A collaboration between Symfony and the PHP Foundation.
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Mcp\Tests\Unit\Fixtures;
+
+use Mcp\Server\Transport\InMemoryTransport;
+
+/**
+ * Records what the protocol sends, in the shape of the session's outgoing queue.
+ */
+final class RecordingTransport extends InMemoryTransport
+{
+    /** @var list<array{message: string, context: array<string, mixed>}> */
+    public array $sent = [];
+
+    public function send(string $data, array $context): void
+    {
+        $this->sent[] = ['message' => $data, 'context' => $context];
+    }
+}