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