Struct HttpTransport 

pub struct HttpTransport { /* private fields */ }

HTTP transport for MCP servers

Implements the Streamable HTTP transport from the MCP specification.

Construction

There are two ways to create an HttpTransport:

• HttpTransport::new(router) — wraps an McpRouter, with full support for per-session notification bridging, sampling, and .layer() middleware.

• HttpTransport::from_service(service) — wraps any Service<RouterRequest> (e.g., McpProxy). The service is cloned for each session. Notification bridging and sampling are not set up automatically; the caller should configure these on the service before passing it in. .layer() is not supported in this mode.

Implementations

impl HttpTransport

pub fn new(router: McpRouter) -> Self

Create a new HTTP transport wrapping an MCP router.

Supports per-session notification bridging, sampling, and .layer() middleware.

pub fn from_service<S>(service: S) -> Self

where S: Service<RouterRequest, Response = RouterResponse, Error = Infallible> + Clone + Send + 'static, S::Future: Send,

Create an HTTP transport from a pre-built service.

This accepts any Service<RouterRequest> implementation, such as McpProxy. The service is cloned for each HTTP session.

Notification bridging and sampling are not set up automatically. The caller should configure these on the service before passing it in.

.layer() is not supported when using from_service() — wrap the service with middleware before passing it in.

Example

use tower_mcp::transport::http::HttpTransport;
use tower_mcp::proxy::McpProxy;

let proxy: McpProxy = /* ... */;
let transport = HttpTransport::from_service(proxy);
transport.serve("127.0.0.1:3000").await?;

pub fn with_notifications( router: McpRouter, notification_rx: NotificationReceiver, ) -> Self

Create an HTTP transport that drains a caller-owned notification channel and fans the items out to every live session’s SSE stream.

This mirrors GenericStdioTransport::with_notifications and is the supported way to push server-originated notifications (e.g. notifications/resources/updated) from outside any request handler — background tasks, lifecycle hooks, anything async that needs to notify subscribed clients.

Per-session notification channels (in-handler ctx.send_log(), progress updates) are unaffected. The external channel runs in parallel and broadcasts to every active session; MCP clients are expected to ignore notifications they didn’t subscribe to.

Example

use tower_mcp::{BoxError, McpRouter};
use tower_mcp::context::{ServerNotification, notification_channel};
use tower_mcp::transport::http::HttpTransport;

#[tokio::main]
async fn main() -> Result<(), BoxError> {
    let (notif_tx, notif_rx) = notification_channel(256);

    let router = McpRouter::new().server_info("my-server", "1.0.0");

    // Hold onto notif_tx in your application state so background tasks
    // can push notifications. tx is `Clone`.
    let pusher = notif_tx.clone();
    tokio::spawn(async move {
        let _ = pusher.send(ServerNotification::ResourceUpdated {
            uri: "claude://chats/123".to_string(),
        }).await;
    });

    let transport = HttpTransport::with_notifications(router, notif_rx);
    transport.serve("127.0.0.1:3000").await?;
    Ok(())
}

pub fn external_notifications( self, notification_rx: NotificationReceiver, ) -> Self

Attach a caller-owned notification receiver after construction.

Useful when wrapping a pre-built service via from_service, where setting a sender on the router isn’t part of the flow. See with_notifications for the typical router-based path.

pub fn with_sampling(self) -> Self

Enable sampling support for this transport.

When sampling is enabled, tool handlers can use ctx.sample() to request LLM completions from connected clients. The server sends each request on the SSE response stream of the POST that caused it, and the client responds via a separate POST. These associated streams are not replayed; use session affinity while a request is in flight.

Example

use tower_mcp::{BoxError, McpRouter, ToolBuilder, CallToolResult, CreateMessageParams, SamplingMessage};
use tower_mcp::extract::{Context, RawArgs};
use tower_mcp::transport::http::HttpTransport;

#[tokio::main]
async fn main() -> Result<(), BoxError> {
    let tool = ToolBuilder::new("ai-tool")
        .extractor_handler((), |ctx: Context, RawArgs(_): RawArgs| async move {
            // Request LLM completion from client
            let params = CreateMessageParams::new(
                vec![SamplingMessage::user("Summarize this...")],
                500,
            );
            let result = ctx.sample(params).await?;
            Ok(CallToolResult::text(format!("{:?}", result.content)))
        })
        .build();

    let router = McpRouter::new()
        .server_info("my-server", "1.0.0")
        .tool(tool);

    let transport = HttpTransport::new(router).with_sampling();
    transport.serve("127.0.0.1:3000").await?;
    Ok(())
}

