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 FIRST CALL

To invoke a bundled server as a subprocess:

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

   my $client = AnyEvent::MCP::Client::JSONL->new (
      command => ["mcp-fs", "--stdio", "/home/alice/src/project"],
   );
   my $cv = AE::cv;
   $client->call (workdirs => {}, sub { $cv->send (@_) });
   my ($success, $result) = $cv->recv;
   $success or die "$result->{message}\n";
   !$result->{isError} or die "$result->{content}[0]{text}\n";

This assumes the project exists and C<mcp-fs> is installed on C<PATH>.
The subprocess is created on first use and reaped when it exits.
The caller owns the client and its event loop; callbacks also distinguish
protocol success from tool errors. See
L<AnyEvent::MCP::Server::Simple/TUTORIAL> for a complete echo server
and a matching client example.

=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_in> may also be an
L<AnyEvent::Handle>, whose buffered input is preserved. 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 blessed);

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

   $self->close if $self->{read_handle};
   weaken $self;
   $self->{read_handle} = blessed $fh_in && $fh_in->isa ("AnyEvent::Handle")
      ? $fh_in : AnyEvent::Handle->new (%{$self->{handle} || {}}, fh => $fh_in);
   $self->{read_handle}{rbuf_max} = $self->{max_size} || 16 * 1024 * 1024;
   $self->{read_handle}->on_read (undef);
   $self->{read_handle}->on_error (sub { $self->close ($_[2]) if $self });
   $self->{read_handle}->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.8
