fix(mcp,fastmcp): serve per-user workflows from MCP front ends

docs/source/run-workflows/fastmcp-server.md

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

docs/source/run-workflows/mcp-server.md

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

packages/nvidia_nat_fastmcp/src/nat/plugins/fastmcp/server/front_end_plugin.py

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

packages/nvidia_nat_fastmcp/src/nat/plugins/fastmcp/server/front_end_plugin_worker.py

@@ -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."""
@@ -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)
@@ -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():
@@ -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]: