MCP/Client/HTTP.pm

=head1 NAME

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

=head1 SYNOPSIS

   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

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

C<url> is the complete endpoint URL, including any URL cookie.
C<bearer> supplies an Authorization bearer token. Without it the URL
alone is used.

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

Options are flat. C<timeout>, C<proxy>, C<tls_ctx>, C<handle_params>,
C<sessionid>, C<on_prepare>, C<on_connect>, C<tcp_connect> and C<headers>
are AnyEvent::HTTP options. C<timeout> defaults to 300 seconds of inactivity.
Protocol and authentication headers are supplied by this module.
Redirects and automatic HTTP retries are disabled, so calls and credentials
are not replayed at a different endpoint.

C<max_size> 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.

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.

=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,
      );
   }

   die "HTTP client requires a URL\n" unless defined $kv{url};
   !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 ($data) = @_;
      my $response = JSON::XS::decode_json $data;
      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->(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->($buffer);
                  } 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.4
Committed:	Sun Oct 11 00:53:55 2026 UTC (3 weeks, 3 days ago) by root
Branch:	MAIN
Changes since 1.3:	+7 -14 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			my $client = AnyEvent::MCP::Client::HTTP->new (
8			name => "example", url => $url, bearer => $token,
9			timeout => 30,
10			);
11
12	root	1.3	=head1 DESCRIPTION
13
14			Use this transport to talk to an MCP server through HTTP. The caller supplies
15			the endpoint and, if needed, credentials; the common Client layer handles
16			discovery and tool calls. Unlike a browser, this client does not follow
17			redirects or silently retry requests. A redirected request could disclose
18			credentials to another host, and retrying a tool call could perform an
19			operation twice.
20
21	root	1.1	=head1 METHODS
22
23			=head2 $client = AnyEvent::MCP::Client::HTTP->new (%config)
24
25			C<url> is the complete endpoint URL, including any URL cookie.
26	root	1.4	C<bearer> supplies an Authorization bearer token. Without it the URL
27			alone is used.
28	root	1.1
29	root	1.4	C<name>, when supplied, loads C<url>, C<bearer> and C<timeout> defaults
30			from C<[client NAME]> in the global configuration.
31	root	1.1	Explicit arguments override them, including explicitly undefined credentials.
32			Without a name no global configuration is accessed.
33
34			Options are flat. C<timeout>, C<proxy>, C<tls_ctx>, C<handle_params>,
35			C<sessionid>, C<on_prepare>, C<on_connect>, C<tcp_connect> and C<headers>
36			are AnyEvent::HTTP options. C<timeout> defaults to 300 seconds of inactivity.
37			Protocol and authentication headers are supplied by this module.
38			Redirects and automatic HTTP retries are disabled, so calls and credentials
39			are not replayed at a different endpoint.
40
41			C<max_size> bounds a JSON response or an individual buffered SSE event,
42			defaulting to 16 MiB. SSE notifications are delivered incrementally rather
43			than after the HTTP response ends.
44
45			Legacy session IDs are retained from initialization. An expired session fails
46			outstanding requests; a later operation can initialize again. Legacy GET
47			streams and the old separate SSE endpoint transport are not implemented.
48
49			=cut
50
51			package AnyEvent::MCP::Client::HTTP;
52
53			use common::sense;
54			use parent "AnyEvent::MCP::Client";
55			use AnyEvent::HTTP ();
56			use JSON::XS ();
57			use MIME::Base64 ();
58			use Scalar::Util qw(weaken);
59
60			sub new {
61			my ($class, %kv) = @_;
62
63			if (defined $kv{name}) {
64			require AnyEvent::MCP::Config;
65			my $settings = AnyEvent::MCP::Config->global->{section}{"client $kv{name}"} \|\| {};
66			%kv = (
67			(map { exists $settings->{$_} ? ($_ => $settings->{$_}) : () }
68	root	1.4	qw(url bearer timeout)),
69	root	1.1	%kv,
70			);
71			}
72
73			die "HTTP client requires a URL\n" unless defined $kv{url};
74	root	1.4	!exists $kv{username} && !exists $kv{password}
75			or die "HTTP Basic authentication is unsupported\n";
76	root	1.1	die "Invalid bearer token\n" if defined $kv{bearer} && $kv{bearer} !~ /\A[^\s]+\z/;
77
78			$class->SUPER::new (timeout => 300, %kv)
79			}
80
81			sub _send {
82			my ($self, $message, $done) = @_;
83			my $key = exists $message->{id} && !exists $message->{result} && !exists $message->{error}
84			? $message->{id} : "send:" . ++$self->{send_id};
85			my $generation = $self->{generation};
86			my %headers = (
87			%{$self->{headers} \|\| {}},
88			"content-type" => "application/json",
89			accept => "application/json, text/event-stream",
90			);
91
92			$headers{authorization} = "Bearer $self->{bearer}" if defined $self->{bearer};
93			$headers{"mcp-protocol-version"} = $self->{protocol_version} if defined $self->{protocol_version};
94			$headers{"mcp-session-id"} = $self->{session}
95			if $self->{protocol_version} lt "2026" && defined $self->{session};
96
97			if ($self->{protocol_version} ge "2026" && exists $message->{method} && exists $message->{id}) {
98			$headers{"mcp-method"} = $message->{method};
99			if ($message->{method} =~ /\A(?:tools\/call\|prompts\/get\|resources\/read)\z/) {
100			my $name = exists $message->{params}{name} ? $message->{params}{name} : $message->{params}{uri};
101			$headers{"mcp-name"} = $name =~ /[^\x20-\x7e]/
102	root	1.2	? "=?base64?" . MIME::Base64::encode_base64 (AnyEvent::MCP::utf8enc ($name), "") . "?="
103	root	1.1	: $name;
104			}
105			}
106
107			my ($buffer, $sse, $received, $failure) = ("", 0, 0);
108			my $limit = $self->{max_size} \|\| 16 * 1024 * 1024;
109			weaken $self;
110
111			my $deliver = sub {
112			my ($data) = @_;
113			my $response = JSON::XS::decode_json $data;
114			if (ref $response eq "HASH" && !exists $response->{method}) {
115			die "HTTP response id mismatch\n"
116			unless exists $message->{id} && defined $response->{id}
117			&& "$response->{id}" eq "$message->{id}";
118			$received = 1;
119			}
120			$self->_receive ($response);
121			};
122
123			my $guard;
124			my $ok = eval {
125			$guard = AnyEvent::HTTP::http_request POST => $self->{url},
126			(map { exists $self->{$_} ? ($_ => $self->{$_}) : () }
127			qw(timeout proxy tls_ctx handle_params sessionid on_prepare on_connect tcp_connect)),
128			headers => \%headers,
129			body => JSON::XS::encode_json ($message),
130			recurse => 0,
131			persistent => 0,
132			on_header => sub {
133			my ($hdr) = @_;
134			return 0 unless $self && $self->{generation} == $generation;
135			$sse = $hdr->{"content-type"} =~ m{\Atext/event-stream(?:\s*;\|\z)}i;
136			if ($message->{method} eq "initialize" && $hdr->{Status} >= 200 && $hdr->{Status} < 300
137			&& defined $hdr->{"mcp-session-id"}) {
138			$self->{session} = $hdr->{"mcp-session-id"};
139			}
140			1
141			},
142			on_body => sub {
143			my ($chunk, $hdr) = @_;
144			return 0 unless $self && $self->{generation} == $generation;
145			$buffer .= $chunk;
146
147			my $ok = eval {
148			if ($sse) {
149			while ($buffer =~ s/\A(.*?)(?:\r\n\r\n\|\n\n\|\r\r)//s) {
150			my $event = $1;
151			die "SSE event exceeds size limit\n" if length $event > $limit;
152			my @data = $event =~ /^data:[ \t]?(.*?)(?:\r?$)/mg;
153			$deliver->(join "\n", @data) if @data;
154			}
155			}
156			die "HTTP response exceeds size limit\n" if length $buffer > $limit;
157			1
158			};
159			$failure = $@ unless $ok;
160			$ok
161			},
162			sub {
163			my (undef, $hdr) = @_;
164			return unless $self && $self->{generation} == $generation;
165			delete $self->{http_requests}{$key};
166
167			if ($hdr->{Status} == 404 && defined $self->{session} && !$received) {
168			return $self->close ("MCP session expired")
169			}
170
171			my $ok = eval {
172			die $failure if $failure;
173			if (!$sse && length $buffer) {
174			my $response = eval { JSON::XS::decode_json $buffer };
175			if (ref $response eq "HASH" && ($hdr->{Status} >= 200 && $hdr->{Status} < 300
176			\|\| ref $response->{error} eq "HASH")) {
177			$deliver->($buffer);
178			} else {
179			die "HTTP $hdr->{Status} $hdr->{Reason}\n"
180			}
181			}
182
183			die "HTTP $hdr->{Status} $hdr->{Reason}\n"
184			unless $hdr->{Status} >= 200 && $hdr->{Status} < 300;
185			die "HTTP response has no matching JSON-RPC reply\n"
186			if exists $message->{id} && exists $message->{method} && !$received;
187			1
188			};
189			$done->($ok ? (1, undef) : (0, { transport => 1, message => "$@" }));
190			};
191			1
192			};
193			if ($ok) {
194			$self->{http_requests}{$key} = $guard if $self && $guard;
195			} else {
196			$done->(0, { transport => 1, message => "$@" });
197			}
198			}
199
200			sub _abort {
201			my ($self, $id) = @_;
202			delete $self->{http_requests}{$id};
203			}
204
205			sub _shutdown {
206			my ($self) = @_;
207			delete @$self{qw(http_requests session)};
208			}
209
210			1