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_ready => $callback->($client)

Called once after a successful MCP discovery or initialization, including
reconnections. It does not fire for repeated C<connect> calls on an already
connected client.

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