| 1 |
=head1 NAME |
| 2 |
|
| 3 |
AnyEvent::MCP::Client::JSONL - MCP byte streams and stdio subprocesses |
| 4 |
|
| 5 |
=head1 SYNOPSIS |
| 6 |
|
| 7 |
use AnyEvent::MCP::Client::JSONL; |
| 8 |
|
| 9 |
my $client = AnyEvent::MCP::Client::JSONL->new ( |
| 10 |
command => [$program, @args], |
| 11 |
env => { PERL_ANYEVENT_MCP_MULTIBROKER_PROTOCOL => "stdio" }, |
| 12 |
); |
| 13 |
|
| 14 |
=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 |
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 |
|
| 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 |
=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 |
=head1 METHODS |
| 52 |
|
| 53 |
=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 |
|
| 70 |
=item env => { %environment } |
| 71 |
|
| 72 |
Child environment overrides. Unspecified variables are inherited from |
| 73 |
the parent; the parent environment is unchanged. |
| 74 |
|
| 75 |
=item handle => { %handle } |
| 76 |
|
| 77 |
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 |
|
| 91 |
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 |
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 |
An accepted multibroker idle shutdown is the exception: its child can exit |
| 100 |
and perform cleanup without receiving SIGTERM. |
| 101 |
|
| 102 |
=back |
| 103 |
|
| 104 |
=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 |
use Scalar::Util qw(weaken blessed); |
| 114 |
|
| 115 |
sub attach { |
| 116 |
my ($self, $fh_in, $fh_out) = @_; |
| 117 |
|
| 118 |
$self->close if $self->{read_handle}; |
| 119 |
weaken $self; |
| 120 |
$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 |
$self->{write_handle} = $fh_out ? AnyEvent::Handle->new ( |
| 127 |
%{$self->{handle} || {}}, |
| 128 |
fh => $fh_out, |
| 129 |
on_read => undef, |
| 130 |
on_error => sub { $self->close ($_[2]) if $self }, |
| 131 |
on_eof => sub { $self->close ("MCP output closed") if $self }, |
| 132 |
) : $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 |
|
| 175 |
weaken $self; |
| 176 |
|
| 177 |
$self->{read_handle}->push_read (line => sub { |
| 178 |
my ($handle, $line) = @_; |
| 179 |
|
| 180 |
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 |
kill TERM => $pid unless $self->{idle_shutdown_expected}; |
| 210 |
} |
| 211 |
delete $self->{child}; |
| 212 |
} |
| 213 |
|
| 214 |
1 |
