barrel_mcp_http_engine (barrel_mcp v3.0.1)

View Source

Transport-neutral MCP HTTP engine.

Holds the protocol logic for both the simple HTTP transport and the Streamable HTTP transport (POST/GET/DELETE/OPTIONS, SSE, sessions, CORS, Origin validation, authentication, the OAuth protected-resource-metadata endpoint and async tool calls) WITHOUT any dependency on a concrete HTTP server.

A binding (the built-in barrel_mcp_http_listener h1/h2 server, or an external adapter such as Livery's) reads the request line and body, then calls handle/6 with:

  • Method: the request method binary (<<"POST">> …).
  • Path: the request target (query string allowed; it is stripped here).
  • Headers: a [{binary(), binary()}] proplist. Lookups are case-insensitive.
  • Body: the full request body (<<>> when none).
  • Responder: a map of I/O closures (see below).
  • Config: the engine configuration (see the config() type).

The Responder abstracts response delivery so the engine never touches a socket:

   #{reply        => fun((Status, Headers, Body) -> ok),
     stream_start => fun((Status, Headers) -> ok),
     stream_chunk => fun((iodata()) -> ok | {error, term()}),
     stream_end   => fun(() -> ok)}

Headers passed to the closures is a [{binary(), binary()}] proplist (lowercase names). A streaming (SSE) response is stream_start then repeated stream_chunk then stream_end.

handle/6 runs in the calling (per-request) process. For a long-lived GET SSE stream it blocks in a receive loop until the session is terminated or the binding signals a client disconnect by sending the calling process the message mcp_disconnect.

Sections, in file order

  • Entry point and method dispatch: handle/6, route/7, dispatch/6 keyed on the engine mode and the verb.
  • Simple transport: POST only, no sessions.
  • Streamable POST: stream_post/4, the era fork in stream_post_request/6, batches, inbound responses.
  • Modern requests: header and body agreement, stateless dispatch, SSE response streams, subscriptions/listen.
  • Async tool calls: handle_async_tool_call/7 decides the tool-call mode for HTTP (inline, task, escalate, refuse) and waits for the worker; the legacy immediate-task path is handle_long_running_call/10.
  • Session resolution: lookup_session/5, owned_session/2, version negotiation.
  • GET and DELETE: the standalone stream with replay, and session termination.
  • The 2024-11-05 HTTP+SSE pair.
  • SSE, validation, authentication, CORS, Origin and bind helpers, the session manager bootstrap, plumbing.

Processes

handle/6 runs in the request process the listener spawned. A tool worker is spawned by the registry with this process (or a relay) as reply_to; a task collector, a relay and, on the legacy pair, a driver and a stream watcher are the other processes this module starts. The Server Internals guide lists them with who links or monitors whom.

Summary

Functions

Start barrel_mcp_session when the library is embedded without its application, so sessions work under a host supervisor.

Serve one HTTP request. Runs in the caller's process and, for an SSE response, blocks in it until the stream ends. Every binding (the built-in listener, an embedder's adapter) enters here.

Put the PRM URL into the auth config so the bearer challenge can name it (RFC 9728 resource_metadata).

Whether a bind address is loopback, in any of the shapes the listener options accept.

Normalise the resource_metadata option into the document the engine serves and the URL it advertises. Shared by both listeners.

Decide the Origin allow-list from the bind address: a loopback bind may default, a public one must list its origins.

Types

config/0

-type config() ::
          #{mode := stream | simple,
            auth_config := map(),
            session_enabled => boolean(),
            allowed_origins => any | [term()],
            allow_missing_origin => boolean(),
            sse_buffer_size => pos_integer(),
            subscription_keepalive_ms => pos_integer(),
            resource_metadata => undefined | map(),
            _ => _}.

responder/0

-type responder() ::
          #{reply := fun((non_neg_integer(), [{binary(), binary()}], iodata()) -> ok),
            stream_start := fun((non_neg_integer(), [{binary(), binary()}]) -> ok),
            stream_chunk := fun((iodata()) -> ok | {error, term()}),
            stream_end := fun(() -> ok)}.

Functions

ensure_session_manager()

-spec ensure_session_manager() -> ok | {ok, pid()} | {error, term()}.

Start barrel_mcp_session when the library is embedded without its application, so sessions work under a host supervisor.

handle(Method, RawPath, Headers, Body, Responder, Config)

-spec handle(binary(), binary(), [{binary(), binary()}], binary(), responder(), config()) -> ok.

Serve one HTTP request. Runs in the caller's process and, for an SSE response, blocks in it until the stream ends. Every binding (the built-in listener, an embedder's adapter) enters here.

init_auth(AuthOpts)

-spec init_auth(map()) -> {ok, map()} | {error, {auth_provider, module(), term()}}.

inject_resource_metadata_url(AuthConfig, _)

-spec inject_resource_metadata_url(map(), map() | undefined) -> map().

Put the PRM URL into the auth config so the bearer challenge can name it (RFC 9728 resource_metadata).

is_loopback(_)

-spec is_loopback(inet:ip_address() | string() | binary()) -> boolean().

Whether a bind address is loopback, in any of the shapes the listener options accept.

normalize_resource_metadata(M)

-spec normalize_resource_metadata(map() | undefined) -> map() | undefined.

Normalise the resource_metadata option into the document the engine serves and the URL it advertises. Shared by both listeners.

resolve_allowed_origins(Loopback, List)

-spec resolve_allowed_origins(boolean(), any | undefined | [binary()]) ->
                                 {ok, any | [term()]} | {error, allowed_origins_required}.

Decide the Origin allow-list from the bind address: a loopback bind may default, a public one must list its origins.