This is page 5 of 5. Use http://codebase.md/php-mcp/server?lines=false&page={x} to view the full context.
# Directory Structure
```
├── .github
│ └── workflows
│ ├── changelog.yml
│ └── tests.yml
├── .gitignore
├── .php-cs-fixer.php
├── CHANGELOG.md
├── composer.json
├── CONTRIBUTING.md
├── examples
│ ├── 01-discovery-stdio-calculator
│ │ ├── McpElements.php
│ │ └── server.php
│ ├── 02-discovery-http-userprofile
│ │ ├── McpElements.php
│ │ ├── server.php
│ │ └── UserIdCompletionProvider.php
│ ├── 03-manual-registration-stdio
│ │ ├── server.php
│ │ └── SimpleHandlers.php
│ ├── 04-combined-registration-http
│ │ ├── DiscoveredElements.php
│ │ ├── ManualHandlers.php
│ │ └── server.php
│ ├── 05-stdio-env-variables
│ │ ├── EnvToolHandler.php
│ │ └── server.php
│ ├── 06-custom-dependencies-stdio
│ │ ├── McpTaskHandlers.php
│ │ ├── server.php
│ │ └── Services.php
│ ├── 07-complex-tool-schema-http
│ │ ├── EventTypes.php
│ │ ├── McpEventScheduler.php
│ │ └── server.php
│ └── 08-schema-showcase-streamable
│ ├── SchemaShowcaseElements.php
│ └── server.php
├── LICENSE
├── phpunit.xml
├── README.md
├── src
│ ├── Attributes
│ │ ├── CompletionProvider.php
│ │ ├── McpPrompt.php
│ │ ├── McpResource.php
│ │ ├── McpResourceTemplate.php
│ │ ├── McpTool.php
│ │ └── Schema.php
│ ├── Configuration.php
│ ├── Context.php
│ ├── Contracts
│ │ ├── CompletionProviderInterface.php
│ │ ├── EventStoreInterface.php
│ │ ├── LoggerAwareInterface.php
│ │ ├── LoopAwareInterface.php
│ │ ├── ServerTransportInterface.php
│ │ ├── SessionHandlerInterface.php
│ │ └── SessionInterface.php
│ ├── Defaults
│ │ ├── ArrayCache.php
│ │ ├── BasicContainer.php
│ │ ├── DefaultUuidSessionIdGenerator.php
│ │ ├── EnumCompletionProvider.php
│ │ ├── FileCache.php
│ │ ├── InMemoryEventStore.php
│ │ ├── ListCompletionProvider.php
│ │ └── SystemClock.php
│ ├── Dispatcher.php
│ ├── Elements
│ │ ├── RegisteredElement.php
│ │ ├── RegisteredPrompt.php
│ │ ├── RegisteredResource.php
│ │ ├── RegisteredResourceTemplate.php
│ │ └── RegisteredTool.php
│ ├── Exception
│ │ ├── ConfigurationException.php
│ │ ├── DiscoveryException.php
│ │ ├── McpServerException.php
│ │ ├── ProtocolException.php
│ │ └── TransportException.php
│ ├── Protocol.php
│ ├── Registry.php
│ ├── Server.php
│ ├── ServerBuilder.php
│ ├── Session
│ │ ├── ArraySessionHandler.php
│ │ ├── CacheSessionHandler.php
│ │ ├── Session.php
│ │ ├── SessionManager.php
│ │ └── SubscriptionManager.php
│ ├── Transports
│ │ ├── HttpServerTransport.php
│ │ ├── StdioServerTransport.php
│ │ └── StreamableHttpServerTransport.php
│ └── Utils
│ ├── Discoverer.php
│ ├── DocBlockParser.php
│ ├── HandlerResolver.php
│ ├── SchemaGenerator.php
│ └── SchemaValidator.php
└── tests
├── Fixtures
│ ├── Discovery
│ │ ├── DiscoverablePromptHandler.php
│ │ ├── DiscoverableResourceHandler.php
│ │ ├── DiscoverableTemplateHandler.php
│ │ ├── DiscoverableToolHandler.php
│ │ ├── EnhancedCompletionHandler.php
│ │ ├── InvocablePromptFixture.php
│ │ ├── InvocableResourceFixture.php
│ │ ├── InvocableResourceTemplateFixture.php
│ │ ├── InvocableToolFixture.php
│ │ ├── NonDiscoverableClass.php
│ │ └── SubDir
│ │ └── HiddenTool.php
│ ├── Enums
│ │ ├── BackedIntEnum.php
│ │ ├── BackedStringEnum.php
│ │ ├── PriorityEnum.php
│ │ ├── StatusEnum.php
│ │ └── UnitEnum.php
│ ├── General
│ │ ├── CompletionProviderFixture.php
│ │ ├── DocBlockTestFixture.php
│ │ ├── InvokableHandlerFixture.php
│ │ ├── PromptHandlerFixture.php
│ │ ├── RequestAttributeChecker.php
│ │ ├── ResourceHandlerFixture.php
│ │ ├── ToolHandlerFixture.php
│ │ └── VariousTypesHandler.php
│ ├── Middlewares
│ │ ├── ErrorMiddleware.php
│ │ ├── FirstMiddleware.php
│ │ ├── HeaderMiddleware.php
│ │ ├── RequestAttributeMiddleware.php
│ │ ├── SecondMiddleware.php
│ │ ├── ShortCircuitMiddleware.php
│ │ └── ThirdMiddleware.php
│ ├── Schema
│ │ └── SchemaGenerationTarget.php
│ ├── ServerScripts
│ │ ├── HttpTestServer.php
│ │ ├── StdioTestServer.php
│ │ └── StreamableHttpTestServer.php
│ └── Utils
│ ├── AttributeFixtures.php
│ ├── DockBlockParserFixture.php
│ └── SchemaGeneratorFixture.php
├── Integration
│ ├── DiscoveryTest.php
│ ├── HttpServerTransportTest.php
│ ├── SchemaGenerationTest.php
│ ├── StdioServerTransportTest.php
│ └── StreamableHttpServerTransportTest.php
├── Mocks
│ ├── Clients
│ │ ├── MockJsonHttpClient.php
│ │ ├── MockSseClient.php
│ │ └── MockStreamHttpClient.php
│ └── Clock
│ └── FixedClock.php
├── Pest.php
├── TestCase.php
└── Unit
├── Attributes
│ ├── CompletionProviderTest.php
│ ├── McpPromptTest.php
│ ├── McpResourceTemplateTest.php
│ ├── McpResourceTest.php
│ └── McpToolTest.php
├── ConfigurationTest.php
├── Defaults
│ ├── EnumCompletionProviderTest.php
│ └── ListCompletionProviderTest.php
├── DispatcherTest.php
├── Elements
│ ├── RegisteredElementTest.php
│ ├── RegisteredPromptTest.php
│ ├── RegisteredResourceTemplateTest.php
│ ├── RegisteredResourceTest.php
│ └── RegisteredToolTest.php
├── ProtocolTest.php
├── RegistryTest.php
├── ServerBuilderTest.php
├── ServerTest.php
├── Session
│ ├── ArraySessionHandlerTest.php
│ ├── CacheSessionHandlerTest.php
│ ├── SessionManagerTest.php
│ └── SessionTest.php
└── Utils
├── DocBlockParserTest.php
├── HandlerResolverTest.php
└── SchemaValidatorTest.php
```
# Files
--------------------------------------------------------------------------------
/src/Transports/StreamableHttpServerTransport.php:
--------------------------------------------------------------------------------
```php
<?php
declare(strict_types=1);
namespace PhpMcp\Server\Transports;
use Evenement\EventEmitterTrait;
use PhpMcp\Server\Contracts\EventStoreInterface;
use PhpMcp\Server\Contracts\LoggerAwareInterface;
use PhpMcp\Server\Contracts\LoopAwareInterface;
use PhpMcp\Server\Contracts\ServerTransportInterface;
use PhpMcp\Server\Exception\McpServerException;
use PhpMcp\Server\Exception\TransportException;
use PhpMcp\Schema\JsonRpc\Message;
use PhpMcp\Schema\JsonRpc\BatchRequest;
use PhpMcp\Schema\JsonRpc\BatchResponse;
use PhpMcp\Schema\JsonRpc\Error;
use PhpMcp\Schema\JsonRpc\Parser;
use PhpMcp\Schema\JsonRpc\Request;
use PhpMcp\Schema\JsonRpc\Response;
use Psr\Http\Message\ServerRequestInterface;
use Psr\Log\LoggerInterface;
use Psr\Log\NullLogger;
use React\EventLoop\Loop;
use React\EventLoop\LoopInterface;
use React\Http\HttpServer;
use React\Http\Message\Response as HttpResponse;
use React\Promise\Deferred;
use React\Promise\PromiseInterface;
use React\Socket\SocketServer;
use React\Stream\ThroughStream;
use Throwable;
use function React\Promise\resolve;
use function React\Promise\reject;
class StreamableHttpServerTransport implements ServerTransportInterface, LoggerAwareInterface, LoopAwareInterface
{
use EventEmitterTrait;
protected LoggerInterface $logger;
protected LoopInterface $loop;
private ?SocketServer $socket = null;
private ?HttpServer $http = null;
private bool $listening = false;
private bool $closing = false;
private ?EventStoreInterface $eventStore;
/**
* Stores Deferred objects for POST requests awaiting a direct JSON response.
* Keyed by a unique pendingRequestId.
* @var array<string, Deferred>
*/
private array $pendingRequests = [];
/**
* Stores active SSE streams.
* Key: streamId
* Value: ['stream' => ThroughStream, 'sessionId' => string, 'context' => array]
* @var array<string, array{stream: ThroughStream, sessionId: string, context: array}>
*/
private array $activeSseStreams = [];
private ?ThroughStream $getStream = null;
/**
* @param bool $enableJsonResponse If true, the server will return JSON responses instead of starting an SSE stream.
* @param bool $stateless If true, the server will not emit client_connected events.
* @param EventStoreInterface $eventStore If provided, the server will replay events to the client.
* @param array<callable(\Psr\Http\Message\ServerRequestInterface, callable): (\Psr\Http\Message\ResponseInterface|\React\Promise\PromiseInterface)> $middlewares Middlewares to be applied to the HTTP server.
* This can be useful for simple request/response scenarios without streaming.
*/
public function __construct(
private readonly string $host = '127.0.0.1',
private readonly int $port = 8080,
private string $mcpPath = '/mcp',
private ?array $sslContext = null,
private readonly bool $enableJsonResponse = true,
private readonly bool $stateless = false,
?EventStoreInterface $eventStore = null,
private array $middlewares = []
) {
$this->logger = new NullLogger();
$this->loop = Loop::get();
$this->mcpPath = '/' . trim($mcpPath, '/');
$this->eventStore = $eventStore;
foreach ($this->middlewares as $mw) {
if (!is_callable($mw)) {
throw new \InvalidArgumentException('All provided middlewares must be callable.');
}
}
}
protected function generateId(): string
{
return bin2hex(random_bytes(16)); // 32 hex characters
}
public function setLogger(LoggerInterface $logger): void
{
$this->logger = $logger;
}
public function setLoop(LoopInterface $loop): void
{
$this->loop = $loop;
}
public function listen(): void
{
if ($this->listening) {
throw new TransportException('StreamableHttp transport is already listening.');
}
if ($this->closing) {
throw new TransportException('Cannot listen, transport is closing/closed.');
}
$listenAddress = "{$this->host}:{$this->port}";
$protocol = $this->sslContext ? 'https' : 'http';
try {
$this->socket = new SocketServer(
$listenAddress,
$this->sslContext ?? [],
$this->loop
);
$handlers = array_merge($this->middlewares, [$this->createRequestHandler()]);
$this->http = new HttpServer($this->loop, ...$handlers);
$this->http->listen($this->socket);
$this->socket->on('error', function (Throwable $error) {
$this->logger->error('Socket server error (StreamableHttp).', ['error' => $error->getMessage()]);
$this->emit('error', [new TransportException("Socket server error: {$error->getMessage()}", 0, $error)]);
$this->close();
});
$this->logger->info("Server is up and listening on {$protocol}://{$listenAddress} 🚀");
$this->logger->info("MCP Endpoint: {$protocol}://{$listenAddress}{$this->mcpPath}");
$this->listening = true;
$this->closing = false;
$this->emit('ready');
} catch (Throwable $e) {
$this->logger->error("Failed to start StreamableHttp listener on {$listenAddress}", ['exception' => $e]);
throw new TransportException("Failed to start StreamableHttp listener on {$listenAddress}: {$e->getMessage()}", 0, $e);
}
}
private function createRequestHandler(): callable
{
return function (ServerRequestInterface $request) {
$path = $request->getUri()->getPath();
$method = $request->getMethod();
$this->logger->debug("Request received", ['method' => $method, 'path' => $path, 'target' => $this->mcpPath]);
if ($path !== $this->mcpPath) {
$error = Error::forInvalidRequest("Not found: {$path}");
return new HttpResponse(404, ['Content-Type' => 'application/json'], json_encode($error));
}
$corsHeaders = [
'Access-Control-Allow-Origin' => '*',
'Access-Control-Allow-Methods' => 'GET, POST, DELETE, OPTIONS',
'Access-Control-Allow-Headers' => 'Content-Type, Mcp-Session-Id, Last-Event-ID, Authorization',
];
if ($method === 'OPTIONS') {
return new HttpResponse(204, $corsHeaders);
}
$addCors = function (HttpResponse $r) use ($corsHeaders) {
foreach ($corsHeaders as $key => $value) {
$r = $r->withAddedHeader($key, $value);
}
return $r;
};
try {
return match ($method) {
'GET' => $this->handleGetRequest($request)->then($addCors, fn($e) => $addCors($this->handleRequestError($e, $request))),
'POST' => $this->handlePostRequest($request)->then($addCors, fn($e) => $addCors($this->handleRequestError($e, $request))),
'DELETE' => $this->handleDeleteRequest($request)->then($addCors, fn($e) => $addCors($this->handleRequestError($e, $request))),
default => $addCors($this->handleUnsupportedRequest($request)),
};
} catch (Throwable $e) {
return $addCors($this->handleRequestError($e, $request));
}
};
}
private function handleGetRequest(ServerRequestInterface $request): PromiseInterface
{
if ($this->stateless) {
$error = Error::forInvalidRequest("GET requests (SSE streaming) are not supported in stateless mode.");
return resolve(new HttpResponse(405, ['Content-Type' => 'application/json'], json_encode($error)));
}
$acceptHeader = $request->getHeaderLine('Accept');
if (!str_contains($acceptHeader, 'text/event-stream')) {
$error = Error::forInvalidRequest("Not Acceptable: Client must accept text/event-stream for GET requests.");
return resolve(new HttpResponse(406, ['Content-Type' => 'application/json'], json_encode($error)));
}
$sessionId = $request->getHeaderLine('Mcp-Session-Id');
if (empty($sessionId)) {
$this->logger->warning("GET request without Mcp-Session-Id.");
$error = Error::forInvalidRequest("Mcp-Session-Id header required for GET requests.");
return resolve(new HttpResponse(400, ['Content-Type' => 'application/json'], json_encode($error)));
}
$this->getStream = new ThroughStream();
$this->getStream->on('close', function () use ($sessionId) {
$this->logger->debug("GET SSE stream closed.", ['sessionId' => $sessionId]);
$this->getStream = null;
});
$this->getStream->on('error', function (Throwable $e) use ($sessionId) {
$this->logger->error("GET SSE stream error.", ['sessionId' => $sessionId, 'error' => $e->getMessage()]);
$this->getStream = null;
});
$headers = [
'Content-Type' => 'text/event-stream',
'Cache-Control' => 'no-cache',
'Connection' => 'keep-alive',
'X-Accel-Buffering' => 'no',
];
$response = new HttpResponse(200, $headers, $this->getStream);
if ($this->eventStore) {
$lastEventId = $request->getHeaderLine('Last-Event-ID');
$this->replayEvents($lastEventId, $this->getStream, $sessionId);
}
return resolve($response);
}
private function handlePostRequest(ServerRequestInterface $request): PromiseInterface
{
$deferred = new Deferred();
$acceptHeader = $request->getHeaderLine('Accept');
if (!str_contains($acceptHeader, 'application/json') && !str_contains($acceptHeader, 'text/event-stream')) {
$error = Error::forInvalidRequest("Not Acceptable: Client must accept both application/json or text/event-stream");
$deferred->resolve(new HttpResponse(406, ['Content-Type' => 'application/json'], json_encode($error)));
return $deferred->promise();
}
if (!str_contains($request->getHeaderLine('Content-Type'), 'application/json')) {
$error = Error::forInvalidRequest("Unsupported Media Type: Content-Type must be application/json");
$deferred->resolve(new HttpResponse(415, ['Content-Type' => 'application/json'], json_encode($error)));
return $deferred->promise();
}
$body = $request->getBody()->getContents();
if (empty($body)) {
$this->logger->warning("Received empty POST body");
$error = Error::forInvalidRequest("Empty request body.");
$deferred->resolve(new HttpResponse(400, ['Content-Type' => 'application/json'], json_encode($error)));
return $deferred->promise();
}
try {
$message = Parser::parse($body);
} catch (Throwable $e) {
$this->logger->error("Failed to parse MCP message from POST body", ['error' => $e->getMessage()]);
$error = Error::forParseError("Invalid JSON: " . $e->getMessage());
$deferred->resolve(new HttpResponse(400, ['Content-Type' => 'application/json'], json_encode($error)));
return $deferred->promise();
}
$isInitializeRequest = ($message instanceof Request && $message->method === 'initialize');
$sessionId = null;
if ($this->stateless) {
$sessionId = $this->generateId();
$this->emit('client_connected', [$sessionId]);
} else {
if ($isInitializeRequest) {
if ($request->hasHeader('Mcp-Session-Id')) {
$this->logger->warning("Client sent Mcp-Session-Id with InitializeRequest. Ignoring.", ['clientSentId' => $request->getHeaderLine('Mcp-Session-Id')]);
$error = Error::forInvalidRequest("Invalid request: Session already initialized. Mcp-Session-Id header not allowed with InitializeRequest.", $message->getId());
$deferred->resolve(new HttpResponse(400, ['Content-Type' => 'application/json'], json_encode($error)));
return $deferred->promise();
}
$sessionId = $this->generateId();
$this->emit('client_connected', [$sessionId]);
} else {
$sessionId = $request->getHeaderLine('Mcp-Session-Id');
if (empty($sessionId)) {
$this->logger->warning("POST request without Mcp-Session-Id.");
$error = Error::forInvalidRequest("Mcp-Session-Id header required for POST requests.", $message->getId());
$deferred->resolve(new HttpResponse(400, ['Content-Type' => 'application/json'], json_encode($error)));
return $deferred->promise();
}
}
}
$context = [
'is_initialize_request' => $isInitializeRequest,
];
$nRequests = match (true) {
$message instanceof Request => 1,
$message instanceof BatchRequest => $message->nRequests(),
default => 0,
};
if ($nRequests === 0) {
$deferred->resolve(new HttpResponse(202));
$context['type'] = 'post_202_sent';
} else {
if ($this->enableJsonResponse) {
$pendingRequestId = $this->generateId();
$this->pendingRequests[$pendingRequestId] = $deferred;
$timeoutTimer = $this->loop->addTimer(30, function () use ($pendingRequestId, $sessionId) {
if (isset($this->pendingRequests[$pendingRequestId])) {
$deferred = $this->pendingRequests[$pendingRequestId];
unset($this->pendingRequests[$pendingRequestId]);
$this->logger->warning("Timeout waiting for direct JSON response processing.", ['pending_request_id' => $pendingRequestId, 'session_id' => $sessionId]);
$errorResponse = McpServerException::internalError("Request processing timed out.")->toJsonRpcError($pendingRequestId);
$deferred->resolve(new HttpResponse(500, ['Content-Type' => 'application/json'], json_encode($errorResponse->toArray())));
}
});
$this->pendingRequests[$pendingRequestId]->promise()->finally(function () use ($timeoutTimer) {
$this->loop->cancelTimer($timeoutTimer);
});
$context['type'] = 'post_json';
$context['pending_request_id'] = $pendingRequestId;
} else {
$streamId = $this->generateId();
$sseStream = new ThroughStream();
$this->activeSseStreams[$streamId] = [
'stream' => $sseStream,
'sessionId' => $sessionId,
'context' => ['nRequests' => $nRequests, 'nResponses' => 0]
];
$sseStream->on('close', function () use ($streamId) {
$this->logger->info("POST SSE stream closed by client/server.", ['streamId' => $streamId, 'sessionId' => $this->activeSseStreams[$streamId]['sessionId']]);
unset($this->activeSseStreams[$streamId]);
});
$sseStream->on('error', function (Throwable $e) use ($streamId) {
$this->logger->error("POST SSE stream error.", ['streamId' => $streamId, 'sessionId' => $this->activeSseStreams[$streamId]['sessionId'], 'error' => $e->getMessage()]);
unset($this->activeSseStreams[$streamId]);
});
$headers = [
'Content-Type' => 'text/event-stream',
'Cache-Control' => 'no-cache',
'Connection' => 'keep-alive',
'X-Accel-Buffering' => 'no',
];
if (!empty($sessionId) && !$this->stateless) {
$headers['Mcp-Session-Id'] = $sessionId;
}
$deferred->resolve(new HttpResponse(200, $headers, $sseStream));
$context['type'] = 'post_sse';
$context['streamId'] = $streamId;
$context['nRequests'] = $nRequests;
}
}
$context['stateless'] = $this->stateless;
$context['request'] = $request;
$this->loop->futureTick(function () use ($message, $sessionId, $context) {
$this->emit('message', [$message, $sessionId, $context]);
});
return $deferred->promise();
}
private function handleDeleteRequest(ServerRequestInterface $request): PromiseInterface
{
if ($this->stateless) {
return resolve(new HttpResponse(204));
}
$sessionId = $request->getHeaderLine('Mcp-Session-Id');
if (empty($sessionId)) {
$this->logger->warning("DELETE request without Mcp-Session-Id.");
$error = Error::forInvalidRequest("Mcp-Session-Id header required for DELETE.");
return resolve(new HttpResponse(400, ['Content-Type' => 'application/json'], json_encode($error)));
}
$streamsToClose = [];
foreach ($this->activeSseStreams as $streamId => $streamInfo) {
if ($streamInfo['sessionId'] === $sessionId) {
$streamsToClose[] = $streamId;
}
}
foreach ($streamsToClose as $streamId) {
$this->activeSseStreams[$streamId]['stream']->end();
unset($this->activeSseStreams[$streamId]);
}
if ($this->getStream !== null) {
$this->getStream->end();
$this->getStream = null;
}
$this->emit('client_disconnected', [$sessionId, 'Session terminated by DELETE request']);
return resolve(new HttpResponse(204));
}
private function handleUnsupportedRequest(ServerRequestInterface $request): HttpResponse
{
$error = Error::forInvalidRequest("Method not allowed: {$request->getMethod()}");
$headers = [
'Content-Type' => 'application/json',
'Allow' => 'GET, POST, DELETE, OPTIONS',
];
return new HttpResponse(405, $headers, json_encode($error));
}
private function handleRequestError(Throwable $e, ServerRequestInterface $request): HttpResponse
{
$this->logger->error("Error processing HTTP request", [
'method' => $request->getMethod(),
'path' => $request->getUri()->getPath(),
'exception' => $e->getMessage()
]);
if ($e instanceof TransportException) {
$error = Error::forInternalError("Transport Error: " . $e->getMessage());
return new HttpResponse(500, ['Content-Type' => 'application/json'], json_encode($error));
}
$error = Error::forInternalError("Internal Server Error during HTTP request processing.");
return new HttpResponse(500, ['Content-Type' => 'application/json'], json_encode($error));
}
public function sendMessage(Message $message, string $sessionId, array $context = []): PromiseInterface
{
if ($this->closing) {
return reject(new TransportException('Transport is closing.'));
}
$isInitializeResponse = ($context['is_initialize_request'] ?? false) && ($message instanceof Response);
switch ($context['type'] ?? null) {
case 'post_202_sent':
return resolve(null);
case 'post_sse':
$streamId = $context['streamId'];
if (!isset($this->activeSseStreams[$streamId])) {
$this->logger->error("SSE stream for POST not found.", ['streamId' => $streamId, 'sessionId' => $sessionId]);
return reject(new TransportException("SSE stream {$streamId} not found for POST response."));
}
$stream = $this->activeSseStreams[$streamId]['stream'];
if (!$stream->isWritable()) {
$this->logger->warning("SSE stream for POST is not writable.", ['streamId' => $streamId, 'sessionId' => $sessionId]);
return reject(new TransportException("SSE stream {$streamId} for POST is not writable."));
}
$sentCountThisCall = 0;
if ($message instanceof Response || $message instanceof Error) {
$json = json_encode($message, JSON_UNESCAPED_SLASHES | JSON_UNESCAPED_UNICODE);
$eventId = $this->eventStore ? $this->eventStore->storeEvent($streamId, $json) : null;
$this->sendSseEventToStream($stream, $json, $eventId);
$sentCountThisCall = 1;
} elseif ($message instanceof BatchResponse) {
foreach ($message->getAll() as $singleResponse) {
$json = json_encode($message, JSON_UNESCAPED_SLASHES | JSON_UNESCAPED_UNICODE);
$eventId = $this->eventStore ? $this->eventStore->storeEvent($streamId, $json) : null;
$this->sendSseEventToStream($stream, $json, $eventId);
$sentCountThisCall++;
}
}
if (isset($this->activeSseStreams[$streamId]['context'])) {
$this->activeSseStreams[$streamId]['context']['nResponses'] += $sentCountThisCall;
if ($this->activeSseStreams[$streamId]['context']['nResponses'] >= $this->activeSseStreams[$streamId]['context']['nRequests']) {
$this->logger->info("All expected responses sent for POST SSE stream. Closing.", ['streamId' => $streamId, 'sessionId' => $sessionId]);
$stream->end(); // Will trigger 'close' event.
if ($context['stateless'] ?? false) {
$this->loop->futureTick(function () use ($sessionId) {
$this->emit('client_disconnected', [$sessionId, 'Stateless request completed']);
});
}
}
}
return resolve(null);
case 'post_json':
$pendingRequestId = $context['pending_request_id'];
if (!isset($this->pendingRequests[$pendingRequestId])) {
$this->logger->error("Pending direct JSON request not found.", ['pending_request_id' => $pendingRequestId, 'session_id' => $sessionId]);
return reject(new TransportException("Pending request {$pendingRequestId} not found."));
}
$deferred = $this->pendingRequests[$pendingRequestId];
unset($this->pendingRequests[$pendingRequestId]);
$responseBody = json_encode($message, JSON_UNESCAPED_SLASHES | JSON_UNESCAPED_UNICODE);
$headers = ['Content-Type' => 'application/json'];
if ($isInitializeResponse && !$this->stateless) {
$headers['Mcp-Session-Id'] = $sessionId;
}
$statusCode = $context['status_code'] ?? 200;
$deferred->resolve(new HttpResponse($statusCode, $headers, $responseBody . "\n"));
if ($context['stateless'] ?? false) {
$this->loop->futureTick(function () use ($sessionId) {
$this->emit('client_disconnected', [$sessionId, 'Stateless request completed']);
});
}
return resolve(null);
default:
if ($this->getStream === null) {
$this->logger->error("GET SSE stream not found.", ['sessionId' => $sessionId]);
return reject(new TransportException("GET SSE stream not found."));
}
if (!$this->getStream->isWritable()) {
$this->logger->warning("GET SSE stream is not writable.", ['sessionId' => $sessionId]);
return reject(new TransportException("GET SSE stream not writable."));
}
if ($message instanceof Response || $message instanceof Error) {
$json = json_encode($message, JSON_UNESCAPED_SLASHES | JSON_UNESCAPED_UNICODE);
$eventId = $this->eventStore ? $this->eventStore->storeEvent('GET_STREAM', $json) : null;
$this->sendSseEventToStream($this->getStream, $json, $eventId);
} elseif ($message instanceof BatchResponse) {
foreach ($message->getAll() as $singleResponse) {
$json = json_encode($singleResponse, JSON_UNESCAPED_SLASHES | JSON_UNESCAPED_UNICODE);
$eventId = $this->eventStore ? $this->eventStore->storeEvent('GET_STREAM', $json) : null;
$this->sendSseEventToStream($this->getStream, $json, $eventId);
}
}
return resolve(null);
}
}
private function replayEvents(string $lastEventId, ThroughStream $sseStream, string $sessionId): void
{
if (empty($lastEventId)) {
return;
}
try {
$this->eventStore->replayEventsAfter(
$lastEventId,
function (string $replayedEventId, string $json) use ($sseStream) {
$this->logger->debug("Replaying event", ['replayedEventId' => $replayedEventId]);
$this->sendSseEventToStream($sseStream, $json, $replayedEventId);
}
);
} catch (Throwable $e) {
$this->logger->error("Error during event replay.", ['sessionId' => $sessionId, 'exception' => $e]);
}
}
private function sendSseEventToStream(ThroughStream $stream, string $data, ?string $eventId = null): bool
{
if (! $stream->isWritable()) {
return false;
}
$frame = "event: message\n";
if ($eventId !== null) {
$frame .= "id: {$eventId}\n";
}
$lines = explode("\n", $data);
foreach ($lines as $line) {
$frame .= "data: {$line}\n";
}
$frame .= "\n";
return $stream->write($frame);
}
public function close(): void
{
if ($this->closing) {
return;
}
$this->closing = true;
$this->listening = false;
$this->logger->info('Closing transport...');
if ($this->socket) {
$this->socket->close();
$this->socket = null;
}
foreach ($this->activeSseStreams as $streamId => $streamInfo) {
if ($streamInfo['stream']->isWritable()) {
$streamInfo['stream']->end();
}
}
if ($this->getStream !== null) {
$this->getStream->end();
$this->getStream = null;
}
foreach ($this->pendingRequests as $pendingRequestId => $deferred) {
$deferred->reject(new TransportException('Transport is closing.'));
}
$this->activeSseStreams = [];
$this->pendingRequests = [];
$this->emit('close', ['Transport closed.']);
$this->removeAllListeners();
}
}
```
--------------------------------------------------------------------------------
/src/Utils/SchemaGenerator.php:
--------------------------------------------------------------------------------
```php
<?php
namespace PhpMcp\Server\Utils;
use phpDocumentor\Reflection\DocBlock\Tags\Param;
use PhpMcp\Server\Attributes\Schema;
use PhpMcp\Server\Context;
use ReflectionEnum;
use ReflectionIntersectionType;
use ReflectionMethod;
use ReflectionNamedType;
use ReflectionParameter;
use ReflectionType;
use ReflectionUnionType;
use stdClass;
/**
* Generates JSON Schema for method parameters with intelligent Schema attribute handling.
*
* Priority system:
* 1. Schema attributes (method-level and parameter-level)
* 2. Reflection type information
* 3. DocBlock type information
*/
class SchemaGenerator
{
private DocBlockParser $docBlockParser;
public function __construct(DocBlockParser $docBlockParser)
{
$this->docBlockParser = $docBlockParser;
}
/**
* Generates a JSON Schema object (as a PHP array) for a method's or function's parameters.
*/
public function generate(\ReflectionMethod|\ReflectionFunction $reflection): array
{
$methodSchema = $this->extractMethodLevelSchema($reflection);
if ($methodSchema && isset($methodSchema['definition'])) {
return $methodSchema['definition'];
}
$parametersInfo = $this->parseParametersInfo($reflection);
return $this->buildSchemaFromParameters($parametersInfo, $methodSchema);
}
/**
* Extracts method-level or function-level Schema attribute.
*/
private function extractMethodLevelSchema(\ReflectionMethod|\ReflectionFunction $reflection): ?array
{
$schemaAttrs = $reflection->getAttributes(Schema::class, \ReflectionAttribute::IS_INSTANCEOF);
if (empty($schemaAttrs)) {
return null;
}
$schemaAttr = $schemaAttrs[0]->newInstance();
return $schemaAttr->toArray();
}
/**
* Extracts parameter-level Schema attribute.
*/
private function extractParameterLevelSchema(ReflectionParameter $parameter): array
{
$schemaAttrs = $parameter->getAttributes(Schema::class, \ReflectionAttribute::IS_INSTANCEOF);
if (empty($schemaAttrs)) {
return [];
}
$schemaAttr = $schemaAttrs[0]->newInstance();
return $schemaAttr->toArray();
}
/**
* Builds the final schema from parameter information and method-level schema.
*
* @param array<int, array{
* name: string,
* doc_block_tag: Param|null,
* reflection_param: ReflectionParameter,
* reflection_type_object: ReflectionType|null,
* type_string: string,
* description: string|null,
* required: bool,
* allows_null: bool,
* default_value: mixed|null,
* has_default: bool,
* is_variadic: bool,
* parameter_schema: array<string, mixed>
* }> $parametersInfo
*
* @param array<string, mixed>|null $methodSchema
*
* @return array<string, mixed>
*/
private function buildSchemaFromParameters(array $parametersInfo, ?array $methodSchema): array
{
$schema = [
'type' => 'object',
'properties' => [],
'required' => [],
];
// Apply method-level schema as base
if ($methodSchema) {
$schema = array_merge($schema, $methodSchema);
if (!isset($schema['type'])) {
$schema['type'] = 'object';
}
if (!isset($schema['properties'])) {
$schema['properties'] = [];
}
if (!isset($schema['required'])) {
$schema['required'] = [];
}
}
foreach ($parametersInfo as $paramInfo) {
$paramName = $paramInfo['name'];
$methodLevelParamSchema = $schema['properties'][$paramName] ?? null;
$paramSchema = $this->buildParameterSchema($paramInfo, $methodLevelParamSchema);
$schema['properties'][$paramName] = $paramSchema;
if ($paramInfo['required'] && !in_array($paramName, $schema['required'])) {
$schema['required'][] = $paramName;
} elseif (!$paramInfo['required'] && ($key = array_search($paramName, $schema['required'])) !== false) {
unset($schema['required'][$key]);
$schema['required'] = array_values($schema['required']); // Re-index
}
}
// Clean up empty properties
if (empty($schema['properties'])) {
$schema['properties'] = new stdClass();
}
if (empty($schema['required'])) {
unset($schema['required']);
}
return $schema;
}
/**
* Builds the final schema for a single parameter by merging all three levels.
*
* @param array{
* name: string,
* doc_block_tag: Param|null,
* reflection_param: ReflectionParameter,
* reflection_type_object: ReflectionType|null,
* type_string: string,
* description: string|null,
* required: bool,
* allows_null: bool,
* default_value: mixed|null,
* has_default: bool,
* is_variadic: bool,
* parameter_schema: array<string, mixed>
* } $paramInfo
* @param array<string, mixed>|null $methodLevelParamSchema
*/
private function buildParameterSchema(array $paramInfo, ?array $methodLevelParamSchema = null): array
{
if ($paramInfo['is_variadic']) {
return $this->buildVariadicParameterSchema($paramInfo);
}
$inferredSchema = $this->buildInferredParameterSchema($paramInfo);
// Method-level takes precedence over inferred schema
$mergedSchema = $inferredSchema;
if ($methodLevelParamSchema) {
$mergedSchema = $this->mergeSchemas($inferredSchema, $methodLevelParamSchema);
}
// Parameter-level takes highest precedence
$parameterLevelSchema = $paramInfo['parameter_schema'];
if (!empty($parameterLevelSchema)) {
if (isset($parameterLevelSchema['definition']) && is_array($parameterLevelSchema['definition'])) {
return $parameterLevelSchema['definition'];
}
$mergedSchema = $this->mergeSchemas($mergedSchema, $parameterLevelSchema);
}
return $mergedSchema;
}
/**
* Merge two schemas where the dominant schema takes precedence over the recessive one.
*
* @param array $recessiveSchema The schema with lower precedence
* @param array $dominantSchema The schema with higher precedence
*/
private function mergeSchemas(array $recessiveSchema, array $dominantSchema): array
{
$mergedSchema = array_merge($recessiveSchema, $dominantSchema);
return $mergedSchema;
}
/**
* Builds parameter schema from inferred type and docblock information only.
* Returns empty array for variadic parameters (handled separately).
*/
private function buildInferredParameterSchema(array $paramInfo): array
{
$paramSchema = [];
// Variadic parameters are handled separately
if ($paramInfo['is_variadic']) {
return [];
}
// Infer JSON Schema types
$jsonTypes = $this->inferParameterTypes($paramInfo);
if (count($jsonTypes) === 1) {
$paramSchema['type'] = $jsonTypes[0];
} elseif (count($jsonTypes) > 1) {
$paramSchema['type'] = $jsonTypes;
}
// Add description from docblock
if ($paramInfo['description']) {
$paramSchema['description'] = $paramInfo['description'];
}
// Add default value only if parameter actually has a default
if ($paramInfo['has_default']) {
$paramSchema['default'] = $paramInfo['default_value'];
}
// Handle enums
$paramSchema = $this->applyEnumConstraints($paramSchema, $paramInfo);
// Handle array items
$paramSchema = $this->applyArrayConstraints($paramSchema, $paramInfo);
return $paramSchema;
}
/**
* Builds schema for variadic parameters.
*/
private function buildVariadicParameterSchema(array $paramInfo): array
{
$paramSchema = ['type' => 'array'];
// Apply parameter-level Schema attributes first
if (!empty($paramInfo['parameter_schema'])) {
$paramSchema = array_merge($paramSchema, $paramInfo['parameter_schema']);
// Ensure type is always array for variadic
$paramSchema['type'] = 'array';
}
if ($paramInfo['description']) {
$paramSchema['description'] = $paramInfo['description'];
}
// If no items specified by Schema attribute, infer from type
if (!isset($paramSchema['items'])) {
$itemJsonTypes = $this->mapPhpTypeToJsonSchemaType($paramInfo['type_string']);
$nonNullItemTypes = array_filter($itemJsonTypes, fn($t) => $t !== 'null');
if (count($nonNullItemTypes) === 1) {
$paramSchema['items'] = ['type' => $nonNullItemTypes[0]];
}
}
return $paramSchema;
}
/**
* Infers JSON Schema types for a parameter.
*/
private function inferParameterTypes(array $paramInfo): array
{
$jsonTypes = $this->mapPhpTypeToJsonSchemaType($paramInfo['type_string']);
if ($paramInfo['allows_null'] && strtolower($paramInfo['type_string']) !== 'mixed' && !in_array('null', $jsonTypes)) {
$jsonTypes[] = 'null';
}
if (count($jsonTypes) > 1) {
// Sort but ensure null comes first for consistency
$nullIndex = array_search('null', $jsonTypes);
if ($nullIndex !== false) {
unset($jsonTypes[$nullIndex]);
sort($jsonTypes);
array_unshift($jsonTypes, 'null');
} else {
sort($jsonTypes);
}
}
return $jsonTypes;
}
/**
* Applies enum constraints to parameter schema.
*/
private function applyEnumConstraints(array $paramSchema, array $paramInfo): array
{
$reflectionType = $paramInfo['reflection_type_object'];
if (!($reflectionType instanceof ReflectionNamedType) || $reflectionType->isBuiltin() || !enum_exists($reflectionType->getName())) {
return $paramSchema;
}
$enumClass = $reflectionType->getName();
$enumReflection = new ReflectionEnum($enumClass);
$backingTypeReflection = $enumReflection->getBackingType();
if ($enumReflection->isBacked() && $backingTypeReflection instanceof ReflectionNamedType) {
$paramSchema['enum'] = array_column($enumClass::cases(), 'value');
$jsonBackingType = match ($backingTypeReflection->getName()) {
'int' => 'integer',
'string' => 'string',
default => null,
};
if ($jsonBackingType) {
if (isset($paramSchema['type']) && is_array($paramSchema['type']) && in_array('null', $paramSchema['type'])) {
$paramSchema['type'] = ['null', $jsonBackingType];
} else {
$paramSchema['type'] = $jsonBackingType;
}
}
} else {
// Non-backed enum - use names as enum values
$paramSchema['enum'] = array_column($enumClass::cases(), 'name');
if (isset($paramSchema['type']) && is_array($paramSchema['type']) && in_array('null', $paramSchema['type'])) {
$paramSchema['type'] = ['null', 'string'];
} else {
$paramSchema['type'] = 'string';
}
}
return $paramSchema;
}
/**
* Applies array-specific constraints to parameter schema.
*/
private function applyArrayConstraints(array $paramSchema, array $paramInfo): array
{
if (!isset($paramSchema['type'])) {
return $paramSchema;
}
$typeString = $paramInfo['type_string'];
$allowsNull = $paramInfo['allows_null'];
// Handle object-like arrays using array{} syntax
if (preg_match('/^array\s*{/i', $typeString)) {
$objectSchema = $this->inferArrayItemsType($typeString);
if (is_array($objectSchema) && isset($objectSchema['properties'])) {
$paramSchema = array_merge($paramSchema, $objectSchema);
$paramSchema['type'] = $allowsNull ? ['object', 'null'] : 'object';
}
}
// Handle regular arrays
elseif (in_array('array', $this->mapPhpTypeToJsonSchemaType($typeString))) {
$itemsType = $this->inferArrayItemsType($typeString);
if ($itemsType !== 'any') {
if (is_string($itemsType)) {
$paramSchema['items'] = ['type' => $itemsType];
} else {
if (!isset($itemsType['type']) && isset($itemsType['properties'])) {
$itemsType = array_merge(['type' => 'object'], $itemsType);
}
$paramSchema['items'] = $itemsType;
}
}
if ($allowsNull) {
$paramSchema['type'] = ['array', 'null'];
sort($paramSchema['type']);
} else {
$paramSchema['type'] = 'array';
}
}
return $paramSchema;
}
/**
* Parses detailed information about a method's parameters.
*
* @return array<int, array{
* name: string,
* doc_block_tag: Param|null,
* reflection_param: ReflectionParameter,
* reflection_type_object: ReflectionType|null,
* type_string: string,
* description: string|null,
* required: bool,
* allows_null: bool,
* default_value: mixed|null,
* has_default: bool,
* is_variadic: bool,
* parameter_schema: array<string, mixed>
* }>
*/
private function parseParametersInfo(\ReflectionMethod|\ReflectionFunction $reflection): array
{
$docComment = $reflection->getDocComment() ?: null;
$docBlock = $this->docBlockParser->parseDocBlock($docComment);
$paramTags = $this->docBlockParser->getParamTags($docBlock);
$parametersInfo = [];
foreach ($reflection->getParameters() as $rp) {
$paramName = $rp->getName();
$paramTag = $paramTags['$' . $paramName] ?? null;
$reflectionType = $rp->getType();
if ($reflectionType instanceof ReflectionNamedType && $reflectionType?->getName() === Context::class) {
continue;
}
$typeString = $this->getParameterTypeString($rp, $paramTag);
$description = $this->docBlockParser->getParamDescription($paramTag);
$hasDefault = $rp->isDefaultValueAvailable();
$defaultValue = $hasDefault ? $rp->getDefaultValue() : null;
$isVariadic = $rp->isVariadic();
$parameterSchema = $this->extractParameterLevelSchema($rp);
if ($defaultValue instanceof \BackedEnum) {
$defaultValue = $defaultValue->value;
}
if ($defaultValue instanceof \UnitEnum) {
$defaultValue = $defaultValue->name;
}
$allowsNull = false;
if ($reflectionType && $reflectionType->allowsNull()) {
$allowsNull = true;
} elseif ($hasDefault && $defaultValue === null) {
$allowsNull = true;
} elseif (str_contains($typeString, 'null') || strtolower($typeString) === 'mixed') {
$allowsNull = true;
}
$parametersInfo[] = [
'name' => $paramName,
'doc_block_tag' => $paramTag,
'reflection_param' => $rp,
'reflection_type_object' => $reflectionType,
'type_string' => $typeString,
'description' => $description,
'required' => !$rp->isOptional(),
'allows_null' => $allowsNull,
'default_value' => $defaultValue,
'has_default' => $hasDefault,
'is_variadic' => $isVariadic,
'parameter_schema' => $parameterSchema,
];
}
return $parametersInfo;
}
/**
* Determines the type string for a parameter, prioritizing DocBlock.
*/
private function getParameterTypeString(ReflectionParameter $rp, ?Param $paramTag): string
{
$docBlockType = $this->docBlockParser->getParamTypeString($paramTag);
$isDocBlockTypeGeneric = false;
if ($docBlockType !== null) {
if (in_array(strtolower($docBlockType), ['mixed', 'unknown', ''])) {
$isDocBlockTypeGeneric = true;
}
} else {
$isDocBlockTypeGeneric = true; // No tag or no type in tag implies generic
}
$reflectionType = $rp->getType();
$reflectionTypeString = null;
if ($reflectionType) {
$reflectionTypeString = $this->getTypeStringFromReflection($reflectionType, $rp->allowsNull());
}
// Prioritize Reflection if DocBlock type is generic AND Reflection provides a more specific type
if ($isDocBlockTypeGeneric && $reflectionTypeString !== null && $reflectionTypeString !== 'mixed') {
return $reflectionTypeString;
}
// Otherwise, use the DocBlock type if it was valid and non-generic
if ($docBlockType !== null && !$isDocBlockTypeGeneric) {
// Consider if DocBlock adds nullability missing from reflection
if (stripos($docBlockType, 'null') !== false && $reflectionTypeString && stripos($reflectionTypeString, 'null') === false && !str_ends_with($reflectionTypeString, '|null')) {
// If reflection didn't capture null, but docblock did, append |null (if not already mixed)
if ($reflectionTypeString !== 'mixed') {
return $reflectionTypeString . '|null';
}
}
return $docBlockType;
}
// Fallback to Reflection type even if it was generic ('mixed')
if ($reflectionTypeString !== null) {
return $reflectionTypeString;
}
// Default to 'mixed' if nothing else found
return 'mixed';
}
/**
* Converts a ReflectionType object into a type string representation.
*/
private function getTypeStringFromReflection(?ReflectionType $type, bool $nativeAllowsNull): string
{
if ($type === null) {
return 'mixed';
}
$types = [];
if ($type instanceof ReflectionUnionType) {
foreach ($type->getTypes() as $innerType) {
$types[] = $this->getTypeStringFromReflection($innerType, $innerType->allowsNull());
}
if ($nativeAllowsNull) {
$types = array_filter($types, fn($t) => strtolower($t) !== 'null');
}
$typeString = implode('|', array_unique(array_filter($types)));
} elseif ($type instanceof ReflectionIntersectionType) {
foreach ($type->getTypes() as $innerType) {
$types[] = $this->getTypeStringFromReflection($innerType, false);
}
$typeString = implode('&', array_unique(array_filter($types)));
} elseif ($type instanceof ReflectionNamedType) {
$typeString = $type->getName();
} else {
return 'mixed';
}
$typeString = match (strtolower($typeString)) {
'bool' => 'boolean',
'int' => 'integer',
'float', 'double' => 'number',
'str' => 'string',
default => $typeString,
};
$isNullable = $nativeAllowsNull;
if ($type instanceof ReflectionNamedType && $type->getName() === 'mixed') {
$isNullable = true;
}
if ($type instanceof ReflectionUnionType && !$nativeAllowsNull) {
foreach ($type->getTypes() as $innerType) {
if ($innerType instanceof ReflectionNamedType && strtolower($innerType->getName()) === 'null') {
$isNullable = true;
break;
}
}
}
if ($isNullable && $typeString !== 'mixed' && stripos($typeString, 'null') === false) {
if (!str_ends_with($typeString, '|null') && !str_ends_with($typeString, '&null')) {
$typeString .= '|null';
}
}
// Remove leading backslash from class names, but handle built-ins like 'int' or unions like 'int|string'
if (str_contains($typeString, '\\')) {
$parts = preg_split('/([|&])/', $typeString, -1, PREG_SPLIT_DELIM_CAPTURE);
$processedParts = array_map(fn($part) => str_starts_with($part, '\\') ? ltrim($part, '\\') : $part, $parts);
$typeString = implode('', $processedParts);
}
return $typeString ?: 'mixed';
}
/**
* Maps a PHP type string (potentially a union) to an array of JSON Schema type names.
*/
private function mapPhpTypeToJsonSchemaType(string $phpTypeString): array
{
$normalizedType = strtolower(trim($phpTypeString));
// PRIORITY 1: Check for array{} syntax which should be treated as object
if (preg_match('/^array\s*{/i', $normalizedType)) {
return ['object'];
}
// PRIORITY 2: Check for array syntax first (T[] or generics)
if (
str_contains($normalizedType, '[]') ||
preg_match('/^(array|list|iterable|collection)</i', $normalizedType)
) {
return ['array'];
}
// PRIORITY 3: Handle unions (recursive)
if (str_contains($normalizedType, '|')) {
$types = explode('|', $normalizedType);
$jsonTypes = [];
foreach ($types as $type) {
$mapped = $this->mapPhpTypeToJsonSchemaType(trim($type));
$jsonTypes = array_merge($jsonTypes, $mapped);
}
return array_values(array_unique($jsonTypes));
}
// PRIORITY 4: Handle simple built-in types
return match ($normalizedType) {
'string', 'scalar' => ['string'],
'?string' => ['null', 'string'],
'int', 'integer' => ['integer'],
'?int', '?integer' => ['null', 'integer'],
'float', 'double', 'number' => ['number'],
'?float', '?double', '?number' => ['null', 'number'],
'bool', 'boolean' => ['boolean'],
'?bool', '?boolean' => ['null', 'boolean'],
'array' => ['array'],
'?array' => ['null', 'array'],
'object', 'stdclass' => ['object'],
'?object', '?stdclass' => ['null', 'object'],
'null' => ['null'],
'resource', 'callable' => ['object'],
'mixed' => [],
'void', 'never' => [],
default => ['object'],
};
}
/**
* Infers the 'items' schema type for an array based on DocBlock type hints.
*/
private function inferArrayItemsType(string $phpTypeString): string|array
{
$normalizedType = trim($phpTypeString);
// Case 1: Simple T[] syntax (e.g., string[], int[], bool[], etc.)
if (preg_match('/^(\\??)([\w\\\\]+)\\s*\\[\\]$/i', $normalizedType, $matches)) {
$itemType = strtolower($matches[2]);
return $this->mapSimpleTypeToJsonSchema($itemType);
}
// Case 2: Generic array<T> syntax (e.g., array<string>, array<int>, etc.)
if (preg_match('/^(\\??)array\s*<\s*([\w\\\\|]+)\s*>$/i', $normalizedType, $matches)) {
$itemType = strtolower($matches[2]);
return $this->mapSimpleTypeToJsonSchema($itemType);
}
// Case 3: Nested array<array<T>> syntax or T[][] syntax
if (
preg_match('/^(\\??)array\s*<\s*array\s*<\s*([\w\\\\|]+)\s*>\s*>$/i', $normalizedType, $matches) ||
preg_match('/^(\\??)([\w\\\\]+)\s*\[\]\[\]$/i', $normalizedType, $matches)
) {
$innerType = $this->mapSimpleTypeToJsonSchema(isset($matches[2]) ? strtolower($matches[2]) : 'any');
// Return a schema for array with items being arrays
return [
'type' => 'array',
'items' => [
'type' => $innerType
]
];
}
// Case 4: Object-like array syntax (e.g., array{name: string, age: int})
if (preg_match('/^(\\??)array\s*\{(.+)\}$/is', $normalizedType, $matches)) {
return $this->parseObjectLikeArray($matches[2]);
}
return 'any';
}
/**
* Parses object-like array syntax into a JSON Schema object
*/
private function parseObjectLikeArray(string $propertiesStr): array
{
$properties = [];
$required = [];
// Parse properties from the string, handling nested structures
$depth = 0;
$buffer = '';
for ($i = 0; $i < strlen($propertiesStr); $i++) {
$char = $propertiesStr[$i];
// Track nested braces
if ($char === '{') {
$depth++;
$buffer .= $char;
} elseif ($char === '}') {
$depth--;
$buffer .= $char;
}
// Property separator (comma)
elseif ($char === ',' && $depth === 0) {
// Process the completed property
$this->parsePropertyDefinition(trim($buffer), $properties, $required);
$buffer = '';
} else {
$buffer .= $char;
}
}
// Process the last property
if (!empty($buffer)) {
$this->parsePropertyDefinition(trim($buffer), $properties, $required);
}
if (!empty($properties)) {
return [
'type' => 'object',
'properties' => $properties,
'required' => $required
];
}
return ['type' => 'object'];
}
/**
* Parses a single property definition from an object-like array syntax
*/
private function parsePropertyDefinition(string $propDefinition, array &$properties, array &$required): void
{
// Match property name and type
if (preg_match('/^([a-zA-Z_\x80-\xff][a-zA-Z0-9_\x80-\xff]*)\s*:\s*(.+)$/i', $propDefinition, $matches)) {
$propName = $matches[1];
$propType = trim($matches[2]);
// Add to required properties
$required[] = $propName;
// Check for nested array{} syntax
if (preg_match('/^array\s*\{(.+)\}$/is', $propType, $nestedMatches)) {
$nestedSchema = $this->parseObjectLikeArray($nestedMatches[1]);
$properties[$propName] = $nestedSchema;
}
// Check for array<T> or T[] syntax
elseif (
preg_match('/^array\s*<\s*([\w\\\\|]+)\s*>$/i', $propType, $arrayMatches) ||
preg_match('/^([\w\\\\]+)\s*\[\]$/i', $propType, $arrayMatches)
) {
$itemType = $arrayMatches[1] ?? 'any';
$properties[$propName] = [
'type' => 'array',
'items' => [
'type' => $this->mapSimpleTypeToJsonSchema($itemType)
]
];
}
// Simple type
else {
$properties[$propName] = ['type' => $this->mapSimpleTypeToJsonSchema($propType)];
}
}
}
/**
* Helper method to map basic PHP types to JSON Schema types
*/
private function mapSimpleTypeToJsonSchema(string $type): string
{
return match (strtolower($type)) {
'string' => 'string',
'int', 'integer' => 'integer',
'bool', 'boolean' => 'boolean',
'float', 'double', 'number' => 'number',
'array' => 'array',
'object', 'stdclass' => 'object',
default => in_array(strtolower($type), ['datetime', 'datetimeinterface']) ? 'string' : 'object',
};
}
}
```
--------------------------------------------------------------------------------
/tests/Unit/DispatcherTest.php:
--------------------------------------------------------------------------------
```php
<?php
namespace PhpMcp\Server\Tests\Unit;
use Mockery;
use Mockery\MockInterface;
use PhpMcp\Schema\ClientCapabilities;
use PhpMcp\Server\Context;
use PhpMcp\Server\Configuration;
use PhpMcp\Server\Contracts\CompletionProviderInterface;
use PhpMcp\Server\Contracts\SessionInterface;
use PhpMcp\Server\Dispatcher;
use PhpMcp\Server\Elements\RegisteredPrompt;
use PhpMcp\Server\Elements\RegisteredResource;
use PhpMcp\Server\Elements\RegisteredResourceTemplate;
use PhpMcp\Server\Elements\RegisteredTool;
use PhpMcp\Server\Exception\McpServerException;
use PhpMcp\Schema\Implementation;
use PhpMcp\Schema\JsonRpc\Notification as JsonRpcNotification;
use PhpMcp\Schema\JsonRpc\Request as JsonRpcRequest;
use PhpMcp\Schema\Prompt as PromptSchema;
use PhpMcp\Schema\PromptArgument;
use PhpMcp\Schema\Request\CallToolRequest;
use PhpMcp\Schema\Request\CompletionCompleteRequest;
use PhpMcp\Schema\Request\GetPromptRequest;
use PhpMcp\Schema\Request\InitializeRequest;
use PhpMcp\Schema\Request\ListToolsRequest;
use PhpMcp\Schema\Request\ReadResourceRequest;
use PhpMcp\Schema\Request\ResourceSubscribeRequest;
use PhpMcp\Schema\Request\SetLogLevelRequest;
use PhpMcp\Schema\Resource as ResourceSchema;
use PhpMcp\Schema\ResourceTemplate as ResourceTemplateSchema;
use PhpMcp\Schema\Result\CallToolResult;
use PhpMcp\Schema\Result\CompletionCompleteResult;
use PhpMcp\Schema\Result\EmptyResult;
use PhpMcp\Schema\Result\GetPromptResult;
use PhpMcp\Schema\Result\InitializeResult;
use PhpMcp\Schema\Result\ReadResourceResult;
use PhpMcp\Schema\ServerCapabilities;
use PhpMcp\Schema\Tool as ToolSchema;
use PhpMcp\Server\Registry;
use PhpMcp\Server\Session\SubscriptionManager;
use PhpMcp\Server\Utils\SchemaValidator;
use Psr\Container\ContainerInterface;
use Psr\Log\NullLogger;
use PhpMcp\Schema\Content\TextContent;
use PhpMcp\Schema\Content\PromptMessage;
use PhpMcp\Schema\Enum\LoggingLevel;
use PhpMcp\Schema\Enum\Role;
use PhpMcp\Schema\PromptReference;
use PhpMcp\Schema\Request\ListPromptsRequest;
use PhpMcp\Schema\Request\ListResourcesRequest;
use PhpMcp\Schema\Request\ListResourceTemplatesRequest;
use PhpMcp\Schema\ResourceReference;
use PhpMcp\Server\Protocol;
use PhpMcp\Server\Tests\Fixtures\Enums\StatusEnum;
use React\EventLoop\Loop;
const DISPATCHER_SESSION_ID = 'dispatcher-session-xyz';
const DISPATCHER_PAGINATION_LIMIT = 3;
beforeEach(function () {
/** @var MockInterface&Configuration $configuration */
$this->configuration = Mockery::mock(Configuration::class);
/** @var MockInterface&Registry $registry */
$this->registry = Mockery::mock(Registry::class);
/** @var MockInterface&SubscriptionManager $subscriptionManager */
$this->subscriptionManager = Mockery::mock(SubscriptionManager::class);
/** @var MockInterface&SchemaValidator $schemaValidator */
$this->schemaValidator = Mockery::mock(SchemaValidator::class);
/** @var MockInterface&SessionInterface $session */
$this->session = Mockery::mock(SessionInterface::class);
/** @var MockInterface&ContainerInterface $container */
$this->container = Mockery::mock(ContainerInterface::class);
$this->context = new Context($this->session);
$configuration = new Configuration(
serverInfo: Implementation::make('DispatcherTestServer', '1.0'),
capabilities: ServerCapabilities::make(),
paginationLimit: DISPATCHER_PAGINATION_LIMIT,
logger: new NullLogger(),
loop: Loop::get(),
cache: null,
container: $this->container
);
$this->dispatcher = new Dispatcher(
$configuration,
$this->registry,
$this->subscriptionManager,
$this->schemaValidator
);
});
it('routes to handleInitialize for initialize request', function () {
$request = new JsonRpcRequest(
jsonrpc: '2.0',
id: 1,
method: 'initialize',
params: [
'protocolVersion' => Protocol::LATEST_PROTOCOL_VERSION,
'clientInfo' => ['name' => 'client', 'version' => '1.0'],
'capabilities' => [],
]
);
$this->session->shouldReceive('set')->with('client_info', Mockery::on(fn($value) => $value['name'] === 'client' && $value['version'] === '1.0'))->once();
$this->session->shouldReceive('set')->with('protocol_version', Protocol::LATEST_PROTOCOL_VERSION)->once();
$result = $this->dispatcher->handleRequest($request, $this->context);
expect($result)->toBeInstanceOf(InitializeResult::class);
expect($result->protocolVersion)->toBe(Protocol::LATEST_PROTOCOL_VERSION);
expect($result->serverInfo->name)->toBe('DispatcherTestServer');
});
it('routes to handlePing for ping request', function () {
$request = new JsonRpcRequest('2.0', 'id1', 'ping', []);
$result = $this->dispatcher->handleRequest($request, $this->context);
expect($result)->toBeInstanceOf(EmptyResult::class);
});
it('throws MethodNotFound for unknown request method', function () {
$rawRequest = new JsonRpcRequest('2.0', 'id1', 'unknown/method', []);
$this->dispatcher->handleRequest($rawRequest, $this->context);
})->throws(McpServerException::class, "Method 'unknown/method' not found.");
it('routes to handleNotificationInitialized for initialized notification', function () {
$notification = new JsonRpcNotification('2.0', 'notifications/initialized', []);
$this->session->shouldReceive('set')->with('initialized', true)->once();
$this->dispatcher->handleNotification($notification, $this->session);
});
it('does nothing for unknown notification method', function () {
$rawNotification = new JsonRpcNotification('2.0', 'unknown/notification', []);
$this->session->shouldNotReceive('set');
$this->dispatcher->handleNotification($rawNotification, $this->session);
});
it('can handle initialize request', function () {
$clientInfo = Implementation::make('TestClient', '0.9.9');
$request = InitializeRequest::make(1, Protocol::LATEST_PROTOCOL_VERSION, ClientCapabilities::make(), $clientInfo, []);
$this->session->shouldReceive('set')->with('client_info', $clientInfo->toArray())->once();
$this->session->shouldReceive('set')->with('protocol_version', Protocol::LATEST_PROTOCOL_VERSION)->once();
$result = $this->dispatcher->handleInitialize($request, $this->session);
expect($result->protocolVersion)->toBe(Protocol::LATEST_PROTOCOL_VERSION);
expect($result->serverInfo->name)->toBe('DispatcherTestServer');
expect($result->capabilities)->toBeInstanceOf(ServerCapabilities::class);
});
it('can handle initialize request with older supported protocol version', function () {
$clientInfo = Implementation::make('TestClient', '0.9.9');
$clientRequestedVersion = '2024-11-05';
$request = InitializeRequest::make(1, $clientRequestedVersion, ClientCapabilities::make(), $clientInfo, []);
$this->session->shouldReceive('set')->with('client_info', $clientInfo->toArray())->once();
$this->session->shouldReceive('set')->with('protocol_version', $clientRequestedVersion)->once();
$result = $this->dispatcher->handleInitialize($request, $this->session);
expect($result->protocolVersion)->toBe($clientRequestedVersion);
expect($result->serverInfo->name)->toBe('DispatcherTestServer');
expect($result->capabilities)->toBeInstanceOf(ServerCapabilities::class);
});
it('can handle initialize request with unsupported protocol version', function () {
$clientInfo = Implementation::make('TestClient', '0.9.9');
$unsupportedVersion = '1999-01-01';
$request = InitializeRequest::make(1, $unsupportedVersion, ClientCapabilities::make(), $clientInfo, []);
$this->session->shouldReceive('set')->with('client_info', $clientInfo->toArray())->once();
$this->session->shouldReceive('set')->with('protocol_version', Protocol::LATEST_PROTOCOL_VERSION)->once();
$result = $this->dispatcher->handleInitialize($request, $this->session);
expect($result->protocolVersion)->toBe(Protocol::LATEST_PROTOCOL_VERSION);
expect($result->serverInfo->name)->toBe('DispatcherTestServer');
expect($result->capabilities)->toBeInstanceOf(ServerCapabilities::class);
});
it('can handle tool list request and return paginated tools', function () {
$toolSchemas = [
ToolSchema::make('tool1', ['type' => 'object', 'properties' => []]),
ToolSchema::make('tool2', ['type' => 'object', 'properties' => []]),
ToolSchema::make('tool3', ['type' => 'object', 'properties' => []]),
ToolSchema::make('tool4', ['type' => 'object', 'properties' => []]),
];
$this->registry->shouldReceive('getTools')->andReturn($toolSchemas);
$request = ListToolsRequest::make(1);
$result = $this->dispatcher->handleToolList($request);
expect($result->tools)->toHaveCount(DISPATCHER_PAGINATION_LIMIT);
expect($result->tools[0]->name)->toBe('tool1');
expect($result->nextCursor)->toBeString();
$nextCursor = $result->nextCursor;
$requestPage2 = ListToolsRequest::make(2, $nextCursor);
$resultPage2 = $this->dispatcher->handleToolList($requestPage2);
expect($resultPage2->tools)->toHaveCount(count($toolSchemas) - DISPATCHER_PAGINATION_LIMIT);
expect($resultPage2->tools[0]->name)->toBe('tool4');
expect($resultPage2->nextCursor)->toBeNull();
});
it('can handle tool call request and return result', function () {
$toolName = 'my-calculator';
$args = ['a' => 10, 'b' => 5];
$toolSchema = ToolSchema::make($toolName, ['type' => 'object', 'properties' => ['a' => ['type' => 'integer'], 'b' => ['type' => 'integer']]]);
$registeredToolMock = Mockery::mock(RegisteredTool::class, [$toolSchema, 'MyToolHandler', 'handleTool', false]);
$this->registry->shouldReceive('getTool')->with($toolName)->andReturn($registeredToolMock);
$this->schemaValidator->shouldReceive('validateAgainstJsonSchema')->with($args, $toolSchema->inputSchema)->andReturn([]); // No validation errors
$registeredToolMock->shouldReceive('call')->with($this->container, $args, $this->context)->andReturn([TextContent::make("Result: 15")]);
$request = CallToolRequest::make(1, $toolName, $args);
$result = $this->dispatcher->handleToolCall($request, $this->context);
expect($result)->toBeInstanceOf(CallToolResult::class);
expect($result->content[0]->text)->toBe("Result: 15");
expect($result->isError)->toBeFalse();
});
it('can handle tool call request and throw exception if tool not found', function () {
$this->registry->shouldReceive('getTool')->with('unknown-tool')->andReturn(null);
$request = CallToolRequest::make(1, 'unknown-tool', []);
$this->dispatcher->handleToolCall($request, $this->context);
})->throws(McpServerException::class, "Tool 'unknown-tool' not found.");
it('can handle tool call request and throw exception if argument validation fails', function () {
$toolName = 'strict-tool';
$args = ['param' => 'wrong_type'];
$toolSchema = ToolSchema::make($toolName, ['type' => 'object', 'properties' => ['param' => ['type' => 'integer']]]);
$registeredToolMock = Mockery::mock(RegisteredTool::class, [$toolSchema, 'MyToolHandler', 'handleTool', false]);
$this->registry->shouldReceive('getTool')->with($toolName)->andReturn($registeredToolMock);
$validationErrors = [['pointer' => '/param', 'keyword' => 'type', 'message' => 'Expected integer']];
$this->schemaValidator->shouldReceive('validateAgainstJsonSchema')->with($args, $toolSchema->inputSchema)->andReturn($validationErrors);
$request = CallToolRequest::make(1, $toolName, $args);
try {
$this->dispatcher->handleToolCall($request, $this->context);
} catch (McpServerException $e) {
expect($e->getMessage())->toContain("Invalid parameters for tool 'strict-tool'");
expect($e->getData()['validation_errors'])->toBeArray();
}
});
it('can handle tool call request and return error if tool execution throws exception', function () {
$toolName = 'failing-tool';
$toolSchema = ToolSchema::make($toolName, ['type' => 'object', 'properties' => []]);
$registeredToolMock = Mockery::mock(RegisteredTool::class, [$toolSchema, 'MyToolHandler', 'handleTool', false]);
$this->registry->shouldReceive('getTool')->with($toolName)->andReturn($registeredToolMock);
$this->schemaValidator->shouldReceive('validateAgainstJsonSchema')->andReturn([]);
$registeredToolMock->shouldReceive('call')->andThrow(new \RuntimeException("Tool crashed!"));
$request = CallToolRequest::make(1, $toolName, []);
$result = $this->dispatcher->handleToolCall($request, $this->context);
expect($result->isError)->toBeTrue();
expect($result->content[0]->text)->toBe("Tool execution failed: Tool crashed!");
});
it('can handle tool call request and return error if result formatting fails', function () {
$toolName = 'bad-result-tool';
$toolSchema = ToolSchema::make($toolName, ['type' => 'object', 'properties' => []]);
$registeredToolMock = Mockery::mock(RegisteredTool::class, [$toolSchema, 'MyToolHandler', 'handleTool', false]);
$this->registry->shouldReceive('getTool')->with($toolName)->andReturn($registeredToolMock);
$this->schemaValidator->shouldReceive('validateAgainstJsonSchema')->andReturn([]);
$registeredToolMock->shouldReceive('call')->andThrow(new \JsonException("Unencodable."));
$request = CallToolRequest::make(1, $toolName, []);
$result = $this->dispatcher->handleToolCall($request, $this->context);
expect($result->isError)->toBeTrue();
expect($result->content[0]->text)->toBe("Failed to serialize tool result: Unencodable.");
});
it('can handle resources list request and return paginated resources', function () {
$resourceSchemas = [
ResourceSchema::make('res://1', 'Resource1'),
ResourceSchema::make('res://2', 'Resource2'),
ResourceSchema::make('res://3', 'Resource3'),
ResourceSchema::make('res://4', 'Resource4'),
ResourceSchema::make('res://5', 'Resource5')
];
$this->registry->shouldReceive('getResources')->andReturn($resourceSchemas);
$requestP1 = ListResourcesRequest::make(1);
$resultP1 = $this->dispatcher->handleResourcesList($requestP1);
expect($resultP1->resources)->toHaveCount(DISPATCHER_PAGINATION_LIMIT);
expect(array_map(fn($r) => $r->name, $resultP1->resources))->toEqual(['Resource1', 'Resource2', 'Resource3']);
expect($resultP1->nextCursor)->toBe(base64_encode('offset=3'));
// Page 2
$requestP2 = ListResourcesRequest::make(2, $resultP1->nextCursor);
$resultP2 = $this->dispatcher->handleResourcesList($requestP2);
expect($resultP2->resources)->toHaveCount(2);
expect(array_map(fn($r) => $r->name, $resultP2->resources))->toEqual(['Resource4', 'Resource5']);
expect($resultP2->nextCursor)->toBeNull();
});
it('can handle resources list request and return empty if registry has no resources', function () {
$this->registry->shouldReceive('getResources')->andReturn([]);
$request = ListResourcesRequest::make(1);
$result = $this->dispatcher->handleResourcesList($request);
expect($result->resources)->toBeEmpty();
expect($result->nextCursor)->toBeNull();
});
it('can handle resource template list request and return paginated templates', function () {
$templateSchemas = [
ResourceTemplateSchema::make('tpl://{id}/1', 'Template1'),
ResourceTemplateSchema::make('tpl://{id}/2', 'Template2'),
ResourceTemplateSchema::make('tpl://{id}/3', 'Template3'),
ResourceTemplateSchema::make('tpl://{id}/4', 'Template4'),
];
$this->registry->shouldReceive('getResourceTemplates')->andReturn($templateSchemas);
// Page 1
$requestP1 = ListResourceTemplatesRequest::make(1);
$resultP1 = $this->dispatcher->handleResourceTemplateList($requestP1);
expect($resultP1->resourceTemplates)->toHaveCount(DISPATCHER_PAGINATION_LIMIT);
expect(array_map(fn($rt) => $rt->name, $resultP1->resourceTemplates))->toEqual(['Template1', 'Template2', 'Template3']);
expect($resultP1->nextCursor)->toBe(base64_encode('offset=3'));
// Page 2
$requestP2 = ListResourceTemplatesRequest::make(2, $resultP1->nextCursor);
$resultP2 = $this->dispatcher->handleResourceTemplateList($requestP2);
expect($resultP2->resourceTemplates)->toHaveCount(1);
expect(array_map(fn($rt) => $rt->name, $resultP2->resourceTemplates))->toEqual(['Template4']);
expect($resultP2->nextCursor)->toBeNull();
});
it('can handle resource read request and return resource contents', function () {
$uri = 'file://data.txt';
$resourceSchema = ResourceSchema::make($uri, 'file_resource');
$registeredResourceMock = Mockery::mock(RegisteredResource::class, [$resourceSchema, ['MyResourceHandler', 'read'], false]);
$resourceContents = [TextContent::make('File content')];
$this->registry->shouldReceive('getResource')->with($uri)->andReturn($registeredResourceMock);
$registeredResourceMock->shouldReceive('read')->with($this->container, $uri, $this->context)->andReturn($resourceContents);
$request = ReadResourceRequest::make(1, $uri);
$result = $this->dispatcher->handleResourceRead($request, $this->context);
expect($result)->toBeInstanceOf(ReadResourceResult::class);
expect($result->contents)->toEqual($resourceContents);
});
it('can handle resource read request and throw exception if resource not found', function () {
$this->registry->shouldReceive('getResource')->with('unknown://uri')->andReturn(null);
$request = ReadResourceRequest::make(1, 'unknown://uri');
$this->dispatcher->handleResourceRead($request, $this->context);
})->throws(McpServerException::class, "Resource URI 'unknown://uri' not found.");
it('can handle resource subscribe request and call subscription manager', function () {
$uri = 'news://updates';
$this->session->shouldReceive('getId')->andReturn(DISPATCHER_SESSION_ID);
$this->subscriptionManager->shouldReceive('subscribe')->with(DISPATCHER_SESSION_ID, $uri)->once();
$request = ResourceSubscribeRequest::make(1, $uri);
$result = $this->dispatcher->handleResourceSubscribe($request, $this->session);
expect($result)->toBeInstanceOf(EmptyResult::class);
});
it('can handle prompts list request and return paginated prompts', function () {
$promptSchemas = [
PromptSchema::make('promptA', '', []),
PromptSchema::make('promptB', '', []),
PromptSchema::make('promptC', '', []),
PromptSchema::make('promptD', '', []),
PromptSchema::make('promptE', '', []),
PromptSchema::make('promptF', '', []),
]; // 6 prompts
$this->registry->shouldReceive('getPrompts')->andReturn($promptSchemas);
// Page 1
$requestP1 = ListPromptsRequest::make(1);
$resultP1 = $this->dispatcher->handlePromptsList($requestP1);
expect($resultP1->prompts)->toHaveCount(DISPATCHER_PAGINATION_LIMIT);
expect(array_map(fn($p) => $p->name, $resultP1->prompts))->toEqual(['promptA', 'promptB', 'promptC']);
expect($resultP1->nextCursor)->toBe(base64_encode('offset=3'));
// Page 2
$requestP2 = ListPromptsRequest::make(2, $resultP1->nextCursor);
$resultP2 = $this->dispatcher->handlePromptsList($requestP2);
expect($resultP2->prompts)->toHaveCount(DISPATCHER_PAGINATION_LIMIT); // 3 more
expect(array_map(fn($p) => $p->name, $resultP2->prompts))->toEqual(['promptD', 'promptE', 'promptF']);
expect($resultP2->nextCursor)->toBeNull(); // End of list
});
it('can handle prompt get request and return prompt messages', function () {
$promptName = 'daily-summary';
$args = ['date' => '2024-07-16'];
$promptSchema = PromptSchema::make($promptName, 'summary_prompt', [PromptArgument::make('date', required: true)]);
$registeredPromptMock = Mockery::mock(RegisteredPrompt::class, [$promptSchema, ['MyPromptHandler', 'get'], false]);
$promptMessages = [PromptMessage::make(Role::User, TextContent::make("Summary for 2024-07-16"))];
$this->registry->shouldReceive('getPrompt')->with($promptName)->andReturn($registeredPromptMock);
$registeredPromptMock->shouldReceive('get')->with($this->container, $args, $this->context)->andReturn($promptMessages);
$request = GetPromptRequest::make(1, $promptName, $args);
$result = $this->dispatcher->handlePromptGet($request, $this->context);
expect($result)->toBeInstanceOf(GetPromptResult::class);
expect($result->messages)->toEqual($promptMessages);
expect($result->description)->toBe($promptSchema->description);
});
it('can handle prompt get request and throw exception if required argument is missing', function () {
$promptName = 'needs-topic';
$promptSchema = PromptSchema::make($promptName, '', [PromptArgument::make('topic', required: true)]);
$registeredPromptMock = Mockery::mock(RegisteredPrompt::class, [$promptSchema, ['MyPromptHandler', 'get'], false]);
$this->registry->shouldReceive('getPrompt')->with($promptName)->andReturn($registeredPromptMock);
$request = GetPromptRequest::make(1, $promptName, ['other_arg' => 'value']); // 'topic' is missing
$this->dispatcher->handlePromptGet($request, $this->context);
})->throws(McpServerException::class, "Missing required argument 'topic' for prompt 'needs-topic'.");
it('can handle logging set level request and set log level on session', function () {
$level = LoggingLevel::Debug;
$this->session->shouldReceive('getId')->andReturn(DISPATCHER_SESSION_ID);
$this->session->shouldReceive('set')->with('log_level', 'debug')->once();
$request = SetLogLevelRequest::make(1, $level);
$result = $this->dispatcher->handleLoggingSetLevel($request, $this->session);
expect($result)->toBeInstanceOf(EmptyResult::class);
});
it('can handle completion complete request for prompt and delegate to provider', function () {
$promptName = 'my-completable-prompt';
$argName = 'tagName';
$currentValue = 'php';
$completions = ['php-mcp', 'php-fig'];
$mockCompletionProvider = Mockery::mock(CompletionProviderInterface::class);
$providerClass = get_class($mockCompletionProvider);
$promptSchema = PromptSchema::make($promptName, '', [PromptArgument::make($argName)]);
$registeredPrompt = new RegisteredPrompt(
schema: $promptSchema,
handler: ['MyPromptHandler', 'get'],
isManual: false,
completionProviders: [$argName => $providerClass]
);
$this->registry->shouldReceive('getPrompt')->with($promptName)->andReturn($registeredPrompt);
$this->container->shouldReceive('get')->with($providerClass)->andReturn($mockCompletionProvider);
$mockCompletionProvider->shouldReceive('getCompletions')->with($currentValue, $this->session)->andReturn($completions);
$request = CompletionCompleteRequest::make(1, PromptReference::make($promptName), ['name' => $argName, 'value' => $currentValue]);
$result = $this->dispatcher->handleCompletionComplete($request, $this->session);
expect($result)->toBeInstanceOf(CompletionCompleteResult::class);
expect($result->values)->toEqual($completions);
expect($result->total)->toBe(count($completions));
expect($result->hasMore)->toBeFalse();
});
it('can handle completion complete request for resource template and delegate to provider', function () {
$templateUri = 'item://{itemId}/category/{catName}';
$uriVarName = 'catName';
$currentValue = 'boo';
$completions = ['books', 'boomerangs'];
$mockCompletionProvider = Mockery::mock(CompletionProviderInterface::class);
$providerClass = get_class($mockCompletionProvider);
$templateSchema = ResourceTemplateSchema::make($templateUri, 'item-template');
$registeredTemplate = new RegisteredResourceTemplate(
schema: $templateSchema,
handler: ['MyResourceTemplateHandler', 'get'],
isManual: false,
completionProviders: [$uriVarName => $providerClass]
);
$this->registry->shouldReceive('getResourceTemplate')->with($templateUri)->andReturn($registeredTemplate);
$this->container->shouldReceive('get')->with($providerClass)->andReturn($mockCompletionProvider);
$mockCompletionProvider->shouldReceive('getCompletions')->with($currentValue, $this->session)->andReturn($completions);
$request = CompletionCompleteRequest::make(1, ResourceReference::make($templateUri), ['name' => $uriVarName, 'value' => $currentValue]);
$result = $this->dispatcher->handleCompletionComplete($request, $this->session);
expect($result->values)->toEqual($completions);
});
it('can handle completion complete request and return empty if no provider', function () {
$promptName = 'no-provider-prompt';
$promptSchema = PromptSchema::make($promptName, '', [PromptArgument::make('arg')]);
$registeredPrompt = new RegisteredPrompt(
schema: $promptSchema,
handler: ['MyPromptHandler', 'get'],
isManual: false,
completionProviders: []
);
$this->registry->shouldReceive('getPrompt')->with($promptName)->andReturn($registeredPrompt);
$request = CompletionCompleteRequest::make(1, PromptReference::make($promptName), ['name' => 'arg', 'value' => '']);
$result = $this->dispatcher->handleCompletionComplete($request, $this->session);
expect($result->values)->toBeEmpty();
});
it('can handle completion complete request with ListCompletionProvider instance', function () {
$promptName = 'list-completion-prompt';
$argName = 'category';
$currentValue = 'bl';
$expectedCompletions = ['blog'];
$listProvider = new \PhpMcp\Server\Defaults\ListCompletionProvider(['blog', 'news', 'docs', 'api']);
$promptSchema = PromptSchema::make($promptName, '', [PromptArgument::make($argName)]);
$registeredPrompt = new RegisteredPrompt(
schema: $promptSchema,
handler: ['MyPromptHandler', 'get'],
isManual: false,
completionProviders: [$argName => $listProvider]
);
$this->registry->shouldReceive('getPrompt')->with($promptName)->andReturn($registeredPrompt);
$request = CompletionCompleteRequest::make(1, PromptReference::make($promptName), ['name' => $argName, 'value' => $currentValue]);
$result = $this->dispatcher->handleCompletionComplete($request, $this->session);
expect($result->values)->toEqual($expectedCompletions);
expect($result->total)->toBe(1);
expect($result->hasMore)->toBeFalse();
});
it('can handle completion complete request with EnumCompletionProvider instance', function () {
$promptName = 'enum-completion-prompt';
$argName = 'status';
$currentValue = 'a';
$expectedCompletions = ['archived'];
$enumProvider = new \PhpMcp\Server\Defaults\EnumCompletionProvider(StatusEnum::class);
$promptSchema = PromptSchema::make($promptName, '', [PromptArgument::make($argName)]);
$registeredPrompt = new RegisteredPrompt(
schema: $promptSchema,
handler: ['MyPromptHandler', 'get'],
isManual: false,
completionProviders: [$argName => $enumProvider]
);
$this->registry->shouldReceive('getPrompt')->with($promptName)->andReturn($registeredPrompt);
$request = CompletionCompleteRequest::make(1, PromptReference::make($promptName), ['name' => $argName, 'value' => $currentValue]);
$result = $this->dispatcher->handleCompletionComplete($request, $this->session);
expect($result->values)->toEqual($expectedCompletions);
expect($result->total)->toBe(1);
expect($result->hasMore)->toBeFalse();
});
it('decodeCursor handles null and invalid cursors', function () {
$method = new \ReflectionMethod(Dispatcher::class, 'decodeCursor');
$method->setAccessible(true);
expect($method->invoke($this->dispatcher, null))->toBe(0);
expect($method->invoke($this->dispatcher, 'not_base64_$$$'))->toBe(0);
expect($method->invoke($this->dispatcher, base64_encode('invalid_format')))->toBe(0);
expect($method->invoke($this->dispatcher, base64_encode('offset=123')))->toBe(123);
});
it('encodeNextCursor generates correct cursor or null', function () {
$method = new \ReflectionMethod(Dispatcher::class, 'encodeNextCursor');
$method->setAccessible(true);
$limit = DISPATCHER_PAGINATION_LIMIT;
expect($method->invoke($this->dispatcher, 0, $limit, 10, $limit))->toBe(base64_encode('offset=3'));
expect($method->invoke($this->dispatcher, 0, $limit, $limit, $limit))->toBeNull();
expect($method->invoke($this->dispatcher, $limit, 2, $limit + 2 + 1, $limit))->toBe(base64_encode('offset=' . ($limit + 2)));
expect($method->invoke($this->dispatcher, $limit, 1, $limit + 1, $limit))->toBeNull();
expect($method->invoke($this->dispatcher, 0, 0, 10, $limit))->toBeNull();
});
```
--------------------------------------------------------------------------------
/tests/Integration/StreamableHttpServerTransportTest.php:
--------------------------------------------------------------------------------
```php
<?php
use PhpMcp\Server\Protocol;
use PhpMcp\Server\Tests\Mocks\Clients\MockJsonHttpClient;
use PhpMcp\Server\Tests\Mocks\Clients\MockStreamHttpClient;
use React\ChildProcess\Process;
use React\Http\Browser;
use React\Http\Message\ResponseException;
use React\Stream\ReadableStreamInterface;
use function React\Async\await;
use function React\Promise\resolve;
const STREAMABLE_HTTP_SCRIPT_PATH = __DIR__ . '/../Fixtures/ServerScripts/StreamableHttpTestServer.php';
const STREAMABLE_HTTP_PROCESS_TIMEOUT = 9;
const STREAMABLE_HTTP_HOST = '127.0.0.1';
const STREAMABLE_MCP_PATH = 'mcp_streamable_json_mode';
beforeEach(function () {
if (!is_file(STREAMABLE_HTTP_SCRIPT_PATH)) {
$this->markTestSkipped("Server script not found: " . STREAMABLE_HTTP_SCRIPT_PATH);
}
if (!is_executable(STREAMABLE_HTTP_SCRIPT_PATH)) {
chmod(STREAMABLE_HTTP_SCRIPT_PATH, 0755);
}
$phpPath = PHP_BINARY ?: 'php';
$commandPhpPath = str_contains($phpPath, ' ') ? '"' . $phpPath . '"' : $phpPath;
$commandScriptPath = escapeshellarg(STREAMABLE_HTTP_SCRIPT_PATH);
$this->port = findFreePort();
$jsonModeCommandArgs = [
escapeshellarg(STREAMABLE_HTTP_HOST),
escapeshellarg((string)$this->port),
escapeshellarg(STREAMABLE_MCP_PATH),
escapeshellarg('true'), // enableJsonResponse = true
];
$this->jsonModeCommand = $commandPhpPath . ' ' . $commandScriptPath . ' ' . implode(' ', $jsonModeCommandArgs);
$streamModeCommandArgs = [
escapeshellarg(STREAMABLE_HTTP_HOST),
escapeshellarg((string)$this->port),
escapeshellarg(STREAMABLE_MCP_PATH),
escapeshellarg('false'), // enableJsonResponse = false
];
$this->streamModeCommand = $commandPhpPath . ' ' . $commandScriptPath . ' ' . implode(' ', $streamModeCommandArgs);
$statelessModeCommandArgs = [
escapeshellarg(STREAMABLE_HTTP_HOST),
escapeshellarg((string)$this->port),
escapeshellarg(STREAMABLE_MCP_PATH),
escapeshellarg('true'), // enableJsonResponse = true
escapeshellarg('false'), // useEventStore = false
escapeshellarg('true'), // stateless = true
];
$this->statelessModeCommand = $commandPhpPath . ' ' . $commandScriptPath . ' ' . implode(' ', $statelessModeCommandArgs);
$this->process = null;
});
afterEach(function () {
if ($this->process instanceof Process && $this->process->isRunning()) {
if ($this->process->stdout instanceof ReadableStreamInterface) {
$this->process->stdout->close();
}
if ($this->process->stderr instanceof ReadableStreamInterface) {
$this->process->stderr->close();
}
$this->process->terminate(SIGTERM);
try {
await(delay(0.02));
} catch (\Throwable $e) {
}
if ($this->process->isRunning()) {
$this->process->terminate(SIGKILL);
}
}
$this->process = null;
});
describe('JSON MODE', function () {
beforeEach(function () {
$this->process = new Process($this->jsonModeCommand, getcwd() ?: null, null, []);
$this->process->start();
$this->jsonClient = new MockJsonHttpClient(STREAMABLE_HTTP_HOST, $this->port, STREAMABLE_MCP_PATH);
await(delay(0.2));
});
it('server starts, initializes via POST JSON, calls a tool, and closes', function () {
// 1. Initialize
$initResult = await($this->jsonClient->sendRequest('initialize', [
'protocolVersion' => Protocol::LATEST_PROTOCOL_VERSION,
'clientInfo' => ['name' => 'JsonModeClient', 'version' => '1.0'],
'capabilities' => []
], 'init-json-1'));
expect($initResult['statusCode'])->toBe(200);
expect($initResult['body']['id'])->toBe('init-json-1');
expect($initResult['body'])->not->toHaveKey('error');
expect($initResult['body']['result']['protocolVersion'])->toBe(Protocol::LATEST_PROTOCOL_VERSION);
expect($initResult['body']['result']['serverInfo']['name'])->toBe('StreamableHttpIntegrationServer');
expect($this->jsonClient->sessionId)->toBeString()->not->toBeEmpty();
// 2. Initialized notification
$notifResult = await($this->jsonClient->sendNotification('notifications/initialized'));
expect($notifResult['statusCode'])->toBe(202);
// 3. Call a tool
$toolResult = await($this->jsonClient->sendRequest('tools/call', [
'name' => 'greet_streamable_tool',
'arguments' => ['name' => 'JSON Mode User']
], 'tool-json-1'));
expect($toolResult['statusCode'])->toBe(200);
expect($toolResult['body']['id'])->toBe('tool-json-1');
expect($toolResult['body'])->not->toHaveKey('error');
expect($toolResult['body']['result']['content'][0]['text'])->toBe('Hello, JSON Mode User!');
// Server process is terminated in afterEach
})->group('integration', 'streamable_http_json');
it('return HTTP 400 error response for invalid JSON in POST request', function () {
$malformedJson = '{"jsonrpc":"2.0", "id": "bad-json-post-1", "method": "tools/list", "params": {"broken"}';
$promise = $this->jsonClient->browser->post(
$this->jsonClient->baseUrl,
['Content-Type' => 'application/json', 'Accept' => 'application/json'],
$malformedJson
);
try {
await(timeout($promise, STREAMABLE_HTTP_PROCESS_TIMEOUT - 2));
} catch (ResponseException $e) {
expect($e->getResponse()->getStatusCode())->toBe(400);
$bodyContent = (string) $e->getResponse()->getBody();
$decodedBody = json_decode($bodyContent, true);
expect($decodedBody['jsonrpc'])->toBe('2.0');
expect($decodedBody['id'])->toBe('');
expect($decodedBody['error']['code'])->toBe(-32700);
expect($decodedBody['error']['message'])->toContain('Invalid JSON');
}
})->group('integration', 'streamable_http_json');
it('returns JSON-RPC error result for request for non-existent method', function () {
// 1. Initialize
await($this->jsonClient->sendRequest('initialize', ['protocolVersion' => Protocol::LATEST_PROTOCOL_VERSION, 'clientInfo' => ['name' => 'JsonModeClient', 'version' => '1.0'], 'capabilities' => []], 'init-json-err'));
await($this->jsonClient->sendNotification('notifications/initialized'));
// 2. Request non-existent method
$errorResult = await($this->jsonClient->sendRequest('non/existentToolViaJson', [], 'err-meth-json-1'));
expect($errorResult['statusCode'])->toBe(200);
expect($errorResult['body']['id'])->toBe('err-meth-json-1');
expect($errorResult['body']['error']['code'])->toBe(-32601);
expect($errorResult['body']['error']['message'])->toContain("Method 'non/existentToolViaJson' not found");
})->group('integration', 'streamable_http_json');
it('can handle batch requests correctly', function () {
// 1. Initialize
await($this->jsonClient->sendRequest('initialize', [
'protocolVersion' => Protocol::LATEST_PROTOCOL_VERSION,
'clientInfo' => ['name' => 'JsonModeClient', 'version' => '1.0'],
'capabilities' => []
], 'init-json-batch'));
expect($this->jsonClient->sessionId)->toBeString()->not->toBeEmpty();
await($this->jsonClient->sendNotification('notifications/initialized'));
// 2. Send Batch Request
$batchRequests = [
['jsonrpc' => '2.0', 'id' => 'batch-req-1', 'method' => 'tools/call', 'params' => ['name' => 'greet_streamable_tool', 'arguments' => ['name' => 'Batch Item 1']]],
['jsonrpc' => '2.0', 'method' => 'notifications/something'],
['jsonrpc' => '2.0', 'id' => 'batch-req-2', 'method' => 'tools/call', 'params' => ['name' => 'sum_streamable_tool', 'arguments' => ['a' => 10, 'b' => 20]]],
['jsonrpc' => '2.0', 'id' => 'batch-req-3', 'method' => 'nonexistent/method']
];
$batchResponse = await($this->jsonClient->sendBatchRequest($batchRequests));
$findResponseById = function (array $batch, $id) {
foreach ($batch as $item) {
if (isset($item['id']) && $item['id'] === $id) {
return $item;
}
}
return null;
};
expect($batchResponse['statusCode'])->toBe(200);
expect($batchResponse['body'])->toBeArray()->toHaveCount(3);
$response1 = $findResponseById($batchResponse['body'], 'batch-req-1');
$response2 = $findResponseById($batchResponse['body'], 'batch-req-2');
$response3 = $findResponseById($batchResponse['body'], 'batch-req-3');
expect($response1['result']['content'][0]['text'])->toBe('Hello, Batch Item 1!');
expect($response2['result']['content'][0]['text'])->toBe('30');
expect($response3['error']['code'])->toBe(-32601);
expect($response3['error']['message'])->toContain("Method 'nonexistent/method' not found");
})->group('integration', 'streamable_http_json');
it('can handle tool list request', function () {
await($this->jsonClient->sendRequest('initialize', [
'protocolVersion' => Protocol::LATEST_PROTOCOL_VERSION,
'clientInfo' => ['name' => 'JsonModeClient', 'version' => '1.0'],
'capabilities' => []
], 'init-json-tools'));
await($this->jsonClient->sendNotification('notifications/initialized'));
$toolListResult = await($this->jsonClient->sendRequest('tools/list', [], 'tool-list-json-1'));
expect($toolListResult['statusCode'])->toBe(200);
expect($toolListResult['body']['id'])->toBe('tool-list-json-1');
expect($toolListResult['body']['result']['tools'])->toBeArray();
expect(count($toolListResult['body']['result']['tools']))->toBe(4);
expect($toolListResult['body']['result']['tools'][0]['name'])->toBe('greet_streamable_tool');
expect($toolListResult['body']['result']['tools'][1]['name'])->toBe('sum_streamable_tool');
expect($toolListResult['body']['result']['tools'][2]['name'])->toBe('tool_reads_context');
})->group('integration', 'streamable_http_json');
it('passes request in Context', function () {
await($this->jsonClient->sendRequest('initialize', [
'protocolVersion' => Protocol::LATEST_PROTOCOL_VERSION,
'clientInfo' => ['name' => 'JsonModeClient', 'version' => '1.0'],
'capabilities' => []
], 'init-json-context'));
await($this->jsonClient->sendNotification('notifications/initialized'));
$toolResult = await($this->jsonClient->sendRequest('tools/call', [
'name' => 'tool_reads_context',
'arguments' => []
], 'tool-json-context-1', ['X-Test-Header' => 'TestValue']));
expect($toolResult['statusCode'])->toBe(200);
expect($toolResult['body']['id'])->toBe('tool-json-context-1');
expect($toolResult['body'])->not->toHaveKey('error');
expect($toolResult['body']['result']['content'][0]['text'])->toBe('TestValue');
})->group('integration', 'streamable_http_json');
it('can read a registered resource', function () {
await($this->jsonClient->sendRequest('initialize', [
'protocolVersion' => Protocol::LATEST_PROTOCOL_VERSION,
'clientInfo' => ['name' => 'JsonModeClient', 'version' => '1.0'],
'capabilities' => []
], 'init-json-res'));
await($this->jsonClient->sendNotification('notifications/initialized'));
$resourceResult = await($this->jsonClient->sendRequest('resources/read', ['uri' => 'test://streamable/static'], 'res-read-json-1'));
expect($resourceResult['statusCode'])->toBe(200);
expect($resourceResult['body']['id'])->toBe('res-read-json-1');
$contents = $resourceResult['body']['result']['contents'];
expect($contents[0]['uri'])->toBe('test://streamable/static');
expect($contents[0]['text'])->toBe(\PhpMcp\Server\Tests\Fixtures\General\ResourceHandlerFixture::$staticTextContent);
})->group('integration', 'streamable_http_json');
it('can get a registered prompt', function () {
await($this->jsonClient->sendRequest('initialize', ['protocolVersion' => Protocol::LATEST_PROTOCOL_VERSION, 'clientInfo' => ['name' => 'JsonModeClient', 'version' => '1.0'], 'capabilities' => []], 'init-json-prompt'));
await($this->jsonClient->sendNotification('notifications/initialized'));
$promptResult = await($this->jsonClient->sendRequest('prompts/get', [
'name' => 'simple_streamable_prompt',
'arguments' => ['name' => 'JsonPromptUser', 'style' => 'terse']
], 'prompt-get-json-1'));
expect($promptResult['statusCode'])->toBe(200);
expect($promptResult['body']['id'])->toBe('prompt-get-json-1');
$messages = $promptResult['body']['result']['messages'];
expect($messages[0]['content']['text'])->toBe('Craft a terse greeting for JsonPromptUser.');
})->group('integration', 'streamable_http_json');
it('rejects subsequent requests if client does not send initialized notification', function () {
// 1. Initialize ONLY
await($this->jsonClient->sendRequest('initialize', ['protocolVersion' => Protocol::LATEST_PROTOCOL_VERSION, 'clientInfo' => ['name' => 'JsonModeClient', 'version' => '1.0'], 'capabilities' => []], 'init-json-noack'));
// Client "forgets" to send notifications/initialized back
// 2. Attempt to Call a tool
$toolResult = await($this->jsonClient->sendRequest('tools/call', [
'name' => 'greet_streamable_tool',
'arguments' => ['name' => 'NoAckJsonUser']
], 'tool-json-noack'));
expect($toolResult['statusCode'])->toBe(200); // HTTP is fine
expect($toolResult['body']['id'])->toBe('tool-json-noack');
expect($toolResult['body']['error']['code'])->toBe(-32600); // Invalid Request
expect($toolResult['body']['error']['message'])->toContain('Client session not initialized');
})->group('integration', 'streamable_http_json');
it('returns HTTP 400 error for non-initialize requests without Mcp-Session-Id', function () {
await($this->jsonClient->sendRequest('initialize', ['protocolVersion' => Protocol::LATEST_PROTOCOL_VERSION, 'clientInfo' => ['name' => 'JsonModeClient', 'version' => '1.0'], 'capabilities' => []], 'init-sess-test'));
$this->jsonClient->sessionId = null;
try {
await($this->jsonClient->sendRequest('tools/list', [], 'tools-list-no-session'));
} catch (ResponseException $e) {
expect($e->getResponse()->getStatusCode())->toBe(400);
$bodyContent = (string) $e->getResponse()->getBody();
$decodedBody = json_decode($bodyContent, true);
expect($decodedBody['jsonrpc'])->toBe('2.0');
expect($decodedBody['id'])->toBe('tools-list-no-session');
expect($decodedBody['error']['code'])->toBe(-32600);
expect($decodedBody['error']['message'])->toContain('Mcp-Session-Id header required');
}
})->group('integration', 'streamable_http_json');
});
describe('STREAM MODE', function () {
beforeEach(function () {
$this->process = new Process($this->streamModeCommand, getcwd() ?: null, null, []);
$this->process->start();
$this->streamClient = new MockStreamHttpClient(STREAMABLE_HTTP_HOST, $this->port, STREAMABLE_MCP_PATH);
await(delay(0.2));
});
afterEach(function () {
if ($this->streamClient ?? null) {
$this->streamClient->closeMainSseStream();
}
});
it('server starts, initializes via POST JSON, calls a tool, and closes', function () {
// 1. Initialize Request
$initResponse = await($this->streamClient->sendInitializeRequest([
'protocolVersion' => Protocol::LATEST_PROTOCOL_VERSION,
'clientInfo' => ['name' => 'StreamModeClient', 'version' => '1.0'],
'capabilities' => []
], 'init-stream-1'));
expect($this->streamClient->sessionId)->toBeString()->not->toBeEmpty();
expect($initResponse['id'])->toBe('init-stream-1');
expect($initResponse)->not->toHaveKey('error');
expect($initResponse['result']['protocolVersion'])->toBe(Protocol::LATEST_PROTOCOL_VERSION);
expect($initResponse['result']['serverInfo']['name'])->toBe('StreamableHttpIntegrationServer');
// 2. Send Initialized Notification
$notifResult = await($this->streamClient->sendHttpNotification('notifications/initialized'));
expect($notifResult['statusCode'])->toBe(202);
// 3. Call a tool
$toolResponse = await($this->streamClient->sendRequest('tools/call', [
'name' => 'greet_streamable_tool',
'arguments' => ['name' => 'Stream Mode User']
], 'tool-stream-1'));
expect($toolResponse['id'])->toBe('tool-stream-1');
expect($toolResponse)->not->toHaveKey('error');
expect($toolResponse['result']['content'][0]['text'])->toBe('Hello, Stream Mode User!');
})->group('integration', 'streamable_http_stream');
it('return HTTP 400 error response for invalid JSON in POST request', function () {
$malformedJson = '{"jsonrpc":"2.0", "id": "bad-json-stream-1", "method": "tools/list", "params": {"broken"}';
$postPromise = $this->streamClient->browser->post(
$this->streamClient->baseMcpUrl,
['Content-Type' => 'application/json', 'Accept' => 'text/event-stream'],
$malformedJson
);
try {
await(timeout($postPromise, STREAMABLE_HTTP_PROCESS_TIMEOUT - 2));
} catch (ResponseException $e) {
$httpResponse = $e->getResponse();
$bodyContent = (string) $httpResponse->getBody();
$decodedBody = json_decode($bodyContent, true);
expect($httpResponse->getStatusCode())->toBe(400);
expect($decodedBody['jsonrpc'])->toBe('2.0');
expect($decodedBody['id'])->toBe('');
expect($decodedBody['error']['code'])->toBe(-32700);
expect($decodedBody['error']['message'])->toContain('Invalid JSON');
}
})->group('integration', 'streamable_http_stream');
it('returns JSON-RPC error result for request for non-existent method', function () {
// 1. Initialize
await($this->streamClient->sendInitializeRequest([
'protocolVersion' => Protocol::LATEST_PROTOCOL_VERSION,
'clientInfo' => ['name' => 'StreamModeClient', 'version' => '1.0'],
'capabilities' => []
], 'init-stream-err'));
await($this->streamClient->sendHttpNotification('notifications/initialized'));
// 2. Send Request
$errorResponse = await($this->streamClient->sendRequest('non/existentToolViaStream', [], 'err-meth-stream-1'));
expect($errorResponse['id'])->toBe('err-meth-stream-1');
expect($errorResponse['error']['code'])->toBe(-32601);
expect($errorResponse['error']['message'])->toContain("Method 'non/existentToolViaStream' not found");
})->group('integration', 'streamable_http_stream');
it('can handle batch requests correctly', function () {
await($this->streamClient->sendInitializeRequest([
'protocolVersion' => Protocol::LATEST_PROTOCOL_VERSION,
'clientInfo' => ['name' => 'StreamModeBatchClient', 'version' => '1.0'],
'capabilities' => []
], 'init-stream-batch'));
expect($this->streamClient->sessionId)->toBeString()->not->toBeEmpty();
await($this->streamClient->sendHttpNotification('notifications/initialized'));
$batchRequests = [
['jsonrpc' => '2.0', 'id' => 'batch-req-1', 'method' => 'tools/call', 'params' => ['name' => 'greet_streamable_tool', 'arguments' => ['name' => 'Batch Item 1']]],
['jsonrpc' => '2.0', 'method' => 'notifications/something'],
['jsonrpc' => '2.0', 'id' => 'batch-req-2', 'method' => 'tools/call', 'params' => ['name' => 'sum_streamable_tool', 'arguments' => ['a' => 10, 'b' => 20]]],
['jsonrpc' => '2.0', 'id' => 'batch-req-3', 'method' => 'nonexistent/method']
];
$batchResponseArray = await($this->streamClient->sendBatchRequest($batchRequests));
expect($batchResponseArray)->toBeArray()->toHaveCount(3);
$findResponseById = function (array $batch, $id) {
foreach ($batch as $item) {
if (isset($item['id']) && $item['id'] === $id) {
return $item;
}
}
return null;
};
$response1 = $findResponseById($batchResponseArray, 'batch-req-1');
$response2 = $findResponseById($batchResponseArray, 'batch-req-2');
$response3 = $findResponseById($batchResponseArray, 'batch-req-3');
expect($response1['result']['content'][0]['text'])->toBe('Hello, Batch Item 1!');
expect($response2['result']['content'][0]['text'])->toBe('30');
expect($response3['error']['code'])->toBe(-32601);
expect($response3['error']['message'])->toContain("Method 'nonexistent/method' not found");
})->group('integration', 'streamable_http_stream');
it('passes request in Context', function () {
await($this->streamClient->sendInitializeRequest([
'protocolVersion' => Protocol::LATEST_PROTOCOL_VERSION,
'clientInfo' => ['name' => 'StreamModeClient', 'version' => '1.0'],
'capabilities' => []
], 'init-stream-context'));
expect($this->streamClient->sessionId)->toBeString()->not->toBeEmpty();
await($this->streamClient->sendHttpNotification('notifications/initialized'));
$toolResult = await($this->streamClient->sendRequest('tools/call', [
'name' => 'tool_reads_context',
'arguments' => []
], 'tool-stream-context-1', ['X-Test-Header' => 'TestValue']));
expect($toolResult['id'])->toBe('tool-stream-context-1');
expect($toolResult)->not->toHaveKey('error');
expect($toolResult['result']['content'][0]['text'])->toBe('TestValue');
})->group('integration', 'streamable_http_stream');
it('can handle tool list request', function () {
await($this->streamClient->sendInitializeRequest(['protocolVersion' => Protocol::LATEST_PROTOCOL_VERSION, 'clientInfo' => []], 'init-stream-tools'));
await($this->streamClient->sendHttpNotification('notifications/initialized'));
$toolListResponse = await($this->streamClient->sendRequest('tools/list', [], 'tool-list-stream-1'));
expect($toolListResponse['id'])->toBe('tool-list-stream-1');
expect($toolListResponse)->not->toHaveKey('error');
expect($toolListResponse['result']['tools'])->toBeArray();
expect(count($toolListResponse['result']['tools']))->toBe(4);
expect($toolListResponse['result']['tools'][0]['name'])->toBe('greet_streamable_tool');
expect($toolListResponse['result']['tools'][1]['name'])->toBe('sum_streamable_tool');
expect($toolListResponse['result']['tools'][2]['name'])->toBe('tool_reads_context');
})->group('integration', 'streamable_http_stream');
it('can read a registered resource', function () {
await($this->streamClient->sendInitializeRequest(['protocolVersion' => Protocol::LATEST_PROTOCOL_VERSION, 'clientInfo' => []], 'init-stream-res'));
await($this->streamClient->sendHttpNotification('notifications/initialized'));
$resourceResponse = await($this->streamClient->sendRequest('resources/read', ['uri' => 'test://streamable/static'], 'res-read-stream-1'));
expect($resourceResponse['id'])->toBe('res-read-stream-1');
$contents = $resourceResponse['result']['contents'];
expect($contents[0]['uri'])->toBe('test://streamable/static');
expect($contents[0]['text'])->toBe(\PhpMcp\Server\Tests\Fixtures\General\ResourceHandlerFixture::$staticTextContent);
})->group('integration', 'streamable_http_stream');
it('can get a registered prompt', function () {
await($this->streamClient->sendInitializeRequest(['protocolVersion' => Protocol::LATEST_PROTOCOL_VERSION, 'clientInfo' => []], 'init-stream-prompt'));
await($this->streamClient->sendHttpNotification('notifications/initialized'));
$promptResponse = await($this->streamClient->sendRequest('prompts/get', [
'name' => 'simple_streamable_prompt',
'arguments' => ['name' => 'StreamPromptUser', 'style' => 'formal']
], 'prompt-get-stream-1'));
expect($promptResponse['id'])->toBe('prompt-get-stream-1');
$messages = $promptResponse['result']['messages'];
expect($messages[0]['content']['text'])->toBe('Craft a formal greeting for StreamPromptUser.');
})->group('integration', 'streamable_http_stream');
it('rejects subsequent requests if client does not send initialized notification', function () {
await($this->streamClient->sendInitializeRequest([
'protocolVersion' => Protocol::LATEST_PROTOCOL_VERSION,
'clientInfo' => ['name' => 'StreamModeClient', 'version' => '1.0'],
'capabilities' => []
], 'init-stream-noack'));
$toolResponse = await($this->streamClient->sendRequest('tools/call', [
'name' => 'greet_streamable_tool',
'arguments' => ['name' => 'NoAckStreamUser']
], 'tool-stream-noack'));
expect($toolResponse['id'])->toBe('tool-stream-noack');
expect($toolResponse['error']['code'])->toBe(-32600);
expect($toolResponse['error']['message'])->toContain('Client session not initialized');
})->group('integration', 'streamable_http_stream');
it('returns HTTP 400 error for non-initialize requests without Mcp-Session-Id', function () {
await($this->streamClient->sendInitializeRequest([
'protocolVersion' => Protocol::LATEST_PROTOCOL_VERSION,
'clientInfo' => ['name' => 'StreamModeClient', 'version' => '1.0'],
'capabilities' => []
], 'init-stream-sess-test'));
$validSessionId = $this->streamClient->sessionId;
$this->streamClient->sessionId = null;
try {
await($this->streamClient->sendRequest('tools/list', [], 'tools-list-no-session-stream'));
$this->fail("Expected request to tools/list to fail with 400, but it succeeded.");
} catch (ResponseException $e) {
expect($e->getResponse()->getStatusCode())->toBe(400);
// Body can't be a json since the header accepts only text/event-stream
}
$this->streamClient->sessionId = $validSessionId;
})->group('integration', 'streamable_http_stream');
});
/**
* STATELESS MODE TESTS
*
* Tests for the stateless mode of StreamableHttpServerTransport, which:
* - Generates session IDs internally but doesn't expose them to clients
* - Doesn't require session IDs in requests after initialization
* - Doesn't include session IDs in response headers
* - Disables GET requests (SSE streaming)
* - Makes DELETE requests meaningless (but returns 204)
* - Treats each request as independent (no persistent session state)
*
* This mode is designed to work with clients like OpenAI's MCP implementation
* that have issues with session management in "never require approval" mode.
*/
describe('STATELESS MODE', function () {
beforeEach(function () {
$this->process = new Process($this->statelessModeCommand, getcwd() ?: null, null, []);
$this->process->start();
$this->statelessClient = new MockJsonHttpClient(STREAMABLE_HTTP_HOST, $this->port, STREAMABLE_MCP_PATH);
await(delay(0.2));
});
it('allows tool calls without having to send initialized notification', function () {
// 1. Initialize Request
$initResult = await($this->statelessClient->sendRequest('initialize', [
'protocolVersion' => Protocol::LATEST_PROTOCOL_VERSION,
'clientInfo' => ['name' => 'StatelessModeClient', 'version' => '1.0'],
'capabilities' => []
], 'init-stateless-1'));
expect($initResult['statusCode'])->toBe(200);
expect($initResult['body']['id'])->toBe('init-stateless-1');
expect($initResult['body'])->not->toHaveKey('error');
expect($initResult['body']['result']['protocolVersion'])->toBe(Protocol::LATEST_PROTOCOL_VERSION);
expect($initResult['body']['result']['serverInfo']['name'])->toBe('StreamableHttpIntegrationServer');
expect($this->statelessClient->sessionId)->toBeString()->toBeEmpty();
// 2. Call a tool
$toolResult = await($this->statelessClient->sendRequest('tools/call', [
'name' => 'greet_streamable_tool',
'arguments' => ['name' => 'Stateless Mode User']
], 'tool-stateless-1'));
expect($toolResult['statusCode'])->toBe(200);
expect($toolResult['body']['id'])->toBe('tool-stateless-1');
expect($toolResult['body'])->not->toHaveKey('error');
expect($toolResult['body']['result']['content'][0]['text'])->toBe('Hello, Stateless Mode User!');
})->group('integration', 'streamable_http_stateless');
it('return HTTP 400 error response for invalid JSON in POST request', function () {
$malformedJson = '{"jsonrpc":"2.0", "id": "bad-json-stateless-1", "method": "tools/list", "params": {"broken"}';
$postPromise = $this->statelessClient->browser->post(
$this->statelessClient->baseUrl,
['Content-Type' => 'application/json', 'Accept' => 'application/json'],
$malformedJson
);
try {
await(timeout($postPromise, STREAMABLE_HTTP_PROCESS_TIMEOUT - 2));
} catch (ResponseException $e) {
$httpResponse = $e->getResponse();
$bodyContent = (string) $httpResponse->getBody();
$decodedBody = json_decode($bodyContent, true);
expect($httpResponse->getStatusCode())->toBe(400);
expect($decodedBody['jsonrpc'])->toBe('2.0');
expect($decodedBody['id'])->toBe('');
expect($decodedBody['error']['code'])->toBe(-32700);
expect($decodedBody['error']['message'])->toContain('Invalid JSON');
}
})->group('integration', 'streamable_http_stateless');
it('returns JSON-RPC error result for request for non-existent method', function () {
$errorResult = await($this->statelessClient->sendRequest('non/existentToolViaStateless', [], 'err-meth-stateless-1'));
expect($errorResult['statusCode'])->toBe(200);
expect($errorResult['body']['id'])->toBe('err-meth-stateless-1');
expect($errorResult['body']['error']['code'])->toBe(-32601);
expect($errorResult['body']['error']['message'])->toContain("Method 'non/existentToolViaStateless' not found");
})->group('integration', 'streamable_http_stateless');
it('can handle batch requests correctly', function () {
$batchRequests = [
['jsonrpc' => '2.0', 'id' => 'batch-req-1', 'method' => 'tools/call', 'params' => ['name' => 'greet_streamable_tool', 'arguments' => ['name' => 'Batch Item 1']]],
['jsonrpc' => '2.0', 'method' => 'notifications/something'],
['jsonrpc' => '2.0', 'id' => 'batch-req-2', 'method' => 'tools/call', 'params' => ['name' => 'sum_streamable_tool', 'arguments' => ['a' => 10, 'b' => 20]]],
['jsonrpc' => '2.0', 'id' => 'batch-req-3', 'method' => 'nonexistent/method']
];
$batchResponse = await($this->statelessClient->sendBatchRequest($batchRequests));
$findResponseById = function (array $batch, $id) {
foreach ($batch as $item) {
if (isset($item['id']) && $item['id'] === $id) {
return $item;
}
}
return null;
};
expect($batchResponse['statusCode'])->toBe(200);
expect($batchResponse['body'])->toBeArray()->toHaveCount(3);
$response1 = $findResponseById($batchResponse['body'], 'batch-req-1');
$response2 = $findResponseById($batchResponse['body'], 'batch-req-2');
$response3 = $findResponseById($batchResponse['body'], 'batch-req-3');
expect($response1['result']['content'][0]['text'])->toBe('Hello, Batch Item 1!');
expect($response2['result']['content'][0]['text'])->toBe('30');
expect($response3['error']['code'])->toBe(-32601);
expect($response3['error']['message'])->toContain("Method 'nonexistent/method' not found");
})->group('integration', 'streamable_http_stateless');
it('passes request in Context', function () {
$toolResult = await($this->statelessClient->sendRequest('tools/call', [
'name' => 'tool_reads_context',
'arguments' => []
], 'tool-stateless-context-1', ['X-Test-Header' => 'TestValue']));
expect($toolResult['statusCode'])->toBe(200);
expect($toolResult['body']['id'])->toBe('tool-stateless-context-1');
expect($toolResult['body'])->not->toHaveKey('error');
expect($toolResult['body']['result']['content'][0]['text'])->toBe('TestValue');
})->group('integration', 'streamable_http_stateless');
it('can handle tool list request', function () {
$toolListResult = await($this->statelessClient->sendRequest('tools/list', [], 'tool-list-stateless-1'));
expect($toolListResult['statusCode'])->toBe(200);
expect($toolListResult['body']['id'])->toBe('tool-list-stateless-1');
expect($toolListResult['body'])->not->toHaveKey('error');
expect($toolListResult['body']['result']['tools'])->toBeArray();
expect(count($toolListResult['body']['result']['tools']))->toBe(4);
expect($toolListResult['body']['result']['tools'][0]['name'])->toBe('greet_streamable_tool');
expect($toolListResult['body']['result']['tools'][1]['name'])->toBe('sum_streamable_tool');
expect($toolListResult['body']['result']['tools'][2]['name'])->toBe('tool_reads_context');
})->group('integration', 'streamable_http_stateless');
it('can read a registered resource', function () {
$resourceResult = await($this->statelessClient->sendRequest('resources/read', ['uri' => 'test://streamable/static'], 'res-read-stateless-1'));
expect($resourceResult['statusCode'])->toBe(200);
expect($resourceResult['body']['id'])->toBe('res-read-stateless-1');
$contents = $resourceResult['body']['result']['contents'];
expect($contents[0]['uri'])->toBe('test://streamable/static');
expect($contents[0]['text'])->toBe(\PhpMcp\Server\Tests\Fixtures\General\ResourceHandlerFixture::$staticTextContent);
})->group('integration', 'streamable_http_stateless');
it('can get a registered prompt', function () {
$promptResult = await($this->statelessClient->sendRequest('prompts/get', [
'name' => 'simple_streamable_prompt',
'arguments' => ['name' => 'StatelessPromptUser', 'style' => 'formal']
], 'prompt-get-stateless-1'));
expect($promptResult['statusCode'])->toBe(200);
expect($promptResult['body']['id'])->toBe('prompt-get-stateless-1');
$messages = $promptResult['body']['result']['messages'];
expect($messages[0]['content']['text'])->toBe('Craft a formal greeting for StatelessPromptUser.');
})->group('integration', 'streamable_http_stateless');
it('does not return session ID in response headers in stateless mode', function () {
$promise = $this->statelessClient->browser->post(
$this->statelessClient->baseUrl,
['Content-Type' => 'application/json', 'Accept' => 'application/json'],
json_encode([
'jsonrpc' => '2.0',
'id' => 'init-header-test',
'method' => 'initialize',
'params' => [
'protocolVersion' => Protocol::LATEST_PROTOCOL_VERSION,
'clientInfo' => ['name' => 'StatelessHeaderTest', 'version' => '1.0'],
'capabilities' => []
]
])
);
$response = await(timeout($promise, STREAMABLE_HTTP_PROCESS_TIMEOUT - 2));
expect($response->getStatusCode())->toBe(200);
expect($response->hasHeader('Mcp-Session-Id'))->toBeFalse();
$body = json_decode((string) $response->getBody(), true);
expect($body['id'])->toBe('init-header-test');
expect($body)->not->toHaveKey('error');
})->group('integration', 'streamable_http_stateless');
it('returns HTTP 405 for GET requests (SSE disabled) in stateless mode', function () {
try {
$getPromise = $this->statelessClient->browser->get(
$this->statelessClient->baseUrl,
['Accept' => 'text/event-stream']
);
await(timeout($getPromise, STREAMABLE_HTTP_PROCESS_TIMEOUT - 2));
$this->fail("Expected GET request to fail with 405, but it succeeded.");
} catch (ResponseException $e) {
expect($e->getResponse()->getStatusCode())->toBe(405);
$bodyContent = (string) $e->getResponse()->getBody();
$decodedBody = json_decode($bodyContent, true);
expect($decodedBody['error']['message'])->toContain('GET requests (SSE streaming) are not supported in stateless mode');
}
})->group('integration', 'streamable_http_stateless');
it('returns 204 for DELETE requests in stateless mode (but they are meaningless)', function () {
$deletePromise = $this->statelessClient->browser->delete($this->statelessClient->baseUrl);
$response = await(timeout($deletePromise, STREAMABLE_HTTP_PROCESS_TIMEOUT - 2));
expect($response->getStatusCode())->toBe(204);
expect((string) $response->getBody())->toBeEmpty();
})->group('integration', 'streamable_http_stateless');
it('handles multiple independent tool calls in stateless mode', function () {
$toolResult1 = await($this->statelessClient->sendRequest('tools/call', [
'name' => 'greet_streamable_tool',
'arguments' => ['name' => 'User 1']
], 'tool-multi-1'));
$toolResult2 = await($this->statelessClient->sendRequest('tools/call', [
'name' => 'sum_streamable_tool',
'arguments' => ['a' => 5, 'b' => 10]
], 'tool-multi-2'));
$toolResult3 = await($this->statelessClient->sendRequest('tools/call', [
'name' => 'greet_streamable_tool',
'arguments' => ['name' => 'User 3']
], 'tool-multi-3'));
expect($toolResult1['statusCode'])->toBe(200);
expect($toolResult1['body']['id'])->toBe('tool-multi-1');
expect($toolResult1['body']['result']['content'][0]['text'])->toBe('Hello, User 1!');
expect($toolResult2['statusCode'])->toBe(200);
expect($toolResult2['body']['id'])->toBe('tool-multi-2');
expect($toolResult2['body']['result']['content'][0]['text'])->toBe('15');
expect($toolResult3['statusCode'])->toBe(200);
expect($toolResult3['body']['id'])->toBe('tool-multi-3');
expect($toolResult3['body']['result']['content'][0]['text'])->toBe('Hello, User 3!');
})->group('integration', 'streamable_http_stateless');
});
it('responds to OPTIONS request with CORS headers', function () {
$this->process = new Process($this->jsonModeCommand, getcwd() ?: null, null, []);
$this->process->start();
$this->jsonClient = new MockJsonHttpClient(STREAMABLE_HTTP_HOST, $this->port, STREAMABLE_MCP_PATH);
await(delay(0.1));
$browser = new Browser();
$optionsUrl = $this->jsonClient->baseUrl;
$promise = $browser->request('OPTIONS', $optionsUrl);
$response = await(timeout($promise, STREAMABLE_HTTP_PROCESS_TIMEOUT - 2));
expect($response->getStatusCode())->toBe(204);
expect($response->getHeaderLine('Access-Control-Allow-Origin'))->toBe('*');
expect($response->getHeaderLine('Access-Control-Allow-Methods'))->toContain('POST');
expect($response->getHeaderLine('Access-Control-Allow-Methods'))->toContain('GET');
expect($response->getHeaderLine('Access-Control-Allow-Headers'))->toContain('Mcp-Session-Id');
})->group('integration', 'streamable_http');
it('returns 404 for unknown paths', function () {
$this->process = new Process($this->jsonModeCommand, getcwd() ?: null, null, []);
$this->process->start();
$this->jsonClient = new MockJsonHttpClient(STREAMABLE_HTTP_HOST, $this->port, STREAMABLE_MCP_PATH);
await(delay(0.1));
$browser = new Browser();
$unknownUrl = "http://" . STREAMABLE_HTTP_HOST . ":" . $this->port . "/completely/unknown/path";
$promise = $browser->get($unknownUrl);
try {
await(timeout($promise, STREAMABLE_HTTP_PROCESS_TIMEOUT - 2));
$this->fail("Request to unknown path should have failed with 404.");
} catch (ResponseException $e) {
expect($e->getResponse()->getStatusCode())->toBe(404);
$decodedBody = json_decode((string)$e->getResponse()->getBody(), true);
expect($decodedBody['error']['message'])->toContain('Not found');
}
})->group('integration', 'streamable_http');
it('can delete client session with DELETE request', function () {
$this->process = new Process($this->jsonModeCommand, getcwd() ?: null, null, []);
$this->process->start();
$this->jsonClient = new MockJsonHttpClient(STREAMABLE_HTTP_HOST, $this->port, STREAMABLE_MCP_PATH);
await(delay(0.1));
// 1. Initialize
await($this->jsonClient->sendRequest('initialize', ['protocolVersion' => Protocol::LATEST_PROTOCOL_VERSION, 'clientInfo' => ['name' => 'JsonModeClient', 'version' => '1.0'], 'capabilities' => []], 'init-delete-test'));
$sessionIdForDelete = $this->jsonClient->sessionId;
expect($sessionIdForDelete)->toBeString();
await($this->jsonClient->sendNotification('notifications/initialized'));
// 2. Establish a GET SSE connection
$sseUrl = $this->jsonClient->baseUrl;
$browserForSse = (new Browser())->withTimeout(3);
$ssePromise = $browserForSse->requestStreaming('GET', $sseUrl, [
'Accept' => 'text/event-stream',
'Mcp-Session-Id' => $sessionIdForDelete
]);
$ssePsrResponse = await(timeout($ssePromise, 3));
expect($ssePsrResponse->getStatusCode())->toBe(200);
expect($ssePsrResponse->getHeaderLine('Content-Type'))->toBe('text/event-stream');
$sseStream = $ssePsrResponse->getBody();
assert($sseStream instanceof ReadableStreamInterface);
$isSseStreamClosed = false;
$sseStream->on('close', function () use (&$isSseStreamClosed) {
$isSseStreamClosed = true;
});
// 3. Send DELETE request
$deleteResponse = await($this->jsonClient->sendDeleteRequest());
expect($deleteResponse['statusCode'])->toBe(204);
// 4. Assert that the GET SSE stream was closed
await(delay(0.1));
expect($isSseStreamClosed)->toBeTrue("The GET SSE stream for session {$sessionIdForDelete} was not closed after DELETE request.");
// 5. Assert that the client session was deleted
try {
await($this->jsonClient->sendRequest('tools/list', [], 'tool-list-json-1'));
$this->fail("Expected request to tools/list to fail with 400, but it succeeded.");
} catch (ResponseException $e) {
expect($e->getResponse()->getStatusCode())->toBe(404);
$bodyContent = (string) $e->getResponse()->getBody();
$decodedBody = json_decode($bodyContent, true);
expect($decodedBody['error']['code'])->toBe(-32600);
expect($decodedBody['error']['message'])->toContain('Invalid or expired session');
}
})->group('integration', 'streamable_http_json');
it('executes middleware that adds headers to response', function () {
$this->process = new Process($this->jsonModeCommand, getcwd() ?: null, null, []);
$this->process->start();
$this->jsonClient = new MockJsonHttpClient(STREAMABLE_HTTP_HOST, $this->port, STREAMABLE_MCP_PATH);
await(delay(0.1));
// 1. Send a request and check that middleware-added header is present
$response = await($this->jsonClient->sendRequest('initialize', [
'protocolVersion' => Protocol::LATEST_PROTOCOL_VERSION,
'clientInfo' => ['name' => 'MiddlewareTestClient'],
'capabilities' => []
], 'init-middleware-headers'));
// Check that the response has the header added by middleware
expect($this->jsonClient->lastResponseHeaders)->toContain('X-Test-Middleware: header-added');
})->group('integration', 'streamable_http', 'middleware');
it('executes middleware that modifies request attributes', function () {
$this->process = new Process($this->jsonModeCommand, getcwd() ?: null, null, []);
$this->process->start();
$this->jsonClient = new MockJsonHttpClient(STREAMABLE_HTTP_HOST, $this->port, STREAMABLE_MCP_PATH);
await(delay(0.1));
// 1. Initialize
await($this->jsonClient->sendRequest('initialize', [
'protocolVersion' => Protocol::LATEST_PROTOCOL_VERSION,
'clientInfo' => ['name' => 'MiddlewareAttrTestClient', 'version' => '1.0'],
'capabilities' => []
], 'init-middleware-attr'));
await($this->jsonClient->sendNotification('notifications/initialized'));
// 2. Call tool that checks for middleware-added attribute
$toolResponse = await($this->jsonClient->sendRequest('tools/call', [
'name' => 'check_request_attribute_tool',
'arguments' => []
], 'tool-attr-check'));
expect($toolResponse['body']['result']['content'][0]['text'])->toBe('middleware-value-found: middleware-value');
})->group('integration', 'streamable_http', 'middleware');
it('executes middleware that can short-circuit request processing', function () {
$this->process = new Process($this->jsonModeCommand, getcwd() ?: null, null, []);
$this->process->start();
await(delay(0.1));
$browser = new Browser();
$shortCircuitUrl = "http://" . STREAMABLE_HTTP_HOST . ":" . $this->port . "/" . STREAMABLE_MCP_PATH . "/short-circuit";
$promise = $browser->get($shortCircuitUrl);
try {
$response = await(timeout($promise, STREAMABLE_HTTP_PROCESS_TIMEOUT - 2));
$this->fail("Expected a 418 status code response, but request succeeded");
} catch (ResponseException $e) {
expect($e->getResponse()->getStatusCode())->toBe(418);
$body = (string) $e->getResponse()->getBody();
expect($body)->toBe('Short-circuited by middleware');
} catch (\Throwable $e) {
$this->fail("Short-circuit middleware test failed: " . $e->getMessage());
}
})->group('integration', 'streamable_http', 'middleware');
it('executes multiple middlewares in correct order', function () {
$this->process = new Process($this->jsonModeCommand, getcwd() ?: null, null, []);
$this->process->start();
$this->jsonClient = new MockJsonHttpClient(STREAMABLE_HTTP_HOST, $this->port, STREAMABLE_MCP_PATH);
await(delay(0.1));
// 1. Send a request and check middleware order
await($this->jsonClient->sendRequest('initialize', [
'protocolVersion' => Protocol::LATEST_PROTOCOL_VERSION,
'clientInfo' => ['name' => 'MiddlewareOrderTestClient'],
'capabilities' => []
], 'init-middleware-order'));
// Check that headers from multiple middlewares are present in correct order
expect($this->jsonClient->lastResponseHeaders)->toContain('X-Middleware-Order: third,second,first');
})->group('integration', 'streamable_http', 'middleware');
it('handles middleware that throws exceptions gracefully', function () {
$this->process = new Process($this->jsonModeCommand, getcwd() ?: null, null, []);
$this->process->start();
await(delay(0.1));
$browser = new Browser();
$errorUrl = "http://" . STREAMABLE_HTTP_HOST . ":" . $this->port . "/" . STREAMABLE_MCP_PATH . "/error-middleware";
$promise = $browser->get($errorUrl);
try {
await(timeout($promise, STREAMABLE_HTTP_PROCESS_TIMEOUT - 2));
$this->fail("Error middleware should have thrown an exception.");
} catch (ResponseException $e) {
expect($e->getResponse()->getStatusCode())->toBe(500);
$body = (string) $e->getResponse()->getBody();
// ReactPHP handles exceptions and returns a generic error message
expect($body)->toContain('Internal Server Error');
}
})->group('integration', 'streamable_http', 'middleware');
```