inbound

Inbound request classification for the modern per-request-envelope path.

Pure module: no I/O, no transport, no mcp.server imports. Runs the validation ladder against a decoded JSON-RPC body and returns either an :class:InboundModernRoute (every rung passed) or an :class:InboundLadderRejection (the first rung that failed). Callers map a rejection's code through :data:ERROR_CODE_HTTP_STATUS to pick the HTTP status.

Also hosts the shared header-value codec and the x-mcp-header schema validator so client emit and server validate read the same source of truth.

MCP_PROTOCOL_VERSION_HEADER module-attribute

MCP_PROTOCOL_VERSION_HEADER: Final = 'mcp-protocol-version'

Canonical lowercase name of the HTTP header carrying the MCP protocol version.

MCP_METHOD_HEADER module-attribute

MCP_METHOD_HEADER: Final = 'mcp-method'

Canonical lowercase name of the HTTP header carrying the JSON-RPC method.

MCP_NAME_HEADER module-attribute

MCP_NAME_HEADER: Final = 'mcp-name'

Canonical lowercase name of the HTTP header carrying the resource name (tool/prompt/resource URI).

X_MCP_HEADER_KEY module-attribute

X_MCP_HEADER_KEY: Final = 'x-mcp-header'

JSON-Schema property annotation that designates an Mcp-Param-* HTTP header.

NAME_BEARING_METHODS module-attribute

NAME_BEARING_METHODS: Final[Mapping[str, str]] = (
    MappingProxyType(
        {
            "tools/call": "name",
            "prompts/get": "name",
            "resources/read": "uri",
        }
    )
)

Method → params key whose value is mirrored as the Mcp-Name HTTP header.

Shared by client emit (which header to send) and server validate (which body field to compare against), so both ends agree on the field by construction.

encode_header_value

encode_header_value(value: str) -> str

Wrap value in the =?base64?...?= sentinel when it would not survive an HTTP field round-trip.

Plain printable ASCII without leading/trailing whitespace passes verbatim; anything else (control chars, non-ASCII, edge whitespace, or a value that already looks like the sentinel) is base64-wrapped so the receiver can recover the exact bytes.

Source code in src/mcp/shared/inbound.py

def encode_header_value(value: str) -> str:
    """Wrap `value` in the `=?base64?...?=` sentinel when it would not survive an HTTP field round-trip.

    Plain printable ASCII without leading/trailing whitespace passes verbatim;
    anything else (control chars, non-ASCII, edge whitespace, or a value that
    already looks like the sentinel) is base64-wrapped so the receiver can
    recover the exact bytes.
    """
    if _HEADER_SAFE.fullmatch(value) and value == value.strip() and not _B64_SENTINEL.fullmatch(value):
        return value
    return f"=?base64?{base64.b64encode(value.encode('utf-8')).decode('ascii')}?="

decode_header_value

decode_header_value(value: str | None) -> str | None

Inverse of :func:encode_header_value.

Returns the value verbatim unless it carries the =?base64?...?= sentinel, in which case the payload is decoded as UTF-8. A malformed sentinel (bad base64, non-canonical base64, or bad UTF-8) yields None so a corrupt header never matches a body value by accident. None in → None out so callers can pass headers.get(...) directly.

Source code in src/mcp/shared/inbound.py

def decode_header_value(value: str | None) -> str | None:
    """Inverse of :func:`encode_header_value`.

    Returns the value verbatim unless it carries the `=?base64?...?=` sentinel,
    in which case the payload is decoded as UTF-8. A malformed sentinel (bad
    base64, non-canonical base64, or bad UTF-8) yields `None` so a corrupt
    header never matches a body value by accident. `None` in → `None` out so
    callers can pass `headers.get(...)` directly.
    """
    if value is None:
        return None
    m = _B64_SENTINEL.fullmatch(value)
    if m is None:
        return value
    payload = m.group("payload")
    try:
        decoded = base64.b64decode(payload, validate=True)
    except binascii.Error:
        return None
    # Reject non-canonical base64 (e.g. non-zero trailing bits), which
    # `validate=True` tolerates; the encoder only ever emits canonical form.
    if base64.b64encode(decoded).decode("ascii") != payload:
        return None
    try:
        return decoded.decode("utf-8")
    except UnicodeDecodeError:
        return None

find_invalid_x_mcp_header

find_invalid_x_mcp_header(input_schema: Any) -> str | None

Return a reason string if any x-mcp-header annotation in input_schema is invalid; else None.

