Skip to content
Open
Show file tree
Hide file tree
Changes from 4 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 8 additions & 0 deletions docs/source/run-workflows/fastmcp-server.md
Original file line number Diff line number Diff line change
Expand Up @@ -150,6 +150,14 @@ general:

With this configuration, the MCP server is accessible at `http://localhost:9902/api/v1/mcp`.

### Per-User Workflows

A per-user workflow can be served over FastMCP. Each user gets their own workflow instance, built on first use. [MCP authentication](../components/auth/mcp-auth/index.md) prescribes pairing `per_user_mcp_client` with a per-user workflow, such as `per_user_react_agent`.

The user is taken from the Bearer token on the MCP request when [`server_auth`](#authentication) is configured, or from runtime context when it is already set. Per-user workflows require authenticated streamable-http requests so the server can derive a user ID. Without that, tool calls fail with an error saying the user ID could not be determined.

The `/debug/tools/list` route returns an empty tool list for a per-user workflow because no shared workflow instance exists at startup. Use `nat mcp client tool list` to inspect the registered MCP tool and its input schema.

## Inspecting and Running MCP Tools Published by a FastMCP Server

Use `nat mcp client` to inspect and run tools exposed by an MCP server using the FastMCP server runtime.
Expand Down
20 changes: 17 additions & 3 deletions docs/source/run-workflows/mcp-server.md
Original file line number Diff line number Diff line change
Expand Up @@ -93,6 +93,14 @@ The `base_path` must start with a forward slash (`/`) and must not end with a fo
The `base_path` feature requires the `streamable-http` transport. SSE transport does not support custom base paths.
:::

### Per-User Workflows

A per-user workflow can be served over MCP. Each user gets their own workflow instance, built on first use. [MCP authentication](../components/auth/mcp-auth/index.md) prescribes pairing `per_user_mcp_client` with a per-user workflow, such as `per_user_react_agent`.

The user is taken from the Bearer token on the MCP request when [`server_auth`](#authentication) is configured, or from runtime context when it is already set. Per-user workflows require authenticated streamable-http requests so the server can derive a user ID. Without that, tool calls fail with an error saying the user ID could not be determined.

The `/debug/tools/list` route returns an empty tool list for a per-user workflow because no shared workflow instance exists at startup. Use `nat mcp client tool list` to inspect the registered MCP tool and its input schema.

## Displaying MCP Tools published by an MCP server

To list the tools published by the MCP server you can use the `nat mcp client tool list` command. This command acts as an MCP client and connects to the MCP server running on the specified URL (defaults to `http://localhost:9901/mcp` for streamable-http, with backwards compatibility for `http://localhost:9901/sse`).
Expand Down Expand Up @@ -393,9 +401,15 @@ This is useful for health checks and monitoring.

## Security Considerations

### Authentication Limitations
- The `nat mcp serve` command currently starts an MCP server without built-in authentication. Server-side authentication is planned for a future release.
- NeMo Agent Toolkit workflows can still connect to protected third-party MCP servers through the MCP client auth provider. Refer to [MCP Authentication](../components/auth/mcp-auth/index.md) for more information.
### Authentication

MCP servers using the **streamable-http** transport can validate OAuth2 bearer tokens when `server_auth` is configured in the front-end config. Validation uses JWKS or OIDC discovery for JWT access tokens, or token introspection for opaque access tokens. Refer to [MCP Authentication](../components/auth/mcp-auth/index.md) and the protected examples under `examples/MCP/`.

The **SSE** transport does not apply `server_auth`. Use streamable-http when the MCP server must authenticate callers.

Per-user workflows require authenticated requests so the server can derive a user ID from the Bearer token. See [Per-User Workflows](#per-user-workflows).

NeMo Agent Toolkit workflows can still connect to protected third-party MCP servers through the MCP client auth provider.

### Local Development
For local development, you can use `localhost` or `127.0.0.1` as the host (default). This limits access to your local machine only.
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -66,49 +66,52 @@ async def run(self) -> None:
# Get the worker instance
worker = self._get_worker_instance()

# Let the worker create the FastMCP server (allows plugins to customize)
mcp = await worker.create_mcp_server()

# Add routes through the worker (includes health endpoint and function registration)
await worker.add_routes(mcp, builder)

try:
if self.front_end_config.base_path:
if self.front_end_config.transport == "sse":
logger.warning(
"base_path is configured but SSE transport does not support mounting at sub-paths. "
"Use streamable-http transport for base_path support.")
# Let the worker create the FastMCP server (allows plugins to customize)
mcp = await worker.create_mcp_server()

# Add routes through the worker (includes health endpoint and function registration)
await worker.add_routes(mcp, builder)

try:
if self.front_end_config.base_path:
if self.front_end_config.transport == "sse":
logger.warning(
"base_path is configured but SSE transport does not support mounting at sub-paths. "
"Use streamable-http transport for base_path support.")
logger.info("Starting FastMCP server with SSE endpoint at /sse")
await mcp.run_async(transport="sse",
host=self.front_end_config.host,
port=self.front_end_config.port,
log_level=self.front_end_config.log_level.lower())
else:
full_url = f"http://{self.front_end_config.host}:{self.front_end_config.port}{self.front_end_config.base_path}/mcp"
logger.info(
"Mounting FastMCP server at %s/mcp on %s:%s",
self.front_end_config.base_path,
self.front_end_config.host,
self.front_end_config.port,
)
logger.info("FastMCP server URL: %s", full_url)
await self._run_with_mount(mcp, worker)
elif self.front_end_config.transport == "sse":
logger.info("Starting FastMCP server with SSE endpoint at /sse")
await mcp.run_async(transport="sse",
host=self.front_end_config.host,
port=self.front_end_config.port,
log_level=self.front_end_config.log_level.lower())
else:
full_url = f"http://{self.front_end_config.host}:{self.front_end_config.port}{self.front_end_config.base_path}/mcp"
logger.info(
"Mounting FastMCP server at %s/mcp on %s:%s",
self.front_end_config.base_path,
self.front_end_config.host,
self.front_end_config.port,
)
full_url = f"http://{self.front_end_config.host}:{self.front_end_config.port}/mcp"
logger.info("FastMCP server URL: %s", full_url)
await self._run_with_mount(mcp, worker)
elif self.front_end_config.transport == "sse":
logger.info("Starting FastMCP server with SSE endpoint at /sse")
await mcp.run_async(transport="sse",
host=self.front_end_config.host,
port=self.front_end_config.port,
log_level=self.front_end_config.log_level.lower())
else:
full_url = f"http://{self.front_end_config.host}:{self.front_end_config.port}/mcp"
logger.info("FastMCP server URL: %s", full_url)
await mcp.run_async(transport="streamable-http",
host=self.front_end_config.host,
port=self.front_end_config.port,
path="/mcp",
log_level=self.front_end_config.log_level.lower())
except KeyboardInterrupt:
logger.info("FastMCP server shutdown requested (Ctrl+C). Shutting down gracefully.")
await mcp.run_async(transport="streamable-http",
host=self.front_end_config.host,
port=self.front_end_config.port,
path="/mcp",
log_level=self.front_end_config.log_level.lower())
except KeyboardInterrupt:
logger.info("FastMCP server shutdown requested (Ctrl+C). Shutting down gracefully.")
finally:
await worker.cleanup()

async def _run_with_mount(self, mcp: "FastMCP", worker: FastMCPFrontEndPluginWorkerBase) -> None:
"""Run FastMCP server mounted at configured base_path using FastAPI wrapper.
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -52,6 +52,44 @@ def __init__(self, config: Config):
"""
self.full_config = config
self.front_end_config: FastMCPFrontEndConfig = config.general.front_end
self._session_managers: list[SessionManager] = []

def _track_session_manager(self, session_manager: SessionManager) -> SessionManager:
"""Track a session manager so ``cleanup()`` can shut it down."""
self._session_managers.append(session_manager)
return session_manager

def _is_primary_workflow_per_user(self) -> bool:
"""Return True when the configured workflow is registered as per-user."""
from nat.cli.type_registry import GlobalTypeRegistry

workflow_registration = GlobalTypeRegistry.get().get_function(type(self.full_config.workflow))
return workflow_registration.is_per_user

def _resolve_workflow_tool_name(self) -> str:
"""Return the MCP tool name for the configured workflow entry point."""
workflow_config = self.full_config.workflow
alias = getattr(workflow_config, "workflow_alias", None)
return alias if alias else workflow_config.type

async def cleanup(self) -> None:
"""Shut down all tracked session managers.

Attempts shutdown for every manager even when an individual call fails.
Clears the tracked list before re-raising the first shutdown error.
"""
errors: list[BaseException] = []
try:
for session_manager in self._session_managers:
try:
await session_manager.shutdown()
except Exception as exc:
logger.exception("Failed to shut down SessionManager during FastMCP worker cleanup")
errors.append(exc)
finally:
self._session_managers.clear()
if errors:
raise errors[0]

def _setup_health_endpoint(self, mcp: FastMCP):
"""Set up the HTTP health endpoint that exercises FastMCP ping handler."""
Expand Down Expand Up @@ -106,14 +144,27 @@ async def add_routes(self, mcp: FastMCP, builder: WorkflowBuilder):
...

async def _default_add_routes(self, mcp: FastMCP, builder: WorkflowBuilder) -> None:
"""Default implementation for adding routes to FastMCP."""
"""Default route registration for FastMCP.

Creates session managers via SessionManager.create (shared or per-user),
registers workflow tools, and exposes debug endpoints for introspection.
"""
from nat.plugins.fastmcp.server.tool_converter import register_function_with_mcp

# Set up the health endpoint
self._setup_health_endpoint(mcp)

# Build the default workflow
workflow = await builder.build()
if self._is_primary_workflow_per_user():
session_manager = self._track_session_manager(await SessionManager.create(config=self.full_config,
shared_builder=builder))
tool_name = self._resolve_workflow_tool_name()
register_function_with_mcp(mcp, tool_name, session_manager, function=None)
self._setup_debug_endpoints(mcp, {})
return

primary_session_manager = self._track_session_manager(await SessionManager.create(config=self.full_config,
shared_builder=builder))
workflow = primary_session_manager.workflow

# Get all functions from the workflow
functions = await self._get_all_functions(workflow)
Expand All @@ -136,14 +187,11 @@ async def _default_add_routes(self, mcp: FastMCP, builder: WorkflowBuilder) -> N
for function_name, function in functions.items():
if isinstance(function, Workflow):
logger.info("Function %s is a Workflow, using directly", function_name)
session_managers[function_name] = await SessionManager.create(config=self.full_config,
shared_builder=builder,
entry_function=None)
session_managers[function_name] = primary_session_manager
else:
logger.info("Function %s is a regular function, building entry workflow", function_name)
session_managers[function_name] = await SessionManager.create(config=self.full_config,
shared_builder=builder,
entry_function=function_name)
session_managers[function_name] = self._track_session_manager(await SessionManager.create(
config=self.full_config, shared_builder=builder, entry_function=function_name))

# Register each function with FastMCP, passing SessionManager for observability
for function_name, session_manager in session_managers.items():
Expand All @@ -152,8 +200,7 @@ async def _default_add_routes(self, mcp: FastMCP, builder: WorkflowBuilder) -> N
if not session_managers:
raise RuntimeError("No functions found in workflow. Please check your configuration.")

# After registration, expose debug endpoints for tool/schema inspection
debug_functions = {name: sm.workflow for name, sm in session_managers.items()}
debug_functions = {name: sm.workflow for name, sm in session_managers.items() if not sm.is_workflow_per_user}
self._setup_debug_endpoints(mcp, debug_functions)

async def _get_all_functions(self, workflow: Workflow) -> dict[str, Function]:
Expand Down
Loading