Function find_invalid_x_mcp_header

src/mcp/shared/inbound.py:176–206

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 no

(input_schema: Any)

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]
        if not isinstance(header, str) or 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) or 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


# INTERNAL_ERROR is deliberately unmapped (→ HTTP 200): the spec assigns no status to