AnyEvent/MCP/Client.pm

=head1 NAME

AnyEvent::MCP::Client - asynchronous MCP client protocol

=head1 SYNOPSIS

   use AnyEvent::MCP::Client::HTTP;

   my $client = AnyEvent::MCP::Client::HTTP->new (
      url => $url, bearer => $token,
   );

   $client->call (echo => { text => "hello" }, sub {
      my ($success, $result) = @_;
      ...
   });

=head1 DESCRIPTION

This is the common protocol implementation used by
L<AnyEvent::MCP::Client::HTTP> and L<AnyEvent::MCP::Client::JSONL>. Choose a
transport, keep the client object alive, and run an AnyEvent event loop.
You can call a tool without connecting explicitly: the client connects and
discovers or initializes the server as needed. Concurrent requests share that
work instead of starting independent connections.

An interrupted connection does not mean the server did nothing. A tool might
have completed its side effects just before the connection failed. For that
reason, the client never replays a tool call automatically. Applications that
want retries must decide whether the operation is safe to repeat.

Modern clients use discovery and per-request metadata. Legacy clients use
initialization and ordinary request/response. Legacy server-initiated requests
are accepted on JSONL; legacy HTTP GET streams are not implemented.

=head1 METHODS

=over

=item $client = AnyEvent::MCP::Client->new (%config)

Normally construct a transport subclass. Its constructor accepts these
common options as well as the transport-specific options:

=over

=item client_info => { name => $name, version => $version }

Advertised client identity. Defaults to C<AnyEvent::MCP> and this
distribution's version.

=item client_capabilities => { %capabilities }

Supported client interactions, defaulting to an empty hash. Advertising a
capability does not install a handler; supply C<on_request> when needed.

=item versions => [$version...]

Supported protocol revisions in preference order. Defaults to
C<2026-07-28>, C<2025-11-25>, C<2025-06-18> and C<2025-03-26>.
Here, modern means the 2026 discovery/input-challenge protocol; legacy
means the 2025 initialization protocol.

=item request_timeout => $seconds

Optional overall deadline for C<rpc> and C<call>, including connection
and user input. Omitted or zero means no overall deadline. This is separate
from HTTP's inactivity timeout.

=item max_rounds => $count

Maximum input-challenge rounds handled by C<call>, defaulting to 32.

=item name => $name

Optional local configuration name, independent of C<client_info>.
HTTP loads C<[client NAME]> defaults only when this is supplied;
JSONL does not load transport defaults.

=back

Event callbacks are also constructor options; see L</EVENTS>.

=item $client->connect ($done->($success, $result))

Discovers or initializes the server. On success C<$result> is undef.
Any JSON-RPC error from discovery falls back to legacy initialization.
Transport failures do not trigger fallback.
After connecting, C<protocol_version>, C<server_info>, C<server_capabilities>
and C<instructions> are available directly in the client hash.

=item $id = $client->rpc ($method, $params, $done->($success, $result))

Sends a request. C<$params> is a hash reference and is not modified.
C<$id> identifies the local operation for cancellation. Callbacks may run
before this method returns.

On success C<$result> is the entire decoded result, including C<isError>,
C<_meta> and C<resultType>. A result with C<resultType> set to
C<input_required> is returned unchanged. C<$success> reports whether the
JSON-RPC request succeeded, not whether the tool succeeded: a tool result
with C<isError> true still has C<$success> true.
On failure C<$result> is an error hash: JSON-RPC errors retain their fields,
while local errors have C<message> and C<transport> true. Local cancellation
also has C<cancelled> true. Each completion callback runs once.

=item $id = $client->call ($name, $args, $done->($success, $result), %params)

Calls a tool and handles modern C<input_required> results via C<on_request>.
C<$args> is a hash reference of tool arguments. C<%params> supplies
additional protocol parameters, such as C<_meta>. Set
C<< $params{_meta}{"io.modelcontextprotocol/clientCapabilities"} >>
to override the constructor capabilities for this call, allowing
brokers to advertise the upstream capabilities without changing shared state.
The original method and arguments are retained; only server-requested answers
and opaque C<requestState> are added for continuation. No connection failure
causes a retry. Without C<on_request>, an interaction fails locally.

Permission challenges are input requests handled by C<on_request>. To ask
locally through the external permission helper:

   client_capabilities => { elicitation => { form => {} } },
   on_request => sub {
      AnyEvent::MCP::Permissions::Ask->elicitation ($_[1]);
   },

Load L<AnyEvent::MCP::Permissions::Ask> before using that callback. It handles
permission-tagged elicitation only; applications supporting other elicitation
requests dispatch those separately. The client does not silently approve
permissions. C<rpc> returns input challenges intact for broker forwarding.

=item $client->tools ($done->($success, $tools))

Collects all pages of tool definitions into an array reference. Names and
definitions are preserved. Repeated pagination cursors are rejected.

=item $client->notify ($method, $params, $done->($success, $result))