Walks every JSON Schema 2020-12 schema position. An annotation is valid only when it sits on a property statically reachable from the root via a chain of pure properties keys, names a non-empty RFC 9110 token, is on an integer/string/boolean property, and is case-insensitively unique across the whole schema. A None / non-mapping schema has no schema positions and returns None.

Source code in src/mcp/shared/inbound.py

def find_invalid_x_mcp_header(input_schema: Any) -> str | None:
    """Return a reason string if any `x-mcp-header` annotation in `input_schema` is invalid; else `None`.

    Walks every JSON Schema 2020-12 schema position. An annotation is valid
    only when it sits on a property statically reachable from the root via a
    chain of pure `properties` keys, names a non-empty RFC 9110 token, is on
    an integer/string/boolean property, and is case-insensitively unique
    across the whole schema. A `None` / non-mapping schema has no schema
    positions and returns `None`.
    """
    seen: dict[str, str] = {}
    for path, schema in _walk_schema_positions(input_schema):
        if X_MCP_HEADER_KEY not in schema:
            continue
        if not path:  # None (off the pure-properties chain) or () (the root itself)
            return f"{X_MCP_HEADER_KEY} found at a schema position not reachable via a pure `properties` chain"
        where = ".".join(path)
        header = schema[X_MCP_HEADER_KEY]
        # Wrong type and malformed value are distinct failures with distinct messages: the
        # non-str arm returns before any interpolation, because `repr` of an arbitrary
        # schema value is not total (a large `int` exceeds `sys.get_int_max_str_digits`).
        if not isinstance(header, str):
            return f"property {where!r}: {X_MCP_HEADER_KEY} must be a string, not {type(header).__name__}"
        if not _RFC9110_TOKEN.fullmatch(header):
            return f"property {where!r}: {X_MCP_HEADER_KEY} {header!r} is not an RFC 9110 token"
        prop_type = schema.get("type")
        if not isinstance(prop_type, str):
            return (
                f"property {where!r}: {X_MCP_HEADER_KEY} is only permitted on "
                f"integer/string/boolean properties (the type keyword is {type(prop_type).__name__}, not a string)"
            )
        if prop_type not in _X_MCP_HEADER_PRIMITIVE_TYPES:
            return (
                f"property {where!r}: {X_MCP_HEADER_KEY} is only permitted on "
                f"integer/string/boolean properties (got {prop_type!r})"
            )
        lower = header.lower()
        if lower in seen:
            return f"{X_MCP_HEADER_KEY} {header!r} on property {where!r} duplicates property {seen[lower]!r}"
        seen[lower] = where
    return None

MCP_PARAM_HEADER_PREFIX module-attribute

MCP_PARAM_HEADER_PREFIX: Final = 'Mcp-Param-'

Prefix the x-mcp-header token is joined to, forming the per-parameter HTTP header name.

x_mcp_header_map

x_mcp_header_map(
    input_schema: Any,
) -> dict[tuple[str, ...], str]

Map each property carrying a valid x-mcp-header to its annotation token, keyed by property path.

The key is the chain of properties keys from the schema root to the annotated property; a top-level property has a one-element path, a nested one a longer path. Call only on a schema that :func:find_invalid_x_mcp_header accepts; an invalid schema yields an undefined subset.

Source code in src/mcp/shared/inbound.py

def x_mcp_header_map(input_schema: Any) -> dict[tuple[str, ...], str]:
    """Map each property carrying a valid `x-mcp-header` to its annotation token, keyed by property path.

    The key is the chain of `properties` keys from the schema root to the
    annotated property; a top-level property has a one-element path, a nested
    one a longer path. Call only on a schema that
    :func:`find_invalid_x_mcp_header` accepts; an invalid schema yields an
    undefined subset.
    """
    return {path: token for path, token, _ in _annotated_positions(input_schema)}

mcp_param_headers

mcp_param_headers(
    header_map: Mapping[tuple[str, ...], str],
    arguments: Mapping[str, Any],
) -> dict[str, str]

Build the Mcp-Param-* headers a tools/call mirrors from its arguments.

For each (path, token) in header_map, read the value at that property path in arguments and, when it is present and not None, emit Mcp-Param-<token> carrying it: bool as true/false, other scalars via str, each passed through :func:encode_header_value so a non-token value is base64-wrapped. A path that hits a missing key or a non-mapping node is skipped, matching the spec's "omit the header when no value is present", as is a value with no header rendering.

