client.hpp

Go to the documentation of this file.

// Copyright (c) 2025 [caomengxuan666]
 
#pragma once
 
 
#include <atomic>
#include <chrono>
#include <condition_variable>
#include <cstdint>
#include <functional>
#include <memory>
#include <mutex>
#include <optional>
#include <string>
#include <string_view>
#include <unordered_map>
#include <utility>
#include <vector>
 
#include "cxxmcp/core/result.hpp"
#include "cxxmcp/protocol/capabilities.hpp"
#include "cxxmcp/protocol/completion.hpp"
#include "cxxmcp/protocol/elicitation.hpp"
#include "cxxmcp/protocol/logging.hpp"
#include "cxxmcp/protocol/prompt.hpp"
#include "cxxmcp/protocol/resource.hpp"
#include "cxxmcp/protocol/roots.hpp"
#include "cxxmcp/protocol/sampling.hpp"
#include "cxxmcp/protocol/serialization.hpp"
#include "cxxmcp/protocol/tool.hpp"
#include "cxxmcp/protocol/types.hpp"
#include "cxxmcp/request.hpp"
 
namespace mcp::client {
 
struct HttpAuthChallenge;
using HttpAuthRefreshHandler =
    std::function<std::optional<std::string>(const HttpAuthChallenge&)>;
 
struct HttpDpopRequestContext {
  std::string method;
  std::string url;
  std::optional<std::string> access_token;
  std::optional<std::string> nonce;
};
 
struct HttpDpopProof {
  std::string proof;
  std::string authorization_scheme = "DPoP";
};
 
using HttpDpopRequestSigner =
    std::function<core::Result<std::optional<HttpDpopProof>>(
        const HttpDpopRequestContext&)>;
 
struct ClientOptions {
  std::string endpoint;
 
  std::chrono::milliseconds timeout{30000};
};
 
struct ClientHandler;
struct ClientHandlerInterface;
 
using TransportRequestHandler =
    std::function<core::Result<protocol::JsonRpcResponse>(
        const protocol::JsonRpcRequest&)>;
 
using TransportNotificationHandler = std::function<core::Result<core::Unit>(
    const protocol::JsonRpcNotification&)>;
 
class Transport {
 public:
  virtual ~Transport() = default;
 
  virtual core::Result<protocol::JsonRpcResponse> send(
      const protocol::JsonRpcRequest& request) = 0;
 
  virtual core::Result<core::Unit> send_notification(
      const protocol::JsonRpcNotification& notification);
 
  virtual core::Result<core::Unit> start(
      TransportRequestHandler request_handler,
      TransportNotificationHandler notification_handler = {}) {
    (void)request_handler;
    (void)notification_handler;
    return core::Unit{};
  }
 
  virtual void stop() noexcept {}
};
 
class Client {
 public:
#if defined(CXXMCP_ENABLE_HTTP)
  struct StreamableHttpEndpoint {
    std::string uri;
 
    std::string host;
    std::uint16_t port = 80;
    std::string path = "/mcp";
    std::unordered_map<std::string, std::string> headers;
 
    std::optional<std::string> auth_header;
 
    HttpAuthRefreshHandler auth_refresh_handler;
 
    HttpDpopRequestSigner dpop_request_signer;
 
    bool stateless = false;
 
    std::chrono::milliseconds timeout{30000};
  };
#endif
 
  struct StdioEndpoint {
    std::string command;
    std::vector<std::string> args;
    std::string cwd;
    std::unordered_map<std::string, std::string> env;
  };
 
  using Root = protocol::Root;
 
  using LoggingMessageHandler =
      std::function<void(std::string_view, std::string_view)>;
 
  using InitializedHandler = std::function<void()>;
 
  using CancelledHandler =
      std::function<void(const protocol::RequestId&, std::string_view)>;
 
  using ListChangedHandler = std::function<void()>;
 
  using ResourceUpdatedHandler = std::function<void(const std::string&)>;
 
