Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
44 changes: 44 additions & 0 deletions docs/handler-authoring.md
Original file line number Diff line number Diff line change
Expand Up @@ -55,6 +55,50 @@ Pass ``advertise_all=True`` to ``serve()`` / ``create_mcp_server()`` /
storyboards, agents that deliberately signal
``not_supported`` on specific tools).

**Custom handler bases — declare ``advertised_tools``.** The override
filter works perfectly for direct ``ADCPHandler`` subclasses and the
specialized bases (``GovernanceHandler``, ``ContentStandardsHandler``,
etc.). But if you're authoring a *new* specialized base that
introduces its own focused tool set — typically a codegen target like
``adcp.decisioning.handler.PlatformHandler``, or a hand-rolled
``ReadOnlyAnalyticsHandler`` — declare the tool set on the class
itself::

from typing import ClassVar
from adcp.server import ADCPHandler

class ReadOnlyAnalyticsHandler(ADCPHandler):
advertised_tools: ClassVar[set[str]] = {
"get_products",
"get_media_buy_delivery",
}

# ... method bodies ...

The framework auto-registers ``ReadOnlyAnalyticsHandler ->
advertised_tools`` at class definition time via
``ADCPHandler.__init_subclass__``. The override filter then runs
against this set instead of inheriting ``ADCPHandler``'s full
surface.

Equivalent imperative form — same outcome::

from adcp.server import register_handler_tools
register_handler_tools("ReadOnlyAnalyticsHandler", {"get_products", ...})

Without either declaration, ``serve()`` emits a one-time
``UserWarning`` at boot pointing you at the registration paths. The
warning matters because the alternative — silent over-advertisement
of the full ADCPHandler surface — is exactly the discoverability
gap that bites operators in production: ``tools/list`` returns 57
tools when the agent only handles 2.

**What not to build:** don't use ``advertise_all=True`` as a
workaround for missing registration. The flag exists for legitimate
opt-in cases (storyboards, deliberate ``not_supported`` signaling);
using it to silence the registration warning over-advertises every
tool to every buyer.

## The `_impl` pattern (production-grade)

Production agents usually don't put business logic directly on handler
Expand Down
2 changes: 2 additions & 0 deletions src/adcp/server/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -99,6 +99,7 @@ async def get_products(params, context=None):
MCPToolSet,
create_mcp_tools,
get_tools_for_handler,
register_handler_tools,
validate_discovery_set,
)
from adcp.server.proposal import ProposalBuilder, ProposalNotSupported
Expand Down Expand Up @@ -167,6 +168,7 @@ async def get_products(params, context=None):
"create_mcp_tools",
"create_mcp_server",
"get_tools_for_handler",
"register_handler_tools",
"serve",
"validate_discovery_set",
# A2A integration
Expand Down
47 changes: 47 additions & 0 deletions src/adcp/server/base.py
Original file line number Diff line number Diff line change
Expand Up @@ -231,10 +231,57 @@ class ADCPHandler(ABC, Generic[TContext]):
- ContentStandardsHandler: For content standards agents
- SponsoredIntelligenceHandler: For sponsored intelligence agents
- GovernanceHandler: For governance agents

**Tool advertisement** (`advertised_tools` class attribute):

A subclass that introduces a new specialism — i.e., a custom base
that needs its own ``tools/list`` filter rather than inheriting one
from a built-in handler — declares the tool set on the class body::

class PlatformHandler(ADCPHandler):
advertised_tools: ClassVar[set[str]] = {
"get_products",
"create_media_buy",
...
}

The framework registers ``PlatformHandler -> advertised_tools`` with
:func:`adcp.server.mcp_tools.register_handler_tools` at class
definition time. Subclasses that DON'T introduce a new specialism
(a custom ``MyContentAgent(ContentStandardsHandler)``, for example)
inherit their parent's tool set unchanged — no class attr needed.

Hand-written equivalent (no ``advertised_tools`` declaration)::

from adcp.server.mcp_tools import register_handler_tools
register_handler_tools("PlatformHandler", {...})

Either path is fine; codegen targets emit the class attribute so the
declaration sits next to the class definition.
"""

_agent_type: str = "this agent"

def __init_subclass__(cls, **kwargs: Any) -> None:
"""Auto-register subclass-declared tool advertisement.

