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

=over

=item $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.

=item $server->close

Closes the server's connections and cancels outstanding requests.

=back

=head1 EVENTS

=over

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

Called for stream errors before closing the connection.

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

Called after a connection has closed.

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

Called by C<close> after closing all connections.

=back

=head1 MANAGED SHUTDOWN

=over

=item $server->request_idle_exit

Requests a managed shutdown, once, from a spawned stdio multibroker child
whose peer advertised C<anyevent-mcp/multibroker> version 1. The application
decides when it is idle. The method returns true when it sends the notification, or undef when
there is no eligible connection or it has already sent one.
The broker can ignore the notification. When it sends shutdown, the server
answers before draining the stream and closing its connections. Closing
the server lets Simple's blocking starter return; an embedding application
controls its own process lifetime.

C<on_idle_shutdown>, if supplied, is C<$callback->($server)> returning true
to accept or false to decline a late shutdown request (for example, if
application work has resumed). Outstanding MCP requests also prevent
acceptance.

=back

=cut

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

use common::sense;
use parent "AnyEvent::MCP::Server";
use AnyEvent::Handle ();
use AnyEvent ();
use JSON::XS ();
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} = {};
   $handle->{mcp_peer}{mcp_handle} = $handle;
   weaken $handle->{mcp_peer}{mcp_handle};
   $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 request_idle_exit {
   my ($self) = @_;

   return unless $ENV{PERL_ANYEVENT_MCP_MULTIBROKER_PROTOCOL} eq "stdio"
      && !$self->{idle_exit_requested};

   for my $handle (values %{$self->{connections}}) {
      next if $handle->{mcp_closed} || !$handle->{mcp_peer}{aemcp_multibroker};

      ($handle->{mcp_out} || $handle)->push_write (JSON::XS::encode_json ({
         jsonrpc => "2.0", method => "anyevent-mcp/idle-exit", params => {},
      }) . "\n");
      $self->{idle_exit_requested} = 1;
      return 1
   }

   undef
}

