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

   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.6
Committed:	Tue Oct 13 10:17:31 2026 UTC (2 weeks, 1 day ago) by root
Branch:	MAIN
Changes since 1.5:	+17 -10 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	root	1.6	defined $kv{url}
78			or die "HTTP client requires a URL\n";
79	root	1.4	!exists $kv{username} && !exists $kv{password}
80			or die "HTTP Basic authentication is unsupported\n";
81	root	1.1	die "Invalid bearer token\n" if defined $kv{bearer} && $kv{bearer} !~ /\A[^\s]+\z/;
82
83			$class->SUPER::new (timeout => 300, %kv)
84			}
85
86			sub _send {
87			my ($self, $message, $done) = @_;
88	root	1.6
89	root	1.1	my $key = exists $message->{id} && !exists $message->{result} && !exists $message->{error}
90			? $message->{id} : "send:" . ++$self->{send_id};
91			my $generation = $self->{generation};
92			my %headers = (
93			%{$self->{headers} \|\| {}},
94			"content-type" => "application/json",
95	root	1.6	accept => "application/json, text/event-stream",
96	root	1.1	);
97
98			$headers{authorization} = "Bearer $self->{bearer}" if defined $self->{bearer};
99			$headers{"mcp-protocol-version"} = $self->{protocol_version} if defined $self->{protocol_version};
100			$headers{"mcp-session-id"} = $self->{session}
101			if $self->{protocol_version} lt "2026" && defined $self->{session};
102
103			if ($self->{protocol_version} ge "2026" && exists $message->{method} && exists $message->{id}) {
104			$headers{"mcp-method"} = $message->{method};
105			if ($message->{method} =~ /\A(?:tools\/call\|prompts\/get\|resources\/read)\z/) {
106			my $name = exists $message->{params}{name} ? $message->{params}{name} : $message->{params}{uri};
107			$headers{"mcp-name"} = $name =~ /[^\x20-\x7e]/
108	root	1.2	? "=?base64?" . MIME::Base64::encode_base64 (AnyEvent::MCP::utf8enc ($name), "") . "?="
109	root	1.1	: $name;
110			}
111			}
112
113			my ($buffer, $sse, $received, $failure) = ("", 0, 0);
114			my $limit = $self->{max_size} \|\| 16 * 1024 * 1024;
115			weaken $self;
116
117			my $deliver = sub {
118	root	1.6	my ($response) = @_;
119
120	root	1.1	if (ref $response eq "HASH" && !exists $response->{method}) {
121			die "HTTP response id mismatch\n"
122			unless exists $message->{id} && defined $response->{id}
123			&& "$response->{id}" eq "$message->{id}";
124			$received = 1;
125			}
126			$self->_receive ($response);
127			};
128
129			my $guard;
130			my $ok = eval {
131			$guard = AnyEvent::HTTP::http_request POST => $self->{url},
132			(map { exists $self->{$_} ? ($_ => $self->{$_}) : () }
133			qw(timeout proxy tls_ctx handle_params sessionid on_prepare on_connect tcp_connect)),
134	root	1.6	headers => \%headers,
135			body => JSON::XS::encode_json ($message),
136			recurse => 0,
137	root	1.1	persistent => 0,
138	root	1.6	on_header => sub {
139	root	1.1	my ($hdr) = @_;
140	root	1.6
141	root	1.1	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	root	1.6
152	root	1.1	return 0 unless $self && $self->{generation} == $generation;
153			$buffer .= $chunk;
154
155			my $ok = eval {
156			if ($sse) {
157			while ($buffer =~ s/\A(.*?)(?:\r\n\r\n\|\n\n\|\r\r)//s) {
158			my $event = $1;
159			die "SSE event exceeds size limit\n" if length $event > $limit;
160			my @data = $event =~ /^data:[ \t]?(.*?)(?:\r?$)/mg;
161	root	1.6	$deliver->(JSON::XS::decode_json join "\n", @data) if @data;
162	root	1.1	}
163			}
164			die "HTTP response exceeds size limit\n" if length $buffer > $limit;
165			1
166			};
167			$failure = $@ unless $ok;
168			$ok
169			},
170			sub {
171			my (undef, $hdr) = @_;
172	root	1.6
173	root	1.1	return unless $self && $self->{generation} == $generation;
174			delete $self->{http_requests}{$key};
175
176			if ($hdr->{Status} == 404 && defined $self->{session} && !$received) {
177			return $self->close ("MCP session expired")
178			}
179
180			my $ok = eval {
181			die $failure if $failure;
182			if (!$sse && length $buffer) {
183			my $response = eval { JSON::XS::decode_json $buffer };
184			if (ref $response eq "HASH" && ($hdr->{Status} >= 200 && $hdr->{Status} < 300
185			\|\| ref $response->{error} eq "HASH")) {
186	root	1.6	$deliver->($response);
187	root	1.1	} else {
188			die "HTTP $hdr->{Status} $hdr->{Reason}\n"
189			}
190			}
191
192			die "HTTP $hdr->{Status} $hdr->{Reason}\n"
193			unless $hdr->{Status} >= 200 && $hdr->{Status} < 300;
194			die "HTTP response has no matching JSON-RPC reply\n"
195			if exists $message->{id} && exists $message->{method} && !$received;
196			1
197			};
198			$done->($ok ? (1, undef) : (0, { transport => 1, message => "$@" }));
199			};
200			1
201			};
202			if ($ok) {
203			$self->{http_requests}{$key} = $guard if $self && $guard;
204			} else {
205			$done->(0, { transport => 1, message => "$@" });
206			}
207			}
208
209			sub _abort {
210			my ($self, $id) = @_;
211	root	1.6
212	root	1.1	delete $self->{http_requests}{$id};
213			}
214
215			sub _shutdown {
216			my ($self) = @_;
217	root	1.6
218	root	1.1	delete @$self{qw(http_requests session)};
219			}
220
221			1