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