Source code in src/mcp/shared/inbound.py

def mcp_param_headers(header_map: Mapping[tuple[str, ...], str], arguments: Mapping[str, Any]) -> dict[str, str]:
    """Build the `Mcp-Param-*` headers a `tools/call` mirrors from its arguments.

    For each `(path, token)` in `header_map`, read the value at that property
    path in `arguments` and, when it is present and not `None`, emit
    `Mcp-Param-<token>` carrying it: `bool` as `true`/`false`, other scalars via
    `str`, each passed through :func:`encode_header_value` so a non-token value
    is base64-wrapped. A path that hits a missing key or a non-mapping node is
    skipped, matching the spec's "omit the header when no value is present",
    as is a value with no header rendering.
    """
    headers: dict[str, str] = {}
    for path, token in header_map.items():
        value = _value_at_path(arguments, path)
        if value is None or (rendered := _render_header_scalar(value)) is None:
            continue
        headers[f"{MCP_PARAM_HEADER_PREFIX}{token}"] = encode_header_value(rendered)
    return headers

ERROR_CODE_HTTP_STATUS module-attribute

ERROR_CODE_HTTP_STATUS: Final[Mapping[int, int]] = (
    MappingProxyType(
        {
            PARSE_ERROR: 400,
            INVALID_REQUEST: 400,
            INVALID_PARAMS: 400,
            HEADER_MISMATCH: 400,
            MISSING_REQUIRED_CLIENT_CAPABILITY: 400,
            UNSUPPORTED_PROTOCOL_VERSION: 400,
            METHOD_NOT_FOUND: 404,
        }
    )
)

HTTP status to send for a JSON-RPC error.code.

Consulted for classifier-origin and handler-origin errors, so one table decides the wire status regardless of where the error was produced. Unmapped codes fall back to the caller's default (typically 200).

InboundModernRoute dataclass

A modern-protocol request whose envelope passed every ladder rung.

client_info and client_capabilities are the raw envelope values; the classifier checks presence only, not shape. Method existence is not a ladder rung — kernel dispatch is the single source of truth for that.

Source code in src/mcp/shared/inbound.py

@dataclass(frozen=True)
class InboundModernRoute:
    """A modern-protocol request whose envelope passed every ladder rung.

    `client_info` and `client_capabilities` are the raw envelope values;
    the classifier checks presence only, not shape. Method existence is not a
    ladder rung — kernel dispatch is the single source of truth for that.
    """

    protocol_version: str
    client_info: Any
    client_capabilities: Any

InboundLadderRejection dataclass

The first ladder rung that failed, as JSON-RPC error fields.

Source code in src/mcp/shared/inbound.py

@dataclass(frozen=True)
class InboundLadderRejection:
    """The first ladder rung that failed, as JSON-RPC error fields."""

    code: int
    message: str
    data: Any = None

find_duplicated_routing_header

find_duplicated_routing_header(
    headers: Iterable[tuple[str, str]],
) -> str | None

Name of a routing header supplied more than once in raw header lines, or None.

Takes raw (name, value) pairs — a folded mapping hides duplicates. A duplicate is rejected because first-copy and last-copy readers would disagree. Mcp-Param-* duplicates are :func:validate_mcp_param_headers's job.

Source code in src/mcp/shared/inbound.py

def find_duplicated_routing_header(headers: Iterable[tuple[str, str]]) -> str | None:
    """Name of a routing header supplied more than once in raw header lines, or `None`.

    Takes raw `(name, value)` pairs — a folded mapping hides duplicates. A
    duplicate is rejected because first-copy and last-copy readers would
    disagree. `Mcp-Param-*` duplicates are :func:`validate_mcp_param_headers`'s job.
    """
    seen: set[str] = set()
    for name, _ in headers:
        key = name.lower()
        if key in _ROUTING_HEADER_NAMES:
            if key in seen:
                return key
            seen.add(key)
    return None

classify_inbound_request

classify_inbound_request(
    body: Mapping[str, Any],
    *,
    headers: Mapping[str, str] | None = None,
    supported_modern_versions: Sequence[
        str
    ] = MODERN_PROTOCOL_VERSIONS
) -> InboundModernRoute | InboundLadderRejection

Run the modern-protocol validation ladder over a decoded JSON-RPC body.

Rungs, in order — first failure wins:

