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.

=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 });
      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.11
Committed:	Sun Nov 1 22:07:43 2026 UTC (2 days, 18 hours ago) by root
Branch:	MAIN
CVS Tags:	HEAD
Changes since 1.10:	+2 -1 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 8 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} \|\| 8 * 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	weaken (my $input = $handle);
139	$handle->{mcp_out} = AnyEvent::Handle->new (
140	fh => $fh_out,
141	on_error => sub {
142	$self->event (connection_error => $input, $_[2]) if $self && $input;
143	$self->close if $self;
144	},
145	);
146	}
147
148	$handle->{mcp_peer} = {};
149	$handle->{mcp_peer}{mcp_handle} = $handle;
150	weaken $handle->{mcp_peer}{mcp_handle};
151	$self->{connections}{$handle} = $handle;
152
153	$handle
154	}
155
156	sub _read {
157	my ($self, $handle) = @_;
158
159	weaken $self;
160
161	$handle->push_read (line => sub {
162	my ($handle, $line) = @_;
163
164	return unless $self;
165
166	++$handle->{mcp_active};
167	$self->json_rpc (
168	$line,
169	sub {
170	--$handle->{mcp_active};
171	return if $handle->{mcp_closed};
172
173	($handle->{mcp_out} \|\| $handle)->push_write ($_[0] . "\n")
174	if defined $_[0];
175	$self->_drain ($handle) if $handle->{mcp_eof};
176	},
177	peer => $handle->{mcp_peer},
178	on_notify => sub {
179	return if $handle->{mcp_closed};
180
181	($handle->{mcp_out} \|\| $handle)->push_write (JSON::XS::encode_json ($_[1]) . "\n");
182	},
183	);
184	});
185	}
186
187	sub request_idle_exit {
188	my ($self) = @_;
189
190	return unless $ENV{PERL_ANYEVENT_MCP_MULTIBROKER_PROTOCOL} eq "stdio"
191	&& !$self->{idle_exit_requested};
192
193	for my $handle (values %{$self->{connections}}) {
194	next if $handle->{mcp_closed} \|\| !$handle->{mcp_peer}{aemcp_multibroker};
195
196	($handle->{mcp_out} \|\| $handle)->push_write (JSON::XS::encode_json ({
197	jsonrpc => "2.0", method => "anyevent-mcp/idle-exit", params => {},
198	}) . "\n");
199	$self->{idle_exit_requested} = 1;
200	return 1
201	}
202
203	undef
204	}
205
206	sub dispatch {
207	my ($self, $req) = @_;
208
209	if ($req->{method} eq "anyevent-mcp/shutdown") {
210	my $handle = $req->{peer}{mcp_handle};
211	my $busy = grep $_->{mcp_active} > ($_ == $handle ? 1 : 0),
212	values %{$self->{connections}};
213	my $accepted = $ENV{PERL_ANYEVENT_MCP_MULTIBROKER_PROTOCOL} eq "stdio"
214	&& $req->{peer}{aemcp_multibroker} && $self->{idle_exit_requested}
215	&& !$self->{_idle_shutting_down} && !$busy
216	&& (!$self->{on_idle_shutdown} \|\| $self->{on_idle_shutdown}->($self));
217
218	$req->reply ({ accepted => $accepted ? JSON::XS::true : JSON::XS::false });
219	if ($accepted) {
220	$self->{_idle_shutting_down} = 1;
221	my $out = $handle->{mcp_out} \|\| $handle;
222	weaken $self;
223	if (length $out->{wbuf}) {
224	$out->on_drain (sub { $self->close if $self });
225	} else {
226	AnyEvent::postpone (sub { $self->close if $self });
227	}
228	}
229
230	return;
231	}
232
233	$self->SUPER::dispatch ($req)
234	}
235
236	sub _drain {
237	my ($self, $handle) = @_;
238
239	return if $handle->{mcp_active} \|\| $handle->{mcp_closed};
240
241	my $output = $handle->{mcp_out};
242	if (length $output->{wbuf}) {
243	weaken $self;
244	$output->on_drain (sub { $self->_close ($handle) if $self });
245	} else {
246	$self->_close ($handle);
247	}
248	}
249
250	sub _close {
251	my ($self, $handle) = @_;
252
253	return if $handle->{mcp_closed};
254	$handle->{mcp_closed} = 1;
255
256	my $requests = delete $handle->{mcp_peer}{requests} \|\| {};
257	$_->cancel ("Connection closed") for grep defined, values %$requests;
258
259	delete $self->{connections}{$handle};
260	$handle->{mcp_out}->destroy if $handle->{mcp_out};
261	$handle->destroy;
262	$self->event (connection_close => $handle);
263	}
264
265	sub close {
266	my ($self) = @_;
267
268	return if $self->{_closed}++;
269	delete $self->{listener};
270	$self->_close ($_) for values %{$self->{connections} \|\| {}};
271	$self->event (close =>);
272	}
273
274	sub DESTROY {
275	$_[0]->close
276	}
277
278	1