Module `adcp.server.mcp_tools`

MCP server integration helpers.

Provides utilities for registering ADCP handlers with MCP servers.

Note

Function signatures in this module use ADCPHandler[Any] rather than a propagated TContext TypeVar. The rationale: these functions (get_tools_for_handler(), create_mcp_tools(), etc.) treat the handler opaquely — they walk the MRO and dispatch by tool name without ever touching the context argument's typed fields. Binding a TypeVar here would force callers to narrow at the call site for no runtime benefit, and cascade the TypeVar through every plumbing function in :mod:serve(). Any keeps the plumbing honest: the static type says "this code works with any ToolContext subclass," which is exactly true.

Global variables

var SchemaMode: Shape used for Pydantic-derived MCP input discovery schemas.

compact keeps every top-level request field, required nested fields, and a small discovery floor while pruning optional nested detail. It retains only reachable local references. defs keeps the complete Pydantic schema with local references, and inline preserves the historical fully expanded shape for clients that do not resolve references.

Functions

def create_mcp_tools(handler: ADCPHandler[Any], *, advertise_all: bool = False, validation: ValidationHookConfig | None = None, pre_validation_hooks: PreValidationHooks | None = None, response_enhancer: ResponseEnhancer | None = None, adcp_version: str | None = None, schema_mode: SchemaMode = 'compact') ‑> MCPToolSet

Expand source code

def create_mcp_tools(
    handler: ADCPHandler[Any],
    *,
    advertise_all: bool = False,
    validation: ValidationHookConfig | None = None,
    pre_validation_hooks: PreValidationHooks | None = None,
    response_enhancer: ResponseEnhancer | None = None,
    adcp_version: str | None = None,
    schema_mode: SchemaMode = "compact",
) -> MCPToolSet:
    """Create MCP tools from an ADCP handler.

    This is the main entry point for MCP server integration.

    Example with mcp library:
        from mcp.server import Server
        from adcp.server import ContentStandardsHandler, create_mcp_tools

        class MyHandler(ContentStandardsHandler):
            # ... implement methods

        handler = MyHandler()
        tools = create_mcp_tools(handler)

        server = Server("my-content-agent")

        @server.list_tools()
        async def list_tools():
            return tools.tool_definitions

        @server.call_tool()
        async def call_tool(name: str, arguments: dict):
            return await tools.call_tool(name, arguments)

    Args:
        handler: ADCP handler instance.
        advertise_all: When True, advertise every tool the handler type
            supports — even those whose method is still the SDK's
            ``not_supported`` default. See :func:`get_tools_for_handler`.
        validation: Opt-in schema validation config. When supplied,
            every tool caller validates requests and responses against
            the bundled AdCP JSON schemas. See
            :func:`create_tool_caller` for mode semantics.
        pre_validation_hooks: Optional dict mapping tool name to a
            ``(tool_name, args) -> args`` callable or ordered sequence.
            Applied before schema + Pydantic validation. See
            :func:`create_tool_caller`.
        response_enhancer: Optional server-wide :data:`ResponseEnhancer`
            applied to every successful response. See
            :func:`create_tool_caller`.
        adcp_version: Trusted server protocol pin for version-scoped
            ``tools/list`` schemas. Decorator-built handlers carry this pin
            automatically; class-based handlers can pass it here.
        schema_mode: Pydantic-derived input discovery shape. ``"compact"``
            is bounded and complete at the request root; use ``"defs"`` for
            the full referenced schema or ``"inline"`` for legacy clients
            that do not resolve local JSON Schema references.

    Returns:
        MCPToolSet with tool definitions and handlers.
    """
    return MCPToolSet(
        handler,
        advertise_all=advertise_all,
        validation=validation,
        pre_validation_hooks=pre_validation_hooks,
        response_enhancer=response_enhancer,
        adcp_version=adcp_version,
        schema_mode=schema_mode,
    )

Create MCP tools from an ADCP handler.

This is the main entry point for MCP server integration.

Example with mcp library: from mcp.server import Server from adcp.server import ContentStandardsHandler, create_mcp_tools

class MyHandler(ContentStandardsHandler):
    # ... implement methods

handler = MyHandler()
tools = create_mcp_tools(handler)

server = Server("my-content-agent")

@server.list_tools()
async def list_tools():
    return tools.tool_definitions

@server.call_tool()
async def call_tool(name: str, arguments: dict):
    return await tools.call_tool(name, arguments)

Args
-----=
handler: ADCP handler instance.
advertise_all: When True, advertise every tool the handler type supports — even those whose method is still the SDK's not_supported default. See :func:get_tools_for_handler().
validation: Opt-in schema validation config. When supplied, every tool caller validates requests and responses against the bundled AdCP JSON schemas. See :func:create_tool_caller() for mode semantics.
pre_validation_hooks: Optional dict mapping tool name to a (tool_name, args) -> args callable or ordered sequence. Applied before schema + Pydantic validation. See :func:create_tool_caller().
response_enhancer: Optional server-wide :data:ResponseEnhancer applied to every successful response. See :func:create_tool_caller().
adcp_version: Trusted server protocol pin for version-scoped tools/list schemas. Decorator-built handlers carry this pin automatically; class-based handlers can pass it here.
schema_mode: Pydantic-derived input discovery shape. "compact" is bounded and complete at the request root; use "defs" for the full referenced schema or "inline" for legacy clients that do not resolve local JSON Schema references.

Returns -----= MCPToolSet with tool definitions and handlers.

def create_tool_caller(handler: ADCPHandler[Any], method_name: str, *, validation: ValidationHookConfig | None = None, pre_validation_hook: PreValidationHookChain | None = None, default_unnegotiated_adcp_version: str | None = '3.0', response_enhancer: ResponseEnhancer | None = None) ‑> Callable[..., typing.Any]

Expand source code