  using ProgressHandler =
      std::function<void(const protocol::ProgressNotificationParams&)>;
 
  using ElicitationCompleteHandler = std::function<void(std::string_view)>;
 
  using TaskStatusHandler = std::function<void(const protocol::Task&)>;
 
  using RootsListRequestHandler =
      std::function<core::Result<protocol::RootsListResult>()>;
  using RootsListRequestCancellationHandler =
      std::function<core::Result<protocol::RootsListResult>(CancellationToken)>;
 
  using SamplingRequestHandler =
      std::function<core::Result<protocol::CreateMessageResult>(
          const protocol::CreateMessageParams&)>;
  using SamplingRequestCancellationHandler =
      std::function<core::Result<protocol::CreateMessageResult>(
          const protocol::CreateMessageParams&, CancellationToken)>;
 
  using ElicitationRequestHandler =
      std::function<core::Result<protocol::CreateElicitationResult>(
          const protocol::CreateElicitationRequestParam&)>;
  using ElicitationRequestCancellationHandler =
      std::function<core::Result<protocol::CreateElicitationResult>(
          const protocol::CreateElicitationRequestParam&, CancellationToken)>;
 
  using CustomRequestHandler = std::function<core::Result<protocol::Json>(
      const protocol::JsonRpcRequest&)>;
  using CustomRequestCancellationHandler =
      std::function<core::Result<protocol::Json>(
          const protocol::JsonRpcRequest&, CancellationToken)>;
  using ListRootsRequestHandler = RootsListRequestHandler;
  using CreateMessageRequestHandler = SamplingRequestHandler;
  using CreateElicitationRequestHandler = ElicitationRequestHandler;
 
  using RawNotificationHandler =
      std::function<void(const protocol::JsonRpcNotification&)>;
 
#if defined(CXXMCP_ENABLE_HTTP)
  static Client connect_streamable_http(StreamableHttpEndpoint endpoint);
 
  static Client connect_streamable_http(std::string uri);
 
  static Client connect_legacy_sse(StreamableHttpEndpoint endpoint);
 
  static Client connect_legacy_sse(std::string uri);
#endif
 
  static Client connect_stdio(StdioEndpoint endpoint);
 
  explicit Client(std::unique_ptr<Transport> transport);
  Client(const Client&) = delete;
  Client& operator=(const Client&) = delete;
  Client(Client&& other) noexcept;
  Client& operator=(Client&& other) noexcept;
 
  core::Result<protocol::Json> initialize(std::string client_name = "cxxmcp",
                                          std::string client_version = "0");
 
  core::Result<protocol::Json> initialize(std::string client_name,
                                          std::string client_version,
                                          RequestOptions options);
 
  std::optional<protocol::ServerCapabilities> server_capabilities() const;
 
  core::Result<core::Unit> notify_initialized();
 
  core::Result<core::Unit> notify_cancelled(protocol::RequestId request_id,
                                            std::string reason = {});
 
  core::Result<core::Unit> notify_progress(
      protocol::ProgressToken progress_token, double progress,
      std::optional<double> total = std::nullopt, std::string message = {});
 
  core::Result<core::Unit> notify_roots_list_changed();
 
  core::Result<core::Unit> ping();
 
  core::Result<std::vector<protocol::Prompt>> list_prompts();
 
  core::Result<protocol::PromptsListResult> list_prompts_page(
      const protocol::PaginatedRequestParams& params = {});
 
  core::Result<std::vector<protocol::Prompt>> list_all_prompts();
 
  core::Result<protocol::PromptsGetResult> get_prompt(
      const protocol::PromptsGetParams& params);
 
  core::Result<protocol::PromptsGetResult> get_prompt(
      std::string_view name,
      const protocol::Json& arguments = protocol::Json::object());
 
  core::Result<std::vector<protocol::Resource>> list_resources();
 
  core::Result<protocol::ResourcesListResult> list_resources_page(
      const protocol::PaginatedRequestParams& params = {});
 
