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 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.9
Committed:	Wed Oct 21 17:28:16 2026 UTC (8 days, 16 hours ago) by root
Branch:	MAIN
CVS Tags:	rel-0_62, rel-0_61, rel-0_6, rel-0_65, HEAD
Changes since 1.8:	+20 -0 lines
Log Message:	* empty log message *
#	Content
1	=head1 NAME
2
3	AnyEvent::MCP::Server::JSONL - MCP over a newline-delimited byte stream
4
5	=head1 DESCRIPTION
6
7	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
15	=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	=head1 METHODS
36
37	=over
38
39	=item $handle = $server->accept ($fh, $fh_out)
40
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	C<max_size> limits a buffered incoming message, defaulting to 1 MiB.
50
51	=item $server->close
52
53	Closes the server's connections and cancels outstanding requests.
54
55	=back
56
57	=head1 EVENTS
58
59	=over
60
61	=item on_connection_error => $callback->($server, $handle, $message)
62
63	Called for stream errors before closing the connection.
64
65	=item on_connection_close => $callback->($server, $handle)
66
67	Called after a connection has closed.
68
69	=item on_close => $callback->($server)
70
71	Called by C<close> after closing all connections.
72
73	=back
74
75	=head1 MANAGED SHUTDOWN
76
77	=over
78
79	=item $server->request_idle_exit
80
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	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	acceptance.
94
95	=back
96
97	=cut
98
99	package AnyEvent::MCP::Server::JSONL;
100
101	use common::sense;
102	use parent "AnyEvent::MCP::Server";
103	use AnyEvent::Handle ();
104	use AnyEvent ();
105	use JSON::XS ();
106	use Scalar::Util qw(weaken);
107
108	sub accept {
109	my ($self, $fh, $fh_out) = @_;
110
111	delete $self->{_closed};
112	weaken $self;
113
114	my $handle = AnyEvent::Handle->new (
115	fh => $fh,
116	rbuf_max => $self->{max_size} \|\| 1024 * 1024,
117	on_read => sub { $self->_read ($_[0]) if $self },
118	on_error => sub {
119	my ($handle, $fatal, $message) = @_;
120
121	$self->event (connection_error => $handle, $message) if $self;
122	$self->_close ($handle) if $self;
123	},
124	on_eof => sub {
125	return unless $self;
126	my ($handle) = @_;
127
128	if ($handle->{mcp_out}) {
129	$handle->{mcp_eof} = 1;
130	$self->_drain ($handle);
131	} else {
132	$self->_close ($handle);
133	}
134	},
135	);
136
137	if ($fh_out) {
138	$handle->{mcp_out} = AnyEvent::Handle->new (
139	fh => $fh_out,
140	on_error => sub {
141	$self->event (connection_error => $_[0], $_[2]) if $self;
142	$self->close if $self;
143	},
144	);
145	}
146
147	$handle->{mcp_peer} = {};
148	$handle->{mcp_peer}{mcp_handle} = $handle;
149	weaken $handle->{mcp_peer}{mcp_handle};
150	$self->{connections}{$handle} = $handle;
151
152	$handle
153	}
154
155	sub _read {
156	my ($self, $handle) = @_;
157
158	weaken $self;
159
160	$handle->push_read (line => sub {
161	my ($handle, $line) = @_;
162
163	return unless $self;
164
165	++$handle->{mcp_active};
166	$self->json_rpc (
167	$line,
168	sub {
169	--$handle->{mcp_active};
170	return if $handle->{mcp_closed};
171
172	($handle->{mcp_out} \|\| $handle)->push_write ($_[0] . "\n")
173	if defined $_[0];
174	$self->_drain ($handle) if $handle->{mcp_eof};
175	},
176	peer => $handle->{mcp_peer},
177	on_notify => sub {
178	return if $handle->{mcp_closed};
179
180	($handle->{mcp_out} \|\| $handle)->push_write (JSON::XS::encode_json ($_[1]) . "\n");
181	},
182	);
183	});
184	}
185
186	sub request_idle_exit {
187	my ($self) = @_;
188
189	return unless $ENV{PERL_ANYEVENT_MCP_MULTIBROKER_PROTOCOL} eq "stdio"
190	&& !$self->{idle_exit_requested};
191
192	for my $handle (values %{$self->{connections}}) {
193	next if $handle->{mcp_closed} \|\| !$handle->{mcp_peer}{aemcp_multibroker};
194
195	($handle->{mcp_out} \|\| $handle)->push_write (JSON::XS::encode_json ({
196	jsonrpc => "2.0", method => "anyevent-mcp/idle-exit", params => {},
197	}) . "\n");
198	$self->{idle_exit_requested} = 1;
199	return 1
200	}
201
202	undef
203	}
204
205	sub dispatch {
206	my ($self, $req) = @_;
207
208	if ($req->{method} eq "anyevent-mcp/shutdown") {
209	my $handle = $req->{peer}{mcp_handle};
210	my $busy = grep $_->{mcp_active} > ($_ == $handle ? 1 : 0),
211	values %{$self->{connections}};
212	my $accepted = $ENV{PERL_ANYEVENT_MCP_MULTIBROKER_PROTOCOL} eq "stdio"
213	&& $req->{peer}{aemcp_multibroker} && $self->{idle_exit_requested}
214	&& !$self->{_idle_shutting_down} && !$busy
215	&& (!$self->{on_idle_shutdown} \|\| $self->{on_idle_shutdown}->($self));
216
217	$req->reply ({ accepted => $accepted ? JSON::XS::true : JSON::XS::false });
218	if ($accepted) {
219	$self->{_idle_shutting_down} = 1;
220	my $out = $handle->{mcp_out} \|\| $handle;
221	weaken $self;
222	if (length $out->{wbuf}) {
223	$out->on_drain (sub { $self->close if $self });
224	} else {
225	AnyEvent::postpone (sub { $self->close if $self });
226	}
227	}
228
229	return;
230	}
231
232	$self->SUPER::dispatch ($req)
233	}
234
235	sub _drain {
236	my ($self, $handle) = @_;
237
238	return if $handle->{mcp_active} \|\| $handle->{mcp_closed};
239
240	my $output = $handle->{mcp_out};
241	if (length $output->{wbuf}) {
242	weaken $self;
243	$output->on_drain (sub { $self->_close ($handle) if $self });
244	} else {
245	$self->_close ($handle);
246	}
247	}
248
249	sub _close {
250	my ($self, $handle) = @_;
251
252	return if $handle->{mcp_closed};
253	$handle->{mcp_closed} = 1;
254
255	my $requests = delete $handle->{mcp_peer}{requests} \|\| {};
256	$_->cancel ("Connection closed") for grep defined, values %$requests;
257
258	delete $self->{connections}{$handle};
259	$handle->{mcp_out}->destroy if $handle->{mcp_out};
260	$handle->destroy;
261	$self->event (connection_close => $handle);
262	}
263
264	sub close {
265	my ($self) = @_;
266
267	return if $self->{_closed}++;
268	delete $self->{listener};
269	$self->_close ($_) for values %{$self->{connections} \|\| {}};
270	$self->event (close =>);
271	}
272
273	sub DESTROY {
274	$_[0]->close
275	}
276
277	1