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.

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) ‑> 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,
) -> 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`.

    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,
    )

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().

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 (issue #249).** When ``validation`` is
    supplied, the dispatcher validates incoming requests and outgoing
    responses against the bundled 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). Defaults to off on the server side — the client-side
    hooks already catch drift for SDK-built clients, and enabling
    server validation is a deliberate opt-in for authors who want
    dispatcher-level enforcement.

    **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.
    Exceptions from the hook surface as ``INVALID_REQUEST`` — do not raise
    for missing-but-defaultable fields, only for structurally unusable args.

    **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.compat.legacy import LEGACY_ADAPTER_VERSIONS, get_legacy_adapter
    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,
    )

    method = getattr(handler, method_name)
    params_model = _resolve_params_pydantic_model(method)

    # 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 pre_validation_hooks:
            try:
                params = _apply_pre_validation_hooks(
                    pre_validation_hooks, method_name, dict(params)
                )
            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.
        #
        # Strictness gate: setting ``ADCP_STRICT_VERSION_ENVELOPE=1``
        # raises ``VERSION_UNSUPPORTED`` for unsupported claims (the
        # spec-prescribed behaviour). The default (off) logs a warning
        # and falls through to SDK-pin validation — adopters with test
        # fixtures using placeholder version values (``adcp_major_version=4``
        # was a common sentinel before this gate existed) keep working
        # while they migrate. Strict will become the default in 5.3.
        wire_version_rejected = False
        try:
            wire_version = detect_wire_version(params)
        except UnsupportedVersionError as exc:
            wire_version_rejected = True
            if os.environ.get("ADCP_STRICT_VERSION_ENVELOPE", "0") == "1":
                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 they
                            # sent (int for ``adcp_major_version``, str
                            # for ``adcp_version``).
                            details={
                                "claimed_version": exc.wire_value,
                                "supported_versions": list(exc.supported),
                            },
                        )
                    ],
                ) from exc
            logger.warning(
                "Wire-version envelope rejected by detect_wire_version (%s); "
                "falling through to SDK-pin validation. "
                "Set ADCP_STRICT_VERSION_ENVELOPE=1 to raise "
                "VERSION_UNSUPPORTED instead. Strict will become the default "
                "in 5.3.",
                exc,
            )
            wire_version = None

        # 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

        if wire_version is None and not wire_version_rejected:
            wire_version = default_unnegotiated_adcp_version

        ctx.resolved_adcp_version = wire_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 ADCPTaskError(
                    operation=method_name,
                    errors=[
                        Error(
                            code="INVALID_REQUEST",
                            message=(
                                f"Legacy adapter for {method_name!r} at "
                                f"AdCP {wire_version} failed: "
                                f"{type(exc).__name__}: {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

        if isinstance(params, dict):
            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 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)
        # 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):
            _normalize_response_envelope(
                method_name,
                result,
                raw_params,
                adcp_version=post_adapter_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:
                _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=post_adapter_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 ADCPTaskError(
                        operation=method_name,
                        errors=[
                            Error(
                                code="INTERNAL_ERROR",
                                message=(
                                    f"Legacy response normalizer for "
                                    f"{method_name!r} at AdCP "
                                    f"{wire_version} failed: "
                                    f"{type(exc).__name__}: {exc}"
                                ),
                            )
                        ],
                    ) from exc
        if (
            legacy_adapter is not None
            and isinstance(result, dict)
            and result.get("status") == "completed"
        ):
            result.pop("status")
        return result

    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 (issue #249). When validation is supplied, the dispatcher validates incoming requests and outgoing responses against the bundled 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). Defaults to off on the server side — the client-side hooks already catch drift for SDK-built clients, and enabling server validation is a deliberate opt-in for authors who want dispatcher-level enforcement.

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. Exceptions from the hook surface as INVALID_REQUEST — do not raise for missing-but-defaultable fields, only for structurally unusable args.

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) ‑> list[dict[str, typing.Any]]

Expand source code

def get_tools_for_handler(
    handler: ADCPHandler[Any] | type[ADCPHandler[Any]],
    *,
    advertise_all: bool = False,
) -> 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.

    Returns:
        Filtered list of tool definitions.
    """
    _ensure_pydantic_schemas_applied()
    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:
        return candidates

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

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.

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)

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,
    ):
        """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`.
        """
        self.handler = handler
        self._filtered_definitions = get_tools_for_handler(handler, advertise_all=advertise_all)
        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,
            )

    @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().

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.

Functions

Args

Returns

623 ("Typed dispatcher rejects valid request when account is

Args

Returns

Args

Returns

Classes

Args

Instance variables

Methods

Args

Returns

Raises

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