  core::Result<std::vector<protocol::Resource>> list_all_resources();
 
  core::Result<protocol::ResourcesReadResult> read_resource(
      const protocol::ResourcesReadParams& params);
 
  core::Result<protocol::ResourcesReadResult> read_resource(
      std::string_view uri);
 
  core::Result<std::vector<protocol::ResourceTemplate>>
  list_resource_templates();
 
  core::Result<protocol::ResourceTemplatesListResult>
  list_resource_templates_page(
      const protocol::PaginatedRequestParams& params = {});
 
  core::Result<std::vector<protocol::ResourceTemplate>>
  list_all_resource_templates();
 
  core::Result<std::vector<protocol::ToolDefinition>> list_tools();
 
  core::Result<protocol::ToolsListResult> list_tools_page(
      const protocol::PaginatedRequestParams& params = {});
 
  core::Result<std::vector<protocol::ToolDefinition>> list_all_tools();
 
  core::Result<protocol::ToolResult> call_tool(const protocol::ToolCall& call);
 
  core::Result<protocol::CreateTaskResult> call_tool_task(
      const protocol::ToolCall& call);
 
  core::Result<protocol::ToolResult> call_raw(
      std::string_view name,
      const protocol::Json& arguments = protocol::Json::object());
 
  core::Result<protocol::CompleteResult> complete(
      const protocol::CompleteParams& request);
 
  core::Result<protocol::Json> complete(const protocol::Json& request);
 
  core::Result<protocol::CompletionResult> complete_prompt_argument(
      std::string_view prompt_name, std::string_view argument_name,
      std::string current_value,
      protocol::Json context = protocol::Json::object());
 
  core::Result<protocol::CompletionResult> complete_resource_argument(
      std::string_view uri_template, std::string_view argument_name,
      std::string current_value,
      protocol::Json context = protocol::Json::object());
 
  core::Result<std::vector<std::string>> complete_prompt_simple(
      std::string_view prompt_name, std::string_view argument_name,
      std::string current_value,
      protocol::Json context = protocol::Json::object());
 
  core::Result<std::vector<std::string>> complete_resource_simple(
      std::string_view uri_template, std::string_view argument_name,
      std::string current_value,
      protocol::Json context = protocol::Json::object());
 
  core::Result<protocol::CreateMessageResult> create_message(
      const protocol::CreateMessageParams& request);
 
  core::Result<protocol::Json> create_message(const protocol::Json& request);
 
  core::Result<protocol::CreateElicitationResult> create_elicitation(
      const protocol::CreateElicitationRequestParam& request);
 
  core::Result<protocol::Json> create_elicitation(
      const protocol::Json& request);
 
  core::Result<std::vector<protocol::Task>> list_tasks();
 
  core::Result<protocol::TaskListResult> list_tasks_page(
      const protocol::TaskListParams& request = {});
 
  core::Result<std::vector<protocol::Task>> list_all_tasks();
 
  core::Result<protocol::Task> get_task(const protocol::TaskGetParams& request);
 
  core::Result<protocol::Task> get_task(std::string_view task_id);
 
  core::Result<protocol::Task> cancel_task(
      const protocol::TaskCancelParams& request);
 
  core::Result<protocol::Task> cancel_task(std::string_view task_id);
 
  core::Result<protocol::Json> task_result(
      const protocol::TaskResultParams& request);
 
  core::Result<protocol::Json> task_result(std::string_view task_id);
 
  core::Result<core::Unit> set_level(
      const protocol::LoggingSetLevelParams& params);
 
  core::Result<core::Unit> set_level(std::string_view level);
 
  core::Result<core::Unit> subscribe(std::string_view uri);
 
  core::Result<core::Unit> unsubscribe(std::string_view uri);
 
  core::Result<protocol::Json> request(const protocol::JsonRpcRequest& request);
 
  RequestHandle<protocol::Json> request_async(
      std::string method, protocol::Json params = protocol::Json::object(),
      RequestOptions options = {});
 
