Skip to content

Feat: Add stateless mode for StreamableHttpServerTransport #48

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 2 commits into from
Jul 12, 2025
Merged

Feat: Add stateless mode for StreamableHttpServerTransport #48

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
d1fc63e
feat: add stateless mode to StreamableHttpServerTransport
CodeWithKyrian Jul 12, 2025
fe5eac2
docs: add stateless mode documentation to README
CodeWithKyrian Jul 12, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 17 additions & 1 deletion README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -524,7 +524,8 @@ $transport = new StreamableHttpServerTransport(
host: '127.0.0.1', // MCP protocol prohibits 0.0.0.0
port: 8080,
mcpPathPrefix: 'mcp',
enableJsonResponse: false // Use SSE streaming (default)
enableJsonResponse: false, // Use SSE streaming (default)
stateless: false // Enable stateless mode for session-less clients
);

$server->listen($transport);
Expand All @@ -546,12 +547,27 @@ $transport = new StreamableHttpServerTransport(
);
```

**Stateless Mode:**

For clients that have issues with session management, enable stateless mode:

```php
$transport = new StreamableHttpServerTransport(
host: '127.0.0.1',
port: 8080,
stateless: true // Each request is independent
);
```

In stateless mode, session IDs are generated internally but not exposed to clients, and each request is treated as independent without persistent session state.

**Features:**
- **Resumable connections** - clients can reconnect and replay missed events
- **Event sourcing** - all events are stored for replay
- **JSON mode** - optional JSON-only responses for fast tools
- **Enhanced session management** - persistent session state
- **Multiple client support** - designed for concurrent clients
- **Stateless mode** - session-less operation for simple clients

## 📋 Schema Generation and Validation

Expand Down
6 changes: 6 additions & 0 deletions src/Protocol.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -134,6 +134,12 @@ public function processMessage(Request|Notification|BatchRequest $message, strin
return;
}

if ($context['stateless'] ?? false) {
$session->set('initialized', true);
$session->set('protocol_version', self::LATEST_PROTOCOL_VERSION);
$session->set('client_info', ['name' => 'stateless-client', 'version' => '1.0.0']);
}

$response = null;

if ($message instanceof BatchRequest) {
Expand Down
68 changes: 49 additions & 19 deletions src/Transports/StreamableHttpServerTransport.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -75,6 +75,7 @@ public function __construct(
private string $mcpPath = '/mcp',
private ?array $sslContext = null,
private readonly bool $enableJsonResponse = true,
private readonly bool $stateless = false,
?EventStoreInterface $eventStore = null
) {
$this->logger = new NullLogger();
Expand Down Expand Up @@ -171,9 +172,9 @@ private function createRequestHandler(): callable

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))),
'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) {
Expand All @@ -184,6 +185,11 @@ private function createRequestHandler(): callable

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.");
Expand Down Expand Up @@ -264,24 +270,29 @@ private function handlePostRequest(ServerRequestInterface $request): PromiseInte
$isInitializeRequest = ($message instanceof Request && $message->method === 'initialize');
$sessionId = null;

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();
}

if ($this->stateless) {
$sessionId = $this->generateId();
$this->emit('client_connected', [$sessionId]);
} else {
$sessionId = $request->getHeaderLine('Mcp-Session-Id');
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();
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();
}
}
}

Expand Down Expand Up @@ -344,7 +355,7 @@ private function handlePostRequest(ServerRequestInterface $request): PromiseInte
'X-Accel-Buffering' => 'no',
];

if (!empty($sessionId)) {
if (!empty($sessionId) && !$this->stateless) {
$headers['Mcp-Session-Id'] = $sessionId;
}

Expand All @@ -355,6 +366,8 @@ private function handlePostRequest(ServerRequestInterface $request): PromiseInte
}
}

$context['stateless'] = $this->stateless;

$this->loop->futureTick(function () use ($message, $sessionId, $context) {
$this->emit('message', [$message, $sessionId, $context]);
});
Expand All @@ -364,6 +377,10 @@ private function handlePostRequest(ServerRequestInterface $request): PromiseInte

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.");
Expand Down Expand Up @@ -466,6 +483,12 @@ public function sendMessage(Message $message, string $sessionId, array $context
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']);
});
}
}
}

Expand All @@ -483,12 +506,19 @@ public function sendMessage(Message $message, string $sessionId, array $context

$responseBody = json_encode($message, JSON_UNESCAPED_SLASHES | JSON_UNESCAPED_UNICODE);
$headers = ['Content-Type' => 'application/json'];
if ($isInitializeResponse) {
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:
Expand Down
4 changes: 3 additions & 1 deletion tests/Fixtures/ServerScripts/StreamableHttpTestServer.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,10 +27,11 @@ public function log($level, \Stringable|string $message, array $context = []): v
$mcpPath = $argv[3] ?? 'mcp_streamable_test';
$enableJsonResponse = filter_var($argv[4] ?? 'true', FILTER_VALIDATE_BOOLEAN);
$useEventStore = filter_var($argv[5] ?? 'false', FILTER_VALIDATE_BOOLEAN);
$stateless = filter_var($argv[6] ?? 'false', FILTER_VALIDATE_BOOLEAN);

try {
$logger = new NullLogger();
$logger->info("Starting StreamableHttpTestServer on {$host}:{$port}/{$mcpPath}, JSON Mode: " . ($enableJsonResponse ? 'ON' : 'OFF'));
$logger->info("Starting StreamableHttpTestServer on {$host}:{$port}/{$mcpPath}, JSON Mode: " . ($enableJsonResponse ? 'ON' : 'OFF') . ", Stateless: " . ($stateless ? 'ON' : 'OFF'));

$eventStore = $useEventStore ? new InMemoryEventStore() : null;

Expand All @@ -48,6 +49,7 @@ public function log($level, \Stringable|string $message, array $context = []): v
port: $port,
mcpPath: $mcpPath,
enableJsonResponse: $enableJsonResponse,
stateless: $stateless,
eventStore: $eventStore
);

Expand Down
Loading