Reusable SseService for both transports and session manager

Author: fsmonter

src/Http/Controllers/McpController.php

@@ -12,7 +12,10 @@
 
 class McpController extends Controller
 {
-    public function __construct(protected McpHandler $mcpHandler, protected SseService $sseService) {}
+    public function __construct(
+        protected McpHandler $mcpHandler,
+        protected SseService $sse
+    ) {}
 
     /**
      * Handle MCP requests.
@@ -57,14 +60,27 @@ public function __invoke(Request $request): JsonResponse|StreamedResponse|Respon
      *
      * @param  array<array-key, mixed>  $messages  JSON-RPC messages from the client
      */
-    protected function handlePostSseResponse(array $messages): StreamedResponse
-    {
-        $response = $this->sseService->createPostSseResponse(
-            $messages,
-            fn ($message) => $this->mcpHandler->handle($message),
-        );
+    protected function handlePostSseResponse(
+        array $messages
+    ): StreamedResponse {
+        return $this->sse->createSseResponse(function () use ($messages) {
+            $this->sse->sendComment('stream opened');
+
+            if (! isset($messages[0])) {
+                $messages = [$messages];
+            }
+
+            foreach ($messages as $message) {
+                if (! isset($message['id'])) {
+                    logger('Missing ID in message');
 
-        return $response;
+                    continue;
+                }
+
+                $response = $this->mcpHandler->handle($message);
+                $this->sse->sendEvent($response);
+            }
+        });
     }
 
     /**

src/Http/Controllers/McpSSEController.php

@@ -7,82 +7,55 @@
 use Illuminate\Routing\Controller;
 use Illuminate\Support\Facades\Log;
 use Kirschbaum\Loop\McpHandler;
-use Kirschbaum\Loop\Services\McpSseService;
+use Kirschbaum\Loop\Services\SseService;
+use Kirschbaum\Loop\Services\SseSessionManager;
 use Symfony\Component\HttpFoundation\StreamedResponse;
 
 class McpSSEController extends Controller
 {
     public function __construct(
-        protected McpHandler $mcpHandler
+        protected McpHandler $mcpHandler,
+        protected SseService $sse,
+        protected SseSessionManager $sessionManager
     ) {}
 
     /**
      * Handle SSE connection setup (GET request), creates server-client event stream.
      */
     public function connect(Request $request): StreamedResponse
     {
-        $headers = [
-            'Content-Type' => 'text/event-stream',
-            'Cache-Control' => 'no-cache, no-store, max-age=0, must-revalidate',
-            'Pragma' => 'no-cache',
-            'X-Accel-Buffering' => 'no',
-            'Connection' => 'keep-alive',
-        ];
-
         $sessionId = uniqid();
-
         $ssePath = config('loop.sse.path', '/mcp/sse');
         $postEndpoint = $ssePath.'/message/?sessionId='.$sessionId;
 
-        set_time_limit(0);
-        ignore_user_abort(false); // Detects client disconnects
-
-        return response()->stream(function () use ($sessionId, $postEndpoint) {
-            while (ob_get_level() > 0) {
-                ob_end_clean();
-            }
-            ob_implicit_flush(true);
-
-            McpSseService::registerConnection($sessionId);
-
-            echo "event: endpoint\n";
-            echo "data: {$postEndpoint}\n\n";
-            flush();
+        return $this->sse->createSseResponse(function () use ($sessionId, $postEndpoint) {
+            $this->sessionManager->registerConnection($sessionId);
 
-            $debugTime = time();
-            echo "event: debug\n";
-            echo "data: {\"message\":\"Connection established\",\"time\":{$debugTime}}\n\n";
-            flush();
+            $this->sse->sendEvent($postEndpoint, 'endpoint');
+            $this->sse->sendDebug('Connection established');
 
-            $heartbeatInterval = 30; // seconds
+            $heartbeatInterval = 30;
             $lastHeartbeat = time();
             $lastMessageId = -1;
 
             register_shutdown_function(function () use ($sessionId) {
-                McpSseService::removeConnection($sessionId);
+                $this->sessionManager->removeConnection($sessionId);
             });
 
             $connectionStatus = connection_status();
 
             while ($connectionStatus === CONNECTION_NORMAL) {
-                $messages = McpSseService::getMessages($sessionId, $lastMessageId);
+                $messages = $this->sessionManager->getMessages($sessionId, $lastMessageId);
 
                 foreach ($messages as $message) {
                     $lastMessageId = $message['id'];
 
-                    logger(json_encode($message));
-
-                    echo "event: message\n";
-                    echo 'data: '.json_encode($message['data'])."\n\n";
-                    flush();
+                    $this->sse->sendEvent($message['data']);
                 }
 
                 if ((time() - $lastHeartbeat) >= $heartbeatInterval) {
-                    $currentTime = time();
-                    echo "event: ping\n";
-                    echo "data: {$currentTime}\n\n";
-                    flush();
-                    $lastHeartbeat = $currentTime;
+                    $this->sse->sendPing();
+                    $lastHeartbeat = time();
                 }
 
                 $oldStatus = $connectionStatus;
@@ -91,20 +64,19 @@ public function connect(Request $request): StreamedResponse
                 usleep(100000);
             }
 
-            McpSseService::removeConnection($sessionId);
-        }, 200, $headers);
+            $this->sessionManager->removeConnection($sessionId);
+        });
     }
 
     /**
      * Handle POST messages from client
      */
     public function message(Request $request): JsonResponse
     {
-
         $requestData = $request->all();
         $sessionId = $request->query('sessionId');
 
-        if (! McpSseService::sessionExists($sessionId)) {
+        if (! $this->sessionManager->sessionExists($sessionId)) {
             return response()->json([
                 'status' => 'error',
                 'message' => 'Session not found or expired',
@@ -114,7 +86,7 @@ public function message(Request $request): JsonResponse
         try {
             $response = $this->mcpHandler->handle($requestData);
 
-            $success = McpSseService::sendToClient($sessionId, $response);
+            $success = $this->sessionManager->sendToClient($sessionId, $response);
 
             if (! $success) {
                 Log::warning("Failed to store message for client: {$sessionId}");
@@ -139,7 +111,7 @@ public function message(Request $request): JsonResponse
                     'message' => $e->getMessage(),
                 ],
             ];
-            McpSseService::sendToClient($sessionId, $errorResponse);
+            $this->sessionManager->sendToClient($sessionId, $errorResponse);
 
             return response()->json([
                 'status' => 'error',

src/LoopServiceProvider.php

@@ -3,9 +3,9 @@
 namespace Kirschbaum\Loop;
 
 use Kirschbaum\Loop\Commands\LoopMcpServerStartCommand;
-use Kirschbaum\Loop\Services\McpSseService;
 use Kirschbaum\Loop\Services\SseDriverManager;
 use Kirschbaum\Loop\Services\SseService;
+use Kirschbaum\Loop\Services\SseSessionManager;
 use Spatie\LaravelPackageTools\Package;
 use Spatie\LaravelPackageTools\PackageServiceProvider;
 
@@ -42,8 +42,8 @@ public function packageBooted(): void
             return new SseService;
         });
 
-        $this->app->singleton(McpSseService::class, function ($app) {
-            return new McpSseService;
+        $this->app->singleton(SseSessionManager::class, function ($app) {
+            return new SseSessionManager($app->make(SseDriverManager::class));
         });
 
         $this->app->singleton(SseDriverManager::class, function ($app) {