  template <class T, class Parser>
  RequestHandle<T> request_async(std::string method, protocol::Json params,
                                 Parser parser, RequestOptions options = {}) {
    protocol::JsonRpcRequest request;
    request.method = std::move(method);
    request.params = std::move(params);
    request.id = next_request_id();
    if (options.meta.has_value()) {
      request.meta = std::move(options.meta);
    }
    request.transport_headers = std::move(options.headers);
    request.protocol_version_override = std::move(options.protocol_version);
 
    const auto request_id = request.id;
    ++in_flight_requests_;
    auto in_flight_mutex = &in_flight_mutex_;
    auto in_flight_cv = in_flight_cv_;
    auto in_flight_count = &in_flight_requests_;
    return RequestHandle<T>::spawn(
        request_id, options.timeout, options.cancellation_token,
        [this, request_id](std::string reason) mutable {
          return notify_cancelled(std::move(request_id), std::move(reason));
        },
        [this, request = std::move(request), parser = std::move(parser),
         in_flight_mutex, in_flight_cv,
         in_flight_count]() mutable -> core::Result<T> {
          auto payload = raw_request(request);
          core::Result<T> result;
          if (!payload) {
            result = mcp::core::unexpected(payload.error());
          } else {
            result = parser(*payload);
          }
          {
            std::lock_guard lock(*in_flight_mutex);
            --(*in_flight_count);
            in_flight_cv->notify_one();
          }
          return result;
        });
  }
 
  RequestHandle<std::vector<protocol::ToolDefinition>> list_tools_async(
      RequestOptions options = {});
 
  RequestHandle<std::vector<protocol::Prompt>> list_prompts_async(
      RequestOptions options = {});
 
  RequestHandle<std::vector<protocol::Resource>> list_resources_async(
      RequestOptions options = {});
 
  RequestHandle<std::vector<protocol::ResourceTemplate>>
  list_resource_templates_async(RequestOptions options = {});
 
  RequestHandle<protocol::ToolResult> call_tool_async(
      const protocol::ToolCall& call, RequestOptions options = {});
 
  RequestHandle<protocol::ToolResult> call_tool_async(
      std::string_view name,
      const protocol::Json& arguments = protocol::Json::object(),
      RequestOptions options = {});
 
  RequestHandle<protocol::CreateTaskResult> call_tool_task_async(
      const protocol::ToolCall& call, RequestOptions options = {});
 
  RequestHandle<protocol::PromptsGetResult> get_prompt_async(
      const protocol::PromptsGetParams& params, RequestOptions options = {});
 
  RequestHandle<protocol::PromptsGetResult> get_prompt_async(
      std::string_view name,
      const protocol::Json& arguments = protocol::Json::object(),
      RequestOptions options = {});
 
  RequestHandle<protocol::ResourcesReadResult> read_resource_async(
      const protocol::ResourcesReadParams& params, RequestOptions options = {});
 
  RequestHandle<protocol::ResourcesReadResult> read_resource_async(
      std::string_view uri, RequestOptions options = {});
 
  RequestHandle<protocol::CompleteResult> complete_async(
      const protocol::CompleteParams& request, RequestOptions options = {});
 
  RequestHandle<protocol::Json> complete_async(const protocol::Json& request,
                                               RequestOptions options = {});
 
  RequestHandle<protocol::CreateMessageResult> create_message_async(
      const protocol::CreateMessageParams& request,
      RequestOptions options = {});
 
  RequestHandle<protocol::Json> create_message_async(
      const protocol::Json& request, RequestOptions options = {});
 
  RequestHandle<protocol::CreateElicitationResult> create_elicitation_async(
      const protocol::CreateElicitationRequestParam& request,
      RequestOptions options = {});
 
  RequestHandle<protocol::Json> create_elicitation_async(
      const protocol::Json& request, RequestOptions options = {});
 
