fix: client logger timeout

Author: lvluoyue

src/McpServerManager.php

@@ -6,17 +6,19 @@
 use Mcp\Server\Session\Psr16StoreSession;
 use Mcp\Server\Transport\CallbackStream;
 use Mcp\Server\Transport\StdioTransport;
-use Mcp\Server\Transport\StreamableHttpTransport;
+use Luoyue\WebmanMcp\Server\StreamableHttpTransport;
 use Mcp\Server\Session\InMemorySessionStore;
 use Nyholm\Psr7\ServerRequest;
 use Psr\Http\Message\ResponseInterface;
+use Psr\Http\Message\StreamInterface;
 use Psr\Log\LoggerInterface;
 use Psr\Log\NullLogger;
 use support\Cache;
 use support\Container;
 use support\Log;
 use Webman\Http\Response;
 use Workerman\Connection\TcpConnection;
+use Workerman\Coroutine;
 use Workerman\Worker;
 use Generator;
 use function request;
@@ -130,8 +132,16 @@ private function handleHttpRequest(Server $server, string $serviceName): Respons
         /** @var ResponseInterface $response */
         $response = $server->run($transport);
 
-        $body = $response->getBody() instanceof CallbackStream ? "\r\n" : $response->getBody()->getContents();
-        return response($body, $response->getStatusCode(), array_map('current', $response->getHeaders()));
+        return response($this->getResponseBody($response->getBody()), $response->getStatusCode(), array_map('current', $response->getHeaders()));
+    }
+
+    private function getResponseBody(StreamInterface $body): string
+    {
+        if($body instanceof CallbackStream) {
+            Coroutine::defer($body->getContents(...));
+            return "\r\n";
+        }
+        return $body->getContents();
     }
 
 }

src/Server/StreamableHttpTransport.php

@@ -0,0 +1,65 @@
+<?php
+
+namespace Luoyue\WebmanMcp\Server;
+
+use \Mcp\Server\Transport\StreamableHttpTransport as BaseStreamableHttpTransport;
+use Psr\Http\Message\ResponseFactoryInterface;
+use Psr\Http\Message\ServerRequestInterface;
+use Psr\Http\Message\StreamFactoryInterface;
+use Psr\Log\LoggerInterface;
+use Psr\Log\NullLogger;
+use Symfony\Component\Uid\Uuid;
+use Workerman\Connection\TcpConnection;
+use Workerman\Protocols\Http\ServerSentEvents;
+
+class StreamableHttpTransport extends BaseStreamableHttpTransport
+{
+
+    private readonly TcpConnection $connection;
+
+    /**
+     * @param array<string, string> $corsHeaders
+     */
+    public function __construct(
+        private readonly ServerRequestInterface $request,
+        ?ResponseFactoryInterface $responseFactory = null,
+        ?StreamFactoryInterface $streamFactory = null,
+        array $corsHeaders = [],
+        LoggerInterface $logger = new NullLogger(),
+    ) {
+        $this->connection = $request->getAttribute(TcpConnection::class);
+        parent::__construct($request, $responseFactory, $streamFactory, $corsHeaders, $logger);
+    }
+
+    protected function handleFiberTermination(): void
+    {
+        $finalResult = $this->sessionFiber->getReturn();
+
+        if (null !== $finalResult) {
+            try {
+                $encoded = json_encode($finalResult, \JSON_THROW_ON_ERROR);
+                $this->connection->send(new ServerSentEvents([
+                    'event' => 'message',
+                    'data' => $encoded,
+                ]));
+            } catch (\JsonException $e) {
+                $this->logger->error('SSE: Failed to encode final Fiber result.', ['exception' => $e]);
+            }
+        }
+
+        $this->sessionFiber = null;
+    }
+
+    protected function flushOutgoingMessages(?Uuid $sessionId): void
+    {
+        $messages = $this->getOutgoingMessages($sessionId);
+
+        foreach ($messages as $message) {
+            $this->connection->send(new ServerSentEvents([
+                'event' => 'message',
+                'data' => $message['message'],
+            ]));
+        }
+    }
+
+}