MCP/Client/HTTP.pm

=head1 NAME

AnyEvent::MCP::Client::HTTP - MCP over AnyEvent::HTTP

=head1 SYNOPSIS

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

   my $client = AnyEvent::MCP::Client::HTTP->new (
      name => "example", url => $url, bearer => $token,
      timeout => 30,
   );

=head1 DESCRIPTION

Use this transport to talk to an MCP server through HTTP. The caller supplies
the endpoint and, if needed, credentials; the common Client layer handles
discovery and tool calls. Unlike a browser, this client does not follow
redirects or silently retry requests. A redirected request could disclose
credentials to another host, and retrying a tool call could perform an
operation twice.

=head1 FIRST CALL

For a bundled MCP server already listening on HTTP, use the full URL
it prints with C<--print-url>:

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

   my $client = AnyEvent::MCP::Client::HTTP->new (
      url => "http://127.0.0.1:31740/TOKEN/",
   );
   my $cv = AE::cv;
   $client->call (workdirs => {}, sub { $cv->send (@_) });
   my ($success, $result) = $cv->recv;
   $success or die "$result->{message}\n";
   !$result->{isError} or die "$result->{content}[0]{text}\n";

Replace the example URL with the actual credential-bearing URL. The
common L<AnyEvent::MCP::Client> layer negotiates MCP and handles
callbacks. A successful JSON-RPC exchange can still return a tool error
as C<isError>; always inspect both.

=head1 METHODS

=over

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

Accepts the common L<AnyEvent::MCP::Client> options and these transport
options. Options are flat, rather than nested in an HTTP configuration hash.

=over

=item url => $url

Complete endpoint URL, optionally including a URL-path token. Required
unless supplied by the named client configuration.

=item bearer => $token

Supplies C<Authorization: Bearer TOKEN>. Without it, the URL alone is
used; a bearer token is not inferred from a URL. The server must accept
the selected credential. With the bundled servers, the URL-path token
and bearer header present the same static credential.

=item name => $name

Loads C<url>, C<bearer> and C<timeout> defaults from C<[client NAME]>.
Explicit constructor arguments override them, including explicitly undefined
credentials. Without a name no global configuration is accessed.

=item timeout => $seconds

HTTP inactivity timeout, defaulting to 300 seconds. Use the common
C<request_timeout> option for an overall tool-call deadline.

=item headers => { %headers }

Additional HTTP headers. Protocol headers and an explicitly configured
bearer header are supplied by this module.

=item max_size => $bytes

Bounds a JSON response or an individual buffered SSE event, defaulting to
16 MiB. SSE notifications are delivered incrementally rather than after
the HTTP response ends.

=back

C<proxy>, C<tls_ctx>, C<handle_params>, C<sessionid>, C<on_prepare>,
C<on_connect> and C<tcp_connect> are passed to L<AnyEvent::HTTP>.
No HTTP cookies or Basic authentication are supplied by this module.
Redirects and automatic HTTP retries are disabled.

Legacy session IDs are retained from initialization. An expired session fails
outstanding requests; a later operation can initialize again. Legacy GET
streams and the old separate SSE endpoint transport are not implemented.

=back

=cut

package AnyEvent::MCP::Client::HTTP;

use common::sense;
use parent "AnyEvent::MCP::Client";
use AnyEvent::HTTP ();
use JSON::XS ();
use MIME::Base64 ();
use Scalar::Util qw(weaken);

