ViewVC Help
View File | Revision Log | Show Annotations | Download File
/cvs/cvsroot/AnyEvent-MCP/lib/AnyEvent/MCP/Client.pm
Revision: 1.2
Committed: Fri Oct 9 09:36:48 2026 UTC (2 weeks, 5 days ago) by root
Branch: MAIN
Changes since 1.1: +11 -4 lines
Log Message:
*** empty log message ***

File Contents

# User Rev Content
1 root 1.1 =head1 NAME
2    
3     AnyEvent::MCP::Client - asynchronous MCP client protocol
4    
5     =head1 SYNOPSIS
6    
7     use AnyEvent::MCP::Client::HTTP;
8    
9     my $client = AnyEvent::MCP::Client::HTTP->new (
10     url => $url, bearer => $token,
11     );
12    
13     $client->call (echo => { text => "hello" }, sub {
14     my ($success, $result) = @_;
15     ...
16     });
17    
18     =head1 DESCRIPTION
19    
20 root 1.2 This is the common protocol implementation used by
21     L<AnyEvent::MCP::Client::HTTP> and L<AnyEvent::MCP::Client::JSONL>. Choose a
22     transport, keep the client object alive, and run an AnyEvent event loop.
23     You can call a tool without connecting explicitly: the client connects and
24     discovers or initializes the server as needed. Concurrent requests share that
25     work instead of starting independent connections.
26    
27     An interrupted connection does not mean the server did nothing. A tool might
28     have completed its side effects just before the connection failed. For that
29     reason, the client never replays a tool call automatically. Applications that
30     want retries must decide whether the operation is safe to repeat.
31 root 1.1
32     Modern clients use discovery and per-request metadata. Legacy clients use
33     initialization and ordinary request/response. Legacy server-initiated requests
34     are accepted on JSONL; legacy HTTP GET streams are not implemented.
35    
36     =head1 METHODS
37    
38     =head2 $client = AnyEvent::MCP::Client->new (%config)
39    
40     Normally construct a transport subclass. C<client_info> is the advertised
41     client identity; C<client_capabilities> declares supported interactions.
42     C<versions> lists supported revisions in preference order. C<request_timeout>
43     is an optional overall deadline in seconds, including connection and user input.
44     C<max_rounds> bounds tool-call input continuations, defaulting to 32.
45    
46     C<name> is an optional local configuration name. HTTP loads defaults from
47     C<[client NAME]> only when it is supplied. It is independent of the advertised
48     C<client_info>. Unnamed clients do not access global configuration.
49    
50     =head2 $client->connect ($done->($success, $result))
51    
52     Discovers or initializes the server. On success C<$result> is undef.
53     Any JSON-RPC error from discovery falls back to legacy initialization.
54     Transport failures do not trigger fallback.
55     After connecting, C<protocol_version>, C<server_info>, C<server_capabilities>
56     and C<instructions> are available directly in the client hash.
57    
58     =head2 $id = $client->rpc ($method, $params, $done->($success, $result))
59    
60     Sends a request. C<$params> is a hash reference and is not modified.
61     C<$id> identifies the local operation for cancellation. Callbacks may run
62     before this method returns.
63    
64     On success C<$result> is the entire decoded result, including C<isError>,
65     C<_meta> and C<resultType>. An C<input_required> result is returned unchanged.
66     On failure C<$result> is an error hash: JSON-RPC errors retain their fields,
67     while local errors have C<message> and C<transport> true. Local cancellation
68     also has C<cancelled> true. Each completion callback runs once.
69    
70     =head2 $id = $client->call ($name, $args, $done->($success, $result), %params)
71    
72     Calls a tool and handles modern C<input_required> results via C<on_request>.
73     C<%params> supplies additional protocol parameters, such as C<_meta>. Explicit
74     per-request client capabilities override the constructor capabilities, allowing
75     brokers to advertise the upstream capabilities without changing shared state.
76     The original method and arguments are retained; only server-requested answers
77     and opaque C<requestState> are added for continuation. No connection failure
78     causes a retry. Without C<on_request>, an interaction fails locally.
79    
80     Permission challenges are input requests handled by C<on_request>. To ask
81     locally through the external permission helper:
82    
83     client_capabilities => { elicitation => { form => {} } },
84     on_request => sub {
85     AnyEvent::MCP::Permissions::Ask->elicitation ($_[1]);
86     },
87    
88     Load L<AnyEvent::MCP::Permissions::Ask> before using that callback. It handles
89     permission-tagged elicitation only; applications supporting other elicitation
90     requests dispatch those separately. The client does not silently approve
91     permissions. C<rpc> returns input challenges intact for broker forwarding.
92    
93     =head2 $client->tools ($done->($success, $tools))
94    
95     Collects all pages of tool definitions into an array reference. Names and
96     definitions are preserved. Repeated pagination cursors are rejected.
97    
98     =head2 $client->notify ($method, $params, $done->($success, $result))
99    
100     Sends a notification after connecting. C<$done> is optional. Success means
101     the transport accepted it, not that the server acted on it.
102    
103     =head2 $client->cancel ($id, $reason)
104    
105     Cancels a pending local operation, completes its callback with an error and
106     sends best-effort protocol cancellation for an outstanding wire request.
107     C<$reason> is optional. Late responses are ignored.
108    
109     =head2 $client->close
110    
111     Closes transport resources, cancels input handlers and fails pending callbacks.
112     New operations can reconnect. Previously sent calls are not repeated.
113    
114     =head1 EVENTS
115    
116     =head2 on_request => $callback->($client, $req)
117    
118     Handles an input request asynchronously. C<$req> is a
119     L<AnyEvent::MCP::Client::Request> with C<method> and C<params>.
120     Declare the corresponding capabilities explicitly in C<client_capabilities>.
121     For a broker, C<rpc> can instead return the complete input challenge upstream.
122    
123     =head2 on_notification => $callback->($client, $method, $params)
124    
125     Receives server notifications, including progress and tool-list changes.
126    
127     =cut
128    
129     package AnyEvent::MCP::Client;
130    
131     use common::sense;
132     use AnyEvent::MCP ();
133     use AnyEvent ();
134     use AnyEvent::MCP::Client::Request ();
135     use Scalar::Util qw(weaken);
136    
137     our @ISA = qw(AnyEvent::MCP::Base);
138    
139     sub new {
140     my $class = shift;
141    
142     $class->SUPER::new (
143     client_info => { name => "AnyEvent::MCP", version => $AnyEvent::MCP::VERSION },
144     client_capabilities => {},
145     versions => [qw(2026-07-28 2025-11-25 2025-06-18 2025-03-26)],
146     @_,
147     )
148     }
149    
150     sub connect {
151     my ($self, $done) = @_;
152    
153     return $done->(1, undef) if $self->{connected};
154     if ($self->{connecting}) {
155     push @{$self->{connecting}}, $done;
156     return
157     }
158    
159     $self->{connecting} = [$done];
160     weaken $self;
161     $self->_open (sub {
162     my ($success, $error) = @_;
163     return unless $self;
164     return $self->_connected (0, $error) unless $success;
165    
166     my ($modern) = grep $_ ge "2026", @{$self->{versions}};
167     return $self->_initialize unless $modern;
168    
169     $self->{protocol_version} = $modern;
170     $self->_request ("server/discover", {}, sub {
171     my ($success, $result) = @_;
172     return unless $self;
173    
174     return $self->_initialize if !$success && !$result->{transport};
175     return $self->_connected (0, $result) unless $success;
176    
177     return $self->_connected (0, { transport => 1, message => "Invalid discovery result" })
178     unless ref $result->{supportedVersions} eq "ARRAY" && ref $result->{capabilities} eq "HASH";
179    
180     my %supported = map { $_ => 1 } @{$result->{supportedVersions}};
181     my ($version) = grep $supported{$_}, @{$self->{versions}};
182     return $self->_connected (0, { transport => 1, message => "No supported protocol version" })
183     unless defined $version;
184     return $self->_initialize if $version lt "2026";
185    
186     $self->{protocol_version} = $version;
187     $self->{server_info} = $result->{_meta}{"io.modelcontextprotocol/serverInfo"};
188     $self->{server_capabilities} = $result->{capabilities};
189     $self->{instructions} = $result->{instructions};
190     $self->_connected (1, undef);
191     });
192     });
193     }
194    
195     sub _initialize {
196     my ($self) = @_;
197    
198     my ($version) = grep $_ lt "2026", @{$self->{versions}};
199     return $self->_connected (0, { transport => 1, message => "Server requires unsupported legacy initialization" })
200     unless defined $version;
201    
202     $self->{protocol_version} = $version;
203     weaken $self;
204     $self->_request ("initialize", {
205     protocolVersion => $version,
206     clientInfo => $self->{client_info},
207     capabilities => $self->{client_capabilities},
208     }, sub {
209     my ($success, $result) = @_;
210     return unless $self;
211     return $self->_connected (0, $result) unless $success;
212    
213     my $version = $result->{protocolVersion};
214     return $self->_connected (0, { transport => 1, message => "Unsupported negotiated protocol version" })
215     unless grep $_ eq $version && $_ lt "2026", @{$self->{versions}};
216    
217     $self->{protocol_version} = $version;
218     $self->{server_info} = $result->{serverInfo};
219     $self->{server_capabilities} = $result->{capabilities};
220     $self->{instructions} = $result->{instructions};
221     $self->_send ({ jsonrpc => "2.0", method => "notifications/initialized" },
222     sub { $self->_connected (@_) if $self });
223     });
224     }
225    
226     sub _connected {
227     my ($self, $success, $error) = @_;
228    
229     my $waiters = delete $self->{connecting} or return;
230     if ($success) {
231     $self->{connected} = 1;
232     } else {
233     $self->_shutdown;
234     delete @$self{qw(connected protocol_version server_info server_capabilities instructions)};
235     }
236     $_->($success, $error) for @$waiters;
237     }
238    
239     sub _request {
240     my ($self, $method, $params, $done, $operation) = @_;
241     my $id = ++$self->{wire_id};
242     my %params = %$params;
243    
244     if ($self->{protocol_version} ge "2026") {
245     $params{_meta} = {
246     %{$params{_meta} || {}},
247     "io.modelcontextprotocol/protocolVersion" => $self->{protocol_version},
248     "io.modelcontextprotocol/clientInfo" => $self->{client_info},
249     "io.modelcontextprotocol/clientCapabilities" => exists $params{_meta}{"io.modelcontextprotocol/clientCapabilities"}
250     ? $params{_meta}{"io.modelcontextprotocol/clientCapabilities"} : $self->{client_capabilities},
251     };
252     }
253    
254     $self->{pending}{$id} = $done;
255     $operation->{wire_id} = $id if $operation;
256     my $generation = $self->{generation};
257     weaken $self;
258     $self->_send ({ jsonrpc => "2.0", id => $id, method => $method, params => \%params }, sub {
259     my ($success, $error) = @_;
260     return if !$self || $self->{generation} != $generation || $success;
261     my $done = delete $self->{pending}{$id} or return;
262     $done->(0, $error);
263     });
264     $id
265     }
266    
267     sub _operation {
268     my ($self, $method, $params, $done, $interactions) = @_;
269     my $id = ++$self->{operation_id};
270     my $op = $self->{operations}{$id} = {
271     method => $method, params => $params, done => $done, interactions => $interactions,
272     };
273    
274     weaken $self;
275     if ($self->{request_timeout}) {
276     $op->{timer} = AnyEvent->timer (after => $self->{request_timeout}, cb => sub {
277     $self->cancel ($id, "Request timed out") if $self;
278     });
279     }
280    
281     $self->connect (sub {
282     my ($success, $error) = @_;
283     return unless $self && $self->{operations}{$id};
284     return $self->_finish ($id, 0, $error) unless $success;
285     $self->_continue ($id, $params);
286     });
287     $id
288     }
289    
290     sub rpc {
291     my ($self, $method, $params, $done) = @_;
292    
293     $self->_operation ($method, $params, $done, 0)
294     }
295    
296     sub call {
297     my ($self, $name, $args, $done, %params) = @_;
298    
299     $self->_operation ("tools/call", { %params, name => $name, arguments => $args }, $done, 1)
300     }
301    
302     sub _continue {
303     my ($self, $id, $params) = @_;
304     my $op = $self->{operations}{$id} or return;
305    
306     weaken $self;
307     $self->_request ($op->{method}, $params, sub {
308     my ($success, $result) = @_;
309     return unless $self && $self->{operations}{$id};
310     delete $op->{wire_id};
311    
312     if ($success && $op->{interactions} && $result->{resultType} eq "input_required") {
313     return $self->_input ($id, $result)
314     }
315     $self->_finish ($id, $success, $result);
316     }, $op);
317     }
318    
319     sub _input {
320     my ($self, $id, $result) = @_;
321     my $op = $self->{operations}{$id} or return;
322     my $input = $result->{inputRequests};
323    
324     return $self->_finish ($id, 0, { transport => 1, message => "Input required without a client handler" })
325     unless $self->{on_request} && ref $input eq "HASH" && keys %$input;
326     return $self->_finish ($id, 0, { transport => 1, message => "Too many input rounds" })
327     if ++$op->{rounds} > ($self->{max_rounds} || 32);
328    
329     weaken $self;
330     my %responses;
331     my $remaining = keys %$input;
332     $op->{input} = [];
333     for my $key (sort keys %$input) {
334     last unless $self->{operations}{$id};
335     my $request = $input->{$key};
336     unless (ref $request eq "HASH" && defined $request->{method} && ref ($request->{params} || {}) eq "HASH") {
337     return $self->_finish ($id, 0, { transport => 1, message => "Invalid input request" })
338     }
339    
340     my $req = AnyEvent::MCP::Client::Request->new (
341     client => $self, id => $key, method => $request->{method}, params => $request->{params} || {},
342     protocol_version => $self->{protocol_version},
343     operation => $id,
344     on_reply => sub {
345     my ($success, $response) = @_;
346     return unless $self->{operations}{$id};
347     return $self->_finish ($id, 0, $response) unless $success;
348     $responses{$key} = $response;
349     return if --$remaining;
350    
351     delete $op->{input};
352     my %params = (%{$op->{params}}, inputResponses => \%responses);
353     $params{requestState} = $result->{requestState} if exists $result->{requestState};
354     $self->_continue ($id, \%params);
355     },
356     );
357     push @{$op->{input}}, $req;
358     $self->event (request => $req);
359     }
360     }
361    
362     sub _finish {
363     my ($self, $id, $success, $result) = @_;
364    
365     my $op = delete $self->{operations}{$id} or return;
366     delete $self->{pending}{$op->{wire_id}} if exists $op->{wire_id};
367     $_->cancel for @{$op->{input} || []};
368     delete $op->{timer};
369     $op->{done}->($success, $result);
370     }
371    
372     sub cancel {
373     my ($self, $id, $reason) = @_;
374     my $op = $self->{operations}{$id} or return;
375     my $wire = $op->{wire_id};
376     $reason = "Request cancelled" unless defined $reason;
377    
378     $self->_finish ($id, 0, { transport => 1, cancelled => 1, message => $reason });
379     if (defined $wire && $self->{connected}) {
380     $self->_send ({
381     jsonrpc => "2.0", method => "notifications/cancelled",
382     params => { requestId => $wire, reason => $reason },
383     }, sub {});
384     $self->_abort ($wire);
385     }
386     }
387    
388     sub tools {
389     my ($self, $done) = @_;
390     my (@tools, %seen);
391     weaken $self;
392     my $page;
393     $page = sub {
394     my ($cursor) = @_;
395     $self->rpc ("tools/list", defined $cursor ? { cursor => $cursor } : {}, sub {
396     my ($success, $result) = @_;
397     unless ($success && ref $result->{tools} eq "ARRAY") {
398     undef $page;
399     return $done->(0, $success ? { transport => 1, message => "Invalid tools/list result" } : $result)
400     }
401    
402     push @tools, @{$result->{tools}};
403     if (defined $result->{nextCursor} && length $result->{nextCursor}) {
404     if ($seen{$result->{nextCursor}}++) {
405     undef $page;
406     return $done->(0, { transport => 1, message => "Repeated tools/list cursor" })
407     }
408     return $page->($result->{nextCursor})
409     }
410    
411     undef $page;
412     $done->(1, \@tools);
413     });
414     };
415     $page->();
416     }
417    
418     sub notify {
419     my ($self, $method, $params, $done) = @_;
420     $done ||= sub {};
421     weaken $self;
422    
423     $self->connect (sub {
424     my ($success, $error) = @_;
425     return unless $self;
426     return $done->(0, $error) unless $success;
427     $self->_send ({ jsonrpc => "2.0", method => $method, params => $params }, $done);
428     });
429     }
430    
431     sub _receive {
432     my ($self, $message) = @_;
433    
434     die "Invalid JSON-RPC message\n"
435     unless ref $message eq "HASH" && $message->{jsonrpc} eq "2.0";
436    
437     if (exists $message->{method}) {
438     die "Invalid server method\n" if ref $message->{method};
439     if (!exists $message->{id}) {
440     $self->event (notification => $message->{method}, $message->{params} || {});
441     return
442     }
443    
444     die "Server-initiated request on modern connection\n" if $self->{protocol_version} ge "2026";
445     weaken $self;
446     my $req = AnyEvent::MCP::Client::Request->new (
447     client => $self, id => $message->{id}, method => $message->{method}, params => $message->{params} || {},
448     on_reply => sub {
449     my ($success, $result) = @_;
450     return unless $self;
451     delete $self->{incoming}{$message->{id}};
452     $self->_send ({
453     jsonrpc => "2.0", id => $message->{id},
454     ($success ? "result" : "error") => $result,
455     }, sub {});
456     },
457     );
458     $self->{incoming}{$req->{id}} = $req;
459     if ($req->{method} eq "ping") {
460     $req->reply ({});
461     } elsif ($self->{on_request}) {
462     $self->event (request => $req);
463     } else {
464     $req->error (-32601, "Method not supported");
465     }
466     return
467     }
468    
469     die "Invalid JSON-RPC response\n"
470     unless defined $message->{id} && !ref $message->{id}
471     && (exists $message->{result}) != (exists $message->{error})
472     && (exists $message->{result} ? ref $message->{result} eq "HASH" : ref $message->{error} eq "HASH");
473    
474     my $done = delete $self->{pending}{$message->{id}} or return;
475     $done->(exists $message->{result}, exists $message->{result} ? $message->{result} : $message->{error});
476     }
477    
478     sub _open {
479     $_[1]->(1, undef)
480     }
481    
482     sub _abort {
483     }
484    
485     sub close {
486     my ($self, $message) = @_;
487     ++$self->{generation};
488     $self->_shutdown;
489     delete @$self{qw(connected protocol_version server_info server_capabilities instructions)};
490    
491     my $error = { transport => 1, message => defined $message ? $message : "Client closed" };
492     my $connecting = delete $self->{connecting} || [];
493     my $pending = delete $self->{pending} || {};
494     my $incoming = delete $self->{incoming} || {};
495     $_->cancel for values %$incoming;
496     $self->_finish ($_, 0, $error) for keys %{$self->{operations} || {}};
497     $_->(0, $error) for values %$pending;
498     $_->(0, $error) for @$connecting;
499     }
500    
501     sub DESTROY {
502     $_[0]->close
503     }
504    
505     1