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

Expand source code

def create_tool_caller(
    handler: ADCPHandler,
    method_name: str,
) -> Callable[[dict[str, Any]], Any]:
    """Create a tool caller function for an ADCP handler method.

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

    Returns:
        Async callable that invokes the handler method
    """
    method = getattr(handler, method_name)

    async def call_tool(params: dict[str, Any]) -> Any:
        context = ToolContext()
        result = await method(params, context)
        # Convert Pydantic models to dicts for MCP serialization
        if hasattr(result, "model_dump"):
            return result.model_dump(exclude_none=True)
        return result

    return call_tool

Create a tool caller function for an ADCP handler method.

Args

handler

The ADCP handler instance

method_name

Name of the method to call

Returns

Async callable that invokes the handler method

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[[dict[str, Any]], 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.