Sends a notification after connecting. C<$done> is optional. Success means
the transport accepted it, not that the server acted on it.

=item $client->cancel ($id, $reason)

Cancels a pending local operation, completes its callback with an error and
sends best-effort protocol cancellation for an outstanding wire request.
C<$reason> is optional. Late responses are ignored.

=item $client->close

Closes transport resources, cancels input handlers and fails pending callbacks.
New operations can reconnect. Previously sent calls are not repeated.

=back

=head1 EVENTS

=over

=item on_request => $callback->($client, $req)

Handles an input request asynchronously. C<$req> is a
L<AnyEvent::MCP::Client::Request> with C<method> and C<params>.
Declare the corresponding capabilities explicitly in C<client_capabilities>.
For a broker, C<rpc> can instead return the complete input challenge upstream.

=item on_notification => $callback->($client, $method, $params)

Receives server notifications, including progress and tool-list changes.

=item on_close => $callback->($client, $error)

Called when an established connection closes unexpectedly, with a transport
error hash. An explicit C<close> without a reason does not trigger this event.

=back

=cut

package AnyEvent::MCP::Client;

use common::sense;
use AnyEvent::MCP ();
use AnyEvent ();
use AnyEvent::MCP::Client::Request ();
use Scalar::Util qw(weaken);

our @ISA = qw(AnyEvent::MCP::Base);

sub new {
   my $class = shift;

   $class->SUPER::new (
      client_info         => { name => "AnyEvent::MCP", version => $AnyEvent::MCP::VERSION },
      client_capabilities => {},
      versions            => [qw(2026-07-28 2025-11-25 2025-06-18 2025-03-26)],
      @_,
   )
}

sub connect {
   my ($self, $done) = @_;

   return $done->(1, undef) if $self->{connected};
   if ($self->{connecting}) {
      push @{$self->{connecting}}, $done;
      return
   }

   $self->{connecting} = [$done];
   weaken $self;
   $self->_open (sub {
      my ($success, $error) = @_;

      return unless $self;
      return $self->_connected (0, $error) unless $success;

      my ($modern) = grep $_ ge "2026", @{$self->{versions}};
      return $self->_initialize unless $modern;

      $self->{protocol_version} = $modern;
      $self->_request ("server/discover", {}, sub {
         my ($success, $result) = @_;

         return unless $self;

         return $self->_initialize if !$success && !$result->{transport};
         return $self->_connected (0, $result) unless $success;

         return $self->_connected (0, { transport => 1, message => "Invalid discovery result" })
            unless ref $result->{supportedVersions} eq "ARRAY" && ref $result->{capabilities} eq "HASH";

         my %supported = map { $_ => 1 } @{$result->{supportedVersions}};
         my ($version) = grep $supported{$_}, @{$self->{versions}};
         return $self->_connected (0, { transport => 1, message => "No supported protocol version" })
            unless defined $version;
         return $self->_initialize if $version lt "2026";

         $self->{protocol_version} = $version;
         $self->{server_info} = $result->{_meta}{"io.modelcontextprotocol/serverInfo"};
         $self->{server_capabilities} = $result->{capabilities};
         $self->{instructions} = $result->{instructions};
         $self->_connected (1, undef);
      });
   });
}

sub _initialize {
   my ($self) = @_;

   my ($version) = grep $_ lt "2026", @{$self->{versions}};
   return $self->_connected (0, { transport => 1, message => "Server requires unsupported legacy initialization" })
      unless defined $version;

   $self->{protocol_version} = $version;
   weaken $self;
   $self->_request ("initialize", {
      protocolVersion => $version,
      clientInfo      => $self->{client_info},
      capabilities    => $self->{client_capabilities},
   }, sub {
      my ($success, $result) = @_;

      return unless $self;
      return $self->_connected (0, $result) unless $success;

      my $version = $result->{protocolVersion};
      return $self->_connected (0, { transport => 1, message => "Unsupported negotiated protocol version" })
         unless grep $_ eq $version && $_ lt "2026", @{$self->{versions}};

      $self->{protocol_version} = $version;
      $self->{server_info} = $result->{serverInfo};
      $self->{server_capabilities} = $result->{capabilities};
      $self->{instructions} = $result->{instructions};
      $self->_send ({ jsonrpc => "2.0", method => "notifications/initialized" },
         sub { $self->_connected (@_) if $self });
   });
}

sub _connected {
   my ($self, $success, $error) = @_;

   my $waiters = delete $self->{connecting} or return;
   if ($success) {
      $self->{connected} = 1;
   } else {
      $self->_shutdown;
      delete @$self{qw(connected protocol_version server_info server_capabilities instructions)};
   }
   $_->($success, $error) for @$waiters;
}