sub dispatch {
   my ($self, $req) = @_;

   if ($req->{method} eq "anyevent-mcp/shutdown") {
      my $handle = $req->{peer}{mcp_handle};
      my $busy = grep $_->{mcp_active} > ($_ == $handle ? 1 : 0),
                      values %{$self->{connections}};
      my $accepted = $ENV{PERL_ANYEVENT_MCP_MULTIBROKER_PROTOCOL} eq "stdio"
         && $req->{peer}{aemcp_multibroker} && $self->{idle_exit_requested}
         && !$self->{_idle_shutting_down} && !$busy
         && (!$self->{on_idle_shutdown} || $self->{on_idle_shutdown}->($self));

      $req->reply ({ accepted => $accepted ? JSON::XS::true : JSON::XS::false });
      if ($accepted) {
         $self->{_idle_shutting_down} = 1;
         my $out = $handle->{mcp_out} || $handle;
         weaken $self;
         if (length $out->{wbuf}) {
            $out->on_drain (sub { $self->close if $self });
         } else {
            AnyEvent::postpone (sub { $self->close if $self });
         }
      }

      return;
   }

   $self->SUPER::dispatch ($req)
}

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.8
Committed:	Fri Oct 16 05:48:34 2026 UTC (12 days ago) by root
Branch:	MAIN
Changes since 1.7:	+13 -9 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	root	1.6	=over
18
19			=item $handle = $server->accept ($fh, $fh_out)
20	root	1.1
21			Starts handling C<$fh>. If C<$fh_out> is supplied, it receives outgoing
22			messages; otherwise C<$fh> is used in both directions. Both handles belong
23			to the server until the connection closes.
24
25			With separate input/output handles, input EOF waits for pending replies
26			and drains the output before closing. EOF on a bidirectional socket closes
27			it immediately and cancels outstanding requests.
28
29			C<max_size> limits a buffered incoming message, defaulting to 1 MiB.
30
31	root	1.6	=item $server->close
32	root	1.1
33			Closes the server's connections and cancels outstanding requests.
34
35	root	1.6	=back
36
37	root	1.1	=head1 EVENTS
38
39	root	1.6	=over
40
41			=item on_connection_error => $callback->($server, $handle, $message)
42	root	1.1
43			Called for stream errors before closing the connection.
44
45	root	1.6	=item on_connection_close => $callback->($server, $handle)
46	root	1.1
47			Called after a connection has closed.
48
49	root	1.6	=item on_close => $callback->($server)
50	root	1.1
51			Called by C<close> after closing all connections.
52
53	root	1.6	=back
54
55	root	1.5	=head1 MANAGED SHUTDOWN
56
57	root	1.6	=over
58
59			=item $server->request_idle_exit
60	root	1.3
61			Requests a managed shutdown, once, from a spawned stdio multibroker child
62			whose peer advertised C<anyevent-mcp/multibroker> version 1. The application
63	root	1.6	decides when it is idle. The method returns true when it sends the notification, or undef when
64			there is no eligible connection or it has already sent one.
65			The broker can ignore the notification. When it sends shutdown, the server
66			answers before draining the stream and closing its connections. Closing
67			the server lets Simple's blocking starter return; an embedding application
68			controls its own process lifetime.
69
70			C<on_idle_shutdown>, if supplied, is C<$callback->($server)> returning true
71			to accept or false to decline a late shutdown request (for example, if
72			application work has resumed). Outstanding MCP requests also prevent
73			acceptance.
74
75			=back
76	root	1.3
77	root	1.1	=cut
78
79			package AnyEvent::MCP::Server::JSONL;
80
81			use common::sense;
82			use parent "AnyEvent::MCP::Server";
83			use AnyEvent::Handle ();
84	root	1.3	use AnyEvent ();
85			use JSON::XS ();
86	root	1.1	use Scalar::Util qw(weaken);
87
88			sub accept {
89			my ($self, $fh, $fh_out) = @_;
90
91			delete $self->{_closed};
92			weaken $self;
93
94			my $handle = AnyEvent::Handle->new (
95			fh => $fh,
96			rbuf_max => $self->{max_size} \|\| 1024 * 1024,
97			on_read => sub { $self->_read ($_[0]) if $self },
98			on_error => sub {
99			my ($handle, $fatal, $message) = @_;
100
101			$self->event (connection_error => $handle, $message) if $self;
102			$self->_close ($handle) if $self;
103			},
104			on_eof => sub {
105			return unless $self;
106			my ($handle) = @_;
107
108			if ($handle->{mcp_out}) {
109			$handle->{mcp_eof} = 1;
110			$self->_drain ($handle);
111			} else {
112			$self->_close ($handle);
113			}
114			},
115			);
116
117			if ($fh_out) {
118			$handle->{mcp_out} = AnyEvent::Handle->new (
119			fh => $fh_out,
120			on_error => sub {
121			$self->event (connection_error => $_[0], $_[2]) if $self;
122			$self->close if $self;
123			},
124			);
125			}
126
127			$handle->{mcp_peer} = {};
128	root	1.3	$handle->{mcp_peer}{mcp_handle} = $handle;
129			weaken $handle->{mcp_peer}{mcp_handle};
130	root	1.1	$self->{connections}{$handle} = $handle;
131
132			$handle
133			}
134
135			sub _read {
136			my ($self, $handle) = @_;
137
138			weaken $self;
139
140			$handle->push_read (line => sub {
141			my ($handle, $line) = @_;
142	root	1.4
143	root	1.1	return unless $self;
144
145			++$handle->{mcp_active};
146	root	1.8	$self->json_rpc (
147			$line,
148			sub {
149			--$handle->{mcp_active};
150			return if $handle->{mcp_closed};
151
152			($handle->{mcp_out} \|\| $handle)->push_write ($_[0] . "\n")
153			if defined $_[0];
154			$self->_drain ($handle) if $handle->{mcp_eof};
155			},
156	root	1.4	peer => $handle->{mcp_peer},
157	root	1.1	on_notify => sub {
158			return if $handle->{mcp_closed};
159
160			($handle->{mcp_out} \|\| $handle)->push_write (JSON::XS::encode_json ($_[1]) . "\n");
161			},
162			);
163			});
164			}
165
166	root	1.3	sub request_idle_exit {
167			my ($self) = @_;
168
169			return unless $ENV{PERL_ANYEVENT_MCP_MULTIBROKER_PROTOCOL} eq "stdio"
170			&& !$self->{idle_exit_requested};
171
172	root	1.7	for my $handle (values %{$self->{connections}}) {
173	root	1.3	next if $handle->{mcp_closed} \|\| !$handle->{mcp_peer}{aemcp_multibroker};
174
175			($handle->{mcp_out} \|\| $handle)->push_write (JSON::XS::encode_json ({
176			jsonrpc => "2.0", method => "anyevent-mcp/idle-exit", params => {},
177			}) . "\n");
178			$self->{idle_exit_requested} = 1;
179			return 1
180			}
181
182			undef
183			}
184
185			sub dispatch {
186			my ($self, $req) = @_;
187
188			if ($req->{method} eq "anyevent-mcp/shutdown") {
189			my $handle = $req->{peer}{mcp_handle};
190			my $busy = grep $_->{mcp_active} > ($_ == $handle ? 1 : 0),
191	root	1.7	values %{$self->{connections}};
192	root	1.3	my $accepted = $ENV{PERL_ANYEVENT_MCP_MULTIBROKER_PROTOCOL} eq "stdio"
193			&& $req->{peer}{aemcp_multibroker} && $self->{idle_exit_requested}
194			&& !$self->{_idle_shutting_down} && !$busy
195			&& (!$self->{on_idle_shutdown} \|\| $self->{on_idle_shutdown}->($self));
196
197			$req->reply ({ accepted => $accepted ? JSON::XS::true : JSON::XS::false });
198			if ($accepted) {
199			$self->{_idle_shutting_down} = 1;
200			my $out = $handle->{mcp_out} \|\| $handle;
201			weaken $self;
202			if (length $out->{wbuf}) {
203			$out->on_drain (sub { $self->close if $self });
204			} else {
205			AnyEvent::postpone (sub { $self->close if $self });
206			}
207			}
208	root	1.8
209			return;
210	root	1.3	}
211
212			$self->SUPER::dispatch ($req)
213			}
214
215	root	1.1	sub _drain {
216			my ($self, $handle) = @_;
217
218			return if $handle->{mcp_active} \|\| $handle->{mcp_closed};
219	root	1.8
220	root	1.1	my $output = $handle->{mcp_out};
221			if (length $output->{wbuf}) {
222			weaken $self;
223			$output->on_drain (sub { $self->_close ($handle) if $self });
224			} else {
225			$self->_close ($handle);
226			}
227			}
228
229			sub _close {
230			my ($self, $handle) = @_;
231
232			return if $handle->{mcp_closed};
233			$handle->{mcp_closed} = 1;
234
235			my $requests = delete $handle->{mcp_peer}{requests} \|\| {};
236			$_->cancel ("Connection closed") for grep defined, values %$requests;
237
238			delete $self->{connections}{$handle};
239			$handle->{mcp_out}->destroy if $handle->{mcp_out};
240			$handle->destroy;
241			$self->event (connection_close => $handle);
242			}
243
244			sub close {
245			my ($self) = @_;
246
247			return if $self->{_closed}++;
248			delete $self->{listener};
249			$self->_close ($_) for values %{$self->{connections} \|\| {}};
250			$self->event (close =>);
251			}
252
253			sub DESTROY {
254			$_[0]->close
255			}
256
257			1