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 Encode ();
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 (
      Encode::encode_utf8 ("$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 (Encode::encode_utf8 ($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.1
Committed:	Thu Oct 8 05:27:00 2026 UTC (3 weeks, 6 days ago) by root
Branch:	MAIN
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 Encode ();
51			use Scalar::Util qw(weaken);
52
53			sub new {
54			my ($class, %kv) = @_;
55
56			if (defined $kv{name}) {
57			require AnyEvent::MCP::Config;
58			my $settings = AnyEvent::MCP::Config->global->{section}{"client $kv{name}"} \|\| {};
59			%kv = (
60			(map { exists $settings->{$_} ? ($_ => $settings->{$_}) : () }
61			qw(url bearer username password timeout)),
62			%kv,
63			);
64			}
65
66			die "HTTP client requires a URL\n" unless defined $kv{url};
67			die "Bearer and Basic authentication cannot be combined\n"
68			if defined $kv{bearer} && (defined $kv{username} \|\| defined $kv{password});
69			die "Basic authentication requires username and password\n"
70			if defined $kv{username} != defined $kv{password};
71			die "Invalid Basic username\n" if $kv{username} =~ /[:\r\n]/;
72			die "Invalid bearer token\n" if defined $kv{bearer} && $kv{bearer} !~ /\A[^\s]+\z/;
73
74			$class->SUPER::new (timeout => 300, %kv)
75			}
76
77			sub _send {
78			my ($self, $message, $done) = @_;
79			my $key = exists $message->{id} && !exists $message->{result} && !exists $message->{error}
80			? $message->{id} : "send:" . ++$self->{send_id};
81			my $generation = $self->{generation};
82			my %headers = (
83			%{$self->{headers} \|\| {}},
84			"content-type" => "application/json",
85			accept => "application/json, text/event-stream",
86			);
87
88			$headers{authorization} = "Bearer $self->{bearer}" if defined $self->{bearer};
89			$headers{authorization} = "Basic " . MIME::Base64::encode_base64 (
90			Encode::encode_utf8 ("$self->{username}:$self->{password}"), ""
91			) if defined $self->{username};
92			$headers{"mcp-protocol-version"} = $self->{protocol_version} if defined $self->{protocol_version};
93			$headers{"mcp-session-id"} = $self->{session}
94			if $self->{protocol_version} lt "2026" && defined $self->{session};
95
96			if ($self->{protocol_version} ge "2026" && exists $message->{method} && exists $message->{id}) {
97			$headers{"mcp-method"} = $message->{method};
98			if ($message->{method} =~ /\A(?:tools\/call\|prompts\/get\|resources\/read)\z/) {
99			my $name = exists $message->{params}{name} ? $message->{params}{name} : $message->{params}{uri};
100			$headers{"mcp-name"} = $name =~ /[^\x20-\x7e]/
101			? "=?base64?" . MIME::Base64::encode_base64 (Encode::encode_utf8 ($name), "") . "?="
102			: $name;
103			}
104			}
105
106			my ($buffer, $sse, $received, $failure) = ("", 0, 0);
107			my $limit = $self->{max_size} \|\| 16 * 1024 * 1024;
108			weaken $self;
109
110			my $deliver = sub {
111			my ($data) = @_;
112			my $response = JSON::XS::decode_json $data;
113			if (ref $response eq "HASH" && !exists $response->{method}) {
114			die "HTTP response id mismatch\n"
115			unless exists $message->{id} && defined $response->{id}
116			&& "$response->{id}" eq "$message->{id}";
117			$received = 1;
118			}
119			$self->_receive ($response);
120			};
121
122			my $guard;
123			my $ok = eval {
124			$guard = AnyEvent::HTTP::http_request POST => $self->{url},
125			(map { exists $self->{$_} ? ($_ => $self->{$_}) : () }
126			qw(timeout proxy tls_ctx handle_params sessionid on_prepare on_connect tcp_connect)),
127			headers => \%headers,
128			body => JSON::XS::encode_json ($message),
129			recurse => 0,
130			persistent => 0,
131			on_header => sub {
132			my ($hdr) = @_;
133			return 0 unless $self && $self->{generation} == $generation;
134			$sse = $hdr->{"content-type"} =~ m{\Atext/event-stream(?:\s*;\|\z)}i;
135			if ($message->{method} eq "initialize" && $hdr->{Status} >= 200 && $hdr->{Status} < 300
136			&& defined $hdr->{"mcp-session-id"}) {
137			$self->{session} = $hdr->{"mcp-session-id"};
138			}
139			1
140			},
141			on_body => sub {
142			my ($chunk, $hdr) = @_;
143			return 0 unless $self && $self->{generation} == $generation;
144			$buffer .= $chunk;
145
146			my $ok = eval {
147			if ($sse) {
148			while ($buffer =~ s/\A(.*?)(?:\r\n\r\n\|\n\n\|\r\r)//s) {
149			my $event = $1;
150			die "SSE event exceeds size limit\n" if length $event > $limit;
151			my @data = $event =~ /^data:[ \t]?(.*?)(?:\r?$)/mg;
152			$deliver->(join "\n", @data) if @data;
153			}
154			}
155			die "HTTP response exceeds size limit\n" if length $buffer > $limit;
156			1
157			};
158			$failure = $@ unless $ok;
159			$ok
160			},
161			sub {
162			my (undef, $hdr) = @_;
163			return unless $self && $self->{generation} == $generation;
164			delete $self->{http_requests}{$key};
165
166			if ($hdr->{Status} == 404 && defined $self->{session} && !$received) {
167			return $self->close ("MCP session expired")
168			}
169
170			my $ok = eval {
171			die $failure if $failure;
172			if (!$sse && length $buffer) {
173			my $response = eval { JSON::XS::decode_json $buffer };
174			if (ref $response eq "HASH" && ($hdr->{Status} >= 200 && $hdr->{Status} < 300
175			\|\| ref $response->{error} eq "HASH")) {
176			$deliver->($buffer);
177			} else {
178			die "HTTP $hdr->{Status} $hdr->{Reason}\n"
179			}
180			}
181
182			die "HTTP $hdr->{Status} $hdr->{Reason}\n"
183			unless $hdr->{Status} >= 200 && $hdr->{Status} < 300;
184			die "HTTP response has no matching JSON-RPC reply\n"
185			if exists $message->{id} && exists $message->{method} && !$received;
186			1
187			};
188			$done->($ok ? (1, undef) : (0, { transport => 1, message => "$@" }));
189			};
190			1
191			};
192			if ($ok) {
193			$self->{http_requests}{$key} = $guard if $self && $guard;
194			} else {
195			$done->(0, { transport => 1, message => "$@" });
196			}
197			}
198
199			sub _abort {
200			my ($self, $id) = @_;
201			delete $self->{http_requests}{$id};
202			}
203
204			sub _shutdown {
205			my ($self) = @_;
206			delete @$self{qw(http_requests session)};
207			}
208
209			1