def create_tool_caller(
    handler: ADCPHandler[Any],
    method_name: str,
    *,
    validation: ValidationHookConfig | None = None,
    pre_validation_hook: PreValidationHookChain | None = None,
    default_unnegotiated_adcp_version: str | None = DEFAULT_UNNEGOTIATED_ADCP_VERSION,
    response_enhancer: ResponseEnhancer | None = None,
) -> Callable[..., Any]:
    """Create a tool caller function for an ADCP handler method.

    Automatically injects context passthrough: if the request contains a
    ``context`` field, it is echoed back in the response (ADCP requirement).
    Handlers no longer need to call ``inject_context()`` manually.

    **Typed params (closes #214).** When the handler method declares its
    ``params`` parameter as a Pydantic model (e.g.
    ``params: GetProductsRequest``), the dispatcher deserialises the raw
    dict into the model before calling the handler — giving authors
    IDE autocomplete, Pydantic validation at the boundary, and typed
    attribute access instead of ``params.get(...)``. Handlers still
    declaring ``params: dict[str, Any]`` keep working unchanged. A
    Pydantic ``ValidationError`` surfaces as a structured
    ``INVALID_REQUEST`` AdCP error so callers see a spec-typed recovery
    classification rather than a raw stack trace.

    **Schema-driven validation (issues #249, #1102).** Public server
    factories supply strict validation by default, so the dispatcher
    validates incoming requests and outgoing responses against the bundled
    canonical AdCP JSON schemas. Request failures
    raise ``ADCPTaskError(VALIDATION_ERROR)`` before the handler runs,
    so malformed payloads never hit business logic. Response failures
    either raise ``VALIDATION_ERROR`` (strict) or log a warning
    (warn). Adopters can explicitly relax this with
    ``ValidationHookConfig`` or disable it with ``validation=None``. The
    low-level ``create_tool_caller`` seam itself retains a ``None`` default;
    ``serve`` and the MCP/A2A server factories pass the strict default.

    **Pre-validation hooks (issues #614, #859).** When
    ``pre_validation_hook`` is supplied, each hook is called with
    ``(tool_name, shallow_copy_of_args)`` and must return a ``dict`` that
    replaces the wire args before schema validation and Pydantic
    ``model_validate`` run. Pass either one hook or an ordered sequence of
    hooks; sequences run left-to-right. The framework passes a shallow
    copy of the incoming params dict to each hook, so a hook may mutate
    its argument freely or return a brand-new dict — either style is safe.
    The original wire params are captured before the copy is made, so
    context echo always reflects what the buyer sent. Use this to apply
    spec-mandated defaults for pre-v3 buyers that omit required fields
    (e.g. ``buying_mode``, ``format_id`` shape coercion, ``asset_type``
    inference). The hook runs on every call; keep it fast.
    Deliberate ``AdcpError`` / ``ADCPTaskError`` rejections preserve their
    structured fields. Other hook exceptions surface as ``INVALID_REQUEST``
    naming the failing hook.

    **Unknown-field policy (issue #858).** When
    ``validation=ValidationHookConfig(unknown_fields=...)`` is supplied,
    unsupported top-level tool arguments are handled after pre-validation
    hooks and legacy adapters, but before request schema validation and
    Pydantic coercion. ``"reject"`` raises ``INVALID_REQUEST``,
    ``"strip"`` removes unsupported keys, and ``"ignore"`` preserves the
    current permissive behavior.

    .. note::
        For the specific case of buyers omitting ``account``, see issue
        #623 ("Typed dispatcher rejects valid request when ``account`` is
        omitted") — that will be the canonical spec-level fix for that
        field. Once #623 lands you can drop any ``account`` placeholder
        hook entry.

    Args:
        handler: The ADCP handler instance
        method_name: Name of the method to call
        validation: Optional :class:`ValidationHookConfig` with
            per-side modes (``strict`` / ``warn`` / ``off``). Omitting
            it disables server-side schema validation entirely.
        pre_validation_hook: Optional callable or ordered sequence of
            callables ``(tool_name, args) -> args`` invoked on the raw wire
            dict before schema + Pydantic validation. See the
            **Pre-validation hooks** section above.
        default_unnegotiated_adcp_version: Release-precision version to use
            when the buyer supplies no version envelope. MCP uses ``"3.0"``
            for legacy compatibility. A2A passes ``None`` so omitted version
            means the current SDK wire shape.
        response_enhancer: Optional server-wide :data:`ResponseEnhancer`
            applied to every successful response after context echo and
            before schema validation. See :data:`ResponseEnhancer` for the
            two supported arities and the failure/idempotency semantics.

    Returns:
        Async callable ``call_tool(params, context=None)``. The ``context``
        parameter is optional — transports that can extract caller identity
        from their auth layer (A2A's ``ServerCallContext.user``, custom
        FastMCP auth middleware, etc.) should pass a populated
        :class:`ToolContext` so the server middleware layer (idempotency
        per-principal scoping, audit logging) gets the real principal. When
        no context is supplied, a bare :class:`ToolContext` is used.
    """
    from pydantic import ValidationError

    from adcp._version import normalize_to_release_precision
    from adcp.canonical_formats import (
        CanonicalFormatLegacyResolutionError,
        CreativeDialect,
        CreativeDialectError,
        LegacyCreativeProjectionError,
        normalize_legacy_creative_request,
        project_canonical_response_to_legacy,
        resolve_creative_dialect,
    )
    from adcp.compat.legacy import LEGACY_ADAPTER_VERSIONS, get_legacy_adapter
    from adcp.compat.legacy.errors import legacy_adapter_task_error
    from adcp.exceptions import ADCPTaskError
    from adcp.server.helpers import inject_context
    from adcp.types import Error
    from adcp.validation.envelope import UnsupportedVersionError, detect_wire_version
    from adcp.validation.schema_errors import build_adcp_validation_error_payload
    from adcp.validation.schema_validator import (
        format_issues,
        validate_request,
        validate_response,
    )

    adopter_method_name = {
        "build_creative": "build_creative_legacy",
        "list_creative_formats": "list_creative_formats_legacy",
        "preview_creative": "preview_creative_legacy",
    }.get(method_name, method_name)
    method = getattr(handler, adopter_method_name)
    params_model = _resolve_params_pydantic_model(method)
    frozen_reporting_feed = (
        method_name == "get_reporting_status"
        and getattr(handler, "reporting_feed_store", None) is not None
    )
    pinned_reporting_status = frozen_reporting_feed or (
        method_name == "get_reporting_status"
        and getattr(handler, "reporting_status_handler", None) is not None
    )

    # Opt-in server-side schema modes. ``None`` keeps validation off
    # entirely (zero overhead on the hot path) — the TS-port default for
    # ``createAdcpServer`` is the same: validation is an explicit opt-in.
    request_mode = validation.requests if validation is not None else None
    response_mode = validation.responses if validation is not None else None
    unknown_field_policy = _normalize_unknown_field_policy(
        validation.unknown_fields if validation is not None else None
    )
    pre_validation_hooks = _flatten_pre_validation_hooks(pre_validation_hook)

    async def call_tool(params: dict[str, Any], context: ToolContext | None = None) -> Any:
        ctx = context if context is not None else ToolContext()

        raw_params = params  # Preserve original wire params for context echo.

        if frozen_reporting_feed and params.get("view") == "periods":
            from adcp.reporting.feed.errors import ReportingFeedError
            from adcp.reporting.feed.request import FeedRequest

            try:
                FeedRequest.parse(params)
            except ReportingFeedError as exc:
                raise ADCPTaskError(
                    operation=method_name, errors=[Error(code=exc.code, message=str(exc))]
                ) from None

        if method_name == "sync_reporting_receipts":
            from adcp.reporting.receipts.errors import ReportingReceiptError
            from adcp.reporting.receipts.wire import ReceiptBatch

            # Financial batches require the raw shape even with generic
            # validation disabled. Coercion must not erase supplied empty
            # arrays, read-only fields, timestamps, or request extensions.
            try:
                receipt_batch = ReceiptBatch.parse(raw_params)
                raw_params = receipt_batch.request
            except ReportingReceiptError as exc:
                raise ADCPTaskError(
                    operation=method_name, errors=[Error(code=exc.code, message=str(exc))]
                ) from None

        if pre_validation_hooks:
            try:
                params = _apply_pre_validation_hooks(
                    pre_validation_hooks,
                    method_name,
                    (
                        copy.deepcopy(params)
                        if method_name == "sync_reporting_receipts"
                        else dict(params)
                    ),
                )
                if method_name == "sync_reporting_receipts":
                    # A rewrite would change the identity the buyer authorized.
                    # Hooks may inspect this task; financial mutations fail closed.
                    try:
                        if ReceiptBatch.parse(params) != receipt_batch:
                            raise ReportingReceiptError("INVALID_REQUEST")
                    except ReportingReceiptError:
                        raise ADCPTaskError(
                            operation=method_name,
                            errors=[
                                Error(
                                    code="INVALID_REQUEST",
                                    message="receipt requests cannot be rewritten",
                                )
                            ],
                        ) from None
            except PreValidationHookError as exc:
                raise ADCPTaskError(
                    operation=method_name,
                    errors=[
                        Error(
                            code="INVALID_REQUEST",
                            message=str(exc),
                        )
                    ],
                ) from exc

        # Wire-version detection: read ``adcp_version`` / ``adcp_major_version``
        # off the post-hook params (legacy buyers may rely on a hook to
        # populate the envelope, so this runs after pre_validation_hook).
        # ``None`` initially means the buyer didn't claim a version.
        # After legacy shape probes run, native unnegotiated traffic is
        # pinned to 3.0 compatibility because those buyers predate the
        # release-precision ``adcp_version`` field and the 3.1 status split.
        # An explicit unsupported claim always fails before validation or
        # handler dispatch. Only an omitted envelope may continue into the
        # legacy shape probes below.
        try:
            wire_version = detect_wire_version(params)
        except UnsupportedVersionError as exc:
            raise ADCPTaskError(
                operation=method_name,
                errors=[
                    Error(
                        code="VERSION_UNSUPPORTED",
                        message=str(exc),
                        # Preserve the wire field's original type so buyer
                        # telemetry sees the same shape it sent (int for
                        # ``adcp_major_version``, str for ``adcp_version``).
                        details={
                            "claimed_version": exc.wire_value,
                            "supported_versions": list(exc.supported),
                        },
                    )
                ],
            ) from exc

        # Shape-based legacy detection (issue: real v2.5 buyers can't
        # send ``adcp_version`` — the field didn't exist in the v2.5
        # schema). When the envelope is empty and a legacy adapter
        # registers an ``is_legacy_shape`` probe, run it. A match
        # promotes ``wire_version`` to the probe's version so the
        # adapter path below fires normally. Bias is conservative:
        # probes return ``True`` only on fields v3 doesn't emit
        # (``brand_manifest``, ``creative_ids``, bare-string
        # ``format_id``). False positives downgrade a real v3 buyer
        # to legacy validation, which is the worst outcome.
        if wire_version is None:
            for candidate in LEGACY_ADAPTER_VERSIONS:
                candidate_adapter = get_legacy_adapter(candidate, method_name)
                if candidate_adapter is None:
                    continue
                probe = candidate_adapter.is_legacy_shape
                if probe is None:
                    continue
                try:
                    matched = probe(params) if isinstance(params, dict) else False
                except Exception:  # noqa: BLE001 — defensive: probes are pure-ish
                    matched = False
                if matched:
                    logger.info(
                        "Detected %s wire shape for %s (no envelope version "
                        "supplied); routing through legacy adapter.",
                        candidate,
                        method_name,
                    )
                    wire_version = candidate
                    break

        # A major-only envelope (``adcp_major_version: 3``) does not select
        # the 3.0 release. Discovery must still advertise the current 3.x
        # native surface; only release-precision ``adcp_version`` can suppress
        # a release-scoped feature.
        wire_release_was_negotiated = isinstance(params, dict) and isinstance(
            params.get("adcp_version"), str
        )
        if wire_version is None:
            wire_version = default_unnegotiated_adcp_version

        claimed_release = params.get("adcp_version")
        bridged_response_version: str | None = None
        if isinstance(claimed_release, str) and claimed_release:
            normalized_claim = normalize_to_release_precision(claimed_release)
            if normalized_claim != wire_version:
                bridged_response_version = normalized_claim

        ctx.resolved_adcp_version = wire_version

        if pinned_reporting_status:
            from adcp._version import is_adcp_version_at_least, resolve_adcp_version

            mounted_version = default_unnegotiated_adcp_version or resolve_adcp_version(None)
            requested_version = wire_version or resolve_adcp_version(None)
            if is_adcp_version_at_least(mounted_version, "3.2-rc.6") != is_adcp_version_at_least(
                requested_version, "3.2-rc.6"
            ):
                # Frozen rc.3 complete periods may carry next_expected_at;
                # rc.6 forbids it. Do not return preserved old bytes under an
                # incompatible advertised schema, even with validation off.
                raise ADCPTaskError(
                    operation=method_name,
                    errors=[
                        Error(
                            code="VERSION_UNSUPPORTED",
                            message=(
                                "Pin the reporting mount and client to the same AdCP version; "
                                "continue an existing walk on its original version, or start "
                                "a new walk without a cursor or checkpoint after changing version."
                            ),
                            details={"mounted_version": mounted_version},
                        )
                    ],
                )

        # Legacy-version routing: if the buyer claims (or shape-detected)
        # a version handled via the adapter path (e.g. ``"2.5"``),
        # validate the params against the legacy schema first, *then*
        # translate to the current shape. Pre-adapter validation
        # surfaces structural errors with the legacy schema's field
        # paths — far easier for the buyer to act on than a v3
        # field-path error after a confusing translation. Post-adapter
        # validation (further down) catches translator bugs against
        # the SDK pin.
        legacy_adapter: Any = None
        if wire_version in LEGACY_ADAPTER_VERSIONS:
            legacy_adapter = get_legacy_adapter(wire_version, method_name)
            if legacy_adapter is None:
                raise ADCPTaskError(
                    operation=method_name,
                    errors=[
                        Error(
                            code="INVALID_REQUEST",
                            message=(
                                f"Tool {method_name!r} is not available on "
                                f"AdCP {wire_version}; upgrade to a "
                                f"supported version or call a tool exposed "
                                f"on this legacy surface."
                            ),
                            details={"legacy_version": wire_version},
                        )
                    ],
                )

            # Pre-adapter validation against the legacy schema.
            # Only runs when validation is enabled at all
            # (``request_mode != off`` AND a config is supplied) — keeps
            # the zero-overhead path for adopters who haven't opted in.
            # ``strict`` rejects; ``warn`` logs and proceeds so the
            # adapter still gets to translate (matching the existing
            # post-adapter contract).
            if request_mode is not None and request_mode != "off":
                pre_outcome = validate_request(method_name, params, version=wire_version)
                if not pre_outcome.valid:
                    summary = format_issues(pre_outcome.issues)
                    if request_mode == "strict":
                        payload = build_adcp_validation_error_payload(
                            method_name, "request", pre_outcome.issues
                        )
                        # Annotate with the wire version so adopter
                        # telemetry knows which schema rejected.
                        payload_details = dict(payload.get("details") or {})
                        payload_details["claimed_version"] = wire_version
                        payload["details"] = payload_details
                        raise ADCPTaskError(
                            operation=method_name,
                            errors=[Error(**payload)],
                        )
                    logger.warning(
                        "Schema validation warning (pre-adapter %s) for %s: %s",
                        wire_version,
                        method_name,
                        summary,
                    )

            try:
                params = legacy_adapter.adapt_request(params)
            except Exception as exc:
                raise legacy_adapter_task_error(
                    operation=method_name,
                    wire_version=wire_version,
                    phase="request",
                    exception=exc,
                ) from exc
            # Adapter output is validated against the SDK pin
            # (catches translator bugs with v3 field paths). The
            # ``post_adapter_validator_version`` name documents
            # which side of the adapter this value plays.
            post_adapter_validator_version: str | None = None
        else:
            post_adapter_validator_version = wire_version
        response_validator_version = None if legacy_adapter is not None else wire_version

        creative_boundary_tools = {
            "create_media_buy",
            "get_creative_delivery",
            "get_media_buy_delivery",
            "get_media_buys",
            "get_products",
            "list_creatives",
            "sync_creatives",
            "update_media_buy",
        }
        legacy_projection_sources: list[Any] = []
        if method_name in creative_boundary_tools and isinstance(params, dict) and wire_version:
            try:
                dialect = (
                    CreativeDialect.LEGACY
                    if not str(wire_version).startswith("3.")
                    else resolve_creative_dialect(
                        wire_version,
                        capabilities=_creative_capabilities_for_handler(handler),
                        request=params,
                    )
                )
                if dialect is CreativeDialect.LEGACY:
                    params = normalize_legacy_creative_request(
                        params,
                        legacy_format_converter=getattr(handler, "legacy_format_converter", None),
                        projection_sources=legacy_projection_sources,
                    )
                    # The normalized object is canonical handler input, not a
                    # valid instance of the caller's legacy wire schema.
                    post_adapter_validator_version = None
            except (CreativeDialectError, LegacyCreativeProjectionError) as exc:
                raise ADCPTaskError(
                    operation=method_name,
                    errors=[
                        Error(
                            code="INVALID_REQUEST",
                            message=str(exc),
                            suggestion=(
                                "Publish a canonical format declaration or configure the "
                                "server's legacy_format_converter"
                            ),
                        )
                    ],
                ) from exc

        if (
            isinstance(params, dict)
            and method_name != "sync_reporting_receipts"
            and not frozen_reporting_feed
        ):
            params = _apply_unknown_field_policy(
                method_name,
                params,
                policy=unknown_field_policy,
                version=post_adapter_validator_version,
                params_model=params_model,
            )

        if request_mode is not None and request_mode != "off":
            outcome = validate_request(method_name, params, version=post_adapter_validator_version)
            if not outcome.valid:
                summary = format_issues(outcome.issues)
                if request_mode == "strict":
                    payload = build_adcp_validation_error_payload(
                        method_name, "request", outcome.issues
                    )
                    raise ADCPTaskError(
                        operation=method_name,
                        errors=[Error(**payload)],
                    )
                logger.warning(
                    "Schema validation warning (request) for %s: %s",
                    method_name,
                    summary,
                )

        call_params: Any = params
        if method_name == "sync_reporting_receipts":
            call_params = raw_params
        elif frozen_reporting_feed:
            call_params = params
        elif params_model is not None and isinstance(params, dict):
            try:
                call_params = params_model.model_validate(params)
            except ValidationError as exc:
                # Surface as a structured AdCP error so MCP clients see
                # INVALID_REQUEST with a field-level pointer instead of
                # a raw Pydantic traceback. translate_error maps this
                # to ToolError (MCP) / ServerError (A2A) per transport.
                #
                # Strip ``input``/``ctx``/``url`` from the Pydantic error
                # details — they echo the raw offending value verbatim
                # (``input`` in particular). In multi-hop agent chains the
                # response flows through intermediaries, so echoing the
                # user-supplied value is a PII/secret-leak vector: a
                # mistyped API key or secret-shaped idempotency_key could
                # land in the broker's logs. The field path in
                # ``Error.field`` is all clients need to programmatically
                # locate the bad value in their own request.
                errors_list = exc.errors(
                    include_input=False, include_context=False, include_url=False
                )
                # Narrow discriminated-union failures to the variant
                # the user actually intended (Stability AI Emma P2:
                # 60-line dump → focused error). For non-union
                # failures the function is a no-op.
                #
                # Defensive: if the narrowing helper itself raises
                # (heuristic edge case, future pydantic format
                # change), keep the original error list rather than
                # 500'ing the wire path. The narrowed-error UX is a
                # nice-to-have; correctness is surfacing SOME error.
                try:
                    errors_list = list(narrow_union_errors(errors_list))
                except Exception:
                    logger.warning(
                        "narrow_union_errors raised on %s — passing through "
                        "unfiltered errors. This is a bug in the narrowing "
                        "heuristic, NOT in the validation itself.",
                        method_name,
                        exc_info=True,
                    )
                first: dict[str, Any] = dict(errors_list[0]) if errors_list else {}
                field_path = ".".join(str(loc) for loc in first.get("loc", ()))
                message = first.get("msg", "validation failed")
                suggestion = (
                    f"Invalid value for field {field_path!r}: {message}"
                    if field_path
                    else f"Request validation failed: {message}"
                )
                raise ADCPTaskError(
                    operation=method_name,
                    errors=[
                        Error(
                            code="INVALID_REQUEST",
                            field=field_path or None,
                            message=suggestion,
                            details={"validation_errors": errors_list},
                        )
                    ],
                ) from exc
        result = await method(call_params, ctx)
        if method_name == "get_adcp_capabilities":
            if isinstance(result, dict):
                from adcp.canonical_formats.dialect import canonical_creatives_capability
                from adcp.server.responses import _apply_canonical_creatives_capability

                # Capture only the adopter's raw, version-independent feature
                # declaration. Do not retain the negotiated response: 3.0
                # intentionally removes this field and must not poison another
                # caller's 3.1 routing decision.
                declared_canonical = canonical_creatives_capability(result)
                if declared_canonical is None:
                    supported_protocols = result.get("supported_protocols")
                    if isinstance(supported_protocols, list) and "media_buy" in supported_protocols:
                        # The canonical framework default is the declaration
                        # even when a negotiated 3.0 response must suppress the
                        # 3.1-only feature field on the wire.
                        declared_canonical = True
                if declared_canonical is not None:
                    setattr(
                        handler,
                        "_declared_creative_capabilities",
                        {
                            "media_buy": {
                                "features": {
                                    "canonical_creatives": declared_canonical,
                                }
                            }
                        },
                    )
                _apply_canonical_creatives_capability(
                    result,
                    # Discovery without a release envelope advertises the
                    # server's current native surface. Only an explicit
                    # negotiated 3.0 request suppresses the 3.1 feature.
                    adcp_version=(wire_version if wire_release_was_negotiated else None),
                )
        if (
            method_name in creative_boundary_tools
            and wire_version
            and (
                not str(wire_version).startswith("3.")
                or resolve_creative_dialect(
                    wire_version,
                    capabilities=_creative_capabilities_for_handler(handler),
                    request=raw_params,
                )
                is CreativeDialect.LEGACY
            )
        ):
            try:
                result = project_canonical_response_to_legacy(
                    result,
                    resolver=getattr(handler, "canonical_format_legacy_resolver", None),
                    sources=legacy_projection_sources,
                )
            except CanonicalFormatLegacyResolutionError as exc:
                raise ADCPTaskError(
                    operation=method_name,
                    errors=[
                        Error(
                            code="INTERNAL_ERROR",
                            message=str(exc),
                            suggestion=(
                                "Configure canonical_format_legacy_resolver or return a "
                                "same-process projection retaining its original tuple"
                            ),
                        )
                    ],
                ) from exc
        # Convert Pydantic models to JSON-safe dicts for MCP serialization
        if hasattr(result, "model_dump"):
            result = result.model_dump(mode="json", exclude_none=True)
        # ADCP requires echoing context from request to response — read
        # from the raw dict the transport sent, not from the validated
        # model (which won't carry the wire ``context`` field).
        if isinstance(result, dict):
            if bridged_response_version is not None:
                # Old rc.7 clients have no stable validator. Retain the wire
                # alias for their response lookup while routing and validating
                # internally against the compatible stable contract.
                result["adcp_version"] = bridged_response_version
            _normalize_response_envelope(
                method_name,
                result,
                raw_params,
                adcp_version=response_validator_version,
            )
            inject_context(raw_params, result)
            # Run the seller's response enhancer AFTER ``inject_context``
            # (so it sees the credential-stripped echo envelope and can't
            # re-introduce a credential) and BEFORE ``validate_response``
            # (so any conformance-breaking mutation surfaces as a
            # VALIDATION_ERROR rather than shipping malformed). This single
            # site covers framework tools, custom tools
            # (``get_task_status`` / ``list_tasks``), and
            # ``get_adcp_capabilities`` on both MCP and A2A. The L3 error
            # envelope is enhanced on the dedicated error paths
            # (``build_mcp_error_result`` / ``_send_adcp_error``), so skip
            # it here to avoid a double pass.
            if "adcp_error" not in result and method_name != "sync_reporting_receipts":
                _apply_response_enhancer(response_enhancer, method_name, result, ctx)

        if response_mode is not None and response_mode != "off" and isinstance(result, dict):
            # Skip validation when the handler returned the AdCP L3
            # error envelope (``{"adcp_error": {...}}``). That envelope
            # has its own shape enforced by the ``Error`` builder; the
            # per-tool response schema would false-positive on it and
            # convert a real protocol error into a fake VALIDATION_ERROR.
            if "adcp_error" not in result:
                outcome = validate_response(method_name, result, version=response_validator_version)
                if not outcome.valid:
                    summary = format_issues(outcome.issues)
                    logger.warning(
                        "Schema validation warning (response) for %s: %s",
                        method_name,
                        summary,
                    )
                    if response_mode == "strict":
                        payload = build_adcp_validation_error_payload(
                            method_name, "response", outcome.issues
                        )
                        raise ADCPTaskError(
                            operation=method_name,
                            errors=[Error(**payload)],
                        )

        # Legacy adapter response rewrite: when the buyer is on a legacy
        # wire shape and the adapter declares a ``normalize_response``
        # callable, translate the current-shape response back so the
        # buyer sees the dict shape they expected. Runs *after*
        # validation (which validated the current shape) so a malformed
        # legacy rewrite doesn't mask handler bugs.
        if legacy_adapter is not None and legacy_adapter.normalize_response is not None:
            if isinstance(result, dict) and "adcp_error" not in result:
                try:
                    result = legacy_adapter.normalize_response(result)
                except Exception as exc:
                    raise legacy_adapter_task_error(
                        operation=method_name,
                        wire_version=wire_version,
                        phase="response",
                        exception=exc,
                    ) from exc
        if (
            legacy_adapter is not None
            and isinstance(result, dict)
            and result.get("status") == "completed"
        ):
            result.pop("status")
        return result

    if frozen_reporting_feed:
        setattr(call_tool, "_adcp_frozen_reporting_feed", True)
    return call_tool