sub new {
   my ($class, %kv) = @_;

   if (defined $kv{name}) {
      require AnyEvent::MCP::Config;
      my $settings = AnyEvent::MCP::Config->global->{section}{"client $kv{name}"} || {};
      %kv = (
         (map { exists $settings->{$_} ? ($_ => $settings->{$_}) : () }
            qw(url bearer timeout)),
         %kv,
      );
   }

   defined $kv{url}
      or die "HTTP client requires a URL\n";
   !exists $kv{username} && !exists $kv{password}
      or die "HTTP Basic authentication is unsupported\n";
   die "Invalid bearer token\n" if defined $kv{bearer} && $kv{bearer} !~ /\A[^\s]+\z/;

   $class->SUPER::new (timeout => 300, %kv)
}

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

   my $key = exists $message->{id} && !exists $message->{result} && !exists $message->{error}
      ? $message->{id} : "send:" . ++$self->{send_id};
   my $generation = $self->{generation};
   my %headers = (
      %{$self->{headers} || {}},
      "content-type" => "application/json",
      accept         => "application/json, text/event-stream",
   );

   $headers{authorization} = "Bearer $self->{bearer}" if defined $self->{bearer};
   $headers{"mcp-protocol-version"} = $self->{protocol_version} if defined $self->{protocol_version};
   $headers{"mcp-session-id"} = $self->{session}
      if $self->{protocol_version} lt "2026" && defined $self->{session};

   if ($self->{protocol_version} ge "2026" && exists $message->{method} && exists $message->{id}) {
      $headers{"mcp-method"} = $message->{method};
      if ($message->{method} =~ /\A(?:tools\/call|prompts\/get|resources\/read)\z/) {
         my $name = exists $message->{params}{name} ? $message->{params}{name} : $message->{params}{uri};
         $headers{"mcp-name"} = $name =~ /[^\x20-\x7e]/
            ? "=?base64?" . MIME::Base64::encode_base64 (AnyEvent::MCP::utf8enc ($name), "") . "?="
            : $name;
      }
   }

   my ($buffer, $sse, $received, $failure) = ("", 0, 0);
   my $limit = $self->{max_size} || 16 * 1024 * 1024;
   weaken $self;

   my $deliver = sub {
      my ($response) = @_;

      if (ref $response eq "HASH" && !exists $response->{method}) {
         die "HTTP response id mismatch\n"
            unless exists $message->{id} && defined $response->{id}
               && "$response->{id}" eq "$message->{id}";
         $received = 1;
      }
      $self->_receive ($response);
   };

   my $guard;
   my $ok = eval {
      $guard = AnyEvent::HTTP::http_request POST => $self->{url},
         (map { exists $self->{$_} ? ($_ => $self->{$_}) : () }
            qw(timeout proxy tls_ctx handle_params sessionid on_prepare on_connect tcp_connect)),
         headers    => \%headers,
         body       => JSON::XS::encode_json ($message),
         recurse    => 0,
         persistent => 0,
         on_header  => sub {
            my ($hdr) = @_;

            return 0 unless $self && $self->{generation} == $generation;
            $sse = $hdr->{"content-type"} =~ m{\Atext/event-stream(?:\s*;|\z)}i;
            if ($message->{method} eq "initialize" && $hdr->{Status} >= 200 && $hdr->{Status} < 300
               && defined $hdr->{"mcp-session-id"}) {
               $self->{session} = $hdr->{"mcp-session-id"};
            }
            1
         },
         on_body => sub {
            my ($chunk, $hdr) = @_;

            return 0 unless $self && $self->{generation} == $generation;
            $buffer .= $chunk;

            my $ok = eval {
               if ($sse) {
                  while ($buffer =~ s/\A(.*?)(?:\r\n\r\n|\n\n|\r\r)//s) {
                     my $event = $1;
                     die "SSE event exceeds size limit\n" if length $event > $limit;
                     my @data = $event =~ /^data:[ \t]?(.*?)(?:\r?$)/mg;
                     $deliver->(JSON::XS::decode_json join "\n", @data) if @data;
                  }
               }
               die "HTTP response exceeds size limit\n" if length $buffer > $limit;
               1
            };
            $failure = $@ unless $ok;
            $ok
         },
         sub {
            my (undef, $hdr) = @_;

            return unless $self && $self->{generation} == $generation;
            delete $self->{http_requests}{$key};

            if ($hdr->{Status} == 404 && defined $self->{session} && !$received) {
               return $self->close ("MCP session expired")
            }

            my $ok = eval {
               die $failure if $failure;
               if (!$sse && length $buffer) {
                  my $response = eval { JSON::XS::decode_json $buffer };
                  if (ref $response eq "HASH" && ($hdr->{Status} >= 200 && $hdr->{Status} < 300
                     || ref $response->{error} eq "HASH")) {
                     $deliver->($response);
                  } else {
                     die "HTTP $hdr->{Status} $hdr->{Reason}\n"
                  }
               }

               die "HTTP $hdr->{Status} $hdr->{Reason}\n"
                  unless $hdr->{Status} >= 200 && $hdr->{Status} < 300;
               die "HTTP response has no matching JSON-RPC reply\n"
                  if exists $message->{id} && exists $message->{method} && !$received;
               1
            };
            $done->($ok ? (1, undef) : (0, { transport => 1, message => "$@" }));
         };
      1
   };
   if ($ok) {
      $self->{http_requests}{$key} = $guard if $self && $guard;
   } else {
      $done->(0, { transport => 1, message => "$@" });
   }
}

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

   delete $self->{http_requests}{$id};
}

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

   delete @$self{qw(http_requests session)};
}

