MCP/Client/JSONL.pm

=head1 NAME

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

=head1 SYNOPSIS

   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

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

C<command> is an argument array for direct execution, or a string explicitly
run through C</bin/sh -c>. The child inherits stderr. C<env> contains child
environment overrides. No command runs until connecting.

C<handle> contains AnyEvent::Handle options, excluding the filehandle and
read/error/EOF callbacks managed here. C<max_size> bounds an incoming JSON
line, defaulting to 16 MiB.

C<name> is available for application identification but does not load transport
defaults. The library does not infer a subprocess command or environment.

=head2 $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.

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