pub fn require_sessions(self) -> Self

Require strict session management.

When enabled, requests without an mcp-session-id header are rejected with a SessionRequired error (-32006). Clients must complete the initialize handshake and include the session ID on all subsequent requests, as specified by the MCP 2025-11-25 spec.

By default, sessions are optional for compatibility with clients (Codex CLI, Cursor, etc.) that don’t carry the session ID forward after initialization.

Example

use tower_mcp::McpRouter;
use tower_mcp::transport::http::HttpTransport;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let router = McpRouter::new().server_info("my-server", "1.0.0");
    let transport = HttpTransport::new(router).require_sessions();
    transport.serve("127.0.0.1:3000").await?;
    Ok(())
}

pub fn protocol_support(self, support: ProtocolSupport) -> Self

Set the exact protocol versions this transport accepts and advertises.

By default, every protocol implementation compiled into tower-mcp is enabled. This setting can narrow that set per server instance. Versions are advertised by server/discover in the order supplied.

pub fn protocol_versions<I, S>( self, versions: I, ) -> Result<Self, ProtocolSupportError>

where I: IntoIterator<Item = S>, S: Into<String>,

Construct and set an exact runtime protocol-version allow-list.

Returns an error when the list is empty, duplicated, or names a version whose Cargo feature was not compiled.

pub fn sse_responses(self, enabled: bool) -> Self

Enable SSE-wrapping for synchronous JSON-RPC responses.

When enabled, synchronous responses (initialize, tools/list, tools/call, etc.) are returned with Content-Type: text/event-stream and formatted as an SSE message event:

event: message
data: {"jsonrpc":"2.0","id":1,"result":{...}}

This matches the behavior of rmcp’s StreamableHttpService, which always uses SSE format for all responses. The MCP Streamable HTTP spec allows both bare JSON and SSE for synchronous responses; this option is provided for compatibility with clients that expect rmcp’s SSE-always behavior.

Known divergence from rmcp: rmcp’s StreamableHttpService always uses SSE for synchronous responses by default. tower-mcp defaults to bare JSON (the spec-correct choice, matching the SHOULD in the 2025-11-25 spec). Use .sse_responses(true) to match rmcp’s behavior when targeting clients written against rmcp.

The existing SSE notification stream (GET /) and subscriptions/listen stream (2026-07-28+) are unaffected by this flag.

Default: false (bare JSON, Content-Type: application/json).

Example

let transport = HttpTransport::new(router).sse_responses(true);

pub fn stamp_server_info(self, enabled: bool) -> Self

Whether 2026-07-28 stateless responses carry server identity in _meta["io.modelcontextprotocol/serverInfo"].

Per SEP-2575, servers SHOULD identify themselves in each result’s _meta “unless specifically configured not to do so” – this is that configuration. Only applies to the version-gated 2026-07-28 stateless dispatch path (stateless feature); other protocol versions and transports are unaffected, and identity there is carried by initialize’s top-level serverInfo instead.

Only takes effect when the transport was built from an McpRouter (HttpTransport::new); a transport built from a pre-built service (HttpTransport::from_service) has no router to read identity from and never stamps, regardless of this setting.

Default: true.

Example

let transport = HttpTransport::new(router).stamp_server_info(false);

pub fn max_body_size(self, bytes: usize) -> Self

Set the maximum accepted POST body size in bytes.

Requests whose body exceeds the limit are rejected with HTTP 413 (Payload Too Large) before any JSON parsing or dispatch happens. A Content-Length header above the limit short-circuits without reading the body; chunked bodies are capped while streaming.

Default: 4 MiB (DEFAULT_MAX_BODY_SIZE), matching rmcp.

Interplay with axum’s DefaultBodyLimit

axum’s built-in DefaultBodyLimit (2 MB by default) only applies to body-consuming extractors such as Bytes, String, and Json. The MCP endpoint consumes the raw Request and reads the body itself, so DefaultBodyLimit never applies to it; this transport-level limit is the only bound on the MCP POST body. Layering DefaultBodyLimit onto the router returned by into_router does not change the MCP endpoint’s behavior.

Example

use tower_mcp::McpRouter;
use tower_mcp::transport::http::HttpTransport;

let router = McpRouter::new().server_info("my-server", "1.0.0");
// Accept request bodies up to 1 MiB.
let transport = HttpTransport::new(router).max_body_size(1024 * 1024);

