Skip to content

Commit 9f99a94

Browse files
CodeWithKyrianCodeWithKyrian
committed
feat: Enhance session management and transport handling
- Added a lottery configuration for session garbage collection in `mcp.php`. - Updated `LaravelHttpTransport` to include session garbage collection on message requests. - Refactored `LaravelStreamableHttpTransport` to streamline message handling and improve context management. - Adjusted command output formatting for better readability when starting the MCP server.
1 parent 944d78c commit 9f99a94

File tree

6 files changed

+71
-86
lines changed

6 files changed

+71
-86
lines changed

‎config/mcp.php‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -115,6 +115,7 @@
115115
'session' => [
116116
'driver' => env('MCP_SESSION_DRIVER', 'cache'), // 'array' or 'cache'
117117
'ttl' => (int) env('MCP_SESSION_TTL', 3600), // Session lifetime in seconds
118+
'lottery' => [2, 100], // 2% chance of garbage collection
118119
],
119120

120121
/*

‎samples/basic/bootstrap/app.php‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -6,13 +6,13 @@
66

77
return Application::configure(basePath: dirname(__DIR__))
88
->withRouting(
9-
web: __DIR__.'/../routes/web.php',
10-
commands: __DIR__.'/../routes/console.php',
9+
web: __DIR__ . '/../routes/web.php',
10+
commands: __DIR__ . '/../routes/console.php',
1111
health: '/up',
1212
)
1313
->withMiddleware(function (Middleware $middleware) {
14-
//
1514
$middleware->validateCsrfTokens(except: [
15+
'mcp',
1616
'mcp/*',
1717
]);
1818
})

‎samples/basic/config/mcp.php‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -99,7 +99,7 @@
9999
'domain' => env('MCP_HTTP_INTEGRATED_DOMAIN'),
100100
'sse_poll_interval' => (int) env('MCP_HTTP_INTEGRATED_SSE_POLL_SECONDS', 1),
101101
'cors_origin' => env('MCP_HTTP_INTEGRATED_CORS_ORIGIN', '*'),
102-
'enable_json_response' => (bool) env('MCP_HTTP_INTEGRATED_JSON_RESPONSE', true),
102+
'enable_json_response' => (bool) env('MCP_HTTP_INTEGRATED_JSON_RESPONSE', false),
103103
'json_response_timeout' => (int) env('MCP_HTTP_INTEGRATED_JSON_TIMEOUT', 30),
104104
'event_store' => env('MCP_HTTP_INTEGRATED_EVENT_STORE'), // FQCN or null
105105
],

‎src/Commands/ServeCommand.php‎

Lines changed: 3 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -81,9 +81,9 @@ private function handleStdioTransport(Server $server): int
8181
}
8282

8383
$this->info('Starting MCP server');
84-
$this->line(" \t- Transport: STDIO");
85-
$this->line(" \t- Communication: STDIN/STDOUT");
86-
$this->line(" \t- Mode: JSON-RPC over Standard I/O");
84+
$this->line(" - Transport: STDIO");
85+
$this->line(" - Communication: STDIN/STDOUT");
86+
$this->line(" - Mode: JSON-RPC over Standard I/O");
8787

8888
try {
8989
$transport = new StdioServerTransport;
@@ -139,8 +139,6 @@ private function handleSseHttpTransport(Server $server, string $host, int $port,
139139
return Command::FAILURE;
140140
}
141141

142-
$this->info("MCP Server (Legacy HTTP) stopped.");
143-
144142
return Command::SUCCESS;
145143
}
146144

@@ -171,8 +169,6 @@ private function handleStreamableHttpTransport(Server $server, string $host, int
171169
return Command::FAILURE;
172170
}
173171

174-
$this->info("MCP Server (Streamable HTTP) stopped.");
175-
176172
return Command::SUCCESS;
177173
}
178174

‎src/Transports/LaravelHttpTransport.php‎

Lines changed: 12 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -74,6 +74,8 @@ public function sendMessage(Message $message, string $sessionId, array $context
7474
*/
7575
public function handleMessageRequest(Request $request): Response
7676
{
77+
$this->collectSessionGarbage();
78+
7779
if (!$request->isJson()) {
7880
Log::warning('Received POST request with invalid Content-Type');
7981

@@ -201,14 +203,23 @@ private function sendSseEvent(string $event, string $data, ?string $id = null):
201203
/**
202204
* Flush output buffer
203205
*/
204-
private function flushOutput(): void
206+
protected function flushOutput(): void
205207
{
206208
if (function_exists('ob_flush')) {
207209
@ob_flush();
208210
}
209211
@flush();
210212
}
211213

214+
protected function collectSessionGarbage(): void
215+
{
216+
$lottery = config('mcp.session.lottery', [2, 100]);
217+
218+
if (random_int(1, $lottery[1]) <= $lottery[0]) {
219+
$this->sessionManager->gc();
220+
}
221+
}
222+
212223
/**
213224
* 'Closes' the transport.
214225
*/

‎src/Transports/LaravelStreamableHttpTransport.php‎

Lines changed: 51 additions & 74 deletions
Original file line numberDiff line numberDiff line change
@@ -30,14 +30,7 @@ class LaravelStreamableHttpTransport implements ServerTransportInterface
3030
public function __construct(
3131
protected SessionManager $sessionManager,
3232
protected ?EventStoreInterface $eventStore = null
33-
) {
34-
$this->on('message', function (Message $message, string $sessionId) {
35-
$session = $this->sessionManager->getSession($sessionId);
36-
if ($session !== null) {
37-
$session->save();
38-
}
39-
});
40-
}
33+
) {}
4134

4235
protected function generateId(): string
4336
{
@@ -59,14 +52,14 @@ public function sendMessage(Message $message, string $sessionId, array $context
5952

6053
$eventId = null;
6154
if ($this->eventStore && isset($context['type']) && in_array($context['type'], ['get_sse', 'post_sse'])) {
62-
$streamKey = $context['type'] === 'get_sse' ? "get_stream_{$sessionId}" : $context['streamId'] ?? "post_stream_{$sessionId}";
63-
$eventId = $this->eventStore->storeEvent($streamKey, $rawMessage);
55+
$streamId = $context['streamId'];
56+
$eventId = $this->eventStore->storeEvent($streamId, $rawMessage);
6457
}
6558

6659
$messageData = [
6760
'id' => $eventId ?? $this->generateId(),
6861
'data' => $rawMessage,
69-
'context' => $context['type'] ?? 'get_sse',
62+
'context' => $context,
7063
'timestamp' => time()
7164
];
7265

@@ -161,35 +154,26 @@ protected function handleJsonResponse(Message $message, string $sessionId, array
161154
$context['type'] = 'post_json';
162155
$this->emit('message', [$message, $sessionId, $context]);
163156

164-
$maxWaitTime = config('mcp.transports.http_integrated.json_response_timeout', 30);
165-
$pollInterval = 0.1; // 100ms
166-
$waitedTime = 0;
167-
168-
while ($waitedTime < $maxWaitTime) {
169-
$messages = $this->dequeueMessagesForContext($sessionId, 'post_json');
170-
171-
if (!empty($messages)) {
172-
$responseMessage = $messages[0];
173-
$data = $responseMessage['data'];
157+
$messages = $this->dequeueMessagesForContext($sessionId, 'post_json');
174158

175-
$headers = [
176-
'Content-Type' => 'application/json',
177-
...$this->getCorsHeaders()
178-
];
159+
if (empty($messages)) {
160+
$error = Error::forInternalError('Internal error');
161+
return response()->json($error, 500, $this->getCorsHeaders());
162+
}
179163

180-
if ($context['is_initialize_request'] ?? false) {
181-
$headers['Mcp-Session-Id'] = $sessionId;
182-
}
164+
$responseMessage = $messages[0];
165+
$data = $responseMessage['data'];
183166

184-
return response()->make($data, 200, $headers);
185-
}
167+
$headers = [
168+
'Content-Type' => 'application/json',
169+
...$this->getCorsHeaders()
170+
];
186171

187-
usleep((int)($pollInterval * 1000000));
188-
$waitedTime += $pollInterval;
172+
if ($context['is_initialize_request'] ?? false) {
173+
$headers['Mcp-Session-Id'] = $sessionId;
189174
}
190175

191-
$error = Error::forInternalError('Request timeout');
192-
return response()->json($error, 504, $this->getCorsHeaders());
176+
return response()->make($data, 200, $headers);
193177
} catch (Throwable $e) {
194178
Log::error('JSON response mode error', ['exception' => $e]);
195179
$error = Error::forInternalError('Internal error');
@@ -202,46 +186,29 @@ protected function handleJsonResponse(Message $message, string $sessionId, array
202186
*/
203187
protected function handleSseResponse(Message $message, string $sessionId, int $nRequests, array $context): StreamedResponse
204188
{
205-
$streamId = $this->generateId();
206-
$context['type'] = 'post_sse';
207-
$context['streamId'] = $streamId;
208-
$context['nRequests'] = $nRequests;
209-
210-
$this->emit('message', [$message, $sessionId, $context]);
211-
212-
return response()->stream(function () use ($sessionId, $nRequests, $streamId) {
213-
$responsesSent = 0;
214-
$maxWaitTime = 30; // 30 seconds timeout
215-
$pollInterval = 0.1; // 100ms
216-
$waitedTime = 0;
189+
$headers = array_merge([
190+
'Content-Type' => 'text/event-stream',
191+
'Cache-Control' => 'no-cache',
192+
'Connection' => 'keep-alive',
193+
'X-Accel-Buffering' => 'no',
194+
], $this->getCorsHeaders());
217195

218-
while ($responsesSent < $nRequests && $waitedTime < $maxWaitTime) {
219-
if (connection_aborted()) {
220-
break;
221-
}
196+
if ($context['is_initialize_request'] ?? false) {
197+
$headers['Mcp-Session-Id'] = $sessionId;
198+
}
222199

223-
$messages = $this->dequeueMessagesForContext($sessionId, 'post_sse', $streamId);
200+
return response()->stream(function () use ($sessionId, $nRequests, $message, $context) {
201+
$streamId = $this->generateId();
202+
$context['type'] = 'post_sse';
203+
$context['streamId'] = $streamId;
204+
$context['nRequests'] = $nRequests;
224205

225-
foreach ($messages as $messageData) {
226-
$this->sendSseEvent($messageData['data'], $messageData['id']);
227-
$responsesSent++;
206+
$this->emit('message', [$message, $sessionId, $context]);
228207

229-
if ($responsesSent >= $nRequests) {
230-
break;
231-
}
232-
}
208+
$messages = $this->dequeueMessagesForContext($sessionId, 'post_sse', $streamId);
233209

234-
if ($responsesSent < $nRequests) {
235-
usleep((int)($pollInterval * 1000000));
236-
$waitedTime += $pollInterval;
237-
}
238-
}
239-
}, headers: array_merge([
240-
'Content-Type' => 'text/event-stream',
241-
'Cache-Control' => 'no-cache',
242-
'Connection' => 'keep-alive',
243-
'X-Accel-Buffering' => 'no',
244-
], $this->getCorsHeaders()));
210+
$this->sendSseEvent($messages[0]['data'], $messages[0]['id']);
211+
}, headers: $headers);
245212
}
246213

247214
/**
@@ -325,20 +292,21 @@ public function handleDeleteRequest(Request $request): Response
325292
/**
326293
* Dequeue messages for specific context, requeue others
327294
*/
328-
protected function dequeueMessagesForContext(string $sessionId, string $context, ?string $streamId = null): array
295+
protected function dequeueMessagesForContext(string $sessionId, string $type, ?string $streamId = null): array
329296
{
330297
$allMessages = $this->sessionManager->dequeueMessages($sessionId);
331298
$contextMessages = [];
332299
$requeueMessages = [];
333300

334301
foreach ($allMessages as $rawMessage) {
335302
$messageData = json_decode($rawMessage, true);
303+
$context = $messageData['context'] ?? [];
336304

337-
if ($messageData && isset($messageData['context'])) {
338-
$matchesContext = $messageData['context'] === $context;
305+
if ($messageData) {
306+
$matchesContext = $context['type'] === $type;
339307

340-
if ($context === 'post_sse' && $streamId) {
341-
$matchesContext = $matchesContext && isset($messageData['streamId']) && $messageData['streamId'] === $streamId;
308+
if ($type === 'post_sse' && $streamId) {
309+
$matchesContext = $matchesContext && isset($context['streamId']) && $context['streamId'] === $streamId;
342310
}
343311

344312
if ($matchesContext) {
@@ -412,6 +380,15 @@ private function flushOutput(): void
412380
@flush();
413381
}
414382

383+
protected function collectSessionGarbage(): void
384+
{
385+
$lottery = config('mcp.session.lottery', [2, 100]);
386+
387+
if (random_int(1, $lottery[1]) <= $lottery[0]) {
388+
$this->sessionManager->gc();
389+
}
390+
}
391+
415392
/**
416393
* Get CORS headers
417394
*/

0 commit comments

Comments
 (0)