nat.plugins.mcp.server.tool_converter#

Attributes#

logger

_USE_PYDANTIC_DEFAULT

Functions#

_use_pydantic_default(→ object)

Mark an omitted FastMCP argument for removal before model validation.

_build_mcp_field_annotation(→ Any)

Build a field annotation for FastMCP's intermediate argument model.

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

is_field_optional(→ tuple[bool, Any])

Determine if a Pydantic field is optional and extract its default value for MCP signatures.

create_function_wrapper(function_name, ...[, ...])

Create a wrapper function that exposes a NAT Function as an MCP tool using SessionManager.

get_function_description(→ str)

Retrieve a human-readable description for a NAT function or workflow.

register_function_with_mcp(→ None)

Register a NAT Function as an MCP tool using SessionManager.

Module Contents#

logger#
_USE_PYDANTIC_DEFAULT#
_use_pydantic_default() → object#

Mark an omitted FastMCP argument for removal before model validation.

_build_mcp_field_annotation(field: pydantic.fields.FieldInfo) → Any#

Build a field annotation for FastMCP’s intermediate argument model.

FastMCP materializes default-factory values before calling the wrapper. Use a private marker factory in that intermediate model so the declared schema remains responsible for applying its original factory and tracking whether the caller supplied the field.

_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:

  1. Replaces non-alphanumeric/underscore characters with _.

  2. Prepends _ when the result starts with a digit.

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

is_field_optional(
field: pydantic.fields.FieldInfo,
) → tuple[bool, Any]#

Determine if a Pydantic field is optional and extract its default value for MCP signatures.

For MCP tool signatures, we need to distinguish:

  • Required fields: marked with Parameter.empty

  • Optional with concrete default: use that default

  • Optional with factory: omit the signature default; the annotated field uses an internal marker so the declared schema can apply its factory later

Args:

field: The Pydantic FieldInfo to check

Returns:

A tuple of (is_optional, default_value): - (False, Parameter.empty) for required fields - (True, actual_default) for optional fields with explicit defaults - (True, Parameter.empty) for optional fields with default_factory

create_function_wrapper(
function_name: str,
session_manager: nat.runtime.session.SessionManager,
schema: type[pydantic.BaseModel],
memory_profiler: MemoryProfiler | None = None,
)#

Create a wrapper function that exposes a NAT Function as an MCP tool using SessionManager.

Here SessionManager.run() which is used to create a Runner that automatically handles observability (emits intermediate step events, starts exporters, etc).

Args:

function_name (str): The name of the function/tool session_manager (SessionManager): SessionManager wrapping the function/workflow schema (type[BaseModel]): The input schema of the function memory_profiler: Optional memory profiler to track requests

Returns:

A wrapper function suitable for registration with MCP

get_function_description(
function: nat.builder.function_base.FunctionBase,
) → str#

Retrieve a human-readable description for a NAT function or workflow.

The description is determined using the following precedence:

  1. If the function is a Workflow and has a ‘description’ attribute, use it.

  2. If the Workflow’s config has a ‘description’, use it.

  3. If the Workflow’s config has a ‘topic’, use it.

  4. If the function is a regular Function, use its ‘description’ attribute.

Args:

function: The NAT FunctionBase instance (Function or Workflow).

Returns:

The best available description string for the function.

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

Register a NAT Function as an MCP tool using SessionManager.

Each function is wrapped in a SessionManager so that all calls go through Runner that automatically handles observability.

Args:

mcp: The FastMCP instance function_name: The name to register the function under session_manager: SessionManager wrapping the function/workflow memory_profiler: Optional memory profiler to track requests