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.

=head1 MANAGED SHUTDOWN

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