NVIDIA · BarneyChambers · Aug 22, 2026 · Aug 22, 2026 · Aug 22, 2026 · Aug 22, 2026
@@ -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.

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

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

@@ -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]:

@@ -22,6 +22,7 @@
 from inspect import Signature
 from typing import Any
 
+from mcp.server.fastmcp.server import Context
 from pydantic import BaseModel
 from pydantic.fields import FieldInfo
 from pydantic_core import PydanticUndefined
@@ -186,6 +187,32 @@ def _is_chat_request_schema(schema: Any) -> bool:
     return schema_name == "ChatRequest" or "ChatRequest" in schema_qualname
 
 
+async def _run_through_session_manager(session_manager: "SessionManager", payload: Any, ctx: Any | None = None) -> Any:
+    """Execute a payload through SessionManager, including per-user workflows."""
+    if session_manager.is_workflow_per_user:
+        from nat.builder.context import Context
+        from nat.runtime.user_manager import UserManager
+
+        user_id = Context.get().user_id
+        http_connection = None
+        if ctx is not None:
+            try:
+                http_connection = ctx.request_context.request
+                if user_id is None and http_connection is not None:
+                    user_info = UserManager.extract_user_from_connection(http_connection)
+                    if user_info is not None:
+                        user_id = user_info.get_user_id()
+            except ValueError:
+                pass
+
+        async with session_manager.session(user_id=user_id, http_connection=http_connection) as session:
+            async with session.run(payload) as runner:
+                return await runner.result()
+
+    async with session_manager.run(payload) as runner:
+        return await runner.result()
+
+
 def create_function_wrapper(
     function_name: str,
     session_manager: "SessionManager",
@@ -200,7 +227,7 @@ def create_function_wrapper(
     """
     signature, alias_map = _build_signature_from_schema(input_schema)
 
-    async def wrapper_func(**kwargs: Any) -> Any:
+    async def wrapper_func(ctx: Context | None = None, **kwargs: Any) -> Any:
         if _is_chat_request_schema(input_schema):
             from nat.data_models.api_server import ChatRequest  # type: ignore[reportMissingImports]
 
@@ -214,8 +241,7 @@ async def wrapper_func(**kwargs: Any) -> Any:
             payload = input_schema.model_validate(cleaned_kwargs) if hasattr(input_schema,
                                                                              "model_validate") else cleaned_kwargs
 
-        async with session_manager.run(payload) as runner:
-            result = await runner.result()
+        result = await _run_through_session_manager(session_manager, payload, ctx=ctx)
 
         if isinstance(result, str):
             return result
@@ -224,7 +250,10 @@ async def wrapper_func(**kwargs: Any) -> Any:
         return str(result)
 
     wrapper_func.__signature__ = signature  # type: ignore[attr-defined]
-    wrapper_func.__annotations__ = _build_annotations_from_schema(input_schema)
+    annotations = _build_annotations_from_schema(input_schema)
+    annotations["ctx"] = Context | None
+    annotations["return"] = Any
+    wrapper_func.__annotations__ = annotations
     wrapper_func.__name__ = function_name
     wrapper_func.__doc__ = "Auto-generated wrapper for a NeMo Agent Toolkit workflow."
     return wrapper_func
@@ -273,19 +302,18 @@ def register_function_with_mcp(mcp: FastMCP,
     """
     logger.info("Registering function %s with FastMCP", function_name)
 
-    # Get the workflow from the session manager
-    workflow = session_manager.workflow
-
-    # Prefer the function's schema/description when available, fall back to workflow
-    target_function = function or workflow
+    if session_manager.is_workflow_per_user:
+        input_schema = session_manager.get_workflow_input_schema()
+        workflow_config = session_manager.config.workflow
+        function_description = getattr(workflow_config, "description", None) or function_name
+    else:
+        workflow = session_manager.workflow
+        target_function = function or workflow
+        input_schema = getattr(target_function, "input_schema", workflow.input_schema)
+        function_description = get_function_description(target_function)
 
-    # Get the input schema from the most specific object available
-    input_schema = getattr(target_function, "input_schema", workflow.input_schema)
     logger.info("Function %s has input schema: %s", function_name, input_schema)
 
-    # Get function description
-    function_description = get_function_description(target_function)
-
     # Create and register the wrapper function with FastMCP
     wrapper_func = create_function_wrapper(function_name, session_manager, input_schema)
     mcp.tool(name=function_name, description=function_description)(wrapper_func)