nat.plugins.a365.tooling.register

Registration for A365 tooling integration with NAT MCP client.

Attributes

logger
_AUTH_HTTP_STATUSES

Classes

A365MCPToolingFunctionGroup

Composite FunctionGroup that aggregates functions from multiple MCP servers.

Functions

_server_display_name(→ str)	Return a deterministic, human-readable name for an MCP server.
_raise_classified_discovery_error(→ NoReturn)	Classify a discovery error as auth vs. SDK and raise the matching A365 exception.
_handle_server_registration_error(→ None)	Apply the configured on_server_registration_error policy.
a365_mcp_tooling_function_group(config, builder)	Register MCP servers discovered from A365 tooling as NAT function groups.

Module Contents

logger

_AUTH_HTTP_STATUSES

_server_display_name(server: object) → str

Return a deterministic, human-readable name for an MCP server.

Falls back to unknown:<hostname> when the gateway response omits mcp_server_name so two unnamed servers can be told apart in logs and server_auth_providers lookups. Prevents the “two servers, both called ‘unknown’, both eligible for the same override” footgun.

_raise_classified_discovery_error(error: Exception) → NoReturn

Classify a discovery error as auth vs. SDK and raise the matching A365 exception.

Preference order:

1. aiohttp.ClientResponseError with status 401/403 -> A365AuthenticationError.

2. Substring match on the error message (“authentication” / “unauthorized” / “forbidden”) -> A365AuthenticationError (covers SDK wrappers that hide the underlying HTTP status from us).

3. Everything else -> A365SDKError with a WARN that lets ops detect when the substring matcher is missing real failure types in production.

_handle_server_registration_error(

*,

server_name: str,

error: Exception,

policy: nat.plugins.a365.tooling.tooling_config.ServerRegistrationErrorPolicy,

skipped_servers: list[tuple[str, str]],

) → None

Apply the configured on_server_registration_error policy.

fail_fast raises A365SDKError so the caller’s async with aborts. The two tolerant policies record (server_name, error_repr) into skipped_servers so the composite group exposes which servers were lost.

class A365MCPToolingFunctionGroup(

config: nat.plugins.a365.tooling.tooling_config.A365MCPToolingConfig,

mcp_groups: list[nat.builder.function.FunctionGroup],

skipped_servers: list[tuple[str, str]] | None = None,

)

Bases: nat.builder.function.FunctionGroup

Composite FunctionGroup that aggregates functions from multiple MCP servers.

Instead of merging functions into a single group, this class delegates to multiple MCP FunctionGroups and aggregates their results. This preserves the original function bindings and avoids double-wrapping.

Skipped-server metadata: skipped_servers records (name, error_repr) tuples for any servers that failed registration under the skip_with_warning / skip_silently policies. Use it for monitoring or health-check surfaces (/health endpoints, dashboards) so quiet degradation is observable.

Initialize the composite function group.

Args:

config: The A365 MCP tooling configuration mcp_groups: List of MCP FunctionGroups to aggregate skipped_servers: (name, error_repr) tuples recorded under tolerant policies; empty under fail_fast.

_mcp_groups

skipped_servers: list[tuple[str, str]] = []

static _aggregate(

mcp_groups: list[nat.builder.function.FunctionGroup],

method_name: str,

filter_fn: collections.abc.Callable[[collections.abc.Sequence[str]], collections.abc.Awaitable[collections.abc.Sequence[str]]] | None,

) → dict[str, nat.builder.function.Function]

Async:

Merge method_name outputs across MCP groups, warning on name collisions.

Function names from MCP groups are already namespaced (e.g. mcp_client__tool_name), so collisions are unusual – they typically indicate the same tool_overrides.alias was applied to two servers that each expose the original tool name. When that happens, the later definition wins (preserving prior dict-update behavior) but a WARN surfaces the ambiguity so the operator can disambiguate via per-server overrides.

async get_all_functions(

filter_fn: collections.abc.Callable[[collections.abc.Sequence[str]], collections.abc.Awaitable[collections.abc.Sequence[str]]] | None = None,

) → dict[str, nat.builder.function.Function]

Aggregate all functions from all MCP groups (collision-aware).

async get_accessible_functions(

filter_fn: collections.abc.Callable[[collections.abc.Sequence[str]], collections.abc.Awaitable[collections.abc.Sequence[str]]] | None = None,

) → dict[str, nat.builder.function.Function]

Aggregate accessible functions from all MCP groups (collision-aware).

async get_included_functions(

filter_fn: collections.abc.Callable[[collections.abc.Sequence[str]], collections.abc.Awaitable[collections.abc.Sequence[str]]] | None = None,

) → dict[str, nat.builder.function.Function]

Aggregate included functions from all MCP groups (collision-aware).

async get_excluded_functions(

filter_fn: collections.abc.Callable[[collections.abc.Sequence[str]], collections.abc.Awaitable[collections.abc.Sequence[str]]] | None = None,

) → dict[str, nat.builder.function.Function]

Aggregate excluded functions from all MCP groups (collision-aware).

async a365_mcp_tooling_function_group(

config: nat.plugins.a365.tooling.tooling_config.A365MCPToolingConfig,

builder: nat.builder.builder.Builder,

)

Register MCP servers discovered from A365 tooling as NAT function groups.

This function:

1. Uses A365 tooling service to discover configured MCP servers

2. Creates MCP client function groups for each discovered server

3. Returns a composite function group containing all discovered tools

Args:

config: A365MCPToolingConfig with agent ID and auth token builder: NAT Builder instance

Returns:

FunctionGroup containing tools from all discovered MCP servers

Raises:

OptionalImportError: If nvidia-nat-mcp package is not installed A365AuthenticationError: If authentication fails when resolving auth token or discovering servers A365SDKError: If MCP server discovery fails