1. params._meta is a mapping carrying every reserved envelope key (protocol version, client info, client capabilities), and the protocol version is a string → else :data:~mcp_types.jsonrpc.INVALID_PARAMS.
2. When headers is given, MCP-Protocol-Version equals the envelope's protocol version, Mcp-Method equals body.method, and — for the methods in :data:NAME_BEARING_METHODS — Mcp-Name equals the named body param → else :data:~mcp_types.jsonrpc.HEADER_MISMATCH. Runs before the supported-version rung so a client that disagrees with itself is told so, rather than told the body's version is unsupported.
3. The envelope's protocol version is in supported_modern_versions → else :data:~mcp_types.jsonrpc.UNSUPPORTED_PROTOCOL_VERSION with data = {"supported": [...], "requested": <value>}.

Method existence is not a rung: kernel dispatch owns that decision so custom-registered methods route and the answer lives in one place.

Parameters:

Name	Type	Description	Default
body	Mapping[str, Any]	The decoded JSON-RPC request mapping. Envelope shape (jsonrpc / id) is not checked here.	required
headers	Mapping[str, str] | None	Transport headers keyed by lowercase name, or None to skip the header rung (non-HTTP callers).	None
supported_modern_versions	Sequence[str]	Modern protocol revisions this server accepts on the per-request-envelope path.	MODERN_PROTOCOL_VERSIONS

Source code in src/mcp/shared/inbound.py

def classify_inbound_request(
    body: Mapping[str, Any],
    *,
    headers: Mapping[str, str] | None = None,
    supported_modern_versions: Sequence[str] = MODERN_PROTOCOL_VERSIONS,
) -> InboundModernRoute | InboundLadderRejection:
    """Run the modern-protocol validation ladder over a decoded JSON-RPC body.

    Rungs, in order — first failure wins:

    1. `params._meta` is a mapping carrying every reserved envelope key
       (protocol version, client info, client capabilities), and the protocol
       version is a string → else :data:`~mcp_types.jsonrpc.INVALID_PARAMS`.
    2. When `headers` is given, `MCP-Protocol-Version` equals the envelope's
       protocol version, `Mcp-Method` equals `body.method`, and — for the
       methods in :data:`NAME_BEARING_METHODS` — `Mcp-Name` equals the named
       body param → else :data:`~mcp_types.jsonrpc.HEADER_MISMATCH`. Runs
       before the supported-version rung so a client that disagrees with itself
       is told so, rather than told the body's version is unsupported.
    3. The envelope's protocol version is in `supported_modern_versions` →
       else :data:`~mcp_types.jsonrpc.UNSUPPORTED_PROTOCOL_VERSION` with
       `data = {"supported": [...], "requested": <value>}`.

    Method existence is *not* a rung: kernel dispatch owns that decision so
    custom-registered methods route and the answer lives in one place.

    Args:
        body: The decoded JSON-RPC request mapping. Envelope shape
            (`jsonrpc` / `id`) is not checked here.
        headers: Transport headers keyed by lowercase name, or `None` to
            skip the header rung (non-HTTP callers).
        supported_modern_versions: Modern protocol revisions this server
            accepts on the per-request-envelope path.
    """
    try:
        meta = body["params"]["_meta"]
        protocol_version = meta[PROTOCOL_VERSION_META_KEY]
        client_info = meta[CLIENT_INFO_META_KEY]
        client_capabilities = meta[CLIENT_CAPABILITIES_META_KEY]
    except (KeyError, TypeError):
        return InboundLadderRejection(
            code=INVALID_PARAMS,
            message="params._meta must carry the reserved protocol-version, client-info and "
            "client-capabilities envelope keys",
        )
    if not isinstance(protocol_version, str):
        # A shape defect, not a version-negotiation outcome: -32022 is the one
        # code auto-negotiating clients do NOT fall back from, and the typed
        # rung-3 payload itself requires a string `requested`.
        return InboundLadderRejection(
            code=INVALID_PARAMS,
            message="the protocol-version envelope value must be a string",
        )

    if headers is not None:
        if headers.get(MCP_PROTOCOL_VERSION_HEADER) != protocol_version:
            return InboundLadderRejection(
                code=HEADER_MISMATCH,
                message=f"{MCP_PROTOCOL_VERSION_HEADER} header does not match the request envelope's protocol version",
            )
        method: Any = body.get("method")
        if headers.get(MCP_METHOD_HEADER) != method:
            return InboundLadderRejection(
                code=HEADER_MISMATCH,
                message=f"{MCP_METHOD_HEADER} header does not match the request body's method",
            )
        name_key = NAME_BEARING_METHODS.get(method)
        if name_key is not None:
            # Rung 1 already proved body["params"] is a mapping.
            body_value = body["params"].get(name_key)
            if body_value is not None and decode_header_value(headers.get(MCP_NAME_HEADER)) != body_value:
                return InboundLadderRejection(
                    code=HEADER_MISMATCH,
                    message=f"{MCP_NAME_HEADER} header does not match the request body's {name_key!r} parameter",
                )

    if protocol_version not in supported_modern_versions:
        return InboundLadderRejection(
            code=UNSUPPORTED_PROTOCOL_VERSION,
            message="Unsupported protocol version",
            data=UnsupportedProtocolVersionErrorData(
                supported=list(supported_modern_versions), requested=protocol_version
            ).model_dump(mode="json"),
        )

    return InboundModernRoute(
        protocol_version=protocol_version,
        client_info=client_info,
        client_capabilities=client_capabilities,
    )

