MCP/Client/JSONL.pm

=head1 NAME

AnyEvent::MCP::Client::JSONL - MCP byte streams and stdio subprocesses

=head1 SYNOPSIS

   use AnyEvent::MCP::Client::JSONL;

   my $client = AnyEvent::MCP::Client::JSONL->new (
      command => [$program, @args],
      env => { PERL_ANYEVENT_MCP_MULTIBROKER_PROTOCOL => "stdio" },
   );

=head1 DESCRIPTION

Use this transport when the MCP server is a subprocess or an already
connected byte stream. It reads one JSON-RPC message per line, so stdout
on a conventional stdio transport must contain protocol messages only.
The C<command> mode here connects the child's stdin and stdout to a
socketpair; diagnostic output belongs on stderr. A stray banner or log
message on the protocol stream can break the connection.

An argument array executes the program directly; a command string runs
through a shell. Prefer the array form when no shell syntax is needed.
The process starts on the first connection, not when the client object
is constructed.

=head1 METHODS

=over

=item $client = AnyEvent::MCP::Client::JSONL->new (%config)

Accepts the common L<AnyEvent::MCP::Client> options and these transport
options:

=over

=item command => [$program, $arg...]

=item command => $shell_command

Argument array for direct execution, or a string run through C</bin/sh -c>.
The child inherits stderr. No command runs until connecting.
Omit this option when using C<attach> instead.

=item env => { %environment }

Child environment overrides. Unspecified variables are inherited from
the parent; the parent environment is unchanged.

=item handle => { %handle }

L<AnyEvent::Handle> options. The filehandle and read/error/EOF callbacks
are managed by this transport.

=item max_size => $bytes

Bounds a buffered incoming JSON line, defaulting to 16 MiB.

=back

C<name> does not load transport defaults. The library does not infer a
subprocess command or environment.

=item $client->attach ($fh_in, $fh_out)

Uses existing handles instead of a command. C<$fh_out> defaults to C<$fh_in>
for a bidirectional socket. The client takes ownership. Attach before sending
requests. Closing an attached connection requires attaching new handles before
reconnecting.

Closing a subprocess connection sends SIGTERM and keeps its child watcher
alive until the child has been reaped. It does not restart an outstanding call.
An accepted multibroker idle shutdown is the exception: its child can exit
and perform cleanup without receiving SIGTERM.

=back

=cut

package AnyEvent::MCP::Client::JSONL;

use common::sense;
use parent "AnyEvent::MCP::Client";
use AnyEvent::Handle ();
use AnyEvent::Util ();
use JSON::XS ();
use Scalar::Util qw(weaken);

sub attach {
   my ($self, $fh_in, $fh_out) = @_;

   $self->close if $self->{read_handle};
   weaken $self;
   $self->{read_handle} = AnyEvent::Handle->new (
      %{$self->{handle} || {}},
      fh       => $fh_in,
      rbuf_max => $self->{max_size} || 16 * 1024 * 1024,
      on_read  => undef,
      on_error => sub { $self->close ($_[2]) if $self },
      on_eof   => sub { $self->close ("MCP stream closed") if $self },
   );
   $self->{write_handle} = $fh_out ? AnyEvent::Handle->new (
      %{$self->{handle} || {}},
      fh       => $fh_out,
      on_read  => undef,
      on_error => sub { $self->close ($_[2]) if $self },
      on_eof   => sub { $self->close ("MCP output closed") if $self },
   ) : $self->{read_handle};
   $self->_read;
}

sub _open {
   my ($self, $done) = @_;

   return $done->(1, undef) if $self->{read_handle};
   return $done->(0, { transport => 1, message => "No command or attached stream" })
      unless $self->{command};

   my $ok = eval {
      local %ENV = (%ENV, %{$self->{env} || {}});
      my ($parent, $child) = AnyEvent::Util::portable_socketpair;
      die "Cannot create MCP socketpair: $!\n" unless defined $parent && defined $child;

      my $pid;
      my $cv = AnyEvent::Util::run_cmd (
         ref $self->{command} eq "ARRAY" ? $self->{command} : ["/bin/sh", "-c", $self->{command}],
         "<" => $child, ">" => $child, '$$' => \$pid, close_all => 1,
      );
      close $child;
      $self->{pid} = $pid;
      $self->attach ($parent);

      weaken $self;
      $cv->cb (sub {
         my $status = $_[0]->recv;
         if ($self && $self->{pid} == $pid) {
            delete $self->{pid};
            delete $self->{child};
            $self->close ("MCP child exited (status $status)");
         }
      });
      $self->{child} = $cv;
      1
   };
   $done->($ok ? (1, undef) : (0, { transport => 1, message => "$@" }));
}

