Module adcp.server.mcp_tools

MCP server integration helpers.

Provides utilities for registering ADCP handlers with MCP servers.

Functions

def create_mcp_tools(handler: ADCPHandler) ‑> MCPToolSet

Expand source code

def create_mcp_tools(handler: ADCPHandler) -> 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

    Returns:
        MCPToolSet with tool definitions and handlers
    """
    return MCPToolSet(handler)

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

Returns

MCPToolSet with tool definitions and handlers

def create_tool_caller(handler: ADCPHandler, method_name: str) ‑> Callable[..., typing.Any]

Expand source code

def create_tool_caller(
    handler: ADCPHandler,
    method_name: str,
) -> 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.

    Args:
        handler: The ADCP handler instance
        method_name: Name of the method to call

    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 adcp.server.helpers import inject_context

    method = getattr(handler, method_name)

    async def call_tool(params: dict[str, Any], context: ToolContext | None = None) -> Any:
        ctx = context if context is not None else ToolContext()
        result = await method(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
        if isinstance(result, dict):
            inject_context(params, result)
        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.

Args

handler

The ADCP handler instance

method_name

Name of the method to call

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

Expand source code

def get_tools_for_handler(handler: ADCPHandler | type[ADCPHandler]) -> list[dict[str, Any]]:
    """Return tool definitions filtered by handler type.

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

    Args:
        handler: The handler instance or class

    Returns:
        Filtered list of tool definitions
    """
    cls = handler if isinstance(handler, type) else type(handler)
    for base in cls.__mro__:
        if base.__name__ in _HANDLER_TOOLS:
            allowed = _HANDLER_TOOLS[base.__name__] | _PROTOCOL_TOOLS
            return [tool for tool in ADCP_TOOL_DEFINITIONS if tool["name"] in allowed]

    return [tool for tool in ADCP_TOOL_DEFINITIONS if tool["name"] in _PROTOCOL_TOOLS]

Return tool definitions filtered by handler type.

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

Args

handler

The handler instance or class

Returns

Filtered list of tool definitions

Classes

class MCPToolSet (handler: ADCPHandler)

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):
        """Create tool set from handler.

        Args:
            handler: ADCP handler instance
        """
        self.handler = handler
        self._filtered_definitions = get_tools_for_handler(handler)
        self._tools: dict[str, Callable[..., Any]] = {}

        # Create tool callers only for filtered tools
        for tool_def in self._filtered_definitions:
            name = tool_def["name"]
            self._tools[name] = create_tool_caller(handler, name)

    @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

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.