[Client] Parse CRLF- and CR-delimited SSE events

src/Client/Transport/HttpTransport.php

@@ -230,16 +230,19 @@ private function processSSEStream(): void
             }
         }
 
-        while (false !== ($pos = strpos($this->sseBuffer, "\n\n"))) {
-            $event = substr($this->sseBuffer, 0, $pos);
-            $this->sseBuffer = substr($this->sseBuffer, $pos + 2);
-
+        while (null !== ($event = $this->extractSSEEvent())) {
             if (!empty(trim($event))) {
                 $this->processSSEEvent($event);
             }
         }
 
-        if ($this->activeStream->eof() && empty($this->sseBuffer)) {
+        if ($this->activeStream->eof()) {
+            // The stream ended without a trailing blank line: dispatch what is left.
+            if (!empty(trim($this->sseBuffer))) {
+                $this->processSSEEvent($this->sseBuffer);
+            }
+
+            $this->sseBuffer = '';
             $this->activeStream = null;
         }
     }
@@ -273,14 +276,45 @@ private function abortSseStream(string $reason): void
         }
     }
 
+    /**
+     * Take the next complete event off the buffer, or null if none is complete yet.
+     *
+     * Per the SSE specification, lines are terminated by CRLF, LF or CR, so an
+     * event is delimited by any pair of those. Servers built on sse-starlette
+     * (the MCP Python SDK) use CRLF.
+     */
+    private function extractSSEEvent(): ?string
+    {
+        $position = null;
+        $length = 0;
+
+        foreach (["\r\n\r\n", "\n\n", "\r\r"] as $delimiter) {
+            $found = strpos($this->sseBuffer, $delimiter);
+
+            if (false !== $found && (null === $position || $found < $position)) {
+                $position = $found;
+                $length = \strlen($delimiter);
+            }
+        }
+
+        if (null === $position) {
+            return null;
+        }
+
+        $event = substr($this->sseBuffer, 0, $position);
+        $this->sseBuffer = substr($this->sseBuffer, $position + $length);
+
+        return $event;
+    }
+
     /**
      * Parse a single SSE event and handle the message.
      */
     private function processSSEEvent(string $event): void
     {
         $data = '';
 
-        foreach (explode("\n", $event) as $line) {
+        foreach (preg_split("/\r\n|\r|\n/", $event) ?: [] as $line) {
             if (str_starts_with($line, 'data:')) {
                 $data .= trim(substr($line, 5));
             }

tests/Unit/Client/Transport/HttpTransportTest.php

@@ -11,14 +11,19 @@
 
 namespace Mcp\Tests\Unit\Client\Transport;
 
+use Mcp\Client;
 use Mcp\Client\State\ClientState;
 use Mcp\Client\Transport\HttpTransport;
 use Mcp\Exception\InvalidArgumentException;
 use Mcp\Schema\JsonRpc\Error;
 use Nyholm\Psr7\Factory\Psr17Factory;
+use Nyholm\Psr7\Response;
+use PHPUnit\Framework\Attributes\DataProvider;
 use PHPUnit\Framework\Attributes\TestDox;
 use PHPUnit\Framework\TestCase;
 use Psr\Http\Client\ClientInterface;
+use Psr\Http\Message\RequestInterface;
+use Psr\Http\Message\ResponseInterface;
 use Psr\Http\Message\StreamInterface;
 
 final class HttpTransportTest extends TestCase
@@ -30,6 +35,65 @@ protected function setUp(): void
         $this->factory = new Psr17Factory();
     }
 
+    /**
+     * @return iterable<string, array{string}>
+     */
+    public static function frameProvider(): iterable
+    {
+        yield 'LF line endings' => ["event: message\ndata: %s\n\n"];
+        // sse-starlette (and therefore every MCP Python SDK server) defaults to CRLF.
+        yield 'CRLF line endings' => ["event: message\r\ndata: %s\r\n\r\n"];
+        yield 'CR line endings' => ["event: message\rdata: %s\r\r"];
+        yield 'with an id field' => ["id: 1\r\nevent: message\r\ndata: %s\r\n\r\n"];
+        yield 'no trailing blank line' => ["event: message\ndata: %s\n"];
+        yield 'preceded by a comment' => [": ping\n\nevent: message\ndata: %s\n\n"];
+    }
+
+    #[DataProvider('frameProvider')]
+    #[TestDox('initialization succeeds for an SSE response framed as: $_dataName')]
+    public function testInitializeParsesSseFraming(string $frame): void
+    {
+        $httpClient = new class($frame) implements ClientInterface {
+            public function __construct(private readonly string $frame)
+            {
+            }
+
+            public function sendRequest(RequestInterface $request): ResponseInterface
+            {
+                $decoded = json_decode((string) $request->getBody(), true);
+
+                if ('initialize' !== ($decoded['method'] ?? null)) {
+                    return new Response(202);
+                }
+
+                $payload = json_encode([
+                    'jsonrpc' => '2.0',
+                    'id' => $decoded['id'],
+                    'result' => [
+                        'protocolVersion' => '2025-11-25',
+                        'capabilities' => ['tools' => ['listChanged' => false]],
+                        'serverInfo' => ['name' => 'test-server', 'version' => '1.0.0'],
+                    ],
+                ]);
+
+                return new Response(200, [
+                    'Content-Type' => 'text/event-stream',
+                    'Mcp-Session-Id' => 'abc123',
+                ], \sprintf($this->frame, $payload));
+            }
+        };
+
+        $client = Client::builder()
+            ->setClientInfo('test-client', '1.0.0')
+            ->setInitTimeout(1)
+            ->build();
+
+        $client->connect(new HttpTransport('http://localhost/mcp', [], $httpClient, $this->factory, $this->factory));
+
+        $this->assertTrue($client->isConnected());
+        $this->assertSame('test-server', $client->getServerInfo()?->name);
+    }
+
     #[TestDox('SSE stream is aborted before the buffer can exceed the configured cap')]
     public function testSseBufferIsBoundedByConfiguredCap(): void
     {