validate_mcp_param_headers

validate_mcp_param_headers(
    input_schema: Any,
    arguments: Mapping[str, Any],
    headers: Mapping[str, str],
) -> InboundLadderRejection | None

Compare a tools/call request's Mcp-Param-* headers against its body arguments.

Each annotated property's header and argument must agree: present together and equal after sentinel decoding, or absent together (null counts as absent). Returns the first failure as a HEADER_MISMATCH rejection, else None.

A header whose argument is absent or unrenderable is deliberately rejected: the spec's purpose clause is exactly an intermediary routing on a value the body never carried. A duplicated recognized header is rejected — first-copy and last-copy readers would disagree. A schema :func:find_invalid_x_mcp_header rejects validates nothing: conforming clients drop the tool and emit no headers.

Source code in src/mcp/shared/inbound.py

def validate_mcp_param_headers(
    input_schema: Any,
    arguments: Mapping[str, Any],
    headers: Mapping[str, str],
) -> InboundLadderRejection | None:
    """Compare a `tools/call` request's `Mcp-Param-*` headers against its body arguments.

    Each annotated property's header and argument must agree: present together
    and equal after sentinel decoding, or absent together (`null` counts as
    absent). Returns the first failure as a `HEADER_MISMATCH` rejection, else `None`.

    A header whose argument is absent or unrenderable is deliberately rejected:
    the spec's purpose clause is exactly an intermediary routing on a value the
    body never carried. A duplicated recognized header is rejected — first-copy
    and last-copy readers would disagree. A schema :func:`find_invalid_x_mcp_header`
    rejects validates nothing: conforming clients drop the tool and emit no headers.
    """
    if find_invalid_x_mcp_header(input_schema) is not None:
        return None
    folded: dict[str, str] = {}
    duplicated: set[str] = set()
    for name, value in headers.items():
        key = name.lower()
        if key in folded:
            duplicated.add(key)
        folded[key] = value
    for path, token, schema in _annotated_positions(input_schema):
        header_name = f"{MCP_PARAM_HEADER_PREFIX}{token}"
        key = header_name.lower()
        raw = folded.get(key)
        value = _value_at_path(arguments, path)
        argument = ".".join(path)
        if raw is not None and key in duplicated:
            return InboundLadderRejection(
                code=HEADER_MISMATCH,
                message=f"{header_name} header appears more than once",
            )
        if value is None:
            if raw is not None:
                return InboundLadderRejection(
                    code=HEADER_MISMATCH,
                    message=f"{header_name} header is present but the request body's {argument!r} argument is absent",
                )
            continue
        rendered = _render_header_scalar(value)
        if rendered is None:
            # Unrenderable value: a conforming client omitted the header, so one claiming it can never match.
            if raw is not None:
                return InboundLadderRejection(
                    code=HEADER_MISMATCH,
                    message=f"{header_name} header does not match the request body's {argument!r} argument",
                )
            continue
        if raw is None:
            return InboundLadderRejection(
                code=HEADER_MISMATCH,
                message=f"{header_name} header is missing but the request body's {argument!r} argument is present",
            )
        decoded = decode_header_value(raw)
        if decoded is None:
            return InboundLadderRejection(
                code=HEADER_MISMATCH,
                message=f"{header_name} header carries a malformed base64 sentinel value",
            )
        if not _mcp_param_value_matches(schema.get("type"), value, rendered, decoded):
            return InboundLadderRejection(
                code=HEADER_MISMATCH,
                message=f"{header_name} header does not match the request body's {argument!r} argument",
            )
    return None