Reads ``cls.__dict__["advertised_tools"]`` (subclass-defined-only
— inherited values don't trigger re-registration) and routes
through :func:`adcp.server.mcp_tools.register_handler_tools`.
Only fires when the subclass declares the attribute on its own
class body; intermediate subclasses (multi-level hierarchy)
register at the level that introduces the attribute.

The lazy import avoids a base.py ↔ mcp_tools.py circular —
mcp_tools imports ADCPHandler at module load, so register is
looked up only when a subclass is actually being created.
"""
super().__init_subclass__(**kwargs)
if "advertised_tools" in cls.__dict__:
from adcp.server.mcp_tools import register_handler_tools

register_handler_tools(cls.__name__, cls.__dict__["advertised_tools"])

def _not_supported(self, operation: str) -> NotImplementedResponse:
"""Create a not-supported response that includes the agent type."""
return not_supported(f"{operation} is not supported by {self._agent_type}")
Expand Down
71 changes: 66 additions & 5 deletions src/adcp/server/mcp_tools.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@
from __future__ import annotations

import copy
import difflib
import logging
from collections.abc import Callable, Iterable
from typing import Any
Expand Down Expand Up @@ -1027,6 +1028,58 @@ def validate_discovery_set(tools: Iterable[str]) -> None:
assert not _unknown, f"{_handler_name} references unknown tools: {_unknown}"


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)


# ============================================================================
# Pydantic schema generation — spec-accurate input schemas
# ============================================================================
Expand Down Expand Up @@ -1325,10 +1378,18 @@ def _apply_pydantic_schemas() -> None:
_apply_pydantic_schemas()


_SDK_BASE_CLASS_NAMES: frozenset[str] = frozenset(_HANDLER_TOOLS.keys())
"""Names of the SDK's own base classes. Used to detect whether a method
is an SDK default (inherited from one of these) or a subclass override.
Kept alongside ``_HANDLER_TOOLS`` so they can't drift."""
def _is_sdk_base_class(cls_name: str) -> bool:
"""True when ``cls_name`` is registered in ``_HANDLER_TOOLS``.

Used during MRO walks to identify the nearest SDK base whose
method baselines a subclass override. Reads ``_HANDLER_TOOLS``
live so that handler classes registered after import time —
via :func:`register_handler_tools` or
:meth:`ADCPHandler.__init_subclass__` reading
``advertised_tools`` — participate in override detection without
requiring a frozen-set rebuild.
"""
return cls_name in _HANDLER_TOOLS


def _is_method_overridden(handler_cls: type, method_name: str) -> bool:
Expand Down Expand Up @@ -1375,7 +1436,7 @@ def _is_method_overridden(handler_cls: type, method_name: str) -> bool:
sdk_base: type | None = None
base_method: Any | None = None
for base in handler_cls.__mro__[1:]:
if base.__name__ not in _SDK_BASE_CLASS_NAMES:
if not _is_sdk_base_class(base.__name__):
continue
found = base.__dict__.get(method_name)
if found is None:
Expand Down
96 changes: 95 additions & 1 deletion src/adcp/server/serve.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,14 +20,19 @@ async def get_adcp_capabilities(self, params, context=None):

import logging
import os
import warnings
from collections.abc import Awaitable, Callable
from dataclasses import dataclass
from typing import TYPE_CHECKING, Any, Literal

logger = logging.getLogger("adcp.server")

from adcp.server.base import ADCPHandler, ToolContext
from adcp.server.mcp_tools import create_tool_caller, get_tools_for_handler
from adcp.server.mcp_tools import (
_HANDLER_TOOLS,
create_tool_caller,
get_tools_for_handler,
)

if TYPE_CHECKING:
from collections.abc import Sequence
Expand Down Expand Up @@ -203,6 +208,14 @@ def _log_advertised_tools(

Registered at ``INFO`` because operators routinely tail this; the
delta at ``DEBUG`` because it's noisy on fully-implemented handlers.

Also fires a one-time ``UserWarning`` at boot when the handler
class introduces a new specialism (a custom subclass that's not in
the framework's tool registry and doesn't declare
``advertised_tools``) but ``advertise_all`` is False — closes the
silent over-advertisement gap where adopters see the full
``ADCPHandler`` tool surface inherited via MRO when they meant to
declare a focused subset.
"""
registered_set = set(registered)
full_defs = get_tools_for_handler(handler, advertise_all=True)
Expand All @@ -219,6 +232,87 @@ def _log_advertised_tools(
if unadvertised and not advertise_all:
logger.debug("%s server unadvertised tools: %s", transport, ", ".join(unadvertised))

# Stacklevel walks: warnings.warn → _warn_if_unregistered_subclass →
# _log_advertised_tools → operator's call site. The MCP path adds one
# extra frame (_register_handler_tools); A2A calls _log_advertised_tools
# directly from create_a2a_server.
caller_stacklevel = 4 if transport == "mcp" else 3
_warn_if_unregistered_subclass(
handler, advertise_all=advertise_all, stacklevel=caller_stacklevel
)


#: Bases whose tool set is broad-by-design — when an adopter subclass
#: lands on one of these via MRO without registering its own
#: ``advertised_tools``, the result is over-advertisement (the broad
#: base's full set inherited unintentionally). Naming the rule rather
#: than checking ``base.__name__ != "ADCPHandler"`` inline so future
#: broad bases (a hypothetical ``UniversalHandler``) get added to one
#: place — and a reviewer's first question becomes "is this base
#: broad-by-design?" not "what's special about ADCPHandler?".
_BROAD_SURFACE_BASES: frozenset[str] = frozenset({"ADCPHandler"})


def _warn_if_unregistered_subclass(
handler: ADCPHandler[Any], *, advertise_all: bool, stacklevel: int = 4
) -> None:
"""Emit a one-time ``UserWarning`` when a custom handler base bypasses
the tool-discovery registry.

The trigger: the concrete handler class itself isn't in
``_HANDLER_TOOLS``, has no ``advertised_tools`` declaration of its
own, and inherits its tool set from a broad-surface base (see
:data:`_BROAD_SURFACE_BASES`) rather than a specialized base like
``GovernanceHandler``. That combination almost always means the
adopter meant to declare a focused tool set but forgot to register
it; the framework over-advertises by silently falling through to
the broad base's full surface.

Suppressed when ``advertise_all=True`` — that's the explicit "yes,
advertise everything" opt-in.
"""
if advertise_all:
return
cls = type(handler)
if cls.__name__ in _HANDLER_TOOLS:
return
if "advertised_tools" in cls.__dict__:
# Should already have been auto-registered via __init_subclass__,
# but defensively skip the warning if the attribute exists.
return
# Walk MRO looking for a specialized (non-broad-surface) SDK base.
# If one is found, the adopter is subclassing a focused base and
# inheriting its tool set — that's the documented pattern, no
# warning needed.
has_specialized_parent = any(
base.__name__ in _HANDLER_TOOLS and base.__name__ not in _BROAD_SURFACE_BASES
for base in cls.__mro__
)
if has_specialized_parent:
return
# Default stacklevel=4 covers the MCP path (warn → this fn →
# _log_advertised_tools → _register_handler_tools → caller). The A2A
# path lacks _register_handler_tools and passes stacklevel=3.
warnings.warn(
f"Handler class {cls.__name__!r} subclasses ADCPHandler directly "
f"but isn't registered in the framework's tool-discovery "
f"registry. tools/list will inherit the full ADCPHandler tool "
f"surface — this almost always means over-advertising for a "
f"new specialism.\n\n"
f"Pick one:\n"
f" (a) declare ``advertised_tools: set[str] = {{...}}`` on "
f"{cls.__name__} (auto-registers via __init_subclass__)\n"
f" (b) call adcp.server.mcp_tools.register_handler_tools("
f"{cls.__name__!r}, {{...}}) before serve()\n"
f" (c) pass advertise_all=True to serve() to acknowledge the "
f"full advertisement\n\n"
f"Decisioning-platform adopters: codegen via "
f"`uv run python scripts/generate_decisioning_handler.py` "
f"emits the declaration for you.",
UserWarning,
stacklevel=stacklevel,
)


async def _dispatch_with_middleware(
middleware: tuple[SkillMiddleware, ...] | Sequence[SkillMiddleware],
Expand Down
Loading
Loading