sub _request {
   my ($self, $method, $params, $done, $operation) = @_;

   my $id = ++$self->{wire_id};
   my %params = %$params;

   if ($self->{protocol_version} ge "2026") {
      $params{_meta} = {
         %{$params{_meta} || {}},
         "io.modelcontextprotocol/protocolVersion"    => $self->{protocol_version},
         "io.modelcontextprotocol/clientInfo"         => $self->{client_info},
         "io.modelcontextprotocol/clientCapabilities" => exists $params{_meta}{"io.modelcontextprotocol/clientCapabilities"}
            ? $params{_meta}{"io.modelcontextprotocol/clientCapabilities"} : $self->{client_capabilities},
      };
   }

   $self->{pending}{$id} = $done;
   $operation->{wire_id} = $id if $operation;
   my $generation = $self->{generation};
   weaken $self;
   $self->_send ({ jsonrpc => "2.0", id => $id, method => $method, params => \%params }, sub {
      my ($success, $error) = @_;

      return if !$self || $self->{generation} != $generation || $success;
      my $done = delete $self->{pending}{$id} or return;
      $done->(0, $error);
   });
   $id
}

sub _operation {
   my ($self, $method, $params, $done, $interactions) = @_;

   my $id = ++$self->{operation_id};
   my $op = $self->{operations}{$id} = {
      method       => $method,
      params       => $params,
      done         => $done,
      interactions => $interactions,
   };

   weaken $self;
   if ($self->{request_timeout}) {
      $op->{timer} = AnyEvent->timer (after => $self->{request_timeout}, cb => sub {
         $self->cancel ($id, "Request timed out") if $self;
      });
   }

   $self->connect (sub {
      my ($success, $error) = @_;

      return unless $self && $self->{operations}{$id};
      return $self->_finish ($id, 0, $error) unless $success;
      $self->_continue ($id, $params);
   });
   $id
}

sub rpc {
   my ($self, $method, $params, $done) = @_;

   $self->_operation ($method, $params, $done, 0)
}

sub call {
   my ($self, $name, $args, $done, %params) = @_;

   $self->_operation ("tools/call", { %params, name => $name, arguments => $args }, $done, 1)
}

sub _continue {
   my ($self, $id, $params) = @_;

   my $op = $self->{operations}{$id} or return;

   weaken $self;
   $self->_request ($op->{method}, $params, sub {
      my ($success, $result) = @_;

      return unless $self && $self->{operations}{$id};
      delete $op->{wire_id};

      if ($success && $op->{interactions} && $result->{resultType} eq "input_required") {
         return $self->_input ($id, $result)
      }
      $self->_finish ($id, $success, $result);
   }, $op);
}

sub _input {
   my ($self, $id, $result) = @_;

   my $op = $self->{operations}{$id} or return;
   my $input = $result->{inputRequests};

   return $self->_finish ($id, 0, { transport => 1, message => "Input required without a client handler" })
      unless $self->{on_request} && ref $input eq "HASH" && keys %$input;
   return $self->_finish ($id, 0, { transport => 1, message => "Too many input rounds" })
      if ++$op->{rounds} > ($self->{max_rounds} || 32);

   weaken $self;
   my %responses;
   my $remaining = keys %$input;
   $op->{input} = [];
   for my $key (sort keys %$input) {
      last unless $self->{operations}{$id};
      my $request = $input->{$key};
      unless (ref $request eq "HASH" && defined $request->{method} && ref ($request->{params} || {}) eq "HASH") {
         return $self->_finish ($id, 0, { transport => 1, message => "Invalid input request" })
      }

      my $req = AnyEvent::MCP::Client::Request->new (
         client           => $self,
         id               => $key,
         method           => $request->{method},
         params           => $request->{params} || {},
         protocol_version => $self->{protocol_version},
         operation        => $id,
         on_reply         => sub {
            my ($success, $response) = @_;

            return unless $self->{operations}{$id};
            return $self->_finish ($id, 0, $response) unless $success;
            $responses{$key} = $response;
            return if --$remaining;

            delete $op->{input};
            my %params = (%{$op->{params}}, inputResponses => \%responses);
            $params{requestState} = $result->{requestState} if exists $result->{requestState};
            $self->_continue ($id, \%params);
         },
      );
      push @{$op->{input}}, $req;
      $self->event (request => $req);
   }
}

sub _finish {
   my ($self, $id, $success, $result) = @_;

   my $op = delete $self->{operations}{$id} or return;
   delete $self->{pending}{$op->{wire_id}} if exists $op->{wire_id};
   $_->cancel for @{$op->{input} || []};
   delete $op->{timer};
   $op->{done}->($success, $result);
}

sub cancel {
   my ($self, $id, $reason) = @_;

   my $op = $self->{operations}{$id} or return;
   my $wire = $op->{wire_id};
   $reason = "Request cancelled" unless defined $reason;

   $self->_finish ($id, 0, { transport => 1, cancelled => 1, message => $reason });
   if (defined $wire && $self->{connected}) {
      $self->_send ({
         jsonrpc => "2.0", method => "notifications/cancelled",
         params  => { requestId => $wire, reason => $reason },
      }, sub {});
      $self->_abort ($wire);
   }
}

