MCP/Server/JSONL.pm

=head1 NAME

AnyEvent::MCP::Server::JSONL - MCP over a newline-delimited byte stream

=head1 DESCRIPTION

Use this transport for stdio servers or an existing stream connection.
It implements L<AnyEvent::MCP::Server> over newline-delimited UTF-8 JSON-RPC:
one line is one complete message. Keep diagnostic output off the protocol
stream; a banner or log message on stdout is not a JSON-RPC response.

Unlike HTTP, replies can complete in a different order from requests.
The JSON-RPC ids associate them with the correct calls.

=head1 METHODS

=head2 $handle = $server->accept ($fh, $fh_out)

Starts handling C<$fh>. If C<$fh_out> is supplied, it receives outgoing
messages; otherwise C<$fh> is used in both directions. Both handles belong
to the server until the connection closes.

With separate input/output handles, input EOF waits for pending replies
and drains the output before closing. EOF on a bidirectional socket closes
it immediately and cancels outstanding requests.

C<max_size> limits a buffered incoming message, defaulting to 1 MiB.

=head2 $server->close

Closes the server's connections and cancels outstanding requests.

=head1 EVENTS

=head2 on_connection_error => $callback->($server, $handle, $message)

Called for stream errors before closing the connection.

=head2 on_connection_close => $callback->($server, $handle)

Called after a connection has closed.

=head2 on_close => $callback->($server)

Called by C<close> after closing all connections.

=cut

package AnyEvent::MCP::Server::JSONL;

use common::sense;
use parent "AnyEvent::MCP::Server";
use AnyEvent::Handle ();
use Scalar::Util qw(weaken);

sub accept {
   my ($self, $fh, $fh_out) = @_;

   delete $self->{_closed};
   weaken $self;

   my $handle = AnyEvent::Handle->new (
      fh       => $fh,
      rbuf_max => $self->{max_size} || 1024 * 1024,
      on_read  => sub { $self->_read ($_[0]) if $self },
      on_error => sub {
         my ($handle, $fatal, $message) = @_;

         $self->event (connection_error => $handle, $message) if $self;
         $self->_close ($handle) if $self;
      },
      on_eof => sub {
         return unless $self;
         my ($handle) = @_;

         if ($handle->{mcp_out}) {
            $handle->{mcp_eof} = 1;
            $self->_drain ($handle);
         } else {
            $self->_close ($handle);
         }
      },
   );

   if ($fh_out) {
      $handle->{mcp_out} = AnyEvent::Handle->new (
         fh       => $fh_out,
         on_error => sub {
            $self->event (connection_error => $_[0], $_[2]) if $self;
            $self->close if $self;
         },
      );
   }

   $handle->{mcp_peer} = {};
   $self->{connections}{$handle} = $handle;

   $handle
}

sub _read {
   my ($self, $handle) = @_;

   weaken $self;

   $handle->push_read (line => sub {
      my ($handle, $line) = @_;
      return unless $self;

      ++$handle->{mcp_active};
      $self->json_rpc ($line, sub {
         --$handle->{mcp_active};
         return if $handle->{mcp_closed};

         ($handle->{mcp_out} || $handle)->push_write ($_[0] . "\n")
            if defined $_[0];
         $self->_drain ($handle) if $handle->{mcp_eof};
      },
         peer => $handle->{mcp_peer},
         on_notify => sub {
            return if $handle->{mcp_closed};

            ($handle->{mcp_out} || $handle)->push_write (JSON::XS::encode_json ($_[1]) . "\n");
         },
      );
   });
}

sub _drain {
   my ($self, $handle) = @_;

   return if $handle->{mcp_active} || $handle->{mcp_closed};
   my $output = $handle->{mcp_out};
   if (length $output->{wbuf}) {
      weaken $self;
      $output->on_drain (sub { $self->_close ($handle) if $self });
   } else {
      $self->_close ($handle);
   }
}

sub _close {
   my ($self, $handle) = @_;

   return if $handle->{mcp_closed};
   $handle->{mcp_closed} = 1;

   my $requests = delete $handle->{mcp_peer}{requests} || {};
   $_->cancel ("Connection closed") for grep defined, values %$requests;

   delete $self->{connections}{$handle};
   $handle->{mcp_out}->destroy if $handle->{mcp_out};
   $handle->destroy;
   $self->event (connection_close => $handle);
}

sub close {
   my ($self) = @_;

   return if $self->{_closed}++;
   delete $self->{listener};
   $self->_close ($_) for values %{$self->{connections} || {}};
   $self->event (close =>);
}

sub DESTROY {
   $_[0]->close
}

