handler = $handler; } public function send(string $message): void { if ($this->stream instanceof Closure) { $this->sendStreamMessage($message); } $this->reply = $message; } public function run(): Response|StreamedResponse { if (is_callable($this->handler)) { ($this->handler)($this->request->getContent()); } if ($this->stream instanceof Closure) { $stream = $this->stream; return response()->stream(function () use ($stream): void { $result = $stream(); if (! is_iterable($result)) { return; } foreach ($result as $message) { if (connection_aborted() !== 0) { return; } $this->sendStreamMessage((string) $message); } }, 200, $this->getHeaders()); } $response = response($this->reply, $this->statusCode(), $this->getHeaders()); assert($response instanceof Response); return $response; } protected function statusCode(): int { // Must be 202 - https://modelcontextprotocol.io/specification/2026-07-28/basic/transports/streamable-http#sending-messages if ($this->reply === null) { return 202; } $reply = json_decode($this->reply, true); if (! is_array($reply) || ! is_array($reply['error'] ?? null)) { return 200; } return match ($reply['error']['code'] ?? null) { ErrorCode::METHOD_NOT_FOUND->value => 404, ErrorCode::INTERNAL_ERROR->value => 500, default => 400, }; } /** * Register a streaming callback. * * The callback may echo SSE-formatted output directly or return an iterable of message payloads. * * @param Closure(): (iterable|void) $stream */ public function stream(Closure $stream): void { $this->stream = $stream; } protected function sendStreamMessage(string $message): void { echo 'data: '.$message."\n\n"; if (ob_get_level() !== 0) { ob_flush(); } flush(); } /** * @return array */ protected function getHeaders(): array { $headers = [ 'Content-Type' => $this->stream instanceof Closure ? 'text/event-stream' : 'application/json', ]; if ($this->stream instanceof Closure) { $headers['X-Accel-Buffering'] = 'no'; } return $headers; } }