  RequestHandle<std::vector<protocol::Task>> list_tasks_async(
      RequestOptions options = {});
 
  RequestHandle<protocol::Task> get_task_async(
      const protocol::TaskGetParams& request, RequestOptions options = {});
 
  RequestHandle<protocol::Task> get_task_async(std::string_view task_id,
                                               RequestOptions options = {});
 
  RequestHandle<protocol::Task> cancel_task_async(
      const protocol::TaskCancelParams& request, RequestOptions options = {});
 
  RequestHandle<protocol::Task> cancel_task_async(std::string_view task_id,
                                                  RequestOptions options = {});
 
  RequestHandle<protocol::Json> task_result_async(
      const protocol::TaskResultParams& request, RequestOptions options = {});
 
  RequestHandle<protocol::Json> task_result_async(std::string_view task_id,
                                                  RequestOptions options = {});
 
  core::Result<core::Unit> notify(
      const protocol::JsonRpcNotification& notification);
 
  std::vector<protocol::Root> list_roots() const;
 
  Client& set_roots(std::vector<protocol::Root> roots);
 
  Client& set_capabilities(protocol::ClientCapabilities capabilities);
 
  Client& on_initialized(InitializedHandler handler);
 
  Client& on_cancelled(CancelledHandler handler);
 
  Client& on_logging_message(LoggingMessageHandler handler);
 
  Client& on_tool_list_changed(ListChangedHandler handler);
 
  Client& on_prompt_list_changed(ListChangedHandler handler);
 
  Client& on_resource_list_changed(ListChangedHandler handler);
 
  Client& on_resource_updated(ResourceUpdatedHandler handler);
 
  Client& on_progress(ProgressHandler handler);
 
  Client& on_elicitation_complete(ElicitationCompleteHandler handler);
 
  Client& on_task_status(TaskStatusHandler handler);
 
  Client& on_roots_list_changed(ListChangedHandler handler);
 
  Client& on_list_roots_request(ListRootsRequestHandler handler);
  Client& on_list_roots_request(RootsListRequestCancellationHandler handler);
 
  Client& on_create_message_request(CreateMessageRequestHandler handler);
  Client& on_create_message_request(SamplingRequestCancellationHandler handler);
 
  Client& on_create_elicitation_request(
      CreateElicitationRequestHandler handler);
  Client& on_create_elicitation_request(
      ElicitationRequestCancellationHandler handler);
 
  Client& on_custom_request(CustomRequestHandler handler);
  Client& on_custom_request(CustomRequestCancellationHandler handler);
 
  Client& on_roots_list_request(RootsListRequestHandler handler);
  Client& on_roots_list_request(RootsListRequestCancellationHandler handler);
 
  Client& on_sampling_request(SamplingRequestHandler handler);
  Client& on_sampling_request(SamplingRequestCancellationHandler handler);
 
  Client& on_elicitation_request(ElicitationRequestHandler handler);
  Client& on_elicitation_request(ElicitationRequestCancellationHandler handler);
 
  Client& on_raw_notification(RawNotificationHandler handler);
 
  Client& on_custom_notification(RawNotificationHandler handler);
 
  Client& set_handler(const ClientHandler& handler);
 
  Client& set_handler(const ClientHandlerInterface& handler);
 
  core::Result<core::Unit> start();
 
  core::Result<core::Unit> handle_notification(
      const protocol::JsonRpcNotification& notification);
 
  core::Result<protocol::JsonRpcResponse> handle_request(
      const protocol::JsonRpcRequest& request);
 
  core::Result<protocol::Json> raw_request(
      const protocol::JsonRpcRequest& request);
 
  core::Result<core::Unit> raw_notification(
      const protocol::JsonRpcNotification& notification);
 
  void stop() noexcept;
 
  ~Client();
 
 private:
  template <typename Handler>
  Handler copy_handler(Handler Client::* slot) const {
    std::lock_guard<std::mutex> lock(*handlers_mutex_);
    return this->*slot;
  }
 
