nat.plugins.mcp.server.tool_converter#

Attributes#

logger

_USE_PYDANTIC_DEFAULT

Functions#

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#
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: use sentinel so Pydantic can apply the 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, _USE_PYDANTIC_DEFAULT) 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