Module adcp.server.builder

Decorator-based server builder for ADCP.

An alternative to the class-based ADCPHandler for simple agents:

from adcp.server import adcp_server, serve
from adcp.server.responses import capabilities_response, products_response

server = adcp_server("my-seller", version="1.0.0")

@server.get_products
async def get_products(params, context=None):
    return products_response(MY_PRODUCTS)

@server.get_adcp_capabilities
async def capabilities(params, context=None):
    return capabilities_response(["media_buy"])

if __name__ == "__main__":
    serve(server, name="my-seller")

Functions

def adcp_server(name: str, **kwargs: Any) ‑> ADCPServerBuilder

Expand source code

def adcp_server(name: str, **kwargs: Any) -> ADCPServerBuilder:
    """Create a decorator-based ADCP server builder.

    Args:
        name: Server name.
        **kwargs: Additional configuration (e.g., version="1.0.0").

    Returns:
        An ADCPServerBuilder instance.
    """
    return ADCPServerBuilder(name, **kwargs)

Create a decorator-based ADCP server builder.

Args

name

Server name.

**kwargs

Additional configuration (e.g., version="1.0.0").

Returns

An ADCPServerBuilder instance.

Classes

class ADCPServerBuilder (name: str, *, version: str = '1.0.0')

Expand source code

class ADCPServerBuilder:
    """Declarative server builder using decorators.

    Use ``adcp_server()`` to create an instance, then register handlers
    with decorators. The builder can be passed directly to ``serve()``.

    Example::

        server = adcp_server("my-seller")

        @server.get_products
        async def get_products(params, context=None):
            return products_response(MY_PRODUCTS)

        serve(server, name="my-seller")
    """

    def __init__(self, name: str, *, version: str = "1.0.0") -> None:
        self.name = name
        self.version = version
        self._handlers: dict[str, Callable[..., Any]] = {}

    def __getattr__(self, task_name: str) -> Callable[..., Any]:
        """Return a decorator that registers a handler for the given task."""
        if task_name.startswith("_"):
            raise AttributeError(task_name)

        def decorator(fn: Callable[..., Any]) -> Callable[..., Any]:
            if (
                task_name not in HANDLER_TO_DOMAIN
                and task_name != "get_adcp_capabilities"
            ):
                raise ValueError(
                    f"'{task_name}' is not a known ADCP task. "
                    f"Check for typos."
                )
            self._handlers[task_name] = fn
            return fn

        return decorator

    def _detect_domains(self) -> list[str]:
        """Detect which ADCP domains the registered handlers cover."""
        domains: set[str] = set()
        for handler_name in self._handlers:
            domain = HANDLER_TO_DOMAIN.get(handler_name)
            if domain:
                domains.add(domain)
        return sorted(domains)

    def build_handler(self) -> ADCPHandler:
        """Build an ADCPHandler from registered decorators.

        If ``get_adcp_capabilities`` is not registered, it will be
        auto-generated from the detected domains.
        """
        handlers = dict(self._handlers)

        # Auto-generate capabilities if not provided
        if "get_adcp_capabilities" not in handlers:
            domains = self._detect_domains()
            if domains:
                from adcp.server.responses import capabilities_response

                async def auto_capabilities(
                    params: Any, context: Any = None
                ) -> dict[str, Any]:
                    return capabilities_response(domains)

                handlers["get_adcp_capabilities"] = auto_capabilities

        # Create a dynamic subclass
        class DynamicHandler(ADCPHandler):
            pass

        for task_name, fn in handlers.items():
            # Wrap standalone functions to accept self
            async def _bound_method(
                self: Any,
                params: Any,
                context: Any = None,
                _fn: Callable[..., Any] = fn,
            ) -> Any:
                return await _fn(params, context)

            setattr(DynamicHandler, task_name, _bound_method)

        return DynamicHandler()

Declarative server builder using decorators.

Use adcp_server() to create an instance, then register handlers with decorators. The builder can be passed directly to serve().

Example::

server = adcp_server("my-seller")

@server.get_products
async def get_products(params, context=None):
    return products_response(MY_PRODUCTS)

serve(server, name="my-seller")

Methods

def build_handler(self) ‑> ADCPHandler

Expand source code

def build_handler(self) -> ADCPHandler:
    """Build an ADCPHandler from registered decorators.

    If ``get_adcp_capabilities`` is not registered, it will be
    auto-generated from the detected domains.
    """
    handlers = dict(self._handlers)

    # Auto-generate capabilities if not provided
    if "get_adcp_capabilities" not in handlers:
        domains = self._detect_domains()
        if domains:
            from adcp.server.responses import capabilities_response

            async def auto_capabilities(
                params: Any, context: Any = None
            ) -> dict[str, Any]:
                return capabilities_response(domains)

            handlers["get_adcp_capabilities"] = auto_capabilities

    # Create a dynamic subclass
    class DynamicHandler(ADCPHandler):
        pass

    for task_name, fn in handlers.items():
        # Wrap standalone functions to accept self
        async def _bound_method(
            self: Any,
            params: Any,
            context: Any = None,
            _fn: Callable[..., Any] = fn,
        ) -> Any:
            return await _fn(params, context)

        setattr(DynamicHandler, task_name, _bound_method)

    return DynamicHandler()

Build an ADCPHandler from registered decorators.

If get_adcp_capabilities is not registered, it will be auto-generated from the detected domains.