nat.plugins.mcp.cli.commands#

Attributes#

logger

MCPError

Classes#

MCPPingResult

Result of an MCP server ping request.

Functions#

mcp_command()

MCP-related commands.

validate_transport_cli_args(→ bool)

Validate transport and parameter combinations, returning False if invalid.

format_tool(→ dict[str, str | None])

Format an MCP tool into a dictionary for display.

print_tool(→ None)

Print a formatted tool to the console with optional detailed information.

_set_auth_defaults(→ tuple[str | None, str | None, ...)

Set default auth values when --auth flag is used.

_create_mcp_client_config(builder, server_cfg, url, ...)

_create_bearer_token_auth_config(builder, server_cfg, ...)

Create bearer token auth configuration for CLI usage.

list_tools_via_function_group(→ list[dict[str, ...)

List tools by constructing the mcp_client function group and introspecting functions.

list_tools_direct(command, url[, tool_name, ...])

List MCP tools using direct MCP protocol with structured exception handling.

ping_mcp_server(→ MCPPingResult)

Ping an MCP server to check if it's responsive.

mcp_client_command()

MCP client commands.

mcp_client_tool_group()

MCP client tool commands.

mcp_client_tool_list(ctx, direct, url, transport, ...)

List MCP tool names (default) or show detailed tool information.

mcp_client_ping(→ None)

Ping an MCP server to check if it's responsive.

call_tool_direct(→ str)

Call an MCP tool directly via the selected transport.

call_tool_and_print(→ str)

Call an MCP tool either directly or via the function group and return output.

mcp_client_tool_call(→ None)

Call an MCP tool by name with optional JSON arguments.

Module Contents#

logger#
mcp_command()#

MCP-related commands.

MCPError#
validate_transport_cli_args(
transport: str,
command: str | None,
args: str | None,
env: str | None,
) → bool#

Validate transport and parameter combinations, returning False if invalid.

Args:

transport: The transport type (‘sse’, ‘stdio’, or ‘streamable-http’) command: Command for stdio transport args: Arguments for stdio transport env: Environment variables for stdio transport

Returns:

bool: True if valid, False if invalid (error message already displayed)

class MCPPingResult(/, **data: Any)#

Bases: pydantic.BaseModel

Result of an MCP server ping request.

Attributes:

url (str): The MCP server URL that was pinged status (str): Health status - ‘healthy’, ‘unhealthy’, or ‘unknown’ response_time_ms (float | None): Response time in milliseconds, None if request failed completely error (str | None): Error message if the ping failed, None if successful

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

url: str#
status: str#
response_time_ms: float | None#
error: str | None#
format_tool(tool: Any) → dict[str, str | None]#

Format an MCP tool into a dictionary for display.

Extracts name, description, and input schema from various MCP tool object types and normalizes them into a consistent dictionary format for CLI display.

Args:

tool (Any): MCPToolClient or raw MCP Tool object (uses Any due to different types)

Returns:

dict[str, str | None]: Dictionary with name, description, and input_schema as keys

print_tool(
tool_dict: dict[str, str | None],
detail: bool = False,
) → None#

Print a formatted tool to the console with optional detailed information.

Outputs tool information in a user-friendly format to stdout. When detail=True or when description/schema are available, shows full information with separator.

Args:

tool_dict (dict[str, str | None]): Dictionary containing tool information with name, description, and input_schema as keys detail (bool, optional): Whether to force detailed output. Defaults to False.

_set_auth_defaults(
auth: bool,
url: str | None,
auth_redirect_uri: str | None,
auth_user_id: str | None,
auth_scopes: str | None,
) → tuple[str | None, str | None, list[str] | None]#

Set default auth values when –auth flag is used.

Args:

auth: Whether –auth flag was used url: MCP server URL auth_redirect_uri: OAuth2 redirect URI auth_user_id: User ID for authentication auth_scopes: OAuth2 scopes (comma-separated string)

Returns:

Tuple of (auth_redirect_uri, auth_user_id, auth_scopes_list) with defaults applied

async _create_mcp_client_config(
builder,
server_cfg,
url: str | None,
transport: str,
auth_redirect_uri: str | None,
auth_user_id: str | None,
auth_scopes: list[str] | None,
)#
async _create_bearer_token_auth_config(
builder,
server_cfg,
bearer_token: str | None,
bearer_token_env: str | None,
)#

Create bearer token auth configuration for CLI usage.

async list_tools_via_function_group(
command: str | None,
url: str | None,
tool_name: str | None = None,
transport: str = 'sse',
args: list[str] | None = None,
env: dict[str, str] | None = None,
auth_redirect_uri: str | None = None,
auth_user_id: str | None = None,
auth_scopes: list[str] | None = None,
) → list[dict[str, str | None]]#

List tools by constructing the mcp_client function group and introspecting functions.

Mirrors the behavior of list_mcp.py but routes through the registered function group to ensure parity with workflow configuration.

async list_tools_direct(
command,
url,
tool_name=None,
transport='sse',
args=None,
env=None,
)#

List MCP tools using direct MCP protocol with structured exception handling.

Bypasses MCPBuilder and uses raw MCP ClientSession and SSE client directly. Converts raw exceptions to structured MCPErrors for consistent user experience. Used when –direct flag is specified in CLI.

Args:

url (str): MCP server URL to connect to tool_name (str | None, optional): Specific tool name to retrieve. If None, retrieves all available tools. Defaults to None.

Returns:

list[dict[str, str | None]]: List of formatted tool dictionaries, each containing name, description, and input_schema as keys

Note:

This function handles ExceptionGroup by extracting the most relevant exception and converting it to MCPError for consistent error reporting.

async ping_mcp_server(
url: str,
timeout: int,
transport: str = 'streamable-http',
command: str | None = None,
args: list[str] | None = None,
env: dict[str, str] | None = None,
auth_redirect_uri: str | None = None,
auth_user_id: str | None = None,
auth_scopes: list[str] | None = None,
) → MCPPingResult#

Ping an MCP server to check if it’s responsive.

Args:

url (str): MCP server URL to ping timeout (int): Timeout in seconds for the ping request

Returns:

MCPPingResult: Structured result with status, response_time, and any error info

mcp_client_command()#

MCP client commands.

mcp_client_tool_group()#

MCP client tool commands.

mcp_client_tool_list(
ctx,
direct,
url,
transport,
command,
args,
env,
tool,
detail,
json_output,
auth,
auth_redirect_uri,
auth_user_id,
auth_scopes,
)#

List MCP tool names (default) or show detailed tool information.

Use –detail for full output including descriptions and input schemas. If –tool is provided, always shows full output for that specific tool. Use –direct to bypass MCPBuilder and use raw MCP protocol. Use –json-output to get structured JSON data instead of formatted text. Use –auth to enable auth with default settings (streamable-http only, not with –direct). Use –auth-redirect-uri to enable auth for protected MCP servers (streamable-http only, not with –direct).

Args:

ctx (click.Context): Click context object for command invocation direct (bool): Whether to bypass MCPBuilder and use direct MCP protocol url (str): MCP server URL to connect to (default: http://localhost:9901/mcp) tool (str | None): Optional specific tool name to retrieve detailed info for detail (bool): Whether to show full details (description + schema) for all tools json_output (bool): Whether to output tool metadata in JSON format instead of text auth (bool): Whether to enable OAuth2 authentication (streamable-http only, not with –direct) auth_redirect_uri (str | None): redirect URI for auth (streamable-http only, not with –direct) auth_user_id (str | None): User ID for authentication (streamable-http only, not with –direct) auth_scopes (str | None): OAuth2 scopes (comma-separated, streamable-http only, not with –direct)

Examples:

nat mcp client tool list # List tool names only nat mcp client tool list –detail # Show all tools with full details nat mcp client tool list –tool my_tool # Show details for specific tool nat mcp client tool list –json-output # Get JSON format output nat mcp client tool list –direct –url http://… # Use direct protocol with custom URL (no auth) nat mcp client tool list –url https://example.com/mcp/ –auth # With auth using defaults nat mcp client tool list –url https://example.com/mcp/ –transport streamable-http –auth-redirect-uri http://localhost:8000/auth/redirect # With custom auth settings nat mcp client tool list –url https://example.com/mcp/ –transport streamable-http –auth-redirect-uri http://localhost:8000/auth/redirect –auth-user-id myuser # With auth and user ID

mcp_client_ping(
url: str,
transport: str,
command: str | None,
args: str | None,
env: str | None,
timeout: int,
json_output: bool,
auth_redirect_uri: str | None,
auth_user_id: str | None,
auth_scopes: str | None,
) → None#

Ping an MCP server to check if it’s responsive.

This command sends a ping request to the MCP server and measures the response time. It’s useful for health checks and monitoring server availability.

Args:

url (str): MCP server URL to ping (default: http://localhost:9901/mcp) timeout (int): Timeout in seconds for the ping request (default: 60) json_output (bool): Whether to output the result in JSON format auth_redirect_uri (str | None): redirect URI for auth (streamable-http only, not with –direct) auth_user_id (str | None): User ID for auth (streamable-http only, not with –direct) auth_scopes (str | None): OAuth2 scopes (comma-separated, streamable-http only, not with –direct)

Examples:

nat mcp client ping # Ping default server nat mcp client ping –url http://custom-server:9901/mcp # Ping custom server nat mcp client ping –timeout 10 # Use 10 second timeout nat mcp client ping –json-output # Get JSON format output nat mcp client ping –url https://example.com/mcp/ –transport streamable-http –auth # With auth

async call_tool_direct(
command: str | None,
url: str | None,
tool_name: str,
transport: str,
args: list[str] | None,
env: dict[str, str] | None,
tool_args: dict[str, Any] | None,
) → str#

Call an MCP tool directly via the selected transport.

Bypasses the WorkflowBuilder and talks to the MCP server using the raw protocol client for the given transport. Aggregates tool outputs into a plain string suitable for terminal display. Converts transport/protocol exceptions into a structured MCPError for consistency.

Args:

command (str | None): For stdio transport, the command to execute. url (str | None): For sse or streamable-http transports, the server URL. tool_name (str): Name of the tool to call. transport (str): One of 'stdio', 'sse', or 'streamable-http'. args (list[str] | None): For stdio transport, additional command arguments. env (dict[str, str] | None): For stdio transport, environment variables. tool_args (dict[str, Any] | None): JSON-serializable arguments passed to the tool.

Returns:

str: Concatenated textual output from the tool invocation.

Raises:
MCPError: If the connection, initialization, or tool call fails. When the

MCP client package is not installed, a generic Exception is raised with an MCP-like error message.

RuntimeError: If required parameters for the chosen transport are missing

or if the tool returns an error response.

async call_tool_and_print(
command: str | None,
url: str | None,
tool_name: str,
transport: str,
args: list[str] | None,
env: dict[str, str] | None,
tool_args: dict[str, Any] | None,
direct: bool,
auth_redirect_uri: str | None = None,
auth_user_id: str | None = None,
auth_scopes: list[str] | None = None,
bearer_token: str | None = None,
bearer_token_env: str | None = None,
) → str#

Call an MCP tool either directly or via the function group and return output.

When direct is True, uses the raw MCP protocol client (bypassing the builder). Otherwise, constructs the mcp_client function group and invokes the corresponding function, mirroring workflow configuration.

Args:

command (str | None): For stdio transport, the command to execute. url (str | None): For sse or streamable-http transports, the server URL. tool_name (str): Name of the tool to call. transport (str): One of 'stdio', 'sse', or 'streamable-http'. args (list[str] | None): For stdio transport, additional command arguments. env (dict[str, str] | None): For stdio transport, environment variables. tool_args (dict[str, Any] | None): JSON-serializable arguments passed to the tool. direct (bool): If True, bypass WorkflowBuilder and use direct MCP client.

Returns:

str: Stringified tool output suitable for terminal display. May be an empty string when the MCP client package is not installed and direct is False.

Raises:

RuntimeError: If the tool is not found when using the function group. MCPError: Propagated from call_tool_direct when direct mode fails.

mcp_client_tool_call(
tool_name: str,
direct: bool,
url: str | None,
transport: str,
command: str | None,
args: str | None,
env: str | None,
json_args: str | None,
auth: bool,
auth_redirect_uri: str | None,
auth_user_id: str | None,
auth_scopes: str | None,
bearer_token: str | None,
bearer_token_env: str | None,
) → None#

Call an MCP tool by name with optional JSON arguments.

Validates transport parameters, parses --json-args into a dictionary, invokes the tool (either directly or via the function group), and prints the resulting output to stdout. Errors are formatted consistently with other MCP CLI commands.

Args:

tool_name (str): Name of the tool to call. direct (bool): If True, bypass WorkflowBuilder and use the direct MCP client. url (str | None): For sse or streamable-http transports, the server URL. transport (str): One of 'stdio', 'sse', or 'streamable-http'. command (str | None): For stdio transport, the command to execute. args (str | None): For stdio transport, space-separated command arguments. env (str | None): For stdio transport, space-separated KEY=VALUE pairs. json_args (str | None): JSON object string with tool arguments (e.g. ‘{“q”: “hello”}’). auth_redirect_uri (str | None): redirect URI for auth (streamable-http only, not with –direct) auth_user_id (str | None): User ID for authentication (streamable-http only, not with –direct) auth_scopes (str | None): OAuth2 scopes (comma-separated, streamable-http only, not with –direct)

Examples:

nat mcp client tool call echo –json-args ‘{“text”: “Hello”}’ nat mcp client tool call search –direct –url http://localhost:9901/mcp –json-args ‘{“query”: “NVIDIA”}’ # Direct mode (no auth) nat mcp client tool call run –transport stdio –command mcp-server –args “–flag1 –flag2” –env “ENV1=V1 ENV2=V2” –json-args ‘{}’ nat mcp client tool call search –url https://example.com/mcp/ –auth –json-args ‘{“query”: “test”}’ # With auth using defaults nat mcp client tool call search –url https://example.com/mcp/ –transport streamable-http –json-args ‘{“query”: “test”}’ –auth