sub tools {
   my ($self, $done) = @_;

   my (@tools, %seen);
   weaken $self;
   my $page;
   $page = sub {
      my ($cursor) = @_;

      $self->rpc ("tools/list", defined $cursor ? { cursor => $cursor } : {}, sub {
         my ($success, $result) = @_;

         unless ($success && ref $result->{tools} eq "ARRAY") {
            undef $page;
            return $done->(0, $success ? { transport => 1, message => "Invalid tools/list result" } : $result)
         }

         push @tools, @{$result->{tools}};
         if (defined $result->{nextCursor} && length $result->{nextCursor}) {
            if ($seen{$result->{nextCursor}}++) {
               undef $page;
               return $done->(0, { transport => 1, message => "Repeated tools/list cursor" })
            }
            return $page->($result->{nextCursor})
         }

         undef $page;
         $done->(1, \@tools);
      });
   };
   $page->();
}

sub notify {
   my ($self, $method, $params, $done) = @_;

   $done ||= sub {};
   weaken $self;

   $self->connect (sub {
      my ($success, $error) = @_;

      return unless $self;
      return $done->(0, $error) unless $success;
      $self->_send ({ jsonrpc => "2.0", method => $method, params => $params }, $done);
   });
}

sub _receive {
   my ($self, $message) = @_;

   die "Invalid JSON-RPC message\n"
      unless ref $message eq "HASH" && $message->{jsonrpc} eq "2.0";

   if (exists $message->{method}) {
      die "Invalid server method\n" if ref $message->{method};
      if (!exists $message->{id}) {
         $self->event (notification => $message->{method}, $message->{params} || {});
         return
      }

      die "Server-initiated request on modern connection\n" if $self->{protocol_version} ge "2026";
      weaken $self;
      my $req = AnyEvent::MCP::Client::Request->new (
         client   => $self,
         id       => $message->{id},
         method   => $message->{method},
         params   => $message->{params} || {},
         on_reply => sub {
            my ($success, $result) = @_;

            return unless $self;
            delete $self->{incoming}{$message->{id}};
            $self->_send ({
               jsonrpc => "2.0", id => $message->{id},
               ($success ? "result" : "error") => $result,
            }, sub {});
         },
      );
      $self->{incoming}{$req->{id}} = $req;
      if ($req->{method} eq "ping") {
         $req->reply ({});
      } elsif ($self->{on_request}) {
         $self->event (request => $req);
      } else {
         $req->error (-32601, "Method not supported");
      }
      return
   }

   my $field = exists $message->{result} ? "result" : "error";

   defined $message->{id} && !ref $message->{id}
      && (exists $message->{result}) != (exists $message->{error})
      && ref $message->{$field} eq "HASH"
      or die "Invalid JSON-RPC response\n";

   my $done = delete $self->{pending}{$message->{id}} or return;
   $done->(exists $message->{result}, $message->{$field});
}

sub _open {
   $_[1]->(1, undef)
}

sub _abort {
}

sub close {
   my ($self, $message) = @_;

   my $was_connected = $self->{connected};
   ++$self->{generation};
   $self->_shutdown;
   delete @$self{qw(connected protocol_version server_info server_capabilities instructions)};

   my $error = { transport => 1, message => defined $message ? $message : "Client closed" };
   my $connecting = delete $self->{connecting} || [];
   my $pending = delete $self->{pending} || {};
   my $incoming = delete $self->{incoming} || {};
   $_->cancel for values %$incoming;
   $self->_finish ($_, 0, $error) for keys %{$self->{operations} || {}};
   $_->(0, $error) for values %$pending;
   $_->(0, $error) for @$connecting;
   $self->event (close => $error) if $was_connected && defined $message;
}

sub DESTROY {
   $_[0]->close
}