Committed:	Wed Oct 21 17:28:16 2026 UTC (6 days, 2 hours ago) by root
Branch:	MAIN
CVS Tags:	rel-0_62, rel-0_61, rel-0_6, rel-0_65, HEAD
Changes since 1.7:	+23 -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	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.8	=head1 FIRST CALL
29
30			To invoke a bundled server as a subprocess:
31
32			use AnyEvent;
33			use AnyEvent::MCP::Client::JSONL;
34
35			my $client = AnyEvent::MCP::Client::JSONL->new (
36			command => ["mcp-fs", "--stdio", "/home/alice/src/project"],
37			);
38			my $cv = AE::cv;
39			$client->call (workdirs => {}, sub { $cv->send (@_) });
40			my ($success, $result) = $cv->recv;
41			$success or die "$result->{message}\n";
42			!$result->{isError} or die "$result->{content}[0]{text}\n";
43
44			This assumes the project exists and C<mcp-fs> is installed on C<PATH>.
45			The subprocess is created on first use and reaped when it exits.
46			The caller owns the client and its event loop; callbacks also distinguish
47			protocol success from tool errors. See
48			L<AnyEvent::MCP::Server::Simple/TUTORIAL> for a complete echo server
49			and a matching client example.
50
51	root	1.1	=head1 METHODS
52
53	root	1.6	=over
54
55			=item $client = AnyEvent::MCP::Client::JSONL->new (%config)
56
57			Accepts the common L<AnyEvent::MCP::Client> options and these transport
58			options:
59
60			=over
61
62			=item command => [$program, $arg...]
63
64			=item command => $shell_command
65
66			Argument array for direct execution, or a string run through C</bin/sh -c>.
67			The child inherits stderr. No command runs until connecting.
68			Omit this option when using C<attach> instead.
69	root	1.1
70	root	1.6	=item env => { %environment }
71	root	1.1
72	root	1.6	Child environment overrides. Unspecified variables are inherited from
73			the parent; the parent environment is unchanged.
74	root	1.1
75	root	1.6	=item handle => { %handle }
76	root	1.1
77	root	1.6	L<AnyEvent::Handle> options. The filehandle and read/error/EOF callbacks
78			are managed by this transport.
79
80			=item max_size => $bytes
81
82			Bounds a buffered incoming JSON line, defaulting to 16 MiB.
83
84			=back
85
86			C<name> does not load transport defaults. The library does not infer a
87			subprocess command or environment.
88
89			=item $client->attach ($fh_in, $fh_out)
90	root	1.1
91	root	1.7	Uses existing handles instead of a command. C<$fh_in> may also be an
92			L<AnyEvent::Handle>, whose buffered input is preserved. C<$fh_out> defaults to C<$fh_in>
93	root	1.1	for a bidirectional socket. The client takes ownership. Attach before sending
94			requests. Closing an attached connection requires attaching new handles before
95			reconnecting.
96
97			Closing a subprocess connection sends SIGTERM and keeps its child watcher
98			alive until the child has been reaped. It does not restart an outstanding call.
99	root	1.3	An accepted multibroker idle shutdown is the exception: its child can exit
100			and perform cleanup without receiving SIGTERM.
101	root	1.1
102	root	1.6	=back
103
104	root	1.1	=cut
105
106			package AnyEvent::MCP::Client::JSONL;
107
108			use common::sense;
109			use parent "AnyEvent::MCP::Client";
110			use AnyEvent::Handle ();
111			use AnyEvent::Util ();
112			use JSON::XS ();
113	root	1.7	use Scalar::Util qw(weaken blessed);
114	root	1.1
115			sub attach {
116			my ($self, $fh_in, $fh_out) = @_;
117
118			$self->close if $self->{read_handle};
119			weaken $self;
120	root	1.7	$self->{read_handle} = blessed $fh_in && $fh_in->isa ("AnyEvent::Handle")
121			? $fh_in : AnyEvent::Handle->new (%{$self->{handle} \|\| {}}, fh => $fh_in);
122			$self->{read_handle}{rbuf_max} = $self->{max_size} \|\| 16 * 1024 * 1024;
123			$self->{read_handle}->on_read (undef);
124			$self->{read_handle}->on_error (sub { $self->close ($_[2]) if $self });
125			$self->{read_handle}->on_eof (sub { $self->close ("MCP stream closed") if $self });
126	root	1.1	$self->{write_handle} = $fh_out ? AnyEvent::Handle->new (
127			%{$self->{handle} \|\| {}},
128	root	1.4	fh => $fh_out,
129			on_read => undef,
130	root	1.1	on_error => sub { $self->close ($_[2]) if $self },
131	root	1.4	on_eof => sub { $self->close ("MCP output closed") if $self },
132	root	1.1	) : $self->{read_handle};
133			$self->_read;
134			}
135
136			sub _open {
137			my ($self, $done) = @_;
138
139			return $done->(1, undef) if $self->{read_handle};
140			return $done->(0, { transport => 1, message => "No command or attached stream" })
141			unless $self->{command};
142
143			my $ok = eval {
144			local %ENV = (%ENV, %{$self->{env} \|\| {}});
145			my ($parent, $child) = AnyEvent::Util::portable_socketpair;
146			die "Cannot create MCP socketpair: $!\n" unless defined $parent && defined $child;
147
148			my $pid;
149			my $cv = AnyEvent::Util::run_cmd (
150			ref $self->{command} eq "ARRAY" ? $self->{command} : ["/bin/sh", "-c", $self->{command}],
151			"<" => $child, ">" => $child, '$$' => \$pid, close_all => 1,
152			);
153			close $child;
154			$self->{pid} = $pid;
155			$self->attach ($parent);
156
157			weaken $self;
158			$cv->cb (sub {
159			my $status = $_[0]->recv;
160			if ($self && $self->{pid} == $pid) {
161			delete $self->{pid};
162			delete $self->{child};
163			$self->close ("MCP child exited (status $status)");
164			}
165			});
166			$self->{child} = $cv;
167			1
168			};
169			$done->($ok ? (1, undef) : (0, { transport => 1, message => "$@" }));
170			}
171
172			sub _read {
173			my ($self) = @_;
174	root	1.4
175	root	1.1	weaken $self;
176
177			$self->{read_handle}->push_read (line => sub {
178			my ($handle, $line) = @_;
179	root	1.4
180	root	1.1	return unless $self;
181			my $ok = eval {
182			$self->_receive (JSON::XS::decode_json $line);
183			1
184			};
185			return $self->close ("$@") unless $ok;
186			$self->_read if $self->{read_handle} && !$handle->destroyed;
187			});
188			}
189
190			sub _send {
191			my ($self, $message, $done) = @_;
192
193			my $ok = eval {
194			my $handle = $self->{write_handle} or die "MCP stream is not connected\n";
195			$handle->push_write (JSON::XS::encode_json ($message) . "\n");
196			1
197			};
198			$done->($ok ? (1, undef) : (0, { transport => 1, message => "$@" }));
199			}
200
201			sub _shutdown {
202			my ($self) = @_;
203
204			my $read = delete $self->{read_handle};
205			my $write = delete $self->{write_handle};
206			$read->destroy if $read;
207			$write->destroy if $write && $write != $read;
208			if (my $pid = delete $self->{pid}) {
209	root	1.3	kill TERM => $pid unless $self->{idle_shutdown_expected};
210	root	1.1	}
211			delete $self->{child};
212			}
213
214			1