pub fn stateless(self, config: StatelessConfig) -> Self

Enable the legacy SEP-1442 stateless opt-in path.

This activates the SEP-1442-style stateless behavior for clients that do NOT send MCP-Protocol-Version: 2026-07-28. Specifically, when a crate::stateless::StatelessConfig is set:

• Requests without a session ID can be served without an initialize handshake (if crate::stateless::StatelessConfig::optional_sessions is true).
• The server/discover RPC is enabled (if crate::stateless::StatelessConfig::enable_discover is true).
• Protocol version may be required in every request body (if crate::stateless::StatelessConfig::require_protocol_version is true).

Note: this method does NOT control the automatic version-gated stateless path for 2026-07-28+ clients. When the stateless feature is compiled in, any request with MCP-Protocol-Version: 2026-07-28 and no mcp-session-id is dispatched statelessly regardless of whether this method is called. See the crate::stateless module documentation for the full two-path explanation.

Stateful clients (those that send mcp-session-id) continue to work normally on the same transport.

Example

use tower_mcp::McpRouter;
use tower_mcp::transport::http::HttpTransport;
use tower_mcp::stateless::StatelessConfig;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let router = McpRouter::new().server_info("my-server", "1.0.0");
    // Enables the SEP-1442 opt-in path. 2026-07-28 clients are
    // handled statelessly regardless of this call.
    let transport = HttpTransport::new(router)
        .stateless(StatelessConfig::new());
    transport.serve("127.0.0.1:3000").await?;
    Ok(())
}

pub fn disable_origin_validation(self) -> Self

Disable Origin header validation (not recommended for production)

pub fn allowed_origins(self, origins: Vec<String>) -> Self

Set allowed origins for CORS/security validation

pub fn disable_host_validation(self) -> Self

Disable Host header validation (not recommended when binding to a non-loopback interface).

Host validation is the defense-in-depth pair to Origin validation: it rejects requests whose Host header doesn’t match the server’s expected hostname, blocking direct DNS-rebinding attacks where a malicious site resolves its own domain to 127.0.0.1.

pub fn allowed_hosts(self, hosts: Vec<String>) -> Self

Set allowed hosts for the Host header allowlist.

Each entry should be a host:port pair (e.g. "api.example.com", "api.example.com:8443"). Localhost variants (localhost, 127.0.0.1, ::1, with any port) are always accepted regardless of this list.

When the Host header is missing, the validator falls back to the HTTP/2 :authority pseudo-header from request.uri().authority(), since middleware like axum::Router::nest can strip the synthesized Host header before it reaches our handler.

pub fn session_config(self, config: SessionConfig) -> Self

Configure session management (TTL, max sessions, cleanup interval)

pub fn session_ttl(self, ttl: Duration) -> Self

Set session TTL (convenience method)

pub fn max_sessions(self, max: usize) -> Self

Set maximum number of concurrent sessions (convenience method)

pub fn session_store(self, store: Arc<dyn SessionStore>) -> Self

Configure a pluggable SessionStore for persisting session metadata.

The default is an in-process MemorySessionStore — supply an external store (Redis, Postgres, etc.) to share session metadata across server instances behind a load balancer.

Runtime state (broadcast channels, pending requests, service instances) is always kept per-instance; only persistent metadata is mirrored to the store.

Example

use std::sync::Arc;
use tower_mcp::{HttpTransport, McpRouter};
use tower_mcp::session_store::{MemorySessionStore, SessionStore};

let router = McpRouter::new();
let store: Arc<dyn SessionStore> = Arc::new(MemorySessionStore::new());
let transport = HttpTransport::new(router).session_store(store);

pub fn event_store(self, store: Arc<dyn EventStore>) -> Self

Configure a pluggable EventStore for SSE event buffering and stream resumption.

The default is an in-process MemoryEventStore with a 1000-event ring buffer per session — supply an external store (Redis, etc.) so clients can resume SSE streams after reconnecting to a different server instance behind a load balancer (SEP-1699).

Typically paired with a matching session_store so both session metadata and buffered events survive across instances.

Example

use std::sync::Arc;
use tower_mcp::{HttpTransport, McpRouter};
use tower_mcp::event_store::{EventStore, MemoryEventStore};

let router = McpRouter::new();
let store: Arc<dyn EventStore> = Arc::new(MemoryEventStore::new());
let transport = HttpTransport::new(router).event_store(store);

pub fn auto_reinitialize_sessions(self, enabled: bool) -> Self