1
Revision:	1.8
Committed:	Wed Oct 21 17:28:16 2026 UTC (4 days, 3 hours ago) by root
Branch:	MAIN
CVS Tags:	rel-0_62, rel-0_61, rel-0_6, rel-0_65, HEAD
Changes since 1.7:	+22 -0 lines
Log Message:	* empty log message *
#	Content
1	=head1 NAME
2
3	AnyEvent::MCP::Client::HTTP - MCP over AnyEvent::HTTP
4
5	=head1 SYNOPSIS
6
7	use AnyEvent::MCP::Client::HTTP;
8
9	my $client = AnyEvent::MCP::Client::HTTP->new (
10	name => "example", url => $url, bearer => $token,
11	timeout => 30,
12	);
13
14	=head1 DESCRIPTION
15
16	Use this transport to talk to an MCP server through HTTP. The caller supplies
17	the endpoint and, if needed, credentials; the common Client layer handles
18	discovery and tool calls. Unlike a browser, this client does not follow
19	redirects or silently retry requests. A redirected request could disclose
20	credentials to another host, and retrying a tool call could perform an
21	operation twice.
22
23	=head1 FIRST CALL
24
25	For a bundled MCP server already listening on HTTP, use the full URL
26	it prints with C<--print-url>:
27
28	use AnyEvent;
29	use AnyEvent::MCP::Client::HTTP;
30
31	my $client = AnyEvent::MCP::Client::HTTP->new (
32	url => "http://127.0.0.1:31740/TOKEN/",
33	);
34	my $cv = AE::cv;
35	$client->call (workdirs => {}, sub { $cv->send (@_) });
36	my ($success, $result) = $cv->recv;
37	$success or die "$result->{message}\n";
38	!$result->{isError} or die "$result->{content}[0]{text}\n";
39
40	Replace the example URL with the actual credential-bearing URL. The
41	common L<AnyEvent::MCP::Client> layer negotiates MCP and handles
42	callbacks. A successful JSON-RPC exchange can still return a tool error
43	as C<isError>; always inspect both.
44
45	=head1 METHODS
46
47	=over
48
49	=item $client = AnyEvent::MCP::Client::HTTP->new (%config)
50
51	Accepts the common L<AnyEvent::MCP::Client> options and these transport
52	options. Options are flat, rather than nested in an HTTP configuration hash.
53
54	=over
55
56	=item url => $url
57
58	Complete endpoint URL, optionally including a URL-path token. Required
59	unless supplied by the named client configuration.
60
61	=item bearer => $token
62
63	Supplies C<Authorization: Bearer TOKEN>. Without it, the URL alone is
64	used; a bearer token is not inferred from a URL. The server must accept
65	the selected credential. With the bundled servers, the URL-path token
66	and bearer header present the same static credential.
67
68	=item name => $name
69
70	Loads C<url>, C<bearer> and C<timeout> defaults from C<[client NAME]>.
71	Explicit constructor arguments override them, including explicitly undefined
72	credentials. Without a name no global configuration is accessed.
73
74	=item timeout => $seconds
75
76	HTTP inactivity timeout, defaulting to 300 seconds. Use the common
77	C<request_timeout> option for an overall tool-call deadline.
78
79	=item headers => { %headers }
80
81	Additional HTTP headers. Protocol headers and an explicitly configured
82	bearer header are supplied by this module.
83
84	=item max_size => $bytes
85
86	Bounds a JSON response or an individual buffered SSE event, defaulting to
87	16 MiB. SSE notifications are delivered incrementally rather than after
88	the HTTP response ends.
89
90	=back
91
92	C<proxy>, C<tls_ctx>, C<handle_params>, C<sessionid>, C<on_prepare>,
93	C<on_connect> and C<tcp_connect> are passed to L<AnyEvent::HTTP>.
94	No HTTP cookies or Basic authentication are supplied by this module.
95	Redirects and automatic HTTP retries are disabled.
96
97	Legacy session IDs are retained from initialization. An expired session fails
98	outstanding requests; a later operation can initialize again. Legacy GET
99	streams and the old separate SSE endpoint transport are not implemented.
100
101	=back
102
103	=cut
104
105	package AnyEvent::MCP::Client::HTTP;
106
107	use common::sense;
108	use parent "AnyEvent::MCP::Client";
109	use AnyEvent::HTTP ();
110	use JSON::XS ();
111	use MIME::Base64 ();
112	use Scalar::Util qw(weaken);
113
114	sub new {
115	my ($class, %kv) = @_;
116
117	if (defined $kv{name}) {
118	require AnyEvent::MCP::Config;
119	my $settings = AnyEvent::MCP::Config->global->{section}{"client $kv{name}"} \|\| {};
120	%kv = (
121	(map { exists $settings->{$_} ? ($_ => $settings->{$_}) : () }
122	qw(url bearer timeout)),
123	%kv,
124	);
125	}
126
127	defined $kv{url}
128	or die "HTTP client requires a URL\n";
129	!exists $kv{username} && !exists $kv{password}
130	or die "HTTP Basic authentication is unsupported\n";
131	die "Invalid bearer token\n" if defined $kv{bearer} && $kv{bearer} !~ /\A[^\s]+\z/;
132
133	$class->SUPER::new (timeout => 300, %kv)
134	}
135
136	sub _send {
137	my ($self, $message, $done) = @_;
138
139	my $key = exists $message->{id} && !exists $message->{result} && !exists $message->{error}
140	? $message->{id} : "send:" . ++$self->{send_id};
141	my $generation = $self->{generation};
142	my %headers = (
143	%{$self->{headers} \|\| {}},
144	"content-type" => "application/json",
145	accept => "application/json, text/event-stream",
146	);
147
148	$headers{authorization} = "Bearer $self->{bearer}" if defined $self->{bearer};
149	$headers{"mcp-protocol-version"} = $self->{protocol_version} if defined $self->{protocol_version};
150	$headers{"mcp-session-id"} = $self->{session}
151	if $self->{protocol_version} lt "2026" && defined $self->{session};
152
153	if ($self->{protocol_version} ge "2026" && exists $message->{method} && exists $message->{id}) {
154	$headers{"mcp-method"} = $message->{method};
155	if ($message->{method} =~ /\A(?:tools\/call\|prompts\/get\|resources\/read)\z/) {
156	my $name = exists $message->{params}{name} ? $message->{params}{name} : $message->{params}{uri};
157	$headers{"mcp-name"} = $name =~ /[^\x20-\x7e]/
158	? "=?base64?" . MIME::Base64::encode_base64 (AnyEvent::MCP::utf8enc ($name), "") . "?="
159	: $name;
160	}
161	}
162
163	my ($buffer, $sse, $received, $failure) = ("", 0, 0);
164	my $limit = $self->{max_size} \|\| 16 * 1024 * 1024;
165	weaken $self;
166
167	my $deliver = sub {
168	my ($response) = @_;
169
170	if (ref $response eq "HASH" && !exists $response->{method}) {
171	die "HTTP response id mismatch\n"
172	unless exists $message->{id} && defined $response->{id}
173	&& "$response->{id}" eq "$message->{id}";
174	$received = 1;
175	}
176	$self->_receive ($response);
177	};
178
179	my $guard;
180	my $ok = eval {
181	$guard = AnyEvent::HTTP::http_request POST => $self->{url},
182	(map { exists $self->{$_} ? ($_ => $self->{$_}) : () }
183	qw(timeout proxy tls_ctx handle_params sessionid on_prepare on_connect tcp_connect)),
184	headers => \%headers,
185	body => JSON::XS::encode_json ($message),
186	recurse => 0,
187	persistent => 0,
188	on_header => sub {
189	my ($hdr) = @_;
190
191	return 0 unless $self && $self->{generation} == $generation;
192	$sse = $hdr->{"content-type"} =~ m{\Atext/event-stream(?:\s*;\|\z)}i;
193	if ($message->{method} eq "initialize" && $hdr->{Status} >= 200 && $hdr->{Status} < 300
194	&& defined $hdr->{"mcp-session-id"}) {
195	$self->{session} = $hdr->{"mcp-session-id"};
196	}
197	1
198	},
199	on_body => sub {
200	my ($chunk, $hdr) = @_;
201
202	return 0 unless $self && $self->{generation} == $generation;
203	$buffer .= $chunk;
204
205	my $ok = eval {
206	if ($sse) {
207	while ($buffer =~ s/\A(.*?)(?:\r\n\r\n\|\n\n\|\r\r)//s) {
208	my $event = $1;
209	die "SSE event exceeds size limit\n" if length $event > $limit;
210	my @data = $event =~ /^data:[ \t]?(.*?)(?:\r?$)/mg;
211	$deliver->(JSON::XS::decode_json join "\n", @data) if @data;
212	}
213	}
214	die "HTTP response exceeds size limit\n" if length $buffer > $limit;
215	1
216	};
217	$failure = $@ unless $ok;
218	$ok
219	},
220	sub {
221	my (undef, $hdr) = @_;
222
223	return unless $self && $self->{generation} == $generation;
224	delete $self->{http_requests}{$key};
225
226	if ($hdr->{Status} == 404 && defined $self->{session} && !$received) {
227	return $self->close ("MCP session expired")
228	}
229
230	my $ok = eval {
231	die $failure if $failure;
232	if (!$sse && length $buffer) {
233	my $response = eval { JSON::XS::decode_json $buffer };
234	if (ref $response eq "HASH" && ($hdr->{Status} >= 200 && $hdr->{Status} < 300
235	\|\| ref $response->{error} eq "HASH")) {
236	$deliver->($response);
237	} else {
238	die "HTTP $hdr->{Status} $hdr->{Reason}\n"
239	}
240	}
241
242	die "HTTP $hdr->{Status} $hdr->{Reason}\n"
243	unless $hdr->{Status} >= 200 && $hdr->{Status} < 300;
244	die "HTTP response has no matching JSON-RPC reply\n"
245	if exists $message->{id} && exists $message->{method} && !$received;
246	1
247	};
248	$done->($ok ? (1, undef) : (0, { transport => 1, message => "$@" }));
249	};
250	1
251	};
252	if ($ok) {
253	$self->{http_requests}{$key} = $guard if $self && $guard;
254	} else {
255	$done->(0, { transport => 1, message => "$@" });
256	}
257	}
258
259	sub _abort {
260	my ($self, $id) = @_;
261
262	delete $self->{http_requests}{$id};
263	}
264
265	sub _shutdown {
266	my ($self) = @_;
267
268	delete @$self{qw(http_requests session)};
269	}
270
271	1