diff --git a/src/Client/Transport/HttpTransport.php b/src/Client/Transport/HttpTransport.php index dd3255f7..ddb662f7 100644 --- a/src/Client/Transport/HttpTransport.php +++ b/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,6 +276,37 @@ 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. */ @@ -280,7 +314,7 @@ 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)); } diff --git a/tests/Unit/Client/Transport/HttpTransportTest.php b/tests/Unit/Client/Transport/HttpTransportTest.php index 408980d7..6c6119e1 100644 --- a/tests/Unit/Client/Transport/HttpTransportTest.php +++ b/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 + */ + 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 {