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