nat.plugins.fastmcp.server.tool_converter#

Convert NeMo Agent Toolkit functions to FastMCP tools.

Attributes#

`logger`
`_USE_PYDANTIC_DEFAULT`

Functions#

`_sanitize_parameter_name`(→ str)	Sanitize a JSON schema property name into a valid Python identifier.
`_build_name_mapping`(→ dict[str, str])	Map each original schema field name to a unique, valid Python identifier.
`_safe_json_schema`(→ dict[str, Any])	Return a JSON schema for Pydantic models or dict-like schemas.
`_get_field_default`(→ Any)	Return field default or a sentinel to skip default.
`_build_signature_from_schema`(...)	Build a function signature from a Pydantic schema if possible.
`_build_input_schema`(→ Any)	Return an input schema for tool registration.
`_build_annotations_from_schema`(→ dict[str, Any])	Build function annotations from a Pydantic schema if possible.
`_is_chat_request_schema`(→ bool)	Return True when the schema represents a ChatRequest.
`create_function_wrapper`(function_name, ...)	Create a wrapper function for MCP that invokes the workflow via `SessionManager`.
`get_function_description`(→ str \| None)	Retrieve a human-readable description for a NAT function or workflow.
`register_function_with_mcp`(→ None)	Register a NeMo Agent Toolkit function as a FastMCP tool.
`format_schema_for_display`(→ str)	Return a pretty JSON schema string for debug endpoints.

Module Contents#

logger#

_USE_PYDANTIC_DEFAULT#

_sanitize_parameter_name(name: str) → str#

Sanitize a JSON schema property name into a valid Python identifier.

The MCP specification places no restriction on property names, but Python’s inspect.Parameter requires valid identifiers. This function:

Replaces non-alphanumeric/underscore characters with _.
Prepends _ when the result starts with a digit.
Appends _ when the result is a Python keyword (PEP 8 convention).

Args:: name: The original property name from the MCP tool schema.
Returns:: A valid Python identifier derived from name.

_build_name_mapping(field_names: list[str]) → dict[str, str]#

Map each original schema field name to a unique, valid Python identifier.

Handles collisions by appending numeric suffixes when multiple original names sanitize to the same Python identifier (e.g. from and from_ both sanitize to from_; the second one becomes from_2).

Names that are already valid Python identifiers map to themselves.

Args:: field_names: Original field names from the Pydantic model.
Returns:: A dict mapping {original_name: safe_name} for every field.

_safe_json_schema(schema: Any) → dict[str, Any]#: Return a JSON schema for Pydantic models or dict-like schemas.

_get_field_default(field_info: pydantic.fields.FieldInfo) → Any#: Return field default or a sentinel to skip default.

_build_signature_from_schema( schema: Any, ) → tuple[inspect.Signature, dict[str, str]]#

Build a function signature from a Pydantic schema if possible.

Returns:: A tuple of (Signature, alias_map) where alias_map maps sanitized parameter names back to original schema names for names that changed.

_build_input_schema(schema: Any) → Any#: Return an input schema for tool registration.

_build_annotations_from_schema(schema: Any) → dict[str, Any]#: Build function annotations from a Pydantic schema if possible.

_is_chat_request_schema(schema: Any) → bool#: Return True when the schema represents a ChatRequest.

create_function_wrapper( function_name: str, session_manager: nat.runtime.session.SessionManager, input_schema: Any, )#

Create a wrapper function for MCP that invokes the workflow via SessionManager.

Args:: function_name: The name of the function to register. session_manager: The session manager for the workflow. input_schema: Input schema for the workflow/function.

get_function_description( function: nat.builder.function_base.FunctionBase | None, ) → str | None#: Retrieve a human-readable description for a NAT function or workflow.

register_function_with_mcp( mcp: fastmcp.FastMCP, function_name: str, session_manager: nat.runtime.session.SessionManager, function: nat.builder.function_base.FunctionBase | None = None, ) → None#

Register a NeMo Agent Toolkit function as a FastMCP tool.

Each function is wrapped in a SessionManager so that all calls go through the runner, which automatically handles observability.

Args:: mcp: The FastMCP instance. function_name: The name to register the function under. session_manager: SessionManager wrapping the function/workflow. function: Optional function metadata (for description/schema).

format_schema_for_display(schema: Any) → str#: Return a pretty JSON schema string for debug endpoints.