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.4
Committed:	Tue Oct 13 10:17:31 2026 UTC (3 weeks, 3 days ago) by root
Branch:	MAIN
Changes since 1.3:	+2 -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			=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	root	1.4
122	root	1.1	return unless $self;
123
124			++$handle->{mcp_active};
125			$self->json_rpc ($line, sub {
126			--$handle->{mcp_active};
127			return if $handle->{mcp_closed};
128
129			($handle->{mcp_out} \|\| $handle)->push_write ($_[0] . "\n")
130			if defined $_[0];
131			$self->_drain ($handle) if $handle->{mcp_eof};
132			},
133	root	1.4	peer => $handle->{mcp_peer},
134	root	1.1	on_notify => sub {
135			return if $handle->{mcp_closed};
136
137			($handle->{mcp_out} \|\| $handle)->push_write (JSON::XS::encode_json ($_[1]) . "\n");
138			},
139			);
140			});
141			}
142
143	root	1.3	sub request_idle_exit {
144			my ($self) = @_;
145
146			return unless $ENV{PERL_ANYEVENT_MCP_MULTIBROKER_PROTOCOL} eq "stdio"
147			&& !$self->{idle_exit_requested};
148
149			for my $handle (values %{$self->{connections} \|\| {}}) {
150			next if $handle->{mcp_closed} \|\| !$handle->{mcp_peer}{aemcp_multibroker};
151
152			($handle->{mcp_out} \|\| $handle)->push_write (JSON::XS::encode_json ({
153			jsonrpc => "2.0", method => "anyevent-mcp/idle-exit", params => {},
154			}) . "\n");
155			$self->{idle_exit_requested} = 1;
156			return 1
157			}
158
159			undef
160			}
161
162			sub dispatch {
163			my ($self, $req) = @_;
164
165			if ($req->{method} eq "anyevent-mcp/shutdown") {
166			my $handle = $req->{peer}{mcp_handle};
167			my $busy = grep $_->{mcp_active} > ($_ == $handle ? 1 : 0),
168			values %{$self->{connections} \|\| {}};
169			my $accepted = $ENV{PERL_ANYEVENT_MCP_MULTIBROKER_PROTOCOL} eq "stdio"
170			&& $req->{peer}{aemcp_multibroker} && $self->{idle_exit_requested}
171			&& !$self->{_idle_shutting_down} && !$busy
172			&& (!$self->{on_idle_shutdown} \|\| $self->{on_idle_shutdown}->($self));
173
174			$req->reply ({ accepted => $accepted ? JSON::XS::true : JSON::XS::false });
175			if ($accepted) {
176			$self->{_idle_shutting_down} = 1;
177			my $out = $handle->{mcp_out} \|\| $handle;
178			weaken $self;
179			if (length $out->{wbuf}) {
180			$out->on_drain (sub { $self->close if $self });
181			} else {
182			AnyEvent::postpone (sub { $self->close if $self });
183			}
184			}
185			return
186			}
187
188			$self->SUPER::dispatch ($req)
189			}
190
191	root	1.1	sub _drain {
192			my ($self, $handle) = @_;
193
194			return if $handle->{mcp_active} \|\| $handle->{mcp_closed};
195			my $output = $handle->{mcp_out};
196			if (length $output->{wbuf}) {
197			weaken $self;
198			$output->on_drain (sub { $self->_close ($handle) if $self });
199			} else {
200			$self->_close ($handle);
201			}
202			}
203
204			sub _close {
205			my ($self, $handle) = @_;
206
207			return if $handle->{mcp_closed};
208			$handle->{mcp_closed} = 1;
209
210			my $requests = delete $handle->{mcp_peer}{requests} \|\| {};
211			$_->cancel ("Connection closed") for grep defined, values %$requests;
212
213			delete $self->{connections}{$handle};
214			$handle->{mcp_out}->destroy if $handle->{mcp_out};
215			$handle->destroy;
216			$self->event (connection_close => $handle);
217			}
218
219			sub close {
220			my ($self) = @_;
221
222			return if $self->{_closed}++;
223			delete $self->{listener};
224			$self->_close ($_) for values %{$self->{connections} \|\| {}};
225			$self->event (close =>);
226			}
227
228			sub DESTROY {
229			$_[0]->close
230			}
231
232			1