http_transport.hpp

Go to the documentation of this file.

// Copyright (c) 2025 [caomengxuan666]
 
#pragma once
 
#include <chrono>
#include <condition_variable>
#include <cstddef>
#include <cstdint>
#include <deque>
#include <memory>
#include <mutex>
#include <optional>
#include <string>
#include <string_view>
#include <unordered_map>
#include <vector>
 
#if defined(CXXMCP_ENABLE_AUTH)
#include "cxxmcp/auth/server_auth_endpoints.hpp"
#endif
 
#include "cxxmcp/server/transport.hpp"
 
 
namespace mcp::server {
 
struct ProtectedResourceMetadataConfig {
  std::string resource;
  std::vector<std::string> authorization_servers;
  std::vector<std::string> scopes_supported;
  std::optional<std::string> resource_name;
  std::optional<std::string> resource_documentation;
};
 
struct AuthChallengeConfig {
  std::string scheme = "Bearer";
  std::optional<std::string> resource_metadata_url;
  std::optional<std::string> scope;
};
 
struct HttpTransportOptions {
  std::string listen_host = "127.0.0.1";
  int listen_port = 0;
  std::string path = "/mcp";
  std::optional<std::chrono::milliseconds> sse_retry;
  bool enable_sse_polling = false;
  std::chrono::milliseconds sse_disconnect_retry{5000};
  std::vector<std::string> allowed_origins;
  std::vector<std::string> allowed_hosts = {"localhost", "127.0.0.1", "::1"};
  std::size_t max_pending_sse_events = 1024;
  std::size_t max_pending_sse_bytes = 4 * 1024 * 1024;
  std::size_t max_sse_replay_events = 256;
  std::chrono::milliseconds request_timeout{30000};
  std::size_t max_request_body_bytes = 10 * 1024 * 1024;
  std::chrono::milliseconds read_timeout{30000};
  std::chrono::milliseconds write_timeout{30000};
  std::size_t max_sessions = 1024;
  bool stateless = false;
  std::string auth_challenge = std::string(DefaultAuthChallenge);
  AuthChallengeConfig auth_challenge_config;
  ProtectedResourceMetadataConfig protected_resource_metadata;
#if defined(CXXMCP_ENABLE_AUTH)
  ::mcp::auth::AuthorizationServerConfig authorization_server;
#endif
};
 
class HttpTransport final : public Transport {
 public:
  explicit HttpTransport(HttpTransportOptions options);
 
  ~HttpTransport() override;
 
  HttpTransport(const HttpTransport&) = delete;
  HttpTransport& operator=(const HttpTransport&) = delete;
 
  core::Result<core::Unit> start(
      RequestHandler handler,
      NotificationHandler notification_handler = {}) override;
 
  core::Result<protocol::JsonRpcResponse> send_request(
      const protocol::JsonRpcRequest& request) override;
 
  core::Result<protocol::JsonRpcResponse> send_request_to_session(
      std::string_view session_id,
      const protocol::JsonRpcRequest& request) override;
 
  std::optional<protocol::ClientCapabilities> client_capabilities()
      const override;
 
  std::optional<protocol::ClientCapabilities> client_capabilities_for_session(
      std::string_view session_id) const override;
 
  core::Result<core::Unit> send_notification(
      const protocol::JsonRpcNotification& notification) override;
 
  core::Result<core::Unit> send_notification_to_session(
      std::string_view session_id,
      const protocol::JsonRpcNotification& notification) override;
 
  void stop() noexcept override;
 
  void disconnect_session_sse(std::string_view session_id);
 
  std::string_view name() const noexcept override;
 
  void wait_until_ready() override;
 
 private:
  struct HttpServerHolder;
 
  struct PendingRequest {
    std::mutex mutex;
    std::condition_variable cv;
    bool ready = false;
    std::optional<protocol::JsonRpcResponse> response;
    std::optional<core::Error> error;
  };
 
  struct QueuedEvent {
    std::optional<std::uint64_t> event_id;
    std::string payload;
  };
 
  struct ProgressResponseSink {
    std::mutex mutex;
    std::condition_variable cv;
    std::deque<std::string> events;
    bool done = false;
  };
 
  struct SessionState {
    std::string session_id;
    bool initialized = false;
    std::optional<protocol::ClientCapabilities> client_capabilities;
    std::deque<QueuedEvent> pending_notifications;
    std::size_t pending_notification_bytes = 0;
    std::deque<QueuedEvent> replay_notifications;
    std::unordered_map<std::string, std::shared_ptr<PendingRequest>>
        pending_requests;
    std::uint64_t next_notification_event_id = 1;
    std::size_t active_sse_streams = 0;
    bool sse_disconnect_requested = false;
    std::shared_ptr<ProgressResponseSink> active_progress_sink;
  };
 
  void abort_pending_requests_locked(SessionState& session,
                                     std::string message);
  void clear_outbound_events_locked(SessionState& session);
  core::Result<core::Unit> enqueue_outbound_event_locked(SessionState& session,
                                                         QueuedEvent event);
  void remember_replay_event_locked(SessionState& session,
                                    const QueuedEvent& event);
  SessionState* find_session_locked(std::string_view session_id);
  const SessionState* find_session_locked(std::string_view session_id) const;
  SessionState* select_default_session_locked();
  void close_sse_stream(std::string_view session_id) noexcept;
 
  static std::optional<protocol::ProgressToken> extract_progress_token(
      const protocol::JsonRpcRequest& request);
 
  HttpTransportOptions options_;
  std::unique_ptr<HttpServerHolder> server_;
  mutable std::mutex mutex_;
  std::condition_variable notification_cv_;
  std::condition_variable startup_cv_;
  std::unordered_map<std::string, SessionState> sessions_;
  std::uint64_t next_session_id_ = 1;
  bool stopped_ = false;
  bool stop_requested_ = false;
  bool startup_finished_ = false;
  bool initialized_ = false;
  std::string default_session_id_;
};
 
}  // namespace mcp::server