1
Revision:	1.2
Committed:	Fri Oct 9 09:36:49 2026 UTC (3 weeks, 4 days ago) by root
Branch:	MAIN
Changes since 1.1:	+7 -3 lines
Log Message:	* empty log message *
#	User	Rev	Content
1	root	1.1	=head1 NAME
2
3			AnyEvent::MCP::Server::JSONL - MCP over a newline-delimited byte stream
4
5			=head1 DESCRIPTION
6
7	root	1.2	Use this transport for stdio servers or an existing stream connection.
8			It implements L<AnyEvent::MCP::Server> over newline-delimited UTF-8 JSON-RPC:
9			one line is one complete message. Keep diagnostic output off the protocol
10			stream; a banner or log message on stdout is not a JSON-RPC response.
11
12			Unlike HTTP, replies can complete in a different order from requests.
13			The JSON-RPC ids associate them with the correct calls.
14	root	1.1
15			=head1 METHODS
16
17			=head2 $handle = $server->accept ($fh, $fh_out)
18
19			Starts handling C<$fh>. If C<$fh_out> is supplied, it receives outgoing
20			messages; otherwise C<$fh> is used in both directions. Both handles belong
21			to the server until the connection closes.
22
23			With separate input/output handles, input EOF waits for pending replies
24			and drains the output before closing. EOF on a bidirectional socket closes
25			it immediately and cancels outstanding requests.
26
27			C<max_size> limits a buffered incoming message, defaulting to 1 MiB.
28
29			=head2 $server->close
30
31			Closes the server's connections and cancels outstanding requests.
32
33			=head1 EVENTS
34
35			=head2 on_connection_error => $callback->($server, $handle, $message)
36
37			Called for stream errors before closing the connection.
38
39			=head2 on_connection_close => $callback->($server, $handle)
40
41			Called after a connection has closed.
42
43			=head2 on_close => $callback->($server)
44
45			Called by C<close> after closing all connections.
46
47			=cut
48
49			package AnyEvent::MCP::Server::JSONL;
50
51			use common::sense;
52			use parent "AnyEvent::MCP::Server";
53			use AnyEvent::Handle ();
54			use Scalar::Util qw(weaken);
55
56			sub accept {
57			my ($self, $fh, $fh_out) = @_;
58
59			delete $self->{_closed};
60			weaken $self;
61
62			my $handle = AnyEvent::Handle->new (
63			fh => $fh,
64			rbuf_max => $self->{max_size} \|\| 1024 * 1024,
65			on_read => sub { $self->_read ($_[0]) if $self },
66			on_error => sub {
67			my ($handle, $fatal, $message) = @_;
68
69			$self->event (connection_error => $handle, $message) if $self;
70			$self->_close ($handle) if $self;
71			},
72			on_eof => sub {
73			return unless $self;
74			my ($handle) = @_;
75
76			if ($handle->{mcp_out}) {
77			$handle->{mcp_eof} = 1;
78			$self->_drain ($handle);
79			} else {
80			$self->_close ($handle);
81			}
82			},
83			);
84
85			if ($fh_out) {
86			$handle->{mcp_out} = AnyEvent::Handle->new (
87			fh => $fh_out,
88			on_error => sub {
89			$self->event (connection_error => $_[0], $_[2]) if $self;
90			$self->close if $self;
91			},
92			);
93			}
94
95			$handle->{mcp_peer} = {};
96			$self->{connections}{$handle} = $handle;
97
98			$handle
99			}
100
101			sub _read {
102			my ($self, $handle) = @_;
103
104			weaken $self;
105
106			$handle->push_read (line => sub {
107			my ($handle, $line) = @_;
108			return unless $self;
109
110			++$handle->{mcp_active};
111			$self->json_rpc ($line, sub {
112			--$handle->{mcp_active};
113			return if $handle->{mcp_closed};
114
115			($handle->{mcp_out} \|\| $handle)->push_write ($_[0] . "\n")
116			if defined $_[0];
117			$self->_drain ($handle) if $handle->{mcp_eof};
118			},
119			peer => $handle->{mcp_peer},
120			on_notify => sub {
121			return if $handle->{mcp_closed};
122
123			($handle->{mcp_out} \|\| $handle)->push_write (JSON::XS::encode_json ($_[1]) . "\n");
124			},
125			);
126			});
127			}
128
129			sub _drain {
130			my ($self, $handle) = @_;
131
132			return if $handle->{mcp_active} \|\| $handle->{mcp_closed};
133			my $output = $handle->{mcp_out};
134			if (length $output->{wbuf}) {
135			weaken $self;
136			$output->on_drain (sub { $self->_close ($handle) if $self });
137			} else {
138			$self->_close ($handle);
139			}
140			}
141
142			sub _close {
143			my ($self, $handle) = @_;
144
145			return if $handle->{mcp_closed};
146			$handle->{mcp_closed} = 1;
147
148			my $requests = delete $handle->{mcp_peer}{requests} \|\| {};
149			$_->cancel ("Connection closed") for grep defined, values %$requests;
150
151			delete $self->{connections}{$handle};
152			$handle->{mcp_out}->destroy if $handle->{mcp_out};
153			$handle->destroy;
154			$self->event (connection_close => $handle);
155			}
156
157			sub close {
158			my ($self) = @_;
159
160			return if $self->{_closed}++;
161			delete $self->{listener};
162			$self->_close ($_) for values %{$self->{connections} \|\| {}};
163			$self->event (close =>);
164			}
165
166			sub DESTROY {
167			$_[0]->close
168			}
169
170			1