1
Revision:	1.5
Committed:	Tue Oct 13 22:54:21 2026 UTC (2 weeks, 6 days ago) by root
Branch:	MAIN
Changes since 1.4:	+68 -23 lines
Log Message:	* empty log message *
#	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	root	1.5	=over
39	root	1.1
40	root	1.5	=item $client = AnyEvent::MCP::Client->new (%config)
41	root	1.1
42	root	1.5	Normally construct a transport subclass. Its constructor accepts these
43			common options as well as the transport-specific options:
44
45			=over
46
47			=item client_info => { name => $name, version => $version }
48
49			Advertised client identity. Defaults to C<AnyEvent::MCP> and this
50			distribution's version.
51
52			=item client_capabilities => { %capabilities }
53
54			Supported client interactions, defaulting to an empty hash. Advertising a
55			capability does not install a handler; supply C<on_request> when needed.
56
57			=item versions => [$version...]
58
59			Supported protocol revisions in preference order. Defaults to
60			C<2026-07-28>, C<2025-11-25>, C<2025-06-18> and C<2025-03-26>.
61			Here, modern means the 2026 discovery/input-challenge protocol; legacy
62			means the 2025 initialization protocol.
63
64			=item request_timeout => $seconds
65
66			Optional overall deadline for C<rpc> and C<call>, including connection
67			and user input. Omitted or zero means no overall deadline. This is separate
68			from HTTP's inactivity timeout.
69
70			=item max_rounds => $count
71
72			Maximum input-challenge rounds handled by C<call>, defaulting to 32.
73
74			=item name => $name
75
76			Optional local configuration name, independent of C<client_info>.
77			HTTP loads C<[client NAME]> defaults only when this is supplied;
78			JSONL does not load transport defaults.
79
80			=back
81
82			Event callbacks are also constructor options; see L</EVENTS>.
83
84			=item $client->connect ($done->($success, $result))
85	root	1.1
86			Discovers or initializes the server. On success C<$result> is undef.
87			Any JSON-RPC error from discovery falls back to legacy initialization.
88			Transport failures do not trigger fallback.
89			After connecting, C<protocol_version>, C<server_info>, C<server_capabilities>
90			and C<instructions> are available directly in the client hash.
91
92	root	1.5	=item $id = $client->rpc ($method, $params, $done->($success, $result))
93	root	1.1
94			Sends a request. C<$params> is a hash reference and is not modified.
95			C<$id> identifies the local operation for cancellation. Callbacks may run
96			before this method returns.
97
98			On success C<$result> is the entire decoded result, including C<isError>,
99	root	1.5	C<_meta> and C<resultType>. A result with C<resultType> set to
100			C<input_required> is returned unchanged. C<$success> reports whether the
101			JSON-RPC request succeeded, not whether the tool succeeded: a tool result
102			with C<isError> true still has C<$success> true.
103	root	1.1	On failure C<$result> is an error hash: JSON-RPC errors retain their fields,
104			while local errors have C<message> and C<transport> true. Local cancellation
105			also has C<cancelled> true. Each completion callback runs once.
106
107	root	1.5	=item $id = $client->call ($name, $args, $done->($success, $result), %params)
108	root	1.1
109			Calls a tool and handles modern C<input_required> results via C<on_request>.
110	root	1.5	C<$args> is a hash reference of tool arguments. C<%params> supplies
111			additional protocol parameters, such as C<_meta>. Set
112			C<< $params{_meta}{"io.modelcontextprotocol/clientCapabilities"} >>
113			to override the constructor capabilities for this call, allowing
114	root	1.1	brokers to advertise the upstream capabilities without changing shared state.
115			The original method and arguments are retained; only server-requested answers
116			and opaque C<requestState> are added for continuation. No connection failure
117			causes a retry. Without C<on_request>, an interaction fails locally.
118
119			Permission challenges are input requests handled by C<on_request>. To ask
120			locally through the external permission helper:
121
122			client_capabilities => { elicitation => { form => {} } },
123			on_request => sub {
124			AnyEvent::MCP::Permissions::Ask->elicitation ($_[1]);
125			},
126
127			Load L<AnyEvent::MCP::Permissions::Ask> before using that callback. It handles
128			permission-tagged elicitation only; applications supporting other elicitation
129			requests dispatch those separately. The client does not silently approve
130			permissions. C<rpc> returns input challenges intact for broker forwarding.
131
132	root	1.5	=item $client->tools ($done->($success, $tools))
133	root	1.1
134			Collects all pages of tool definitions into an array reference. Names and
135			definitions are preserved. Repeated pagination cursors are rejected.
136
137	root	1.5	=item $client->notify ($method, $params, $done->($success, $result))
138	root	1.1
139			Sends a notification after connecting. C<$done> is optional. Success means
140			the transport accepted it, not that the server acted on it.
141
142	root	1.5	=item $client->cancel ($id, $reason)
143	root	1.1
144			Cancels a pending local operation, completes its callback with an error and
145			sends best-effort protocol cancellation for an outstanding wire request.
146			C<$reason> is optional. Late responses are ignored.
147
148	root	1.5	=item $client->close
149	root	1.1
150			Closes transport resources, cancels input handlers and fails pending callbacks.
151			New operations can reconnect. Previously sent calls are not repeated.
152
153	root	1.5	=back
154
155	root	1.1	=head1 EVENTS
156
157	root	1.5	=over
158
159			=item on_request => $callback->($client, $req)
160	root	1.1
161			Handles an input request asynchronously. C<$req> is a
162			L<AnyEvent::MCP::Client::Request> with C<method> and C<params>.
163			Declare the corresponding capabilities explicitly in C<client_capabilities>.
164			For a broker, C<rpc> can instead return the complete input challenge upstream.
165
166	root	1.5	=item on_notification => $callback->($client, $method, $params)
167	root	1.1
168			Receives server notifications, including progress and tool-list changes.
169
170	root	1.5	=item on_close => $callback->($client, $error)
171	root	1.3
172			Called when an established connection closes unexpectedly, with a transport
173			error hash. An explicit C<close> without a reason does not trigger this event.
174
175	root	1.5	=back
176
177	root	1.1	=cut
178
179			package AnyEvent::MCP::Client;
180
181			use common::sense;
182			use AnyEvent::MCP ();
183			use AnyEvent ();
184			use AnyEvent::MCP::Client::Request ();
185			use Scalar::Util qw(weaken);
186
187			our @ISA = qw(AnyEvent::MCP::Base);
188
189			sub new {
190			my $class = shift;
191
192			$class->SUPER::new (
193			client_info => { name => "AnyEvent::MCP", version => $AnyEvent::MCP::VERSION },
194			client_capabilities => {},
195			versions => [qw(2026-07-28 2025-11-25 2025-06-18 2025-03-26)],
196			@_,
197			)
198			}
199
200			sub connect {
201			my ($self, $done) = @_;
202
203			return $done->(1, undef) if $self->{connected};
204			if ($self->{connecting}) {
205			push @{$self->{connecting}}, $done;
206			return
207			}
208
209			$self->{connecting} = [$done];
210			weaken $self;
211			$self->_open (sub {
212			my ($success, $error) = @_;
213	root	1.4
214	root	1.1	return unless $self;
215			return $self->_connected (0, $error) unless $success;
216
217			my ($modern) = grep $_ ge "2026", @{$self->{versions}};
218			return $self->_initialize unless $modern;
219
220			$self->{protocol_version} = $modern;
221			$self->_request ("server/discover", {}, sub {
222			my ($success, $result) = @_;
223	root	1.4
224	root	1.1	return unless $self;
225
226			return $self->_initialize if !$success && !$result->{transport};
227			return $self->_connected (0, $result) unless $success;
228
229			return $self->_connected (0, { transport => 1, message => "Invalid discovery result" })
230			unless ref $result->{supportedVersions} eq "ARRAY" && ref $result->{capabilities} eq "HASH";
231
232			my %supported = map { $_ => 1 } @{$result->{supportedVersions}};
233			my ($version) = grep $supported{$_}, @{$self->{versions}};
234			return $self->_connected (0, { transport => 1, message => "No supported protocol version" })
235			unless defined $version;
236			return $self->_initialize if $version lt "2026";
237
238			$self->{protocol_version} = $version;
239			$self->{server_info} = $result->{_meta}{"io.modelcontextprotocol/serverInfo"};
240			$self->{server_capabilities} = $result->{capabilities};
241			$self->{instructions} = $result->{instructions};
242			$self->_connected (1, undef);
243			});
244			});
245			}
246
247			sub _initialize {
248			my ($self) = @_;
249
250			my ($version) = grep $_ lt "2026", @{$self->{versions}};
251			return $self->_connected (0, { transport => 1, message => "Server requires unsupported legacy initialization" })
252			unless defined $version;
253
254			$self->{protocol_version} = $version;
255			weaken $self;
256			$self->_request ("initialize", {
257			protocolVersion => $version,
258			clientInfo => $self->{client_info},
259			capabilities => $self->{client_capabilities},
260			}, sub {
261			my ($success, $result) = @_;
262	root	1.4
263	root	1.1	return unless $self;
264			return $self->_connected (0, $result) unless $success;
265
266			my $version = $result->{protocolVersion};
267			return $self->_connected (0, { transport => 1, message => "Unsupported negotiated protocol version" })
268			unless grep $_ eq $version && $_ lt "2026", @{$self->{versions}};
269
270			$self->{protocol_version} = $version;
271			$self->{server_info} = $result->{serverInfo};
272			$self->{server_capabilities} = $result->{capabilities};
273			$self->{instructions} = $result->{instructions};
274			$self->_send ({ jsonrpc => "2.0", method => "notifications/initialized" },
275			sub { $self->_connected (@_) if $self });
276			});
277			}
278
279			sub _connected {
280			my ($self, $success, $error) = @_;
281
282			my $waiters = delete $self->{connecting} or return;
283			if ($success) {
284			$self->{connected} = 1;
285			} else {
286			$self->_shutdown;
287			delete @$self{qw(connected protocol_version server_info server_capabilities instructions)};
288			}
289			$_->($success, $error) for @$waiters;
290			}
291
292			sub _request {
293			my ($self, $method, $params, $done, $operation) = @_;
294	root	1.4
295	root	1.1	my $id = ++$self->{wire_id};
296			my %params = %$params;
297
298			if ($self->{protocol_version} ge "2026") {
299			$params{_meta} = {
300			%{$params{_meta} \|\| {}},
301			"io.modelcontextprotocol/protocolVersion" => $self->{protocol_version},
302			"io.modelcontextprotocol/clientInfo" => $self->{client_info},
303			"io.modelcontextprotocol/clientCapabilities" => exists $params{_meta}{"io.modelcontextprotocol/clientCapabilities"}
304			? $params{_meta}{"io.modelcontextprotocol/clientCapabilities"} : $self->{client_capabilities},
305			};
306			}
307
308			$self->{pending}{$id} = $done;
309			$operation->{wire_id} = $id if $operation;
310			my $generation = $self->{generation};
311			weaken $self;
312			$self->_send ({ jsonrpc => "2.0", id => $id, method => $method, params => \%params }, sub {
313			my ($success, $error) = @_;
314	root	1.4
315	root	1.1	return if !$self \|\| $self->{generation} != $generation \|\| $success;
316			my $done = delete $self->{pending}{$id} or return;
317			$done->(0, $error);
318			});
319			$id
320			}
321
322			sub _operation {
323			my ($self, $method, $params, $done, $interactions) = @_;
324	root	1.4
325	root	1.1	my $id = ++$self->{operation_id};
326			my $op = $self->{operations}{$id} = {
327	root	1.4	method => $method,
328			params => $params,
329			done => $done,
330			interactions => $interactions,
331	root	1.1	};
332
333			weaken $self;
334			if ($self->{request_timeout}) {
335			$op->{timer} = AnyEvent->timer (after => $self->{request_timeout}, cb => sub {
336			$self->cancel ($id, "Request timed out") if $self;
337			});
338			}
339
340			$self->connect (sub {
341			my ($success, $error) = @_;
342	root	1.4
343	root	1.1	return unless $self && $self->{operations}{$id};
344			return $self->_finish ($id, 0, $error) unless $success;
345			$self->_continue ($id, $params);
346			});
347			$id
348			}
349
350			sub rpc {
351			my ($self, $method, $params, $done) = @_;
352
353			$self->_operation ($method, $params, $done, 0)
354			}
355
356			sub call {
357			my ($self, $name, $args, $done, %params) = @_;
358
359			$self->_operation ("tools/call", { %params, name => $name, arguments => $args }, $done, 1)
360			}
361
362			sub _continue {
363			my ($self, $id, $params) = @_;
364	root	1.4
365	root	1.1	my $op = $self->{operations}{$id} or return;
366
367			weaken $self;
368			$self->_request ($op->{method}, $params, sub {
369			my ($success, $result) = @_;
370	root	1.4
371	root	1.1	return unless $self && $self->{operations}{$id};
372			delete $op->{wire_id};
373
374			if ($success && $op->{interactions} && $result->{resultType} eq "input_required") {
375			return $self->_input ($id, $result)
376			}
377			$self->_finish ($id, $success, $result);
378			}, $op);
379			}
380
381			sub _input {
382			my ($self, $id, $result) = @_;
383	root	1.4
384	root	1.1	my $op = $self->{operations}{$id} or return;
385			my $input = $result->{inputRequests};
386
387			return $self->_finish ($id, 0, { transport => 1, message => "Input required without a client handler" })
388			unless $self->{on_request} && ref $input eq "HASH" && keys %$input;
389			return $self->_finish ($id, 0, { transport => 1, message => "Too many input rounds" })
390			if ++$op->{rounds} > ($self->{max_rounds} \|\| 32);
391
392			weaken $self;
393			my %responses;
394			my $remaining = keys %$input;
395			$op->{input} = [];
396			for my $key (sort keys %$input) {
397			last unless $self->{operations}{$id};
398			my $request = $input->{$key};
399			unless (ref $request eq "HASH" && defined $request->{method} && ref ($request->{params} \|\| {}) eq "HASH") {
400			return $self->_finish ($id, 0, { transport => 1, message => "Invalid input request" })
401			}
402
403			my $req = AnyEvent::MCP::Client::Request->new (
404	root	1.4	client => $self,
405			id => $key,
406			method => $request->{method},
407			params => $request->{params} \|\| {},
408	root	1.1	protocol_version => $self->{protocol_version},
409	root	1.4	operation => $id,
410			on_reply => sub {
411	root	1.1	my ($success, $response) = @_;
412	root	1.4
413	root	1.1	return unless $self->{operations}{$id};
414			return $self->_finish ($id, 0, $response) unless $success;
415			$responses{$key} = $response;
416			return if --$remaining;
417
418			delete $op->{input};
419			my %params = (%{$op->{params}}, inputResponses => \%responses);
420			$params{requestState} = $result->{requestState} if exists $result->{requestState};
421			$self->_continue ($id, \%params);
422			},
423			);
424			push @{$op->{input}}, $req;
425			$self->event (request => $req);
426			}
427			}
428
429			sub _finish {
430			my ($self, $id, $success, $result) = @_;
431
432			my $op = delete $self->{operations}{$id} or return;
433			delete $self->{pending}{$op->{wire_id}} if exists $op->{wire_id};
434			$_->cancel for @{$op->{input} \|\| []};
435			delete $op->{timer};
436			$op->{done}->($success, $result);
437			}
438
439			sub cancel {
440			my ($self, $id, $reason) = @_;
441	root	1.4
442	root	1.1	my $op = $self->{operations}{$id} or return;
443			my $wire = $op->{wire_id};
444			$reason = "Request cancelled" unless defined $reason;
445
446			$self->_finish ($id, 0, { transport => 1, cancelled => 1, message => $reason });
447			if (defined $wire && $self->{connected}) {
448			$self->_send ({
449			jsonrpc => "2.0", method => "notifications/cancelled",
450	root	1.4	params => { requestId => $wire, reason => $reason },
451	root	1.1	}, sub {});
452			$self->_abort ($wire);
453			}
454			}
455
456			sub tools {
457			my ($self, $done) = @_;
458	root	1.4
459	root	1.1	my (@tools, %seen);
460			weaken $self;
461			my $page;
462			$page = sub {
463			my ($cursor) = @_;
464	root	1.4
465	root	1.1	$self->rpc ("tools/list", defined $cursor ? { cursor => $cursor } : {}, sub {
466			my ($success, $result) = @_;
467	root	1.4
468	root	1.1	unless ($success && ref $result->{tools} eq "ARRAY") {
469			undef $page;
470			return $done->(0, $success ? { transport => 1, message => "Invalid tools/list result" } : $result)
471			}
472
473			push @tools, @{$result->{tools}};
474			if (defined $result->{nextCursor} && length $result->{nextCursor}) {
475			if ($seen{$result->{nextCursor}}++) {
476			undef $page;
477			return $done->(0, { transport => 1, message => "Repeated tools/list cursor" })
478			}
479			return $page->($result->{nextCursor})
480			}
481
482			undef $page;
483			$done->(1, \@tools);
484			});
485			};
486			$page->();
487			}
488
489			sub notify {
490			my ($self, $method, $params, $done) = @_;
491	root	1.4
492	root	1.1	$done \|\|= sub {};
493			weaken $self;
494
495			$self->connect (sub {
496			my ($success, $error) = @_;
497	root	1.4
498	root	1.1	return unless $self;
499			return $done->(0, $error) unless $success;
500			$self->_send ({ jsonrpc => "2.0", method => $method, params => $params }, $done);
501			});
502			}
503
504			sub _receive {
505			my ($self, $message) = @_;
506
507			die "Invalid JSON-RPC message\n"
508			unless ref $message eq "HASH" && $message->{jsonrpc} eq "2.0";
509
510			if (exists $message->{method}) {
511			die "Invalid server method\n" if ref $message->{method};
512			if (!exists $message->{id}) {
513			$self->event (notification => $message->{method}, $message->{params} \|\| {});
514			return
515			}
516
517			die "Server-initiated request on modern connection\n" if $self->{protocol_version} ge "2026";
518			weaken $self;
519			my $req = AnyEvent::MCP::Client::Request->new (
520	root	1.4	client => $self,
521			id => $message->{id},
522			method => $message->{method},
523			params => $message->{params} \|\| {},
524	root	1.1	on_reply => sub {
525			my ($success, $result) = @_;
526	root	1.4
527	root	1.1	return unless $self;
528			delete $self->{incoming}{$message->{id}};
529			$self->_send ({
530			jsonrpc => "2.0", id => $message->{id},
531			($success ? "result" : "error") => $result,
532			}, sub {});
533			},
534			);
535			$self->{incoming}{$req->{id}} = $req;
536			if ($req->{method} eq "ping") {
537			$req->reply ({});
538			} elsif ($self->{on_request}) {
539			$self->event (request => $req);
540			} else {
541			$req->error (-32601, "Method not supported");
542			}
543			return
544			}
545
546	root	1.4	my $field = exists $message->{result} ? "result" : "error";
547
548			defined $message->{id} && !ref $message->{id}
549			&& (exists $message->{result}) != (exists $message->{error})
550			&& ref $message->{$field} eq "HASH"
551			or die "Invalid JSON-RPC response\n";
552	root	1.1
553			my $done = delete $self->{pending}{$message->{id}} or return;
554	root	1.4	$done->(exists $message->{result}, $message->{$field});
555	root	1.1	}
556
557			sub _open {
558			$_[1]->(1, undef)
559			}
560
561			sub _abort {
562			}
563
564			sub close {
565			my ($self, $message) = @_;
566	root	1.4
567	root	1.3	my $was_connected = $self->{connected};
568	root	1.1	++$self->{generation};
569			$self->_shutdown;
570			delete @$self{qw(connected protocol_version server_info server_capabilities instructions)};
571
572			my $error = { transport => 1, message => defined $message ? $message : "Client closed" };
573			my $connecting = delete $self->{connecting} \|\| [];
574			my $pending = delete $self->{pending} \|\| {};
575			my $incoming = delete $self->{incoming} \|\| {};
576			$_->cancel for values %$incoming;
577			$self->_finish ($_, 0, $error) for keys %{$self->{operations} \|\| {}};
578			$_->(0, $error) for values %$pending;
579			$_->(0, $error) for @$connecting;
580	root	1.3	$self->event (close => $error) if $was_connected && defined $message;
581	root	1.1	}
582
583			sub DESTROY {
584			$_[0]->close
585			}
586
587			1