nat.plugins.mcp.server.front_end_plugin_worker

Attributes

logger

Classes

MCPFrontEndPluginWorkerBase	Base class for MCP front end plugin workers.
MCPFrontEndPluginWorker	Default MCP server worker implementation.

Module Contents

logger

class MCPFrontEndPluginWorkerBase(config: nat.data_models.config.Config)

Bases: abc.ABC

Base class for MCP front end plugin workers.

This abstract base class provides shared utilities and defines the contract for MCP worker implementations. Most users should inherit from MCPFrontEndPluginWorker instead of this class directly.

Initialize the MCP worker with configuration.

Args:

config: The full NAT configuration

full_config

front_end_config: nat.plugins.mcp.server.front_end_config.MCPFrontEndConfig

memory_profiler

_setup_health_endpoint(mcp: mcp.server.fastmcp.FastMCP)

Set up the HTTP health endpoint that exercises MCP ping handler.

abstractmethod create_mcp_server() → mcp.server.fastmcp.FastMCP

Async:

Create and configure the MCP server instance.

This is the main extension point. Plugins can return FastMCP or any subclass to customize server behavior (for example, add authentication, custom transports).

Returns:

FastMCP instance or a subclass with custom behavior

abstractmethod add_routes(

mcp: mcp.server.fastmcp.FastMCP,

builder: nat.builder.workflow_builder.WorkflowBuilder,

)

Async:

Add routes to the MCP server.

Plugins must implement this method. Most plugins can call _default_add_routes() for standard behavior and then add custom enhancements.

Args:

mcp: The FastMCP server instance builder: The workflow builder instance

async _default_add_routes(

mcp: mcp.server.fastmcp.FastMCP,

builder: nat.builder.workflow_builder.WorkflowBuilder,

)

Default route registration logic - reusable by subclasses.

This is a protected helper method that plugins can call to get standard route registration behavior. Plugins typically call this from their add_routes() implementation and then add custom features.

This method: - Sets up the health endpoint - Builds the workflow and extracts all functions - Filters functions based on tool_names config - Registers each function as an MCP tool - Sets up debug endpoints for tool introspection

Args:

mcp: The FastMCP server instance builder: The workflow builder instance

async _get_all_functions(

workflow: nat.builder.workflow.Workflow,

) → dict[str, nat.builder.function.Function]

Get all functions from the workflow.

Args:

workflow: The NAT workflow.

Returns:

Dict mapping function names to Function objects.

async add_root_level_routes(

wrapper_app: fastapi.FastAPI,

mcp: mcp.server.fastmcp.FastMCP,

) → None

Add routes to the wrapper FastAPI app (optional extension point).

This method is called when base_path is configured and a wrapper FastAPI app is created to mount the MCP server. Plugins can override this to add routes to the wrapper app at the root level, outside the mounted MCP server path.

Common use cases: - OAuth discovery endpoints (e.g., /.well-known/oauth-protected-resource) - Health checks at root level - Static file serving - Custom authentication/authorization endpoints

Default implementation does nothing, making this an optional extension point.

Args:

wrapper_app: The FastAPI wrapper application that mounts the MCP server mcp: The FastMCP server instance (already mounted at base_path)

_setup_debug_endpoints(

mcp: mcp.server.fastmcp.FastMCP,

functions: collections.abc.Mapping[str, nat.builder.function_base.FunctionBase],

) → None

Set up HTTP debug endpoints for introspecting tools and schemas.

Exposes:

• GET /debug/tools/list: List tools. Optional query param name (one or more, repeatable or comma separated) selects a subset and returns details for those tools.

• GET /debug/memory/stats: Get current memory profiling statistics (read-only)

class MCPFrontEndPluginWorker(config: nat.data_models.config.Config)

Bases: MCPFrontEndPluginWorkerBase

Default MCP server worker implementation.

Inherit from this class to create custom MCP workers that extend or modify server behavior. Override create_mcp_server() to use a different server type, and override add_routes() to add custom functionality.

Example:

class CustomWorker(MCPFrontEndPluginWorker):

async def create_mcp_server(self):

# Return custom MCP server instance return MyCustomFastMCP(…)

async def add_routes(self, mcp, builder):

# Get default routes await super().add_routes(mcp, builder) # Add custom features self._add_my_custom_features(mcp)

Initialize the MCP worker with configuration.

Args:

config: The full NAT configuration

async create_mcp_server() → mcp.server.fastmcp.FastMCP

Create default MCP server with optional authentication.

Returns:

FastMCP instance configured with settings from NAT config

async add_routes(

mcp: mcp.server.fastmcp.FastMCP,

builder: nat.builder.workflow_builder.WorkflowBuilder,

)

Add default routes to the MCP server.

Args:

mcp: The FastMCP server instance builder: The workflow builder instance