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 EXAMPLE

An application that already owns its tool registry and lifetime can
serve one stdio MCP connection:

   use AnyEvent;
   use AnyEvent::MCP::Server::JSONL;

   my $server = AnyEvent::MCP::Server::JSONL->new (
      toolsets => [$tools],
   );
   $server->accept (\*STDIN, \*STDOUT);
   AE::cv->recv;

C<$tools> is an L<AnyEvent::MCP::Server::Toolset>. Keep stdout reserved
for MCP messages and write diagnostics to stderr. The application is
responsible for its event loop and termination. If you do not need to
manage those details, use L<AnyEvent::MCP::Server::Simple/mcp_stdio>.
The matching client transport is L<AnyEvent::MCP::Client::JSONL>.

=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 8 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. A declined shutdown clears the one-shot notification state,
allowing a later idle-exit request.

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