Enable auto-reinitialization for unknown session IDs.

When a request arrives with an mcp-session-id that is not live locally and has no record in the configured session_store, the transport normally returns a session-not-found error. With this flag enabled, the transport instead spins up a new session claiming that ID and completes the initialize handshake internally with synthetic client info (name = "auto-recovered", empty capabilities).

This lets tolerant clients continue after a server restart without repeating the handshake, at the cost of losing the original client’s identity and negotiated capabilities. Prefer pairing this with a real session_store — the store path runs first and preserves full identity when a record exists.

Disabled by default. This is the pattern established by anubis-mcp #125.

Example

use tower_mcp::{HttpTransport, McpRouter};

let router = McpRouter::new();
let transport = HttpTransport::new(router).auto_reinitialize_sessions(true);

pub fn oauth(self, metadata: ProtectedResourceMetadata) -> Self

Configure OAuth 2.1 Protected Resource Metadata for this transport.

This lower-level method only serves metadata; it does not install token or scope enforcement. Prefer Self::into_oauth_router for a complete, fail-closed MCP resource-server setup.

Example

use tower_mcp::oauth::ProtectedResourceMetadata;
use tower_mcp::transport::http::HttpTransport;
use tower_mcp::McpRouter;

let metadata = ProtectedResourceMetadata::new("https://mcp.example.com")
    .authorization_server("https://auth.example.com")
    .scope("mcp:read");

let router = McpRouter::new().server_info("my-server", "1.0.0");
let transport = HttpTransport::new(router).oauth(metadata);

pub fn into_oauth_router<V>( self, validator: V, metadata: ProtectedResourceMetadata, policy: ScopePolicy, ) -> Result<Router, ProtectedResourceMetadataError>

where V: TokenValidator,

Build a fully protected OAuth resource-server router.

This validates the Protected Resource Metadata, serves it at the path-aware RFC 9728 endpoint, validates bearer tokens, independently enforces the token audience against metadata.resource, and installs fail-closed per-operation scope enforcement.

Errors

Returns an error when the resource metadata is not suitable for an MCP resource server.

pub fn into_oauth_router_with_handle<V>( self, validator: V, metadata: ProtectedResourceMetadata, policy: ScopePolicy, ) -> Result<(Router, SessionHandle), ProtectedResourceMetadataError>

where V: TokenValidator,

Build a fully protected OAuth router and return its session handle.

This is the session-management variant of Self::into_oauth_router.

pub fn into_oauth_router_at<V>( self, path: &str, validator: V, metadata: ProtectedResourceMetadata, policy: ScopePolicy, ) -> Result<Router, ProtectedResourceMetadataError>

where V: TokenValidator,

Build a fully protected OAuth router mounted at path.

The metadata route is derived from metadata.resource, not from the local mount path, so it remains correct for path-based resource URLs.

pub fn into_oauth_router_at_with_handle<V>( self, path: &str, validator: V, metadata: ProtectedResourceMetadata, policy: ScopePolicy, ) -> Result<(Router, SessionHandle), ProtectedResourceMetadataError>

where V: TokenValidator,

Build a path-mounted protected OAuth router and return its session handle.

pub fn layer<L>(self, layer: L) -> Self

where L: Layer<McpRouter> + Send + Sync + 'static, L::Service: Service<RouterRequest, Response = RouterResponse> + Clone + Send + 'static, <L::Service as Service<RouterRequest>>::Error: Display + Send, <L::Service as Service<RouterRequest>>::Future: Send,

Apply a tower middleware layer to MCP request processing.

Panics

Panics if this transport was created via from_service(). When using from_service(), wrap the service with middleware before passing it in.

pub fn into_router(self) -> Router

Build the axum router for this transport.

pub fn into_router_with_handle(self) -> (Router, SessionHandle)

Build the axum router and return a SessionHandle for managing sessions and final subscription streams.

Example

let transport = HttpTransport::new(router);
let (router, handle) = transport.into_router_with_handle();

// Use handle in an admin endpoint
let count = handle.session_count().await;

pub fn into_router_at(self, path: &str) -> Router

Build an axum router mounted at a specific path.

pub fn into_router_at_with_handle(self, path: &str) -> (Router, SessionHandle)

Build an axum router mounted at a specific path and return a SessionHandle for managing sessions and final subscription streams.

pub async fn serve(self, addr: &str) -> Result<()>

Serve the transport on the given address

This is a convenience method that creates a TCP listener and serves the transport.