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
from a child server must contain protocol messages only. Diagnostic output
belongs on stderr. A single stray banner on stdout 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.

=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;
   }
   delete $self->{child};
}

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