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. C<pending> counts operations interrupted by the close, before
their callbacks were failed. 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};
   # The close event runs after outstanding operations have been failed.
   my $outstanding = scalar keys %{$self->{operations} || {}};
   ++$self->{generation};
   $self->_shutdown;
   delete @$self{qw(connected protocol_version server_info server_capabilities instructions)};

   my $error = { transport => 1, pending => $outstanding,
      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.6
Committed:	Fri Oct 16 08:15:32 2026 UTC (3 weeks, 5 days ago) by root
Branch:	MAIN
Changes since 1.5:	+7 -2 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	root	1.6	error hash. C<pending> counts operations interrupted by the close, before
174			their callbacks were failed. An explicit C<close> without a reason does not
175			trigger this event.
176	root	1.3
177	root	1.5	=back
178
179	root	1.1	=cut
180
181			package AnyEvent::MCP::Client;
182
183			use common::sense;
184			use AnyEvent::MCP ();
185			use AnyEvent ();
186			use AnyEvent::MCP::Client::Request ();
187			use Scalar::Util qw(weaken);
188
189			our @ISA = qw(AnyEvent::MCP::Base);
190
191			sub new {
192			my $class = shift;
193
194			$class->SUPER::new (
195			client_info => { name => "AnyEvent::MCP", version => $AnyEvent::MCP::VERSION },
196			client_capabilities => {},
197			versions => [qw(2026-07-28 2025-11-25 2025-06-18 2025-03-26)],
198			@_,
199			)
200			}
201
202			sub connect {
203			my ($self, $done) = @_;
204
205			return $done->(1, undef) if $self->{connected};
206			if ($self->{connecting}) {
207			push @{$self->{connecting}}, $done;
208			return
209			}
210
211			$self->{connecting} = [$done];
212			weaken $self;
213			$self->_open (sub {
214			my ($success, $error) = @_;
215	root	1.4
216	root	1.1	return unless $self;
217			return $self->_connected (0, $error) unless $success;
218
219			my ($modern) = grep $_ ge "2026", @{$self->{versions}};
220			return $self->_initialize unless $modern;
221
222			$self->{protocol_version} = $modern;
223			$self->_request ("server/discover", {}, sub {
224			my ($success, $result) = @_;
225	root	1.4
226	root	1.1	return unless $self;
227
228			return $self->_initialize if !$success && !$result->{transport};
229			return $self->_connected (0, $result) unless $success;
230
231			return $self->_connected (0, { transport => 1, message => "Invalid discovery result" })
232			unless ref $result->{supportedVersions} eq "ARRAY" && ref $result->{capabilities} eq "HASH";
233
234			my %supported = map { $_ => 1 } @{$result->{supportedVersions}};
235			my ($version) = grep $supported{$_}, @{$self->{versions}};
236			return $self->_connected (0, { transport => 1, message => "No supported protocol version" })
237			unless defined $version;
238			return $self->_initialize if $version lt "2026";
239
240			$self->{protocol_version} = $version;
241			$self->{server_info} = $result->{_meta}{"io.modelcontextprotocol/serverInfo"};
242			$self->{server_capabilities} = $result->{capabilities};
243			$self->{instructions} = $result->{instructions};
244			$self->_connected (1, undef);
245			});
246			});
247			}
248
249			sub _initialize {
250			my ($self) = @_;
251
252			my ($version) = grep $_ lt "2026", @{$self->{versions}};
253			return $self->_connected (0, { transport => 1, message => "Server requires unsupported legacy initialization" })
254			unless defined $version;
255
256			$self->{protocol_version} = $version;
257			weaken $self;
258			$self->_request ("initialize", {
259			protocolVersion => $version,
260			clientInfo => $self->{client_info},
261			capabilities => $self->{client_capabilities},
262			}, sub {
263			my ($success, $result) = @_;
264	root	1.4
265	root	1.1	return unless $self;
266			return $self->_connected (0, $result) unless $success;
267
268			my $version = $result->{protocolVersion};
269			return $self->_connected (0, { transport => 1, message => "Unsupported negotiated protocol version" })
270			unless grep $_ eq $version && $_ lt "2026", @{$self->{versions}};
271
272			$self->{protocol_version} = $version;
273			$self->{server_info} = $result->{serverInfo};
274			$self->{server_capabilities} = $result->{capabilities};
275			$self->{instructions} = $result->{instructions};
276			$self->_send ({ jsonrpc => "2.0", method => "notifications/initialized" },
277			sub { $self->_connected (@_) if $self });
278			});
279			}
280
281			sub _connected {
282			my ($self, $success, $error) = @_;
283
284			my $waiters = delete $self->{connecting} or return;
285			if ($success) {
286			$self->{connected} = 1;
287			} else {
288			$self->_shutdown;
289			delete @$self{qw(connected protocol_version server_info server_capabilities instructions)};
290			}
291			$_->($success, $error) for @$waiters;
292			}
293
294			sub _request {
295			my ($self, $method, $params, $done, $operation) = @_;
296	root	1.4
297	root	1.1	my $id = ++$self->{wire_id};
298			my %params = %$params;
299
300			if ($self->{protocol_version} ge "2026") {
301			$params{_meta} = {
302			%{$params{_meta} \|\| {}},
303			"io.modelcontextprotocol/protocolVersion" => $self->{protocol_version},
304			"io.modelcontextprotocol/clientInfo" => $self->{client_info},
305			"io.modelcontextprotocol/clientCapabilities" => exists $params{_meta}{"io.modelcontextprotocol/clientCapabilities"}
306			? $params{_meta}{"io.modelcontextprotocol/clientCapabilities"} : $self->{client_capabilities},
307			};
308			}
309
310			$self->{pending}{$id} = $done;
311			$operation->{wire_id} = $id if $operation;
312			my $generation = $self->{generation};
313			weaken $self;
314			$self->_send ({ jsonrpc => "2.0", id => $id, method => $method, params => \%params }, sub {
315			my ($success, $error) = @_;
316	root	1.4
317	root	1.1	return if !$self \|\| $self->{generation} != $generation \|\| $success;
318			my $done = delete $self->{pending}{$id} or return;
319			$done->(0, $error);
320			});
321			$id
322			}
323
324			sub _operation {
325			my ($self, $method, $params, $done, $interactions) = @_;
326	root	1.4
327	root	1.1	my $id = ++$self->{operation_id};
328			my $op = $self->{operations}{$id} = {
329	root	1.4	method => $method,
330			params => $params,
331			done => $done,
332			interactions => $interactions,
333	root	1.1	};
334
335			weaken $self;
336			if ($self->{request_timeout}) {
337			$op->{timer} = AnyEvent->timer (after => $self->{request_timeout}, cb => sub {
338			$self->cancel ($id, "Request timed out") if $self;
339			});
340			}
341
342			$self->connect (sub {
343			my ($success, $error) = @_;
344	root	1.4
345	root	1.1	return unless $self && $self->{operations}{$id};
346			return $self->_finish ($id, 0, $error) unless $success;
347			$self->_continue ($id, $params);
348			});
349			$id
350			}
351
352			sub rpc {
353			my ($self, $method, $params, $done) = @_;
354
355			$self->_operation ($method, $params, $done, 0)
356			}
357
358			sub call {
359			my ($self, $name, $args, $done, %params) = @_;
360
361			$self->_operation ("tools/call", { %params, name => $name, arguments => $args }, $done, 1)
362			}
363
364			sub _continue {
365			my ($self, $id, $params) = @_;
366	root	1.4
367	root	1.1	my $op = $self->{operations}{$id} or return;
368
369			weaken $self;
370			$self->_request ($op->{method}, $params, sub {
371			my ($success, $result) = @_;
372	root	1.4
373	root	1.1	return unless $self && $self->{operations}{$id};
374			delete $op->{wire_id};
375
376			if ($success && $op->{interactions} && $result->{resultType} eq "input_required") {
377			return $self->_input ($id, $result)
378			}
379			$self->_finish ($id, $success, $result);
380			}, $op);
381			}
382
383			sub _input {
384			my ($self, $id, $result) = @_;
385	root	1.4
386	root	1.1	my $op = $self->{operations}{$id} or return;
387			my $input = $result->{inputRequests};
388
389			return $self->_finish ($id, 0, { transport => 1, message => "Input required without a client handler" })
390			unless $self->{on_request} && ref $input eq "HASH" && keys %$input;
391			return $self->_finish ($id, 0, { transport => 1, message => "Too many input rounds" })
392			if ++$op->{rounds} > ($self->{max_rounds} \|\| 32);
393
394			weaken $self;
395			my %responses;
396			my $remaining = keys %$input;
397			$op->{input} = [];
398			for my $key (sort keys %$input) {
399			last unless $self->{operations}{$id};
400			my $request = $input->{$key};
401			unless (ref $request eq "HASH" && defined $request->{method} && ref ($request->{params} \|\| {}) eq "HASH") {
402			return $self->_finish ($id, 0, { transport => 1, message => "Invalid input request" })
403			}
404
405			my $req = AnyEvent::MCP::Client::Request->new (
406	root	1.4	client => $self,
407			id => $key,
408			method => $request->{method},
409			params => $request->{params} \|\| {},
410	root	1.1	protocol_version => $self->{protocol_version},
411	root	1.4	operation => $id,
412			on_reply => sub {
413	root	1.1	my ($success, $response) = @_;
414	root	1.4
415	root	1.1	return unless $self->{operations}{$id};
416			return $self->_finish ($id, 0, $response) unless $success;
417			$responses{$key} = $response;
418			return if --$remaining;
419
420			delete $op->{input};
421			my %params = (%{$op->{params}}, inputResponses => \%responses);
422			$params{requestState} = $result->{requestState} if exists $result->{requestState};
423			$self->_continue ($id, \%params);
424			},
425			);
426			push @{$op->{input}}, $req;
427			$self->event (request => $req);
428			}
429			}
430
431			sub _finish {
432			my ($self, $id, $success, $result) = @_;
433
434			my $op = delete $self->{operations}{$id} or return;
435			delete $self->{pending}{$op->{wire_id}} if exists $op->{wire_id};
436			$_->cancel for @{$op->{input} \|\| []};
437			delete $op->{timer};
438			$op->{done}->($success, $result);
439			}
440
441			sub cancel {
442			my ($self, $id, $reason) = @_;
443	root	1.4
444	root	1.1	my $op = $self->{operations}{$id} or return;
445			my $wire = $op->{wire_id};
446			$reason = "Request cancelled" unless defined $reason;
447
448			$self->_finish ($id, 0, { transport => 1, cancelled => 1, message => $reason });
449			if (defined $wire && $self->{connected}) {
450			$self->_send ({
451			jsonrpc => "2.0", method => "notifications/cancelled",
452	root	1.4	params => { requestId => $wire, reason => $reason },
453	root	1.1	}, sub {});
454			$self->_abort ($wire);
455			}
456			}
457
458			sub tools {
459			my ($self, $done) = @_;
460	root	1.4
461	root	1.1	my (@tools, %seen);
462			weaken $self;
463			my $page;
464			$page = sub {
465			my ($cursor) = @_;
466	root	1.4
467	root	1.1	$self->rpc ("tools/list", defined $cursor ? { cursor => $cursor } : {}, sub {
468			my ($success, $result) = @_;
469	root	1.4
470	root	1.1	unless ($success && ref $result->{tools} eq "ARRAY") {
471			undef $page;
472			return $done->(0, $success ? { transport => 1, message => "Invalid tools/list result" } : $result)
473			}
474
475			push @tools, @{$result->{tools}};
476			if (defined $result->{nextCursor} && length $result->{nextCursor}) {
477			if ($seen{$result->{nextCursor}}++) {
478			undef $page;
479			return $done->(0, { transport => 1, message => "Repeated tools/list cursor" })
480			}
481			return $page->($result->{nextCursor})
482			}
483
484			undef $page;
485			$done->(1, \@tools);
486			});
487			};
488			$page->();
489			}
490
491			sub notify {
492			my ($self, $method, $params, $done) = @_;
493	root	1.4
494	root	1.1	$done \|\|= sub {};
495			weaken $self;
496
497			$self->connect (sub {
498			my ($success, $error) = @_;
499	root	1.4
500	root	1.1	return unless $self;
501			return $done->(0, $error) unless $success;
502			$self->_send ({ jsonrpc => "2.0", method => $method, params => $params }, $done);
503			});
504			}
505
506			sub _receive {
507			my ($self, $message) = @_;
508
509			die "Invalid JSON-RPC message\n"
510			unless ref $message eq "HASH" && $message->{jsonrpc} eq "2.0";
511
512			if (exists $message->{method}) {
513			die "Invalid server method\n" if ref $message->{method};
514			if (!exists $message->{id}) {
515			$self->event (notification => $message->{method}, $message->{params} \|\| {});
516			return
517			}
518
519			die "Server-initiated request on modern connection\n" if $self->{protocol_version} ge "2026";
520			weaken $self;
521			my $req = AnyEvent::MCP::Client::Request->new (
522	root	1.4	client => $self,
523			id => $message->{id},
524			method => $message->{method},
525			params => $message->{params} \|\| {},
526	root	1.1	on_reply => sub {
527			my ($success, $result) = @_;
528	root	1.4
529	root	1.1	return unless $self;
530			delete $self->{incoming}{$message->{id}};
531			$self->_send ({
532			jsonrpc => "2.0", id => $message->{id},
533			($success ? "result" : "error") => $result,
534			}, sub {});
535			},
536			);
537			$self->{incoming}{$req->{id}} = $req;
538			if ($req->{method} eq "ping") {
539			$req->reply ({});
540			} elsif ($self->{on_request}) {
541			$self->event (request => $req);
542			} else {
543			$req->error (-32601, "Method not supported");
544			}
545			return
546			}
547
548	root	1.4	my $field = exists $message->{result} ? "result" : "error";
549
550			defined $message->{id} && !ref $message->{id}
551			&& (exists $message->{result}) != (exists $message->{error})
552			&& ref $message->{$field} eq "HASH"
553			or die "Invalid JSON-RPC response\n";
554	root	1.1
555			my $done = delete $self->{pending}{$message->{id}} or return;
556	root	1.4	$done->(exists $message->{result}, $message->{$field});
557	root	1.1	}
558
559			sub _open {
560			$_[1]->(1, undef)
561			}
562
563			sub _abort {
564			}
565
566			sub close {
567			my ($self, $message) = @_;
568	root	1.4
569	root	1.3	my $was_connected = $self->{connected};
570	root	1.6	# The close event runs after outstanding operations have been failed.
571			my $outstanding = scalar keys %{$self->{operations} \|\| {}};
572	root	1.1	++$self->{generation};
573			$self->_shutdown;
574			delete @$self{qw(connected protocol_version server_info server_capabilities instructions)};
575
576	root	1.6	my $error = { transport => 1, pending => $outstanding,
577			message => defined $message ? $message : "Client closed" };
578	root	1.1	my $connecting = delete $self->{connecting} \|\| [];
579			my $pending = delete $self->{pending} \|\| {};
580			my $incoming = delete $self->{incoming} \|\| {};
581			$_->cancel for values %$incoming;
582			$self->_finish ($_, 0, $error) for keys %{$self->{operations} \|\| {}};
583			$_->(0, $error) for values %$pending;
584			$_->(0, $error) for @$connecting;
585	root	1.3	$self->event (close => $error) if $was_connected && defined $message;
586	root	1.1	}
587
588			sub DESTROY {
589			$_[0]->close
590			}
591
592			1