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, optionally including a URL-path token.
C<bearer> supplies an explicit C<Authorization: Bearer TOKEN> header.
Without it, the URL alone is used. The server must accept the selected
credential; a bearer token is not inferred from a URL, and no HTTP cookies
or Basic authentication are supplied. When using the bundled MCP servers,
the URL-path token and the bearer header are two ways to present the same
static credential, not two different permission systems.

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