Create a tool caller function for an ADCP handler method.

Automatically injects context passthrough: if the request contains a context field, it is echoed back in the response (ADCP requirement). Handlers no longer need to call inject_context() manually.

Typed params (closes #214). When the handler method declares its params parameter as a Pydantic model (e.g. params: GetProductsRequest), the dispatcher deserialises the raw dict into the model before calling the handler — giving authors IDE autocomplete, Pydantic validation at the boundary, and typed attribute access instead of params.get(…). Handlers still declaring params: dict[str, Any] keep working unchanged. A Pydantic ValidationError surfaces as a structured INVALID_REQUEST AdCP error so callers see a spec-typed recovery classification rather than a raw stack trace.

Schema-driven validation (issues #249, #1102). Public server factories supply strict validation by default, so the dispatcher validates incoming requests and outgoing responses against the bundled canonical AdCP JSON schemas. Request failures raise ADCPTaskError(VALIDATION_ERROR) before the handler runs, so malformed payloads never hit business logic. Response failures either raise VALIDATION_ERROR (strict) or log a warning (warn). Adopters can explicitly relax this with ValidationHookConfig or disable it with validation=None. The low-level create_tool_caller() seam itself retains a None default; serve and the MCP/A2A server factories pass the strict default.

Pre-validation hooks (issues #614, #859). When pre_validation_hook is supplied, each hook is called with (tool_name, shallow_copy_of_args) and must return a dict that replaces the wire args before schema validation and Pydantic model_validate run. Pass either one hook or an ordered sequence of hooks; sequences run left-to-right. The framework passes a shallow copy of the incoming params dict to each hook, so a hook may mutate its argument freely or return a brand-new dict — either style is safe. The original wire params are captured before the copy is made, so context echo always reflects what the buyer sent. Use this to apply spec-mandated defaults for pre-v3 buyers that omit required fields (e.g. buying_mode, format_id shape coercion, asset_type inference). The hook runs on every call; keep it fast. Deliberate AdcpError / ADCPTaskError rejections preserve their structured fields. Other hook exceptions surface as INVALID_REQUEST naming the failing hook.

Unknown-field policy (issue #858). When validation=ValidationHookConfig(unknown_fields=...) is supplied, unsupported top-level tool arguments are handled after pre-validation hooks and legacy adapters, but before request schema validation and Pydantic coercion. "reject" raises INVALID_REQUEST, "strip" removes unsupported keys, and "ignore" preserves the current permissive behavior.

Note

For the specific case of buyers omitting account, see issue

623 ("Typed dispatcher rejects valid request when `account` is

omitted") — that will be the canonical spec-level fix for that field. Once #623 lands you can drop any account placeholder hook entry.

Args
-----=
handler: The ADCP handler instance
method_name: Name of the method to call
validation: Optional :class:ValidationHookConfig with per-side modes (strict / warn / off). Omitting it disables server-side schema validation entirely.
pre_validation_hook: Optional callable or ordered sequence of callables (tool_name, args) -> args invoked on the raw wire dict before schema + Pydantic validation. See the Pre-validation hooks section above.
default_unnegotiated_adcp_version: Release-precision version to use when the buyer supplies no version envelope. MCP uses "3.0" for legacy compatibility. A2A passes None so omitted version means the current SDK wire shape.
response_enhancer: Optional server-wide :data:ResponseEnhancer applied to every successful response after context echo and before schema validation. See :data:ResponseEnhancer for the two supported arities and the failure/idempotency semantics.

Returns -----= Async callable call_tool(params, context=None). The context parameter is optional — transports that can extract caller identity from their auth layer (A2A's ServerCallContext.user, custom FastMCP auth middleware, etc.) should pass a populated :class:ToolContext so the server middleware layer (idempotency per-principal scoping, audit logging) gets the real principal. When no context is supplied, a bare :class:ToolContext is used.

def get_tools_for_handler(handler: ADCPHandler[Any] | type[ADCPHandler[Any]], *, advertise_all: bool = False, adcp_version: str | None = None, schema_mode: SchemaMode = 'compact') ‑> list[dict[str, typing.Any]]

Expand source code

def get_tools_for_handler(
    handler: ADCPHandler[Any] | type[ADCPHandler[Any]],
    *,
    advertise_all: bool = False,
    adcp_version: str | None = None,
    schema_mode: SchemaMode = "compact",
    _include_schemas: bool = True,
) -> list[dict[str, Any]]:
    """Return tool definitions the handler will actually answer.

    Walks the MRO to find the matching handler base class, so subclasses
    (e.g. ``MyGovernanceAgent(GovernanceHandler)``) get the correct tool
    set. ADCPHandler gets all tools. Unknown handlers get only protocol
    discovery (minimum privilege).

    By default, tools whose handler method is still the SDK's
    ``not_supported`` default (the subclass never overrode it) are
    filtered out — there's no point advertising a tool that answers
    every call with ``NOT_SUPPORTED``. This keeps ``tools/list`` small
    and protects agent clients from chasing non-functional tool surface.

    Always-advertised tools:
    - :data:`_PROTOCOL_TOOLS` (``get_adcp_capabilities``) — per-spec
      handshake requirement.
    - :data:`DISCOVERY_TOOLS` — auth-optional discovery tools the spec
      requires agents to expose.

    Escape hatch: pass ``advertise_all=True`` to restore the pre-#220
    behavior and advertise every tool in the handler-type's allowed
    set regardless of override state. Useful for spec-compliance
    storyboard tests and for agents that deliberately want to expose a
    ``not_supported`` tool (e.g. to signal "we know about X but don't
    implement it yet").

    Args:
        handler: The handler instance or class.
        advertise_all: When True, skip the override-based filter and
            advertise every tool allowed for the handler type.
        adcp_version: Trusted server protocol pin used to select request and
            response schemas. When omitted for an instance, the handler's
            ``get_adcp_version()`` / ``_adcp_version`` pin is used when
            available. Class-only introspection retains the current generated
            model surface.
        schema_mode: Pydantic-derived input discovery shape. ``"compact"``
            (default) keeps every root request field, required nested fields,
            scalar optionals, and protected mutation surfaces; ``"defs"``
            keeps the complete schema with local references, and ``"inline"``
            preserves the historical fully expanded schema.
            Version-pinned handlers continue to advertise their exact bundled
            wire schemas.

    Returns:
        Filtered list of tool definitions.
    """
    if schema_mode not in _SCHEMA_MODES:
        raise ValueError(
            f"schema_mode must be one of {sorted(_SCHEMA_MODES)!r}, got {schema_mode!r}"
        )

    cls = handler if isinstance(handler, type) else type(handler)
    instance = handler if not isinstance(handler, type) else None

    candidates: list[dict[str, Any]] = []
    for base in cls.__mro__:
        if base.__name__ in _HANDLER_TOOLS:
            allowed = _HANDLER_TOOLS[base.__name__] | _PROTOCOL_TOOLS
            candidates = [tool for tool in ADCP_TOOL_DEFINITIONS if tool["name"] in allowed]
            break
    else:
        candidates = [tool for tool in ADCP_TOOL_DEFINITIONS if tool["name"] in _PROTOCOL_TOOLS]

    # Per-instance specialism filter (Emma cross-cutting P1). When the
    # handler instance exposes ``advertised_tools_for_instance``, intersect
    # the candidate universe with the per-instance set BEFORE the
    # override-detection filter. This trims tools whose Protocol family
    # the platform didn't claim (sales-only adopter no longer advertises
    # ``acquire_rights``, ``build_creative``, etc.). Falls back to the
    # class-level universe when:
    #
    # * The handler is being inspected by class (no instance) — class-level
    #   advertisement preserves backwards compat for static introspection.
    # * The hook returns an empty set (adopter piloting a novel specialism
    #   slug not in :data:`SPECIALISM_TO_ADVERTISED_TOOLS`); muting the
    #   handler would be a worse foot-gun than over-advertising.
    if instance is not None and hasattr(instance, "advertised_tools_for_instance"):
        try:
            per_instance_set = instance.advertised_tools_for_instance()
        except Exception:
            # Defensive: never let an instance hook crash tools/list.
            per_instance_set = None
        if per_instance_set:
            always_on = _PROTOCOL_TOOLS | DISCOVERY_TOOLS
            candidates = [
                tool
                for tool in candidates
                if tool["name"] in always_on or tool["name"] in per_instance_set
            ]

    if advertise_all:
        selected = candidates
    else:
        always_on = _PROTOCOL_TOOLS | DISCOVERY_TOOLS
        selected = [
            tool
            for tool in candidates
            if tool["name"] in always_on or _is_method_overridden(cls, tool["name"])
        ]

    resolved_version = _resolve_handler_adcp_version(instance, adcp_version)

    if not _include_schemas:
        return [{"name": tool["name"]} for tool in selected]

    if resolved_version is None:
        # Pydantic schema generation is expensive for the full AdCP surface,
        # especially with the 3.2 model graph. Compile only the definitions
        # this handler will advertise.
        _ensure_pydantic_schemas_applied(tool["name"] for tool in selected)
        # The in-memory registry shares memoized schema subtrees to keep server
        # startup compact. Public definitions remain ordinary independently
        # mutable JSON values, matching the pre-memoization behavior.
        definitions = [_copy_json_without_aliases(tool) for tool in selected]
        if schema_mode != "compact":
            alternate_inputs = _generate_pydantic_schemas(
                (definition["name"] for definition in definitions),
                schema_mode=schema_mode,
            )
            for definition in definitions:
                alternate = alternate_inputs.get(definition["name"])
                if alternate is not None:
                    definition["inputSchema"] = alternate
        for definition in definitions:
            if definition["name"] == "sync_reporting_receipts":
                from adcp.reporting.receipts.wire import receipt_schema

                # Also runs when model generation fell back to hand-written
                # stubs. Do not mutate cached upstream/generated schema inputs.
                definition["inputSchema"] = receipt_schema("request")
                definition["outputSchema"] = receipt_schema("sync")
            elif (
                definition["name"] == "get_reporting_status"
                and getattr(instance, "reporting_feed_store", None) is not None
            ):
                from adcp.reporting.feed.request import feed_schema

                definition["inputSchema"] = feed_schema("request")
                definition["outputSchema"] = feed_schema("sync")
        return definitions

    if not list_validator_keys(version=resolved_version):
        raise ValueError(
            f"no bundled AdCP schemas are available for adcp_version={resolved_version!r}"
        )

    # A pinned server advertises the exact bundled wire contract. This also
    # removes tools absent from that release (for example, the compact 3.2
    # lifecycle on a 3.1 endpoint) instead of leaking the process-global
    # current-model surface into tools/list.
    versioned: list[dict[str, Any]] = []
    for tool in selected:
        name = tool["name"]
        input_schema = get_mcp_schema(name, "request", version=resolved_version)
        if input_schema is None:
            continue
        # The current-model schemas can be large and are replaced below by
        # the exact versioned wire schemas. Copy only retained metadata.
        definition = copy.deepcopy(
            {
                key: value
                for key, value in tool.items()
                if key not in {"inputSchema", "outputSchema"}
            }
        )
        if name == "sync_reporting_receipts":
            from adcp.reporting.receipts.wire import receipt_schema

            definition["inputSchema"] = receipt_schema("request", version=resolved_version)
            definition["outputSchema"] = receipt_schema("sync", version=resolved_version)
            versioned.append(definition)
            continue
        if (
            name == "get_reporting_status"
            and getattr(instance, "reporting_feed_store", None) is not None
        ):
            from adcp.reporting.feed.request import feed_schema

            definition["inputSchema"] = feed_schema("request", version=resolved_version)
            definition["outputSchema"] = feed_schema("sync", version=resolved_version)
            versioned.append(definition)
            continue
        definition["inputSchema"] = input_schema
        output_schema = get_mcp_schema(name, "sync", version=resolved_version)
        if output_schema is not None:
            output_schema.setdefault("type", "object")
            definition["outputSchema"] = output_schema
        else:
            definition.pop("outputSchema", None)
        versioned.append(definition)
    return versioned

Return tool definitions the handler will actually answer.

Walks the MRO to find the matching handler base class, so subclasses (e.g. MyGovernanceAgent(GovernanceHandler)) get the correct tool set. ADCPHandler gets all tools. Unknown handlers get only protocol discovery (minimum privilege).

By default, tools whose handler method is still the SDK's not_supported default (the subclass never overrode it) are filtered out — there's no point advertising a tool that answers every call with NOT_SUPPORTED. This keeps tools/list small and protects agent clients from chasing non-functional tool surface.

Always-advertised tools: - :data:_PROTOCOL_TOOLS (get_adcp_capabilities) — per-spec handshake requirement. - :data:DISCOVERY_TOOLS — auth-optional discovery tools the spec requires agents to expose.

Escape hatch: pass advertise_all=True to restore the pre-#220 behavior and advertise every tool in the handler-type's allowed set regardless of override state. Useful for spec-compliance storyboard tests and for agents that deliberately want to expose a not_supported tool (e.g. to signal "we know about X but don't implement it yet").

Args
-----=
handler: The handler instance or class.
advertise_all: When True, skip the override-based filter and advertise every tool allowed for the handler type.
adcp_version: Trusted server protocol pin used to select request and response schemas. When omitted for an instance, the handler's get_adcp_version() / _adcp_version pin is used when available. Class-only introspection retains the current generated model surface.
schema_mode: Pydantic-derived input discovery shape. "compact" (default) keeps every root request field, required nested fields, scalar optionals, and protected mutation surfaces; "defs" keeps the complete schema with local references, and "inline" preserves the historical fully expanded schema. Version-pinned handlers continue to advertise their exact bundled wire schemas.

Returns -----= Filtered list of tool definitions.

def register_handler_tools(handler_name: str, tools: Iterable[str]) ‑> None

Expand source code

def register_handler_tools(handler_name: str, tools: Iterable[str]) -> None:
    """Register a handler-class-name → tool-set mapping with the framework.

    Public seam. ``get_tools_for_handler`` reads ``_HANDLER_TOOLS`` to
    filter ``tools/list`` per handler subclass; without registration, an
    ``ADCPHandler`` subclass that introduces a new specialism would fall
    through to its parent's tool set (typically ``ADCPHandler``'s
    full-spec surface), over-advertising. Codegen targets like
    ``adcp.decisioning.handler.PlatformHandler`` register here at class
    definition time via ``ADCPHandler.__init_subclass__``; hand-written
    custom bases call this directly before ``serve()``.
    Idempotent on equal input — calling twice with the same tool set
    is a no-op so module re-imports / reload-friendly test harnesses
    don't break.
    Conflicts raise. Unknown tool names raise with a closest-match
    suggestion (typo recovery for adopters working from spec memory).
    :param handler_name: The class name of the handler subclass —
        typically ``cls.__name__`` from inside ``__init_subclass__``.
    :param tools: Iterable of AdCP tool names this handler answers
        (members of ``ADCP_TOOL_DEFINITIONS``). Order doesn't matter.
    :raises ValueError: when ``handler_name`` is already registered with
        a different tool set, or when any tool name is not in the AdCP
        spec surface.
    """
    incoming = frozenset(tools)
    existing = _HANDLER_TOOLS.get(handler_name)
    if existing is not None:
        if frozenset(existing) == incoming:
            return
        raise ValueError(
            f"register_handler_tools({handler_name!r}, ...) called twice "
            f"with different tool sets. Existing: {sorted(existing)}; "
            f"incoming: {sorted(incoming)}. The framework can only hold "
            "one mapping per handler class — pick the canonical set."
        )
    unknown = incoming - _ALL_TOOL_NAMES
    if unknown:
        suggestions: list[str] = []
        for bad in sorted(unknown):
            close = difflib.get_close_matches(bad, _ALL_TOOL_NAMES, n=1)
            if close:
                suggestions.append(f"{bad!r} (did you mean {close[0]!r}?)")
            else:
                suggestions.append(repr(bad))
        raise ValueError(
            f"register_handler_tools({handler_name!r}, ...): unknown tool "
            f"name(s) {', '.join(suggestions)}. Tool names must match the "
            "AdCP spec — see ``adcp.server.mcp_tools.ADCP_TOOL_DEFINITIONS``."
        )
    _HANDLER_TOOLS[handler_name] = set(incoming)

Public seam. get_tools_for_handler() reads _HANDLER_TOOLS to filter tools/list per handler subclass; without registration, an ADCPHandler subclass that introduces a new specialism would fall through to its parent's tool set (typically ADCPHandler's full-spec surface), over-advertising. Codegen targets like PlatformHandler register here at class definition time via ADCPHandler.__init_subclass__; hand-written custom bases call this directly before serve(). Idempotent on equal input — calling twice with the same tool set is a no-op so module re-imports / reload-friendly test harnesses don't break. Conflicts raise. Unknown tool names raise with a closest-match suggestion (typo recovery for adopters working from spec memory). :param handler_name: The class name of the handler subclass — typically cls.__name__ from inside __init_subclass__. :param tools: Iterable of AdCP tool names this handler answers (members of ADCP_TOOL_DEFINITIONS). Order doesn't matter. :raises ValueError: when handler_name is already registered with a different tool set, or when any tool name is not in the AdCP spec surface.

def validate_discovery_set(tools: Iterable[str]) ‑> None

Expand source code

def validate_discovery_set(tools: Iterable[str]) -> None:
    """Fail-closed validation for an auth-optional tool set.

    Downstream that extends :data:`DISCOVERY_TOOLS` (``DISCOVERY_TOOLS |
    {"my_public_tool"}``) risks accidentally including a mutation tool,
    which would silently unauthenticate writes over HTTP. This helper
    asserts every name in the set resolves to a known ADCP tool whose
    annotations declare ``readOnlyHint: True`` — it refuses to pass
    anything mutating, destructive, or unknown.

    Call this at server startup on the effective set your middleware
    uses::

        from adcp.server import DISCOVERY_TOOLS, validate_discovery_set

        MY_DISCOVERY = DISCOVERY_TOOLS | {"list_public_formats"}
        validate_discovery_set(MY_DISCOVERY)  # raises early if misconfigured

    :raises ValueError: if any name in ``tools`` is unknown or resolves
        to a non-read-only tool.
    """
    by_name = {t["name"]: t for t in ADCP_TOOL_DEFINITIONS}
    unknown: list[str] = []
    mutating: list[str] = []
    for name in tools:
        tool = by_name.get(name)
        if tool is None:
            unknown.append(name)
            continue
        annotations = tool.get("annotations") or {}
        if not annotations.get("readOnlyHint"):
            mutating.append(name)
    problems: list[str] = []
    if unknown:
        problems.append(f"unknown tool(s): {sorted(unknown)}")
    if mutating:
        problems.append(
            f"non-read-only tool(s) {sorted(mutating)} — adding these to the "
            "auth-optional set would silently unauthenticate mutations"
        )
    if problems:
        raise ValueError("validate_discovery_set rejected the set: " + "; ".join(problems))

Fail-closed validation for an auth-optional tool set.

Downstream that extends :data:DISCOVERY_TOOLS (DISCOVERY_TOOLS | {"my_public_tool"}) risks accidentally including a mutation tool, which would silently unauthenticate writes over HTTP. This helper asserts every name in the set resolves to a known ADCP tool whose annotations declare readOnlyHint: True — it refuses to pass anything mutating, destructive, or unknown.

Call this at server startup on the effective set your middleware uses::

from adcp.server import DISCOVERY_TOOLS, validate_discovery_set

MY_DISCOVERY = DISCOVERY_TOOLS | {"list_public_formats"}
validate_discovery_set(MY_DISCOVERY)  # raises early if misconfigured

:raises ValueError: if any name in tools is unknown or resolves to a non-read-only tool.

Classes

class MCPToolSet (handler: ADCPHandler[Any], *, advertise_all: bool = False, validation: ValidationHookConfig | None = None, pre_validation_hooks: PreValidationHooks | None = None, response_enhancer: ResponseEnhancer | None = None, adcp_version: str | None = None, schema_mode: SchemaMode = 'compact')

Expand source code

class MCPToolSet:
    """Collection of MCP tools from an ADCP handler.

    Provides tool definitions and handlers for registering with an MCP server.
    """

    def __init__(
        self,
        handler: ADCPHandler[Any],
        *,
        advertise_all: bool = False,
        validation: ValidationHookConfig | None = None,
        pre_validation_hooks: PreValidationHooks | None = None,
        response_enhancer: ResponseEnhancer | None = None,
        adcp_version: str | None = None,
        schema_mode: SchemaMode = "compact",
    ):
        """Create tool set from handler.

        Args:
            handler: ADCP handler instance.
            advertise_all: When True, advertise every tool the handler
                type supports — even those whose method is still the
                SDK's ``not_supported`` default. See
                :func:`get_tools_for_handler` for the default behavior
                (override-filtered advertisement).
            validation: Opt-in schema validation config applied to every
                tool caller. See :func:`create_tool_caller`.
            pre_validation_hooks: Optional dict mapping tool name to a
                ``(tool_name, args) -> args`` callable or ordered sequence.
                Applied before schema + Pydantic validation. See
                :func:`create_tool_caller`.
            response_enhancer: Optional server-wide :data:`ResponseEnhancer`
                applied to every successful response. See
                :func:`create_tool_caller`.
            schema_mode: Pydantic-derived input discovery shape. See
                :func:`get_tools_for_handler`.
        """
        self.handler = handler
        resolved_adcp_version = _resolve_handler_adcp_version(handler, adcp_version)
        self._filtered_definitions = get_tools_for_handler(
            handler,
            advertise_all=advertise_all,
            adcp_version=resolved_adcp_version,
            schema_mode=schema_mode,
        )
        self._tools: dict[str, Callable[..., Any]] = {}

        # Create tool callers only for filtered tools
        for tool_def in self._filtered_definitions:
            name = tool_def["name"]
            hook = (pre_validation_hooks or {}).get(name)
            self._tools[name] = create_tool_caller(
                handler,
                name,
                validation=validation,
                pre_validation_hook=hook,
                response_enhancer=response_enhancer,
                default_unnegotiated_adcp_version=(
                    resolved_adcp_version or DEFAULT_UNNEGOTIATED_ADCP_VERSION
                ),
            )

    @property
    def tool_definitions(self) -> list[dict[str, Any]]:
        """Get MCP tool definitions filtered by handler type."""
        return list(self._filtered_definitions)

    async def call_tool(self, name: str, params: dict[str, Any]) -> Any:
        """Call a tool by name.

        Args:
            name: Tool name
            params: Tool parameters

        Returns:
            Tool result

        Raises:
            KeyError: If tool not found
        """
        if name not in self._tools:
            raise KeyError(f"Unknown tool: {name}")
        return await self._tools[name](params)

    def get_tool_names(self) -> list[str]:
        """Get list of available tool names."""
        return list(self._tools.keys())

Collection of MCP tools from an ADCP handler.

Provides tool definitions and handlers for registering with an MCP server.

Create tool set from handler.

Args
-----=
handler: ADCP handler instance.
advertise_all: When True, advertise every tool the handler type supports — even those whose method is still the SDK's not_supported default. See :func:get_tools_for_handler() for the default behavior (override-filtered advertisement).
validation: Opt-in schema validation config applied to every tool caller. See :func:create_tool_caller().
pre_validation_hooks: Optional dict mapping tool name to a (tool_name, args) -> args callable or ordered sequence. Applied before schema + Pydantic validation. See :func:create_tool_caller().
response_enhancer: Optional server-wide :data:ResponseEnhancer applied to every successful response. See :func:create_tool_caller().
schema_mode: Pydantic-derived input discovery shape. See :func:get_tools_for_handler().

Instance variables

prop tool_definitions : list[dict[str, Any]]

Expand source code

@property
def tool_definitions(self) -> list[dict[str, Any]]:
    """Get MCP tool definitions filtered by handler type."""
    return list(self._filtered_definitions)

Get MCP tool definitions filtered by handler type.

Methods

async def call_tool(self, name: str, params: dict[str, Any]) ‑> Any

Expand source code

async def call_tool(self, name: str, params: dict[str, Any]) -> Any:
    """Call a tool by name.

    Args:
        name: Tool name
        params: Tool parameters

    Returns:
        Tool result

    Raises:
        KeyError: If tool not found
    """
    if name not in self._tools:
        raise KeyError(f"Unknown tool: {name}")
    return await self._tools[name](params)

Call a tool by name.

Args
-----=
name: Tool name
params: Tool parameters

Returns -----= Tool result

Raises
-----=
KeyError: If tool not found

def get_tool_names(self) ‑> list[str]

Expand source code

def get_tool_names(self) -> list[str]:
    """Get list of available tool names."""
    return list(self._tools.keys())

Get list of available tool names.

Global variables

Functions

623 ("Typed dispatcher rejects valid request when account is

Classes

Instance variables

Methods

623 ("Typed dispatcher rejects valid request when `account` is