use std::net::SocketAddr;
use std::sync::Arc;
use super::{HttpHandler, HttpSession, Ids};
use crate::adapters::primary::transport::http::config::ServerConfig;
use crate::adapters::primary::transport::http::limits::RateLimiter;
use crate::adapters::primary::transport::http::router::McpRouter;
use crate::adapters::primary::transport::http::session_store::SessionStore;
use crate::adapters::primary::transport::http::{AuthPolicy, Authenticator};
use crate::adapters::primary::transport::origin::OriginPolicy;
use crate::application::server::Server;
use crate::application::server::dispatch::Dispatcher;
use crate::domain::{ProtocolVersion, SupportedVersions};
pub struct HttpBuilder {
server: Server,
addr: SocketAddr,
ceiling: ProtocolVersion,
config: Option<ServerConfig>,
ids: Option<Ids>,
authenticator: Option<Arc<dyn Authenticator + Send + Sync>>,
policy: Option<AuthPolicy>,
origin_policy: Option<OriginPolicy>,
rate_limiter: Option<RateLimiter>,
#[cfg(feature = "legacy-sse")]
legacy_sse: Option<LegacySsePaths>,
}
#[cfg(feature = "legacy-sse")]
#[derive(Debug, Clone)]
struct LegacySsePaths {
stream: String,
message: String,
}
impl core::fmt::Debug for HttpBuilder {
fn fmt(&self, formatter: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
formatter
.debug_struct("HttpBuilder")
.field("addr", &self.addr)
.field("ceiling", &self.ceiling)
.field("config", &self.config)
.field("ids_supplied", &self.ids.is_some())
.finish_non_exhaustive()
}
}
impl HttpBuilder {
#[must_use]
pub fn new(server: Server) -> Self {
Self {
server,
addr: SocketAddr::from(([127, 0, 0, 1], 3000)),
ceiling: default_ceiling(),
config: None,
ids: None,
authenticator: None,
policy: None,
origin_policy: None,
rate_limiter: None,
#[cfg(feature = "legacy-sse")]
legacy_sse: None,
}
}
#[cfg(feature = "legacy-sse")]
#[must_use]
pub fn legacy_sse(self) -> Self {
self.legacy_sse_at(
crate::adapters::primary::transport::legacy_sse::DEFAULT_STREAM_PATH,
crate::adapters::primary::transport::legacy_sse::DEFAULT_MESSAGE_PATH,
)
}
#[cfg(feature = "legacy-sse")]
#[must_use]
pub fn legacy_sse_at(mut self, stream: impl Into<String>, message: impl Into<String>) -> Self {
self.legacy_sse = Some(LegacySsePaths {
stream: stream.into(),
message: message.into(),
});
self
}
#[must_use]
pub const fn addr(mut self, addr: SocketAddr) -> Self {
self.addr = addr;
self
}
#[must_use]
pub const fn version(mut self, version: ProtocolVersion) -> Self {
self.ceiling = version;
self
}
#[must_use]
pub fn config(mut self, config: ServerConfig) -> Self {
self.config = Some(config);
self
}
#[must_use]
pub fn ids(mut self, ids: Ids) -> Self {
self.ids = Some(ids);
self
}
#[must_use]
pub fn with_authenticator(
mut self,
authenticator: Arc<dyn Authenticator + Send + Sync>,
) -> Self {
self.authenticator = Some(authenticator);
self
}
#[must_use]
pub const fn with_policy(mut self, policy: AuthPolicy) -> Self {
self.policy = Some(policy);
self
}
#[must_use]
pub fn with_origin_policy(mut self, origin: OriginPolicy) -> Self {
self.origin_policy = Some(origin);
self
}
#[must_use]
pub fn with_rate_limiter(mut self, limiter: RateLimiter) -> Self {
self.rate_limiter = Some(limiter);
self
}
pub async fn bind(self) -> Result<(tokio::net::TcpListener, ServeReport), ServeError> {
let (listener, bound) = bind(self.addr).await?;
let report = ServeReport {
addr: bound,
ceiling: self.ceiling,
};
Ok((listener, report))
}
pub async fn serve(self) -> Result<ServeReport, ServeError> {
let parts = routes(self)?;
let (listener, bound) = bind(parts.addr).await?;
axum::serve(listener, parts.router)
.await
.map_err(ServeError::Accept)?;
Ok(ServeReport {
addr: bound,
ceiling: parts.ceiling,
})
}
pub async fn serve_on(
self,
listener: tokio::net::TcpListener,
) -> Result<ServeReport, ServeError> {
let parts = routes(self)?;
let bound = listener.local_addr().map_err(|source| ServeError::Bind {
addr: parts.addr,
source,
})?;
axum::serve(listener, parts.router)
.await
.map_err(ServeError::Accept)?;
Ok(ServeReport {
addr: bound,
ceiling: parts.ceiling,
})
}
}
#[derive(Debug)]
pub(super) struct BuilderOutput {
pub(super) router: axum::Router,
pub(super) addr: SocketAddr,
pub(super) ceiling: ProtocolVersion,
}
pub(super) fn routes(builder: HttpBuilder) -> Result<BuilderOutput, ServeError> {
let HttpBuilder {
server,
addr,
ceiling,
config,
ids,
authenticator,
policy,
origin_policy,
rate_limiter,
#[cfg(feature = "legacy-sse")]
legacy_sse,
} = builder;
let ids = match ids {
Some(ids) => ids,
None => default_ids()?,
};
#[cfg(feature = "random-ids")]
let server = {
let mut server = server;
server.ensure_request_state_key(crate::adapters::secondary::id_source::random_state_key);
server
};
let config = match config {
Some(config) => config,
None => ServerConfig::new()?,
};
let dispatcher = Arc::new(Dispatcher::new(Arc::new(server)));
let sessions = Arc::new(SessionStore::<HttpSession>::new(Arc::clone(&ids)));
let handler = HttpHandler::new(Arc::clone(&dispatcher), Arc::clone(&sessions), ceiling);
#[cfg(feature = "legacy-sse")]
let legacy = legacy_sse
.map(|paths| {
let mut router = crate::adapters::primary::transport::legacy_sse::LegacyRouter::new(
config.clone(),
Arc::new(SessionStore::new(ids)),
crate::adapters::primary::transport::legacy_sse::ServerDispatcher::new(
Arc::clone(&dispatcher),
ProtocolVersion::V2024_11_05,
),
)
.with_stream_path(paths.stream)
.with_message_path(paths.message);
if let Some(origin) = origin_policy.clone() {
router = router.with_origins(origin);
}
if let Some(limiter) = rate_limiter.clone() {
router = router.with_rate_limiter(limiter);
}
if let (Some(policy), Some(authenticator)) = (policy, authenticator.clone()) {
router = router.with_auth(policy, authenticator);
}
router.into_router()
})
.transpose()?;
let mut router = McpRouter::new(config, sessions, handler);
if let Some(authenticator) = authenticator {
router = router.with_authenticator(authenticator);
}
if let Some(policy) = policy {
router = router.with_policy(policy);
}
if let Some(origin) = origin_policy {
router = router.with_origin_policy(origin);
}
if let Some(limiter) = rate_limiter {
router = router.with_rate_limiter(limiter);
}
let router = router.into_router().map_err(ServeError::from)?;
#[cfg(feature = "legacy-sse")]
let router = match legacy {
Some(legacy) => router.merge(legacy),
None => router,
};
Ok(BuilderOutput {
router,
addr,
ceiling,
})
}
#[cfg(test)]
#[allow(
clippy::redundant_pub_crate,
reason = "the function is test-only; `pub(crate)` is needed in test builds for the tests to reach it, but in lib builds nothing outside the crate can see it, and the lint does not have a per-cfg answer."
)]
pub(crate) fn test_routes(
server: Server,
ceiling: ProtocolVersion,
config: Option<ServerConfig>,
ids: Option<Ids>,
) -> Result<axum::Router, ServeError> {
routes(HttpBuilder {
server,
addr: SocketAddr::from(([127, 0, 0, 1], 3000)),
ceiling,
config,
ids,
authenticator: None,
policy: None,
origin_policy: None,
rate_limiter: None,
#[cfg(feature = "legacy-sse")]
legacy_sse: None,
})
.map(|parts| parts.router)
}
#[cfg(feature = "random-ids")]
fn default_ids() -> Result<Ids, ServeError> {
use crate::adapters::secondary::id_source::RandomIds;
RandomIds::new()
.map(|ids| Arc::new(ids) as Ids)
.map_err(|error| ServeError::IdSource {
reason: error.to_string(),
})
}
#[cfg(not(feature = "random-ids"))]
fn default_ids() -> Result<Ids, ServeError> {
Err(ServeError::NoIdSource)
}
pub(super) fn default_ceiling() -> ProtocolVersion {
SupportedVersions::all().map_or_else(
|_| ProtocolVersion::latest_known(),
|supported| supported.latest_handshake(),
)
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ServeReport {
pub addr: SocketAddr,
pub ceiling: ProtocolVersion,
}
#[derive(Debug, thiserror::Error)]
pub enum ServeError {
#[error("could not bind {addr}: {source}")]
Bind {
addr: SocketAddr,
#[source]
source: std::io::Error,
},
#[error("the endpoint policy is not usable: {0}")]
Endpoint(#[from] crate::adapters::primary::transport::http::config::EndpointError),
#[error("the accept loop failed: {0}")]
Accept(#[source] std::io::Error),
#[error("the route table could not be built: {0}")]
Routes(#[from] crate::adapters::primary::transport::http::router::RouterError),
#[cfg(feature = "legacy-sse")]
#[error("the legacy HTTP+SSE routes could not be mounted: {0}")]
LegacyRoutes(#[from] crate::adapters::primary::transport::legacy_sse::RouteError),
#[error(
"this build has no id source: build with `random-ids`, or pass one to \
HttpBuilder::ids"
)]
NoIdSource,
#[error("session ids could not be drawn: {reason}")]
IdSource {
reason: String,
},
}
async fn bind(addr: SocketAddr) -> Result<(tokio::net::TcpListener, SocketAddr), ServeError> {
let listener = tokio::net::TcpListener::bind(addr)
.await
.map_err(|source| ServeError::Bind { addr, source })?;
let bound = listener
.local_addr()
.map_err(|source| ServeError::Bind { addr, source })?;
Ok((listener, bound))
}