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. Alternatively C<username>
and C<password> supply Basic authentication. Configuring both is an error.
No authentication options means the URL alone is used.

C<name>, when supplied, loads C<url>, C<bearer>, C<username>, C<password> 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 username password timeout)),
         %kv,
      );
   }

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