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 (7 days, 5 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 *
#	User	Rev	Content
1	root	1.1	=head1 NAME
2
3			AnyEvent::MCP::Client::HTTP - MCP over AnyEvent::HTTP
4
5			=head1 SYNOPSIS
6
7	root	1.7	use AnyEvent::MCP::Client::HTTP;
8
9	root	1.1	my $client = AnyEvent::MCP::Client::HTTP->new (
10			name => "example", url => $url, bearer => $token,
11			timeout => 30,
12			);
13
14	root	1.3	=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	root	1.8	=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	root	1.1	=head1 METHODS
46
47	root	1.7	=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	root	1.1
70	root	1.7	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	root	1.1
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	root	1.7	=back
102
103	root	1.1	=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	root	1.4	qw(url bearer timeout)),
123	root	1.1	%kv,
124			);
125			}
126
127	root	1.6	defined $kv{url}
128			or die "HTTP client requires a URL\n";
129	root	1.4	!exists $kv{username} && !exists $kv{password}
130			or die "HTTP Basic authentication is unsupported\n";
131	root	1.1	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	root	1.6
139	root	1.1	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	root	1.6	accept => "application/json, text/event-stream",
146	root	1.1	);
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	root	1.2	? "=?base64?" . MIME::Base64::encode_base64 (AnyEvent::MCP::utf8enc ($name), "") . "?="
159	root	1.1	: $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	root	1.6	my ($response) = @_;
169
170	root	1.1	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	root	1.6	headers => \%headers,
185			body => JSON::XS::encode_json ($message),
186			recurse => 0,
187	root	1.1	persistent => 0,
188	root	1.6	on_header => sub {
189	root	1.1	my ($hdr) = @_;
190	root	1.6
191	root	1.1	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	root	1.6
202	root	1.1	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	root	1.6	$deliver->(JSON::XS::decode_json join "\n", @data) if @data;
212	root	1.1	}
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	root	1.6
223	root	1.1	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	root	1.6	$deliver->($response);
237	root	1.1	} 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	root	1.6
262	root	1.1	delete $self->{http_requests}{$id};
263			}
264
265			sub _shutdown {
266			my ($self) = @_;
267	root	1.6
268	root	1.1	delete @$self{qw(http_requests session)};
269			}
270
271			1