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.

=head2 $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. No process exits until the broker sends shutdown.
The shutdown request is explicitly answered before the stream is closed.
C<on_idle_shutdown>, if supplied, is a callback returning true to accept or
false to decline a late shutdown request (e.g. if the application became busy).

=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.3
Committed:	Mon Oct 12 05:03:59 2026 UTC (3 weeks, 2 days ago) by root
Branch:	MAIN
Changes since 1.2:	+61 -0 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	root	1.3	=head2 $server->request_idle_exit
48
49			Requests a managed shutdown, once, from a spawned stdio multibroker child
50			whose peer advertised C<anyevent-mcp/multibroker> version 1. The application
51			decides when it is idle. No process exits until the broker sends shutdown.
52			The shutdown request is explicitly answered before the stream is closed.
53			C<on_idle_shutdown>, if supplied, is a callback returning true to accept or
54			false to decline a late shutdown request (e.g. if the application became busy).
55
56	root	1.1	=cut
57
58			package AnyEvent::MCP::Server::JSONL;
59
60			use common::sense;
61			use parent "AnyEvent::MCP::Server";
62			use AnyEvent::Handle ();
63	root	1.3	use AnyEvent ();
64			use JSON::XS ();
65	root	1.1	use Scalar::Util qw(weaken);
66
67			sub accept {
68			my ($self, $fh, $fh_out) = @_;
69
70			delete $self->{_closed};
71			weaken $self;
72
73			my $handle = AnyEvent::Handle->new (
74			fh => $fh,
75			rbuf_max => $self->{max_size} \|\| 1024 * 1024,
76			on_read => sub { $self->_read ($_[0]) if $self },
77			on_error => sub {
78			my ($handle, $fatal, $message) = @_;
79
80			$self->event (connection_error => $handle, $message) if $self;
81			$self->_close ($handle) if $self;
82			},
83			on_eof => sub {
84			return unless $self;
85			my ($handle) = @_;
86
87			if ($handle->{mcp_out}) {
88			$handle->{mcp_eof} = 1;
89			$self->_drain ($handle);
90			} else {
91			$self->_close ($handle);
92			}
93			},
94			);
95
96			if ($fh_out) {
97			$handle->{mcp_out} = AnyEvent::Handle->new (
98			fh => $fh_out,
99			on_error => sub {
100			$self->event (connection_error => $_[0], $_[2]) if $self;
101			$self->close if $self;
102			},
103			);
104			}
105
106			$handle->{mcp_peer} = {};
107	root	1.3	$handle->{mcp_peer}{mcp_handle} = $handle;
108			weaken $handle->{mcp_peer}{mcp_handle};
109	root	1.1	$self->{connections}{$handle} = $handle;
110
111			$handle
112			}
113
114			sub _read {
115			my ($self, $handle) = @_;
116
117			weaken $self;
118
119			$handle->push_read (line => sub {
120			my ($handle, $line) = @_;
121			return unless $self;
122
123			++$handle->{mcp_active};
124			$self->json_rpc ($line, sub {
125			--$handle->{mcp_active};
126			return if $handle->{mcp_closed};
127
128			($handle->{mcp_out} \|\| $handle)->push_write ($_[0] . "\n")
129			if defined $_[0];
130			$self->_drain ($handle) if $handle->{mcp_eof};
131			},
132			peer => $handle->{mcp_peer},
133			on_notify => sub {
134			return if $handle->{mcp_closed};
135
136			($handle->{mcp_out} \|\| $handle)->push_write (JSON::XS::encode_json ($_[1]) . "\n");
137			},
138			);
139			});
140			}
141
142	root	1.3	sub request_idle_exit {
143			my ($self) = @_;
144
145			return unless $ENV{PERL_ANYEVENT_MCP_MULTIBROKER_PROTOCOL} eq "stdio"
146			&& !$self->{idle_exit_requested};
147
148			for my $handle (values %{$self->{connections} \|\| {}}) {
149			next if $handle->{mcp_closed} \|\| !$handle->{mcp_peer}{aemcp_multibroker};
150
151			($handle->{mcp_out} \|\| $handle)->push_write (JSON::XS::encode_json ({
152			jsonrpc => "2.0", method => "anyevent-mcp/idle-exit", params => {},
153			}) . "\n");
154			$self->{idle_exit_requested} = 1;
155			return 1
156			}
157
158			undef
159			}
160
161			sub dispatch {
162			my ($self, $req) = @_;
163
164			if ($req->{method} eq "anyevent-mcp/shutdown") {
165			my $handle = $req->{peer}{mcp_handle};
166			my $busy = grep $_->{mcp_active} > ($_ == $handle ? 1 : 0),
167			values %{$self->{connections} \|\| {}};
168			my $accepted = $ENV{PERL_ANYEVENT_MCP_MULTIBROKER_PROTOCOL} eq "stdio"
169			&& $req->{peer}{aemcp_multibroker} && $self->{idle_exit_requested}
170			&& !$self->{_idle_shutting_down} && !$busy
171			&& (!$self->{on_idle_shutdown} \|\| $self->{on_idle_shutdown}->($self));
172
173			$req->reply ({ accepted => $accepted ? JSON::XS::true : JSON::XS::false });
174			if ($accepted) {
175			$self->{_idle_shutting_down} = 1;
176			my $out = $handle->{mcp_out} \|\| $handle;
177			weaken $self;
178			if (length $out->{wbuf}) {
179			$out->on_drain (sub { $self->close if $self });
180			} else {
181			AnyEvent::postpone (sub { $self->close if $self });
182			}
183			}
184			return
185			}
186
187			$self->SUPER::dispatch ($req)
188			}
189
190	root	1.1	sub _drain {
191			my ($self, $handle) = @_;
192
193			return if $handle->{mcp_active} \|\| $handle->{mcp_closed};
194			my $output = $handle->{mcp_out};
195			if (length $output->{wbuf}) {
196			weaken $self;
197			$output->on_drain (sub { $self->_close ($handle) if $self });
198			} else {
199			$self->_close ($handle);
200			}
201			}
202
203			sub _close {
204			my ($self, $handle) = @_;
205
206			return if $handle->{mcp_closed};
207			$handle->{mcp_closed} = 1;
208
209			my $requests = delete $handle->{mcp_peer}{requests} \|\| {};
210			$_->cancel ("Connection closed") for grep defined, values %$requests;
211
212			delete $self->{connections}{$handle};
213			$handle->{mcp_out}->destroy if $handle->{mcp_out};
214			$handle->destroy;
215			$self->event (connection_close => $handle);
216			}
217
218			sub close {
219			my ($self) = @_;
220
221			return if $self->{_closed}++;
222			delete $self->{listener};
223			$self->_close ($_) for values %{$self->{connections} \|\| {}};
224			$self->event (close =>);
225			}
226
227			sub DESTROY {
228			$_[0]->close
229			}
230
231			1