diff --git a/ci/vale/styles/config/vocabularies/nat/accept.txt b/ci/vale/styles/config/vocabularies/nat/accept.txt index 4bb026c6a..039afa38b 100644 --- a/ci/vale/styles/config/vocabularies/nat/accept.txt +++ b/ci/vale/styles/config/vocabularies/nat/accept.txt @@ -85,6 +85,7 @@ LLM(s?) # https://github.com/logpai/loghub/ Loghub Mem0 +[Mm]iddleware Milvus [Mm]ixin(s?) MLflow diff --git a/docs/source/extend/mcp-server.md b/docs/source/extend/mcp-server.md new file mode 100644 index 000000000..f8a6cd0a4 --- /dev/null +++ b/docs/source/extend/mcp-server.md @@ -0,0 +1,253 @@ + + +# Adding a Custom MCP Server Worker + +:::{note} +We recommend reading the [MCP Server Guide](../workflows/mcp/mcp-server.md) before proceeding with this documentation, to understand how MCP servers work in NVIDIA NeMo Agent toolkit. +::: + +The NVIDIA NeMo Agent toolkit provides a default MCP server worker that publishes your workflow functions as MCP tools. However, you may need to customize the server behavior for enterprise requirements such as authentication, custom endpoints, or telemetry. This guide shows you how to create custom MCP server workers that extend the default implementation. + +## When to Create a Custom Worker + +Create a custom MCP worker when you need to: +- **Add authentication/authorization**: OAuth, API keys, JWT tokens, or custom auth flows +- **Integrate custom transport protocols**: WebSocket, gRPC, or other communication methods +- **Add logging and telemetry**: Custom logging, metrics collection, or distributed tracing +- **Modify server behavior**: Custom middleware, error handling, or protocol extensions +- **Integrate with enterprise systems**: SSO, audit logging, or compliance requirements + +## Creating and Registering a Custom MCP Worker + +To extend the NeMo Agent toolkit with custom MCP workers, you need to create a worker class that inherits from {py:class}`~nat.front_ends.mcp.mcp_front_end_plugin_worker.MCPFrontEndPluginWorker` and override the methods you want to customize. + +This section provides a step-by-step guide to create and register a custom MCP worker with the NeMo Agent toolkit. A custom status endpoint worker is used as an example to demonstrate the process. + +## Step 1: Implement the Worker Class + +Create a new Python file for your worker implementation. The following example shows a minimal worker that adds a custom status endpoint to the MCP server. + +Each worker is instantiated once when `nat mcp serve` runs. The `create_mcp_server()` method executes during initialization, and `add_routes()` runs after the workflow is built. + + +`src/my_package/custom_worker.py`: +```python +import logging + +from mcp.server.fastmcp import FastMCP + +from nat.builder.workflow_builder import WorkflowBuilder +from nat.front_ends.mcp.mcp_front_end_plugin_worker import MCPFrontEndPluginWorker + +logger = logging.getLogger(__name__) + + +class CustomStatusWorker(MCPFrontEndPluginWorker): + """MCP worker that adds a custom status endpoint.""" + + async def add_routes(self, mcp: FastMCP, builder: WorkflowBuilder): + """Register tools and add custom server behavior. + + This method calls the parent implementation to get all default behavior, + then adds custom routes. + + Args: + mcp: The FastMCP server instance + builder: The workflow builder containing functions to expose + """ + # Get all default routes and tool registration + await super().add_routes(mcp, builder) + + # Add a custom status endpoint + @mcp.custom_route("/custom/status", methods=["GET"]) + async def custom_status(_request): + """Custom status endpoint with additional server information.""" + from starlette.responses import JSONResponse + + logger.info("Custom status endpoint called") + return JSONResponse({ + "status": "ok", + "server": mcp.name, + "custom_worker": "CustomStatusWorker" + }) +``` + +**Key components**: +- **Inheritance**: Extend {py:class}`~nat.front_ends.mcp.mcp_front_end_plugin_worker.MCPFrontEndPluginWorker` +- **`super().add_routes()`**: Calls parent to get standard tool registration and default routes +- **`@mcp.custom_route()`**: Adds custom HTTP endpoints to the server +- **Clean inheritance**: Use standard Python `super()` pattern to extend behavior + +## Step 2: Use the Worker in Your Workflow + +Configure your workflow to use the custom worker by specifying the fully qualified class name in the `runner_class` field. + + +`custom_mcp_server_workflow.yml`: +```yaml +general: + front_end: + _type: mcp + runner_class: "my_package.custom_worker.CustomStatusWorker" + name: "my_custom_server" + host: "localhost" + port: 9000 + + +llms: + nim_llm: + _type: nim + model_name: meta/llama-3.3-70b-instruct + +functions: + search: + _type: tavily_internet_search + +workflow: + _type: react_agent + llm_name: nim_llm + tool_names: [search] +``` + +## Step 3: Run and Test Your Server + +Start your server using the NeMo Agent toolkit CLI: + +```bash +nat mcp serve --config_file custom_mcp_server_workflow.yml +``` + +**Expected output**: +``` +INFO: Started server process [12345] +INFO: Waiting for application startup. +INFO: Application startup complete. +INFO: Uvicorn running on http://localhost:9000 (Press CTRL+C to quit) +``` + +**Test the server** with the MCP client: + +```bash +# List available tools +nat mcp client tool list --url http://localhost:9000/mcp + +# Call a tool +nat mcp client tool call search \ + --url http://localhost:9000/mcp \ + --json-args '{"question": "When is the next GTC event?"}' + +# Test the custom status endpoint +curl http://localhost:9000/custom/status +``` + +**Expected response from custom endpoint**: +```json +{ + "status": "ok", + "server": "my_custom_server", + "custom_worker": "CustomStatusWorker" +} +``` + +## Understanding Inheritance and Extension + +### Using `super().add_routes()` + +When extending {py:class}`~nat.front_ends.mcp.mcp_front_end_plugin_worker.MCPFrontEndPluginWorker`, call `super().add_routes()` to get all default functionality: + +- **Health endpoint**: `/health` for server status checks +- **Workflow building**: Processes your workflow configuration +- **Function-to-tool conversion**: Registers NeMo Agent toolkit functions as MCP tools +- **Debug endpoints**: Additional routes for development + +Most workers call `super().add_routes()` first to ensure all standard NeMo Agent toolkit tools are registered, then add custom features: + +```python +async def add_routes(self, mcp: FastMCP, builder: WorkflowBuilder): + # Get all default behavior from parent + await super().add_routes(mcp, builder) + + # Add your custom features + @mcp.custom_route("/my/endpoint", methods=["GET"]) + async def my_endpoint(_request): + return JSONResponse({"custom": "data"}) +``` + +### Overriding `create_mcp_server()` + +Override `create_mcp_server()` when you need to use a different MCP server implementation: + +```python +async def create_mcp_server(self) -> FastMCP: + from my_custom_mcp import CustomFastMCP + + return CustomFastMCP( + name=self.front_end_config.name, + host=self.front_end_config.host, + port=self.front_end_config.port, + # Custom parameters + auth_provider=self.get_auth_provider(), + ) +``` + +**Authentication ownership**: When you override `create_mcp_server()`, your worker controls authentication. If you need custom auth (JWT, OAuth2, API keys), configure it inside `create_mcp_server()`. Any front-end config auth settings are optional hints and may be ignored by your worker. + +### Accessing Configuration + +Your worker has access to configuration through instance variables: + +- **`self.front_end_config`**: MCP server configuration + - `name`: Server name + - `host`: Server host address + - `port`: Server port number + - `debug`: Debug mode flag + +- **`self.full_config`**: Complete NeMo Agent toolkit configuration + - `general`: General settings including front end config + - `llms`: LLM configurations + - `functions`: Function configurations + - `workflow`: Workflow configuration + +**Example using configuration**: + +```python +async def create_mcp_server(self) -> FastMCP: + # Access server name from config + server_name = self.front_end_config.name + + # Customize based on debug mode + if self.front_end_config.debug: + logger.info(f"Creating debug server: {server_name}") + + return FastMCP( + name=server_name, + host=self.front_end_config.host, + port=self.front_end_config.port, + debug=self.front_end_config.debug, + ) +``` + +## Summary + +This guide provides a step-by-step process to create custom MCP server workers in the NeMo Agent toolkit. The custom status worker demonstrates how to: + +1. Extend {py:class}`~nat.front_ends.mcp.mcp_front_end_plugin_worker.MCPFrontEndPluginWorker` +2. Override `add_routes()` and use `super()` to get default behavior +3. Override `create_mcp_server()` to use a different server implementation. When doing so, implement your own authentication and authorization logic within that server. + +Custom workers enable enterprise features like authentication, telemetry, and integration with existing infrastructure without modifying NeMo Agent toolkit core code. diff --git a/docs/source/index.md b/docs/source/index.md index 2938caa45..c166d9ba9 100644 --- a/docs/source/index.md +++ b/docs/source/index.md @@ -121,6 +121,7 @@ Adding an Authentication Provider <./extend/adding-an-authentication-provider.md Integrating AWS Bedrock Models <./extend/integrating-aws-bedrock-models.md> Cursor Rules Developer Guide <./extend/cursor-rules-developer-guide.md> Adding a Telemetry Exporter <./extend/telemetry-exporters.md> +Adding a Custom MCP Server Worker <./extend/mcp-server.md> ``` ```{toctree} diff --git a/docs/source/workflows/mcp/mcp-server.md b/docs/source/workflows/mcp/mcp-server.md index 214da4fc1..9c8bfdbd6 100644 --- a/docs/source/workflows/mcp/mcp-server.md +++ b/docs/source/workflows/mcp/mcp-server.md @@ -59,6 +59,25 @@ nat mcp serve --config_file examples/getting_started/simple_calculator/configs/c --tool_names calculator ``` +### Mounting at Custom Paths +By default, the MCP server is available at the root path (such as `http://localhost:9901/mcp`). You can mount the server at a custom base path by setting `base_path` in your configuration file: + +```yaml +general: + front_end: + _type: mcp + name: "my_server" + base_path: "/api/v1" +``` + +With this configuration, the MCP server will be accessible at `http://localhost:9901/api/v1/mcp`. This is useful when deploying MCP servers that need to be mounted at specific paths for reverse proxy configurations or service mesh architectures. + +The `base_path` must start with a forward slash (`/`) and must not end with a forward slash (`/`). + +:::{note} +The `base_path` feature requires the `streamable-http` transport. SSE transport does not support custom base paths. +::: + ## 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`). diff --git a/src/nat/front_ends/mcp/mcp_front_end_config.py b/src/nat/front_ends/mcp/mcp_front_end_config.py index 140061850..8bd6629d0 100644 --- a/src/nat/front_ends/mcp/mcp_front_end_config.py +++ b/src/nat/front_ends/mcp/mcp_front_end_config.py @@ -17,6 +17,7 @@ from typing import Literal from pydantic import Field +from pydantic import field_validator from pydantic import model_validator from nat.authentication.oauth2.oauth2_resource_server_config import OAuth2ResourceServerConfig @@ -47,10 +48,25 @@ class MCPFrontEndConfig(FrontEndBaseConfig, name="mcp"): description="Transport type for the MCP server (default: streamable-http, backwards compatible with sse)") runner_class: str | None = Field( default=None, description="Custom worker class for handling MCP routes (default: built-in worker)") + base_path: str | None = Field(default=None, + description="Base path to mount the MCP server at (e.g., '/api/v1'). " + "If specified, the server will be accessible at http://host:port{base_path}/mcp. " + "If None, server runs at root path /mcp.") server_auth: OAuth2ResourceServerConfig | None = Field( default=None, description=("OAuth 2.0 Resource Server configuration for token verification.")) + @field_validator('base_path') + @classmethod + def validate_base_path(cls, v: str | None) -> str | None: + """Validate that base_path starts with '/' and doesn't end with '/'.""" + if v is not None: + if not v.startswith('/'): + raise ValueError("base_path must start with '/'") + if v.endswith('/'): + raise ValueError("base_path must not end with '/'") + return v + # Memory profiling configuration enable_memory_profiling: bool = Field(default=False, description="Enable memory profiling and diagnostics (default: False)") diff --git a/src/nat/front_ends/mcp/mcp_front_end_plugin.py b/src/nat/front_ends/mcp/mcp_front_end_plugin.py index c414462e4..11a67964b 100644 --- a/src/nat/front_ends/mcp/mcp_front_end_plugin.py +++ b/src/nat/front_ends/mcp/mcp_front_end_plugin.py @@ -16,12 +16,14 @@ import logging import typing -from nat.authentication.oauth2.oauth2_resource_server_config import OAuth2ResourceServerConfig from nat.builder.front_end import FrontEndBase from nat.builder.workflow_builder import WorkflowBuilder from nat.front_ends.mcp.mcp_front_end_config import MCPFrontEndConfig from nat.front_ends.mcp.mcp_front_end_plugin_worker import MCPFrontEndPluginWorkerBase +if typing.TYPE_CHECKING: + from mcp.server.fastmcp import FastMCP + logger = logging.getLogger(__name__) @@ -43,7 +45,7 @@ def get_worker_class_name(self) -> str: worker_class = self.get_worker_class() return f"{worker_class.__module__}.{worker_class.__qualname__}" - def _get_worker_instance(self) -> MCPFrontEndPluginWorkerBase: + def _get_worker_instance(self): """Get an instance of the worker class.""" # Import the worker class dynamically if specified in config if self.front_end_config.runner_class: @@ -56,61 +58,94 @@ def _get_worker_instance(self) -> MCPFrontEndPluginWorkerBase: return worker_class(self.full_config) - async def _create_token_verifier(self, token_verifier_config: OAuth2ResourceServerConfig): - """Create a token verifier based on configuration.""" - from nat.front_ends.mcp.introspection_token_verifier import IntrospectionTokenVerifier - - if not self.front_end_config.server_auth: - return None - - return IntrospectionTokenVerifier(token_verifier_config) - async def run(self) -> None: """Run the MCP server.""" - # Import FastMCP - from mcp.server.fastmcp import FastMCP - - # Create auth settings and token verifier if auth is required - auth_settings = None - token_verifier = None - # Build the workflow and add routes using the worker async with WorkflowBuilder.from_config(config=self.full_config) as builder: - if self.front_end_config.server_auth: - from mcp.server.auth.settings import AuthSettings - from pydantic import AnyHttpUrl - - server_url = f"http://{self.front_end_config.host}:{self.front_end_config.port}" - - auth_settings = AuthSettings(issuer_url=AnyHttpUrl(self.front_end_config.server_auth.issuer_url), - required_scopes=self.front_end_config.server_auth.scopes, - resource_server_url=AnyHttpUrl(server_url)) - - token_verifier = await self._create_token_verifier(self.front_end_config.server_auth) - - # Create an MCP server with the configured parameters - mcp = FastMCP(name=self.front_end_config.name, - host=self.front_end_config.host, - port=self.front_end_config.port, - debug=self.front_end_config.debug, - auth=auth_settings, - token_verifier=token_verifier) - - # Get the worker instance and set up routes + # Get the worker instance worker = self._get_worker_instance() + # Let the worker create the MCP 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) # Start the MCP server with configurable transport # streamable-http is the default, but users can choose sse if preferred try: - if self.front_end_config.transport == "sse": + # If base_path is configured, mount server at sub-path using FastAPI wrapper + 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 MCP server with SSE endpoint at /sse") + await mcp.run_sse_async() + 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 MCP 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("MCP server URL: %s", full_url) + await self._run_with_mount(mcp) + # Standard behavior - run at root path + elif self.front_end_config.transport == "sse": logger.info("Starting MCP server with SSE endpoint at /sse") await mcp.run_sse_async() else: # streamable-http - logger.info("Starting MCP server with streamable-http endpoint at /mcp/") + full_url = f"http://{self.front_end_config.host}:{self.front_end_config.port}/mcp" + logger.info("MCP server URL: %s", full_url) await mcp.run_streamable_http_async() except KeyboardInterrupt: logger.info("MCP server shutdown requested (Ctrl+C). Shutting down gracefully.") + + async def _run_with_mount(self, mcp: "FastMCP") -> None: + """Run MCP server mounted at configured base_path using FastAPI wrapper. + + Args: + mcp: The FastMCP server instance to mount + """ + import contextlib + + import uvicorn + from fastapi import FastAPI + + @contextlib.asynccontextmanager + async def lifespan(_app: FastAPI): + """Manage MCP server session lifecycle.""" + logger.info("Starting MCP server session manager...") + async with contextlib.AsyncExitStack() as stack: + try: + # Initialize the MCP server's session manager + await stack.enter_async_context(mcp.session_manager.run()) + logger.info("MCP server session manager started successfully") + yield + except Exception as e: + logger.error("Failed to start MCP server session manager: %s", e) + raise + logger.info("MCP server session manager stopped") + + # Create a FastAPI wrapper app with lifespan management + app = FastAPI( + title=self.front_end_config.name, + description="MCP server mounted at custom base path", + lifespan=lifespan, + ) + + # Mount the MCP server's ASGI app at the configured base_path + app.mount(self.front_end_config.base_path, mcp.streamable_http_app()) + + # Configure and start uvicorn server + config = uvicorn.Config( + app, + host=self.front_end_config.host, + port=self.front_end_config.port, + log_level=self.front_end_config.log_level.lower(), + ) + server = uvicorn.Server(config) + await server.serve() diff --git a/src/nat/front_ends/mcp/mcp_front_end_plugin_worker.py b/src/nat/front_ends/mcp/mcp_front_end_plugin_worker.py index e8d9eb914..450ee2b70 100644 --- a/src/nat/front_ends/mcp/mcp_front_end_plugin_worker.py +++ b/src/nat/front_ends/mcp/mcp_front_end_plugin_worker.py @@ -35,7 +35,12 @@ class MCPFrontEndPluginWorkerBase(ABC): - """Base class for MCP front end plugin workers.""" + """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. + """ def __init__(self, config: Config): """Initialize the MCP worker with configuration. @@ -83,15 +88,86 @@ async def health_check(_request: Request): }, status_code=503) + @abstractmethod + async def create_mcp_server(self) -> FastMCP: + """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 async def add_routes(self, mcp: FastMCP, builder: WorkflowBuilder): """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 def _default_add_routes(self, mcp: FastMCP, 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 (WorkflowBuilder): The workflow builder instance + builder: The workflow builder instance """ - pass + from nat.front_ends.mcp.tool_converter import register_function_with_mcp + + # Set up the health endpoint + self._setup_health_endpoint(mcp) + + # Build the workflow and register all functions with MCP + workflow = await builder.build() + + # Get all functions from the workflow + functions = await self._get_all_functions(workflow) + + # Filter functions based on tool_names if provided + if self.front_end_config.tool_names: + logger.info("Filtering functions based on tool_names: %s", self.front_end_config.tool_names) + filtered_functions: dict[str, Function] = {} + for function_name, function in functions.items(): + if function_name in self.front_end_config.tool_names: + # Treat current tool_names as function names, so check if the function name is in the list + filtered_functions[function_name] = function + elif any(function_name.startswith(f"{group_name}.") for group_name in self.front_end_config.tool_names): + # Treat tool_names as function group names, so check if the function name starts with the group name + filtered_functions[function_name] = function + else: + logger.debug("Skipping function %s as it's not in tool_names", function_name) + functions = filtered_functions + + # Register each function with MCP, passing workflow context for observability + for function_name, function in functions.items(): + register_function_with_mcp(mcp, function_name, function, workflow, self.memory_profiler) + + # Add a simple fallback function if no functions were found + if not functions: + raise RuntimeError("No functions found in workflow. Please check your configuration.") + + # After registration, expose debug endpoints for tool/schema inspection + self._setup_debug_endpoints(mcp, functions) async def _get_all_functions(self, workflow: Workflow) -> dict[str, Function]: """Get all functions from the workflow. @@ -225,48 +301,62 @@ async def get_memory_stats(_request: Request): class MCPFrontEndPluginWorker(MCPFrontEndPluginWorkerBase): - """Default MCP front end plugin worker implementation.""" + """Default MCP server worker implementation. - async def add_routes(self, mcp: FastMCP, builder: WorkflowBuilder): - """Add default routes to the MCP server. + 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. - Args: - mcp: The FastMCP server instance - builder (WorkflowBuilder): The workflow builder instance + 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) + """ + + async def create_mcp_server(self) -> FastMCP: + """Create default MCP server with optional authentication. + + Returns: + FastMCP instance configured with settings from NAT config """ - from nat.front_ends.mcp.tool_converter import register_function_with_mcp + # Handle auth if configured + auth_settings = None + token_verifier = None - # Set up the health endpoint - self._setup_health_endpoint(mcp) + if self.front_end_config.server_auth: + from mcp.server.auth.settings import AuthSettings + from pydantic import AnyHttpUrl - # Build the workflow and register all functions with MCP - workflow = await builder.build() + server_url = f"http://{self.front_end_config.host}:{self.front_end_config.port}" + auth_settings = AuthSettings(issuer_url=AnyHttpUrl(self.front_end_config.server_auth.issuer_url), + required_scopes=self.front_end_config.server_auth.scopes, + resource_server_url=AnyHttpUrl(server_url)) - # Get all functions from the workflow - functions = await self._get_all_functions(workflow) + # Create token verifier + from nat.front_ends.mcp.introspection_token_verifier import IntrospectionTokenVerifier - # Filter functions based on tool_names if provided - if self.front_end_config.tool_names: - logger.info("Filtering functions based on tool_names: %s", self.front_end_config.tool_names) - filtered_functions: dict[str, Function] = {} - for function_name, function in functions.items(): - if function_name in self.front_end_config.tool_names: - # Treat current tool_names as function names, so check if the function name is in the list - filtered_functions[function_name] = function - elif any(function_name.startswith(f"{group_name}.") for group_name in self.front_end_config.tool_names): - # Treat tool_names as function group names, so check if the function name starts with the group name - filtered_functions[function_name] = function - else: - logger.debug("Skipping function %s as it's not in tool_names", function_name) - functions = filtered_functions + token_verifier = IntrospectionTokenVerifier(self.front_end_config.server_auth) - # Register each function with MCP, passing workflow context for observability - for function_name, function in functions.items(): - register_function_with_mcp(mcp, function_name, function, workflow, self.memory_profiler) + return FastMCP(name=self.front_end_config.name, + host=self.front_end_config.host, + port=self.front_end_config.port, + debug=self.front_end_config.debug, + auth=auth_settings, + token_verifier=token_verifier) - # Add a simple fallback function if no functions were found - if not functions: - raise RuntimeError("No functions found in workflow. Please check your configuration.") + async def add_routes(self, mcp: FastMCP, builder: WorkflowBuilder): + """Add default routes to the MCP server. - # After registration, expose debug endpoints for tool/schema inspection - self._setup_debug_endpoints(mcp, functions) + Args: + mcp: The FastMCP server instance + builder: The workflow builder instance + """ + # Use the default implementation from base class to add the tools to the MCP server + await self._default_add_routes(mcp, builder)