nat.plugins.fastmcp.server.front_end_plugin_worker#

FastMCP front end worker implementation.

Attributes#

logger

Classes#

`FastMCPFrontEndPluginWorkerBase`	Base class for FastMCP front end plugin workers.
`FastMCPFrontEndPluginWorker`	Default FastMCP server worker implementation.

Module Contents#

logger#

class FastMCPFrontEndPluginWorkerBase(config: nat.data_models.config.Config)#

Bases: abc.ABC

Base class for FastMCP front end plugin workers.

Initialize the FastMCP worker with configuration.

Args:: config: The full NeMo Agent Toolkit configuration.

full_config#

front_end_config: nat.plugins.fastmcp.server.front_end_config.FastMCPFrontEndConfig#

_setup_health_endpoint(mcp: fastmcp.FastMCP)#: Set up the HTTP health endpoint that exercises FastMCP ping handler.

abstractmethod create_mcp_server() → fastmcp.FastMCP#

Async:

Create and configure the FastMCP server instance.

Returns:: FastMCP instance or a subclass with custom behavior

abstractmethod add_routes( mcp: fastmcp.FastMCP, builder: nat.builder.workflow_builder.WorkflowBuilder, )#

Async:

Add routes to the FastMCP server.

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

async _default_add_routes( mcp: fastmcp.FastMCP, builder: nat.builder.workflow_builder.WorkflowBuilder, ) → None#: Default implementation for adding routes to FastMCP.

async _get_all_functions( workflow: nat.builder.workflow.Workflow, ) → dict[str, nat.builder.function.Function]#

Get all functions from the workflow.

Args:: workflow: The NeMo Agent Toolkit workflow.
Returns:: Dict mapping function names to Function objects.

async add_root_level_routes( wrapper_app: fastapi.FastAPI, mcp: 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.

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

_setup_debug_endpoints( mcp: fastmcp.FastMCP, functions: collections.abc.Mapping[str, nat.builder.function_base.FunctionBase], ) → None#: Set up HTTP debug endpoints for introspecting tools and schemas.

class FastMCPFrontEndPluginWorker(config: nat.data_models.config.Config)#

Bases: FastMCPFrontEndPluginWorkerBase

Default FastMCP server worker implementation.

Initialize the FastMCP worker with configuration.

Args:: config: The full NeMo Agent Toolkit configuration.

async create_mcp_server() → fastmcp.FastMCP#

Create default FastMCP server instance.

Returns:: FastMCP instance configured with settings from toolkit config.

async add_routes( mcp: fastmcp.FastMCP, builder: nat.builder.workflow_builder.WorkflowBuilder, )#

Add default routes to the FastMCP server.

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