NAME

MCP::Client::SSE - an incremental Server-Sent Events parser

DESCRIPTION

A 2026-07-28 MCP server may answer a Streamable HTTP request with an SSE stream: notifications first, then the response that ends the request. The bytes of that stream arrive in whatever sizes the network felt like, and none of those sizes have anything to do with where events end. A parser that treats each chunk as a unit will eventually try to decode half a JSON object and take the whole connection down with it.

This class is the fix, and it is deliberately standalone: feed it bytes or text as they arrive, tap events for whole events. It holds an incomplete trailing event back until the rest of it turns up, and it decodes UTF-8 incrementally, so a chunk boundary landing in the middle of a multi-byte character is a non-event rather than an exception.

What counts as an event

Per the SSE specification:

• Events are separated by a blank line — \n\n, \r\n\r\n or \r\r.

• A line beginning with : is a comment. Keepalives (: ping) are comments, and are skipped.

• data: lines accumulate; multiple data: lines in one event are joined with a newline, which is how a server sends multi-line JSON.

• One optional space after the field's colon is part of the syntax, not the value, and is stripped.

• event: names the event type (message when unnamed), id: sets the last event id, retry: a reconnection delay in milliseconds. Unknown fields are ignored.

• An event carrying no data is not dispatched.

Emitted events are hashes: type, data, and id/retry when the event set them.

Closing

close ends the events supply. By default it also parses whatever is left in the buffer, because the last event of an HTTP response body is routinely sent without its trailing blank line, and for MCP that last event is the one carrying the response — dropping it would hang the request that is waiting for it. Pass :!flush for the strict reading, where an unterminated trailing event is discarded.

EXAMPLES

Parsing a response body as it arrives:

use MCP::Client::SSE;
use JSON::Fast;

my $sse = MCP::Client::SSE.new;

# Tap before feeding: the supply is live, and events are dropped, not
# buffered, while nothing is listening.
$sse.events.tap: -> %event {
	my %msg = from-json(%event<data>);
	%msg<id>:exists ?? $correlator.resolve(%msg<id>, %msg<result>)
	                !! $notifications.emit(%msg);
};

react {
	whenever $response.body-byte-stream -> $bytes { $sse.feed($bytes) }
	whenever $response.body-byte-stream.done { $sse.close }
}

Chunk boundaries are irrelevant — these two feeds produce exactly one event:

my $sse = MCP::Client::SSE.new;
my @seen;
$sse.events.tap: { @seen.push($_) };

$sse.feed(qq:to/CHUNK/.chomp);
    : keepalive
    event: message
    data: {"jsonrpc":"2.0","id":1,"res
    CHUNK
$sse.feed(qq:to/CHUNK/);
    ult":{"ok":true}}

    CHUNK

say @seen.elems;        # 1
say @seen[0]<type>;     # message
say @seen[0]<data>;     # {"jsonrpc":"2.0","id":1,"result":{"ok":true}}