sub _read {
   my ($self) = @_;

   weaken $self;

   $self->{read_handle}->push_read (line => sub {
      my ($handle, $line) = @_;

      return unless $self;
      my $ok = eval {
         $self->_receive (JSON::XS::decode_json $line);
         1
      };
      return $self->close ("$@") unless $ok;
      $self->_read if $self->{read_handle} && !$handle->destroyed;
   });
}

sub _send {
   my ($self, $message, $done) = @_;

   my $ok = eval {
      my $handle = $self->{write_handle} or die "MCP stream is not connected\n";
      $handle->push_write (JSON::XS::encode_json ($message) . "\n");
      1
   };
   $done->($ok ? (1, undef) : (0, { transport => 1, message => "$@" }));
}

sub _shutdown {
   my ($self) = @_;

   my $read = delete $self->{read_handle};
   my $write = delete $self->{write_handle};
   $read->destroy if $read;
   $write->destroy if $write && $write != $read;
   if (my $pid = delete $self->{pid}) {
      kill TERM => $pid unless $self->{idle_shutdown_expected};
   }
   delete $self->{child};
}

1
Revision:	1.6
Committed:	Tue Oct 13 22:54:22 2026 UTC (13 days, 18 hours ago) by root
Branch:	MAIN
Changes since 1.5:	+37 -10 lines
Log Message:	* empty log message *
#	User	Rev	Content
1	root	1.1	=head1 NAME
2
3			AnyEvent::MCP::Client::JSONL - MCP byte streams and stdio subprocesses
4
5			=head1 SYNOPSIS
6
7	root	1.6	use AnyEvent::MCP::Client::JSONL;
8
9	root	1.1	my $client = AnyEvent::MCP::Client::JSONL->new (
10			command => [$program, @args],
11			env => { PERL_ANYEVENT_MCP_MULTIBROKER_PROTOCOL => "stdio" },
12			);
13
14	root	1.2	=head1 DESCRIPTION
15
16			Use this transport when the MCP server is a subprocess or an already
17			connected byte stream. It reads one JSON-RPC message per line, so stdout
18	root	1.5	on a conventional stdio transport must contain protocol messages only.
19			The C<command> mode here connects the child's stdin and stdout to a
20			socketpair; diagnostic output belongs on stderr. A stray banner or log
21			message on the protocol stream can break the connection.
22	root	1.2
23			An argument array executes the program directly; a command string runs
24			through a shell. Prefer the array form when no shell syntax is needed.
25			The process starts on the first connection, not when the client object
26			is constructed.
27
28	root	1.1	=head1 METHODS
29
30	root	1.6	=over
31
32			=item $client = AnyEvent::MCP::Client::JSONL->new (%config)
33
34			Accepts the common L<AnyEvent::MCP::Client> options and these transport
35			options:
36
37			=over
38
39			=item command => [$program, $arg...]
40
41			=item command => $shell_command
42
43			Argument array for direct execution, or a string run through C</bin/sh -c>.
44			The child inherits stderr. No command runs until connecting.
45			Omit this option when using C<attach> instead.
46	root	1.1
47	root	1.6	=item env => { %environment }
48	root	1.1
49	root	1.6	Child environment overrides. Unspecified variables are inherited from
50			the parent; the parent environment is unchanged.
51	root	1.1
52	root	1.6	=item handle => { %handle }
53	root	1.1
54	root	1.6	L<AnyEvent::Handle> options. The filehandle and read/error/EOF callbacks
55			are managed by this transport.
56
57			=item max_size => $bytes
58
59			Bounds a buffered incoming JSON line, defaulting to 16 MiB.
60
61			=back
62
63			C<name> does not load transport defaults. The library does not infer a
64			subprocess command or environment.
65
66			=item $client->attach ($fh_in, $fh_out)
67	root	1.1
68			Uses existing handles instead of a command. C<$fh_out> defaults to C<$fh_in>
69			for a bidirectional socket. The client takes ownership. Attach before sending
70			requests. Closing an attached connection requires attaching new handles before
71			reconnecting.
72
73			Closing a subprocess connection sends SIGTERM and keeps its child watcher
74			alive until the child has been reaped. It does not restart an outstanding call.
75	root	1.3	An accepted multibroker idle shutdown is the exception: its child can exit
76			and perform cleanup without receiving SIGTERM.
77	root	1.1
78	root	1.6	=back
79
80	root	1.1	=cut
81
82			package AnyEvent::MCP::Client::JSONL;
83
84			use common::sense;
85			use parent "AnyEvent::MCP::Client";
86			use AnyEvent::Handle ();
87			use AnyEvent::Util ();
88			use JSON::XS ();
89			use Scalar::Util qw(weaken);
90
91			sub attach {
92			my ($self, $fh_in, $fh_out) = @_;
93
94			$self->close if $self->{read_handle};
95			weaken $self;
96			$self->{read_handle} = AnyEvent::Handle->new (
97			%{$self->{handle} \|\| {}},
98	root	1.4	fh => $fh_in,
99	root	1.1	rbuf_max => $self->{max_size} \|\| 16 * 1024 * 1024,
100	root	1.4	on_read => undef,
101	root	1.1	on_error => sub { $self->close ($_[2]) if $self },
102	root	1.4	on_eof => sub { $self->close ("MCP stream closed") if $self },
103	root	1.1	);
104			$self->{write_handle} = $fh_out ? AnyEvent::Handle->new (
105			%{$self->{handle} \|\| {}},
106	root	1.4	fh => $fh_out,
107			on_read => undef,
108	root	1.1	on_error => sub { $self->close ($_[2]) if $self },
109	root	1.4	on_eof => sub { $self->close ("MCP output closed") if $self },
110	root	1.1	) : $self->{read_handle};
111			$self->_read;
112			}
113
114			sub _open {
115			my ($self, $done) = @_;
116
117			return $done->(1, undef) if $self->{read_handle};
118			return $done->(0, { transport => 1, message => "No command or attached stream" })
119			unless $self->{command};
120
121			my $ok = eval {
122			local %ENV = (%ENV, %{$self->{env} \|\| {}});
123			my ($parent, $child) = AnyEvent::Util::portable_socketpair;
124			die "Cannot create MCP socketpair: $!\n" unless defined $parent && defined $child;
125
126			my $pid;
127			my $cv = AnyEvent::Util::run_cmd (
128			ref $self->{command} eq "ARRAY" ? $self->{command} : ["/bin/sh", "-c", $self->{command}],
129			"<" => $child, ">" => $child, '$$' => \$pid, close_all => 1,
130			);
131			close $child;
132			$self->{pid} = $pid;
133			$self->attach ($parent);
134
135			weaken $self;
136			$cv->cb (sub {
137			my $status = $_[0]->recv;
138			if ($self && $self->{pid} == $pid) {
139			delete $self->{pid};
140			delete $self->{child};
141			$self->close ("MCP child exited (status $status)");
142			}
143			});
144			$self->{child} = $cv;
145			1
146			};
147			$done->($ok ? (1, undef) : (0, { transport => 1, message => "$@" }));
148			}
149
150			sub _read {
151			my ($self) = @_;
152	root	1.4
153	root	1.1	weaken $self;
154
155			$self->{read_handle}->push_read (line => sub {
156			my ($handle, $line) = @_;
157	root	1.4
158	root	1.1	return unless $self;
159			my $ok = eval {
160			$self->_receive (JSON::XS::decode_json $line);
161			1
162			};
163			return $self->close ("$@") unless $ok;
164			$self->_read if $self->{read_handle} && !$handle->destroyed;
165			});
166			}
167
168			sub _send {
169			my ($self, $message, $done) = @_;
170
171			my $ok = eval {
172			my $handle = $self->{write_handle} or die "MCP stream is not connected\n";
173			$handle->push_write (JSON::XS::encode_json ($message) . "\n");
174			1
175			};
176			$done->($ok ? (1, undef) : (0, { transport => 1, message => "$@" }));
177			}
178
179			sub _shutdown {
180			my ($self) = @_;
181
182			my $read = delete $self->{read_handle};
183			my $write = delete $self->{write_handle};
184			$read->destroy if $read;
185			$write->destroy if $write && $write != $read;
186			if (my $pid = delete $self->{pid}) {
187	root	1.3	kill TERM => $pid unless $self->{idle_shutdown_expected};
188	root	1.1	}
189			delete $self->{child};
190			}
191
192			1