  template <typename Handler>
  void store_handler(Handler Client::* slot, Handler handler) {
    std::lock_guard<std::mutex> lock(*handlers_mutex_);
    this->*slot = std::move(handler);
  }
 
  core::Result<protocol::Json> send_request(std::string method,
                                            protocol::Json params);
  core::Result<protocol::JsonRpcResponse> send_rpc_request(
      protocol::JsonRpcRequest request);
  core::Result<core::Unit> ensure_transport_started();
  core::Result<core::Unit> record_server_capabilities(
      const protocol::Json& initialize_payload);
  bool server_capabilities_known() const noexcept;
  bool supports_server_completion() const noexcept;
  bool supports_server_logging() const noexcept;
  bool supports_server_resource_subscribe() const noexcept;
  bool supports_server_task_list() const noexcept;
  bool supports_server_task_cancel() const noexcept;
  bool supports_server_tasks() const noexcept;
  bool supports_server_task_tool_call() const noexcept;
  std::optional<protocol::ClientCapabilities> capabilities_snapshot() const;
  std::optional<protocol::ServerCapabilities> server_capabilities_snapshot()
      const;
  CancellationToken begin_request_cancellation(
      const protocol::RequestId& request_id);
  void end_request_cancellation(const protocol::RequestId& request_id) noexcept;
  void cancel_request(const protocol::RequestId& request_id) noexcept;
  std::int64_t next_request_id() noexcept {
    return next_request_id_.fetch_add(1, std::memory_order_relaxed);
  }
 
  std::unique_ptr<Transport> transport_;
  std::atomic<std::int64_t> next_request_id_{1};
  bool transport_started_ = false;
  std::vector<protocol::Root> roots_;
  std::optional<protocol::Json> last_initialize_params_;
  std::optional<protocol::ClientCapabilities> capabilities_;
  std::optional<protocol::ServerCapabilities> server_capabilities_;
  InitializedHandler initialized_handler_;
  CancelledHandler cancelled_handler_;
  LoggingMessageHandler logging_message_handler_;
  ListChangedHandler tool_list_changed_handler_;
  ListChangedHandler prompt_list_changed_handler_;
  ListChangedHandler resource_list_changed_handler_;
  ResourceUpdatedHandler resource_updated_handler_;
  ProgressHandler progress_handler_;
  ElicitationCompleteHandler elicitation_complete_handler_;
  TaskStatusHandler task_status_handler_;
  ListChangedHandler roots_list_changed_handler_;
  RootsListRequestHandler roots_list_request_handler_;
  RootsListRequestCancellationHandler roots_list_request_cancellation_handler_;
  SamplingRequestHandler sampling_request_handler_;
  SamplingRequestCancellationHandler sampling_request_cancellation_handler_;
  ElicitationRequestHandler elicitation_request_handler_;
  ElicitationRequestCancellationHandler
      elicitation_request_cancellation_handler_;
  CustomRequestHandler custom_request_handler_;
  CustomRequestCancellationHandler custom_request_cancellation_handler_;
  RawNotificationHandler raw_notification_handler_;
  mutable std::shared_ptr<std::mutex> state_mutex_ =
      std::make_shared<std::mutex>();
  mutable std::shared_ptr<std::mutex> handlers_mutex_ =
      std::make_shared<std::mutex>();
  std::shared_ptr<std::mutex> active_request_cancellations_mutex_ =
      std::make_shared<std::mutex>();
  std::shared_ptr<std::unordered_map<std::string, CancellationSource>>
      active_request_cancellations_ = std::make_shared<
          std::unordered_map<std::string, CancellationSource>>();
 
  std::atomic<int> in_flight_requests_{0};
  mutable std::mutex in_flight_mutex_;
  std::shared_ptr<std::condition_variable> in_flight_cv_ =
      std::make_shared<std::condition_variable>();
};
 
}  // namespace mcp::client
 
#include "cxxmcp/client/handler.hpp"