diff --git a/docs/clients/cli.mdx b/docs/clients/cli.mdx
new file mode 100644
index 000000000..b12996631
--- /dev/null
+++ b/docs/clients/cli.mdx
@@ -0,0 +1,126 @@
+---
+title: Client CLI
+sidebarTitle: CLI
+description: Query and invoke MCP server tools directly from the terminal with fastmcp list and fastmcp call.
+icon: terminal
+---
+
+import { VersionBadge } from '/snippets/version-badge.mdx'
+
+
+
+MCP servers are designed for programmatic consumption by AI assistants and applications. But during development, you often want to poke at a server directly: check what tools it exposes, call one with test arguments, or verify that a deployment is responding correctly. The FastMCP CLI gives you that direct access with two commands, `fastmcp list` and `fastmcp call`, so you can query and invoke any MCP server without writing a single line of Python.
+
+These commands are also valuable for LLM-based agents that lack native MCP support. An agent that can execute shell commands can use `fastmcp list --json` to discover available tools and `fastmcp call --json` to invoke them, with structured JSON output designed for programmatic consumption.
+
+## Server Targets
+
+Both commands need to know which server to talk to. You provide a "server spec" as the first argument, and FastMCP figures out the transport automatically. You can point at an HTTP URL for a running server, a Python file that defines one, a JSON configuration file that describes one, or a JavaScript file. The CLI resolves the right connection mechanism so you can focus on the query.
+
+```bash
+fastmcp list http://localhost:8000/mcp
+fastmcp list server.py
+fastmcp list mcp-config.json
+```
+
+Python files are handled with particular care. Rather than requiring your script to call `mcp.run()` at the bottom, the CLI routes it through `fastmcp run` internally, which means any Python file that defines a FastMCP server object works as a target with no boilerplate.
+
+For servers that communicate over stdio (common with Node.js-based MCP servers), use the `--command` flag instead of a positional server spec. The string is shell-split into a command and arguments.
+
+```bash
+fastmcp list --command 'npx -y @modelcontextprotocol/server-github'
+```
+
+## Discovering Tools
+
+`fastmcp list` connects to a server and prints every tool it exposes. The default output is compact: each tool appears as a function signature with its parameter names, types, and a description.
+
+```bash
+fastmcp list http://localhost:8000/mcp
+```
+
+The output looks like a Python function signature, making it easy to see at a glance what a tool expects and what it returns. Required parameters appear with just their type annotation, while optional ones show their defaults.
+
+When you need the full JSON Schema for a tool's inputs or outputs -- useful for understanding nested object structures or enum constraints -- opt into them with `--input-schema` or `--output-schema`. These print the raw schema beneath each tool signature.
+
+### Beyond Tools
+
+MCP servers can expose resources and prompts alongside tools. By default, `fastmcp list` only shows tools because they are the most common interaction point. Add `--resources` or `--prompts` to include those in the output.
+
+```bash
+fastmcp list server.py --resources --prompts
+```
+
+Resources appear with their URIs and descriptions. Prompts appear with their argument names so you can see what parameters they accept.
+
+### Machine-Readable Output
+
+The `--json` flag switches from human-friendly text to structured JSON. Each tool includes its name, description, and full input schema (and output schema when present). When combined with `--resources` or `--prompts`, those are included as additional top-level keys.
+
+```bash
+fastmcp list server.py --json
+```
+
+This is the format to use when building automation around MCP servers or feeding tool definitions to an LLM agent that needs to decide which tool to call.
+
+## Calling Tools
+
+`fastmcp call` invokes a single tool on a server. You provide the server spec, the tool name, and arguments as `key=value` pairs. The CLI fetches the tool's schema, coerces your string values to the correct types (integers, floats, booleans, arrays, objects), and makes the call.
+
+```bash
+fastmcp call http://localhost:8000/mcp search query=hello limit=5
+```
+
+Type coercion is driven by the tool's JSON Schema. If a parameter is declared as an integer, the string `"5"` becomes the integer `5`. Booleans accept `true`/`false`, `yes`/`no`, and `1`/`0`. Array and object parameters are parsed as JSON.
+
+For tools with complex or deeply nested arguments, the `key=value` syntax gets unwieldy. You can pass a single JSON object as the argument instead, and the CLI treats it as the full input dictionary.
+
+```bash
+fastmcp call server.py create_item '{"name": "Widget", "tags": ["sale", "new"], "metadata": {"color": "blue"}}'
+```
+
+Alternatively, `--input-json` provides the base argument dictionary. Any `key=value` pairs you add alongside it override keys from the JSON, which is useful for templating a complex call and varying one parameter at a time.
+
+### Error Handling
+
+The CLI validates your call before sending it. If you misspell a tool name, it uses fuzzy matching to suggest corrections. If you omit a required argument, it tells you which ones are missing and prints the tool's signature as a reminder.
+
+When a tool call itself returns an error (the server executed the tool but it failed), the error message is printed and the CLI exits with a non-zero status code, making it straightforward to use in scripts.
+
+### Structured Output
+
+Like `fastmcp list`, the `--json` flag on `fastmcp call` emits structured JSON instead of formatted text. The output includes the content blocks, error status, and structured content when the server provides it. Use this when you need to parse tool results programmatically.
+
+```bash
+fastmcp call server.py get_weather city=London --json
+```
+
+## Authentication
+
+When the server target is an HTTP URL, the CLI automatically enables OAuth authentication. If the server requires it, you will be guided through the OAuth flow (typically opening a browser for authorization). If the server has no auth requirements, the OAuth setup is a silent no-op.
+
+To explicitly disable authentication -- for example, when connecting to a local development server where OAuth setup would just slow you down -- pass `--auth none`.
+
+```bash
+fastmcp call http://localhost:8000/mcp my_tool --auth none
+```
+
+## Transport Override
+
+FastMCP defaults to Streamable HTTP for URL targets. If you are connecting to a server that only supports Server-Sent Events (SSE), use `--transport sse` to force the older transport. This appends `/sse` to the URL path automatically so the client picks the correct protocol.
+
+```bash
+fastmcp list http://localhost:8000 --transport sse
+```
+
+## Interactive Elicitation
+
+Some MCP tools request additional input from the user during execution through a mechanism called elicitation. When a tool sends an elicitation request, the CLI prints the server's question to the terminal and prompts you to respond. Each field in the elicitation schema is presented with its name and expected type, and required fields are clearly marked.
+
+You can type `decline` to skip a question or `cancel` to abort the tool call entirely. This interactive behavior means the CLI works naturally with tools that have multi-step or conversational workflows.
+
+## LLM Agent Integration
+
+For LLM agents that can execute shell commands but lack built-in MCP support, the CLI provides a clean integration path. The agent calls `fastmcp list --json` to get a structured description of every available tool, including full input schemas, and then calls `fastmcp call --json` with the chosen tool and arguments. Both commands return well-formed JSON that is straightforward to parse.
+
+Because the CLI handles connection management, transport selection, and type coercion internally, the agent does not need to understand MCP protocol details. It just needs to read JSON and construct shell commands.
diff --git a/docs/development/v3-notes/v3-features.mdx b/docs/development/v3-notes/v3-features.mdx
index 974d34009..45a9e647a 100644
--- a/docs/development/v3-notes/v3-features.mdx
+++ b/docs/development/v3-notes/v3-features.mdx
@@ -6,6 +6,33 @@ This document tracks major features in FastMCP v3.0 for release notes preparatio
## 3.0.0beta2
+### CLI: `fastmcp list` and `fastmcp call`
+
+New client-side CLI commands for querying and invoking tools on any MCP server — remote URLs, local Python files, MCPConfig JSON, or arbitrary stdio commands. Especially useful for giving LLMs that don't have built-in MCP support access to MCP tools via shell commands.
+
+```bash
+# Discover tools on a server
+fastmcp list http://localhost:8000/mcp
+fastmcp list server.py
+fastmcp list --command 'npx -y @modelcontextprotocol/server-github'
+
+# Call a tool
+fastmcp call server.py greet name=World
+fastmcp call http://localhost:8000/mcp search query=hello limit=5
+fastmcp call server.py create_item '{"name": "Widget", "tags": ["a", "b"]}'
+```
+
+Key features:
+- Tool arguments are auto-coerced using the tool's JSON schema (`limit=5` → int)
+- Single JSON objects work as positional args alongside `key=value` and `--input-json`
+- `--input-schema` / `--output-schema` for full JSON schemas, `--json` for machine-readable output
+- `--transport sse` for SSE servers, `--command` for stdio servers
+- Auto OAuth for HTTP targets (no-ops if server doesn't require auth)
+- Fuzzy tool name matching suggests alternatives on typos
+- Interactive terminal elicitation for tools that request user input mid-execution
+
+Documentation: [Client CLI](/clients/cli)
+
### CLI: Expanded Reload File Watching
The `--reload` flag now watches a comprehensive set of file types, making it suitable for MCP apps with frontend bundles ([#3028](https://github.com/jlowin/fastmcp/pull/3028)). Previously limited to `.py` files, it now watches JavaScript, TypeScript, HTML, CSS, config files, and media assets.
diff --git a/docs/docs.json b/docs/docs.json
index 70b6d2173..de6369d10 100644
--- a/docs/docs.json
+++ b/docs/docs.json
@@ -168,6 +168,7 @@
"group": "Clients",
"pages": [
"clients/client",
+ "clients/cli",
"clients/transports",
{
"group": "Core Operations",
diff --git a/docs/patterns/cli.mdx b/docs/patterns/cli.mdx
index 8662e26e5..72badb870 100644
--- a/docs/patterns/cli.mdx
+++ b/docs/patterns/cli.mdx
@@ -18,6 +18,8 @@ fastmcp --help
| Command | Purpose | Dependency Management |
| ------- | ------- | --------------------- |
+| `list` | List tools on any MCP server | **Supports:** URLs, local files, MCPConfig JSON, stdio commands. **Deps:** N/A (connects to existing servers) |
+| `call` | Call a tool on any MCP server | **Supports:** URLs, local files, MCPConfig JSON, stdio commands. **Deps:** N/A (connects to existing servers) |
| `run` | Run a FastMCP server directly | **Supports:** Local files, factory functions, URLs, fastmcp.json configs, MCP configs. **Deps:** Uses your local environment directly. With `--python`, `--with`, `--project`, or `--with-requirements`: Runs via `uv run` subprocess. With fastmcp.json: Automatically manages dependencies based on configuration |
| `dev` | Run a server with the MCP Inspector for testing | **Supports:** Local files and fastmcp.json configs. **Deps:** Always runs via `uv run` subprocess (never uses your local environment); dependencies must be specified or available in a uv-managed project. With fastmcp.json: Uses configured dependencies |
| `install` | Install a server in MCP client applications | **Supports:** Local files and fastmcp.json configs. **Deps:** Creates an isolated environment; dependencies must be explicitly specified with `--with` and/or `--with-editable`. With fastmcp.json: Uses configured dependencies |
@@ -25,6 +27,138 @@ fastmcp --help
| `project prepare` | Create a persistent uv project from fastmcp.json environment config | **Supports:** fastmcp.json configs only. **Deps:** Creates a uv project directory with all dependencies pre-installed for reuse with `--project` flag |
| `version` | Display version information | N/A |
+## `fastmcp list`
+
+List tools available on any MCP server. This works with remote URLs, local Python files, MCPConfig JSON files, and arbitrary stdio commands. Together with `fastmcp call`, these commands are especially useful for giving LLMs that don't have built-in MCP support access to MCP tools via shell commands.
+
+```bash
+fastmcp list http://localhost:8000/mcp
+fastmcp list server.py
+fastmcp list mcp.json
+fastmcp list --command 'npx -y @modelcontextprotocol/server-github'
+```
+
+By default, the output shows each tool's signature and description. Use `--input-schema` or `--output-schema` to include full JSON schemas, or `--json` for machine-readable output.
+
+### Options
+
+| Option | Flag | Description |
+| ------ | ---- | ----------- |
+| Command | `--command` | Connect to a stdio server command (e.g. `'npx -y @mcp/server'`) |
+| Transport | `--transport`, `-t` | Force transport type for URL targets (`http` or `sse`) |
+| Resources | `--resources` | Also list resources |
+| Prompts | `--prompts` | Also list prompts |
+| Input Schema | `--input-schema` | Show full input schemas |
+| Output Schema | `--output-schema` | Show full output schemas |
+| JSON | `--json` | Output as JSON |
+| Timeout | `--timeout` | Connection timeout in seconds |
+| Auth | `--auth` | Auth method: `oauth` (default for HTTP), a bearer token, or `none` to disable |
+
+### Server Targets
+
+The `` argument accepts:
+
+1. **URLs** — `http://` or `https://` endpoints. Uses Streamable HTTP by default; pass `--transport sse` for SSE servers.
+2. **Python files** — `.py` files are run via `fastmcp run` automatically.
+3. **MCPConfig JSON** — `.json` files with an `mcpServers` key are treated as multi-server configs.
+4. **Stdio commands** — Use `--command` to connect to any MCP server via stdio (e.g. `npx`, `uvx`).
+
+### Examples
+
+```bash
+# List tools on a remote server
+fastmcp list http://localhost:8000/mcp
+
+# List tools from a local Python file
+fastmcp list server.py
+
+# Include full input schemas
+fastmcp list server.py --input-schema
+
+# Machine-readable JSON
+fastmcp list server.py --json
+
+# SSE server
+fastmcp list http://localhost:8000/mcp --transport sse
+
+# Stdio command
+fastmcp list --command 'npx -y @modelcontextprotocol/server-github'
+
+# Include resources and prompts
+fastmcp list server.py --resources --prompts
+```
+
+## `fastmcp call`
+
+Call a tool on any MCP server. Arguments can be passed as `key=value` pairs, a single JSON object, or via `--input-json`.
+
+```bash
+fastmcp call server.py greet name=World
+fastmcp call http://localhost:8000/mcp search query=hello limit=5
+fastmcp call server.py create_item '{"name": "x", "tags": ["a", "b"]}'
+```
+
+Tool arguments are automatically coerced to the correct type based on the tool's input schema — string values like `limit=5` become integers when the schema expects one.
+
+### Options
+
+| Option | Flag | Description |
+| ------ | ---- | ----------- |
+| Command | `--command` | Connect to a stdio server command (e.g. `'npx -y @mcp/server'`) |
+| Transport | `--transport`, `-t` | Force transport type for URL targets (`http` or `sse`) |
+| Input JSON | `--input-json` | JSON string of tool arguments (merged with key=value args) |
+| JSON | `--json` | Output raw JSON result |
+| Timeout | `--timeout` | Connection timeout in seconds |
+| Auth | `--auth` | Auth method: `oauth` (default for HTTP), a bearer token, or `none` to disable |
+
+### Argument Passing
+
+There are three ways to pass arguments:
+
+**Key=value pairs** are the simplest for flat arguments. Values are coerced using the tool's JSON schema (strings become ints, bools, etc.):
+
+```bash
+fastmcp call server.py search query=hello limit=5 verbose=true
+```
+
+**A single JSON object** works when you have structured or nested arguments:
+
+```bash
+fastmcp call server.py create_item '{"name": "Widget", "tags": ["new", "sale"]}'
+```
+
+**`--input-json`** provides a base dict that key=value pairs can override:
+
+```bash
+fastmcp call server.py search --input-json '{"query": "hello", "limit": 5}' limit=10
+```
+
+### Examples
+
+```bash
+# Call a tool with simple args
+fastmcp call server.py greet name=World
+
+# Call with JSON object
+fastmcp call server.py create '{"name": "x", "tags": ["a"]}'
+
+# Get JSON output for scripting
+fastmcp call server.py add a=3 b=4 --json
+
+# Call a tool on a remote server
+fastmcp call http://localhost:8000/mcp search query=hello
+
+# Call via stdio command
+fastmcp call --command 'npx -y @mcp/server' tool_name arg=value
+
+# Disable OAuth for HTTP targets
+fastmcp call http://localhost:8000/mcp search query=hello --auth none
+```
+
+
+If you call a tool that doesn't exist, FastMCP will suggest similar tool names. Use `fastmcp list` to see all available tools on a server.
+
+
## `fastmcp run`
Run a FastMCP server directly or proxy a remote server.
diff --git a/examples/elicitation.py b/examples/elicitation.py
new file mode 100644
index 000000000..368980ea1
--- /dev/null
+++ b/examples/elicitation.py
@@ -0,0 +1,47 @@
+"""
+FastMCP Elicitation Example
+
+Demonstrates tools that ask users for input during execution.
+
+Try it with the CLI:
+
+ fastmcp list examples/elicitation.py
+ fastmcp call examples/elicitation.py greet
+ fastmcp call examples/elicitation.py survey
+"""
+
+from dataclasses import dataclass
+
+from fastmcp import Context, FastMCP
+
+mcp = FastMCP("Elicitation Demo")
+
+
+@mcp.tool
+async def greet(ctx: Context) -> str:
+ """Greet the user by name (asks for their name)."""
+ result = await ctx.elicit("What is your name?", response_type=str)
+
+ if result.action == "accept":
+ return f"Hello, {result.data}!"
+ return "Maybe next time!"
+
+
+@mcp.tool
+async def survey(ctx: Context) -> str:
+ """Run a short survey collecting structured info."""
+
+ @dataclass
+ class SurveyResponse:
+ favorite_color: str
+ lucky_number: int
+
+ result = await ctx.elicit(
+ "Quick survey — tell us about yourself:",
+ response_type=SurveyResponse,
+ )
+
+ if result.action == "accept":
+ resp = result.data
+ return f"Got it — you like {resp.favorite_color} and your lucky number is {resp.lucky_number}."
+ return "Survey skipped."
diff --git a/skills/fastmcp-client-cli/SKILL.md b/skills/fastmcp-client-cli/SKILL.md
new file mode 100644
index 000000000..9742fa5cb
--- /dev/null
+++ b/skills/fastmcp-client-cli/SKILL.md
@@ -0,0 +1,93 @@
+---
+name: fastmcp-client-cli
+description: Query and invoke tools on MCP servers using fastmcp list and fastmcp call. Use when you need to discover what tools a server offers, call tools, or integrate MCP servers into workflows.
+---
+
+# FastMCP CLI: List and Call
+
+Use `fastmcp list` and `fastmcp call` to interact with any MCP server from the command line.
+
+## Listing Tools
+
+```bash
+# Remote server
+fastmcp list http://localhost:8000/mcp
+
+# Local Python file (runs via fastmcp run automatically)
+fastmcp list server.py
+
+# MCPConfig with multiple servers
+fastmcp list mcp.json
+
+# Stdio command (npx, uvx, etc.)
+fastmcp list --command 'npx -y @modelcontextprotocol/server-github'
+
+# Include full input/output schemas
+fastmcp list server.py --input-schema --output-schema
+
+# Machine-readable JSON
+fastmcp list server.py --json
+
+# Include resources and prompts
+fastmcp list server.py --resources --prompts
+```
+
+Default output shows tool signatures and descriptions. Use `--input-schema` or `--output-schema` to include full JSON schemas, `--json` for structured output.
+
+## Calling Tools
+
+```bash
+# Key=value arguments (auto-coerced to correct types)
+fastmcp call server.py greet name=World
+fastmcp call server.py add a=3 b=4
+
+# Single JSON object for complex/nested args
+fastmcp call server.py create_item '{"name": "Widget", "tags": ["a", "b"]}'
+
+# --input-json with key=value overrides
+fastmcp call server.py search --input-json '{"query": "hello", "limit": 5}' limit=10
+
+# JSON output for scripting
+fastmcp call server.py add a=3 b=4 --json
+```
+
+Type coercion is automatic: `limit=5` becomes an integer, `verbose=true` becomes a boolean, based on the tool's input schema.
+
+## Server Targets
+
+All commands accept the same server targets:
+
+| Target | Example |
+|--------|---------|
+| HTTP/HTTPS URL | `http://localhost:8000/mcp` |
+| Python file | `server.py` |
+| MCPConfig JSON | `mcp.json` (must have `mcpServers` key) |
+| Stdio command | `--command 'npx -y @mcp/server'` |
+
+For SSE servers, pass `--transport sse`:
+
+```bash
+fastmcp list http://localhost:8000/mcp --transport sse
+```
+
+## Auth
+
+HTTP targets automatically use OAuth (no-ops if the server doesn't require auth). Disable with `--auth none`:
+
+```bash
+fastmcp call http://server/mcp tool --auth none
+```
+
+## Workflow Pattern
+
+Discover tools first, then call them:
+
+```bash
+# 1. See what's available
+fastmcp list server.py
+
+# 2. Call a tool
+fastmcp call server.py tool_name arg=value
+```
+
+If you call a nonexistent tool, FastMCP suggests close matches.
diff --git a/src/fastmcp/cli/cli.py b/src/fastmcp/cli/cli.py
index c8a358cea..9eb197660 100644
--- a/src/fastmcp/cli/cli.py
+++ b/src/fastmcp/cli/cli.py
@@ -19,6 +19,7 @@ from rich.table import Table
import fastmcp
from fastmcp.cli import run as run_module
+from fastmcp.cli.client import call_command, list_command
from fastmcp.cli.install import install_app
from fastmcp.cli.tasks import tasks_app
from fastmcp.utilities.cli import is_already_in_uv_subprocess, load_and_merge_config
@@ -952,6 +953,10 @@ app.command(install_app)
# Add tasks subcommand group
app.command(tasks_app)
+# Add client query commands
+app.command(list_command, name="list")
+app.command(call_command, name="call")
+
if __name__ == "__main__":
app()
diff --git a/src/fastmcp/cli/client.py b/src/fastmcp/cli/client.py
new file mode 100644
index 000000000..e517e18e4
--- /dev/null
+++ b/src/fastmcp/cli/client.py
@@ -0,0 +1,872 @@
+"""Client-side CLI commands for querying and invoking MCP servers."""
+
+import difflib
+import json
+import os
+import shlex
+import sys
+from pathlib import Path
+from typing import Annotated, Any, Literal
+
+import cyclopts
+import mcp.types
+from rich.console import Console
+
+from fastmcp.client.client import CallToolResult, Client
+from fastmcp.client.elicitation import ElicitResult
+from fastmcp.client.transports.stdio import StdioTransport
+from fastmcp.utilities.logging import get_logger
+
+logger = get_logger("cli.client")
+console = Console()
+
+
+# ---------------------------------------------------------------------------
+# Server spec resolution
+# ---------------------------------------------------------------------------
+
+_JSON_SCHEMA_TYPE_MAP: dict[str, str] = {
+ "string": "str",
+ "integer": "int",
+ "number": "float",
+ "boolean": "bool",
+ "array": "list",
+ "object": "dict",
+ "null": "None",
+}
+
+
+def resolve_server_spec(
+ server_spec: str | None,
+ *,
+ command: str | None = None,
+ transport: str | None = None,
+) -> str | dict[str, Any] | StdioTransport:
+ """Turn CLI inputs into something ``Client()`` accepts.
+
+ Exactly one of ``server_spec`` or ``command`` should be provided.
+
+ Resolution order for ``server_spec``:
+ 1. URLs (``http://``, ``https://``) — passed through as-is.
+ If ``--transport`` is ``sse``, the URL is rewritten to end with ``/sse``
+ so ``infer_transport`` picks the right transport.
+ 2. Existing file paths, or strings ending in ``.py``/``.js``/``.json``.
+ 3. Anything else — error with guidance.
+
+ When ``command`` is provided, the string is shell-split into a
+ ``StdioTransport(command, args)``.
+ """
+
+ if command is not None and server_spec is not None:
+ console.print(
+ "[bold red]Error:[/bold red] Cannot use both a server spec and --command"
+ )
+ sys.exit(1)
+
+ if command is not None:
+ return _build_stdio_from_command(command)
+
+ if server_spec is None:
+ console.print(
+ "[bold red]Error:[/bold red] Provide a server spec or use --command"
+ )
+ sys.exit(1)
+
+ assert isinstance(server_spec, str)
+ spec: str = server_spec
+
+ # 1. URL
+ if spec.startswith(("http://", "https://")):
+ if transport == "sse" and not spec.rstrip("/").endswith("/sse"):
+ spec = spec.rstrip("/") + "/sse"
+ return spec
+
+ # 2. File path (must be a file, not a directory)
+ path = Path(spec)
+ is_file = path.is_file() or (
+ not path.is_dir() and spec.endswith((".py", ".js", ".json"))
+ )
+
+ if is_file:
+ if spec.endswith(".json"):
+ return _resolve_json_spec(path)
+ if spec.endswith(".py"):
+ # Run via `fastmcp run` so scripts don't need mcp.run()
+ resolved_path = path.resolve()
+ return StdioTransport(
+ command="fastmcp",
+ args=["run", str(resolved_path), "--no-banner"],
+ log_file=Path(os.devnull),
+ )
+ # .js — pass through for Client's infer_transport
+ return spec
+
+ # 3. Unrecognised
+ console.print(
+ f"[bold red]Error:[/bold red] Could not resolve server spec: [cyan]{spec}[/cyan]\n\n"
+ "Expected one of:\n"
+ " • A URL (e.g. http://localhost:8000/mcp)\n"
+ " • A Python file (e.g. server.py)\n"
+ " • An MCPConfig (e.g. mcp.json)\n"
+ " • --command (e.g. --command 'npx -y @mcp/server')\n"
+ )
+ sys.exit(1)
+
+
+def _build_stdio_from_command(command_str: str) -> StdioTransport:
+ """Shell-split a command string into a ``StdioTransport``."""
+ try:
+ parts = shlex.split(command_str)
+ except ValueError as exc:
+ console.print(f"[bold red]Error:[/bold red] Invalid command: {exc}")
+ sys.exit(1)
+
+ if not parts:
+ console.print("[bold red]Error:[/bold red] Empty --command")
+ sys.exit(1)
+
+ return StdioTransport(command=parts[0], args=parts[1:], log_file=Path(os.devnull))
+
+
+def _resolve_json_spec(path: Path) -> str | dict[str, Any]:
+ """Disambiguate a ``.json`` server spec."""
+
+ if not path.exists():
+ console.print(
+ f"[bold red]Error:[/bold red] File not found: [cyan]{path}[/cyan]"
+ )
+ sys.exit(1)
+
+ try:
+ data = json.loads(path.read_text())
+ except json.JSONDecodeError as exc:
+ console.print(f"[bold red]Error:[/bold red] Invalid JSON in {path}: {exc}")
+ sys.exit(1)
+
+ if isinstance(data, dict) and "mcpServers" in data:
+ return data
+
+ # Likely a fastmcp.json (MCPServerConfig) — not directly usable as a client target.
+ console.print(
+ f"[bold red]Error:[/bold red] [cyan]{path}[/cyan] is a FastMCP server config, not an MCPConfig.\n"
+ f"Start the server first, then query it:\n\n"
+ f" fastmcp run {path}\n"
+ f" fastmcp list http://localhost:8000/mcp\n"
+ )
+ sys.exit(1)
+
+
+def _is_http_target(resolved: str | dict[str, Any] | StdioTransport) -> bool:
+ """Return True if the resolved target will use an HTTP-based transport.
+
+ MCPConfig dicts are excluded because ``MCPConfigTransport`` manages
+ individual server transports internally and does not support top-level auth.
+ """
+ if isinstance(resolved, str):
+ return resolved.startswith(("http://", "https://"))
+ return False
+
+
+async def _terminal_elicitation_handler(
+ message: str,
+ response_type: type[Any] | None,
+ params: Any,
+ context: Any,
+) -> ElicitResult[dict[str, Any]]:
+ """Prompt the user on the terminal for elicitation responses.
+
+ Prints the server's message and prompts for each field in the schema.
+ The user can type 'decline' or 'cancel' instead of a value to abort.
+ """
+ from mcp.types import ElicitRequestFormParams
+
+ console.print(f"\n[bold yellow]Server asks:[/bold yellow] {message}")
+
+ if not isinstance(params, ElicitRequestFormParams):
+ answer = console.input(
+ "[dim](press Enter to accept, or type 'decline'):[/dim] "
+ )
+ if answer.strip().lower() == "decline":
+ return ElicitResult(action="decline")
+ if answer.strip().lower() == "cancel":
+ return ElicitResult(action="cancel")
+ return ElicitResult(action="accept", content={})
+
+ schema = params.requestedSchema
+ properties = schema.get("properties", {})
+ required = set(schema.get("required", []))
+
+ if not properties:
+ answer = console.input(
+ "[dim](press Enter to accept, or type 'decline'):[/dim] "
+ )
+ if answer.strip().lower() == "decline":
+ return ElicitResult(action="decline")
+ if answer.strip().lower() == "cancel":
+ return ElicitResult(action="cancel")
+ return ElicitResult(action="accept", content={})
+
+ result: dict[str, Any] = {}
+ for field_name, field_schema in properties.items():
+ type_hint = field_schema.get("type", "string")
+ req_marker = " [red]*[/red]" if field_name in required else ""
+ prompt_text = f" [cyan]{field_name}[/cyan] ({type_hint}){req_marker}: "
+
+ raw = console.input(prompt_text)
+ if raw.strip().lower() == "decline":
+ return ElicitResult(action="decline")
+ if raw.strip().lower() == "cancel":
+ return ElicitResult(action="cancel")
+
+ if raw == "" and field_name not in required:
+ continue
+
+ result[field_name] = coerce_value(raw, field_schema)
+
+ return ElicitResult(action="accept", content=result)
+
+
+def _build_client(
+ resolved: str | dict[str, Any] | StdioTransport,
+ *,
+ timeout: float | None = None,
+ auth: str | None = None,
+) -> Client:
+ """Build a ``Client`` from a resolved server spec.
+
+ Applies ``auth='oauth'`` automatically for HTTP-based targets unless
+ the caller explicitly passes ``--auth none`` to disable it.
+
+ ``auth=None`` means "not specified" (use default), ``auth="none"``
+ means "explicitly disabled".
+ """
+ if auth == "none":
+ effective_auth: str | None = None
+ elif auth is not None:
+ effective_auth = auth
+ elif _is_http_target(resolved):
+ effective_auth = "oauth"
+ else:
+ effective_auth = None
+
+ return Client(
+ resolved,
+ timeout=timeout,
+ auth=effective_auth,
+ elicitation_handler=_terminal_elicitation_handler,
+ )
+
+
+# ---------------------------------------------------------------------------
+# Argument coercion
+# ---------------------------------------------------------------------------
+
+
+def coerce_value(raw: str, schema: dict[str, Any]) -> Any:
+ """Coerce a string CLI value according to a JSON-Schema type hint."""
+
+ schema_type = schema.get("type", "string")
+
+ if schema_type == "integer":
+ try:
+ return int(raw)
+ except ValueError:
+ raise ValueError(f"Expected integer, got {raw!r}") from None
+
+ if schema_type == "number":
+ try:
+ return float(raw)
+ except ValueError:
+ raise ValueError(f"Expected number, got {raw!r}") from None
+
+ if schema_type == "boolean":
+ if raw.lower() in ("true", "1", "yes"):
+ return True
+ if raw.lower() in ("false", "0", "no"):
+ return False
+ raise ValueError(f"Expected boolean, got {raw!r}")
+
+ if schema_type in ("array", "object"):
+ try:
+ return json.loads(raw)
+ except json.JSONDecodeError:
+ raise ValueError(f"Expected JSON {schema_type}, got {raw!r}") from None
+
+ # Default: treat as string
+ return raw
+
+
+def parse_tool_arguments(
+ raw_args: tuple[str, ...],
+ input_json: str | None,
+ input_schema: dict[str, Any],
+) -> dict[str, Any]:
+ """Build a tool-call argument dict from CLI inputs.
+
+ A single JSON object argument is treated as the full argument dict.
+ ``--input-json`` provides the base dict; ``key=value`` pairs override.
+ Values are coerced using the tool's ``inputSchema``.
+ """
+
+ # A single positional arg that looks like JSON → treat as input-json
+ if len(raw_args) == 1 and raw_args[0].startswith("{") and input_json is None:
+ input_json = raw_args[0]
+ raw_args = ()
+
+ result: dict[str, Any] = {}
+
+ if input_json is not None:
+ try:
+ parsed = json.loads(input_json)
+ except json.JSONDecodeError as exc:
+ console.print(f"[bold red]Error:[/bold red] Invalid --input-json: {exc}")
+ sys.exit(1)
+ if not isinstance(parsed, dict):
+ console.print(
+ "[bold red]Error:[/bold red] --input-json must be a JSON object"
+ )
+ sys.exit(1)
+ result.update(parsed)
+
+ properties = input_schema.get("properties", {})
+
+ for arg in raw_args:
+ if "=" not in arg:
+ console.print(
+ f"[bold red]Error:[/bold red] Invalid argument [cyan]{arg}[/cyan] — expected key=value"
+ )
+ sys.exit(1)
+ key, value = arg.split("=", 1)
+ prop_schema = properties.get(key, {})
+ try:
+ result[key] = coerce_value(value, prop_schema)
+ except ValueError as exc:
+ console.print(
+ f"[bold red]Error:[/bold red] Argument [cyan]{key}[/cyan]: {exc}"
+ )
+ sys.exit(1)
+
+ return result
+
+
+# ---------------------------------------------------------------------------
+# Tool signature formatting
+# ---------------------------------------------------------------------------
+
+
+def _json_schema_type_to_str(schema: dict[str, Any]) -> str:
+ """Produce a short Python-style type string from a JSON-Schema fragment."""
+
+ if "anyOf" in schema:
+ parts = [_json_schema_type_to_str(s) for s in schema["anyOf"]]
+ return " | ".join(parts)
+
+ schema_type = schema.get("type", "any")
+ if isinstance(schema_type, list):
+ return " | ".join(_JSON_SCHEMA_TYPE_MAP.get(t, t) for t in schema_type)
+
+ return _JSON_SCHEMA_TYPE_MAP.get(schema_type, schema_type)
+
+
+def format_tool_signature(tool: mcp.types.Tool) -> str:
+ """Build ``name(param: type, ...) -> return_type`` from a tool's JSON schemas."""
+
+ params: list[str] = []
+ schema = tool.inputSchema
+ properties = schema.get("properties", {})
+ required = set(schema.get("required", []))
+
+ for prop_name, prop_schema in properties.items():
+ type_str = _json_schema_type_to_str(prop_schema)
+ if prop_name in required:
+ params.append(f"{prop_name}: {type_str}")
+ else:
+ default = prop_schema.get("default")
+ default_repr = repr(default) if default is not None else "..."
+ params.append(f"{prop_name}: {type_str} = {default_repr}")
+
+ sig = f"{tool.name}({', '.join(params)})"
+
+ if tool.outputSchema:
+ ret = _json_schema_type_to_str(tool.outputSchema)
+ sig += f" -> {ret}"
+
+ return sig
+
+
+# ---------------------------------------------------------------------------
+# Output formatting
+# ---------------------------------------------------------------------------
+
+
+def _print_schema(label: str, schema: dict[str, Any]) -> None:
+ """Print a JSON schema with a label."""
+ properties = schema.get("properties", {})
+ if not properties:
+ return
+ console.print(f" [dim]{label}: {json.dumps(schema)}[/dim]")
+
+
+def _format_call_result_text(result: CallToolResult) -> None:
+ """Pretty-print a tool call result to the console."""
+
+ if result.is_error:
+ for block in result.content:
+ if isinstance(block, mcp.types.TextContent):
+ console.print(f"[bold red]Error:[/bold red] {block.text}")
+ else:
+ console.print(f"[bold red]Error:[/bold red] {block}")
+ return
+
+ if result.structured_content is not None:
+ console.print_json(json.dumps(result.structured_content))
+ return
+
+ for block in result.content:
+ if isinstance(block, mcp.types.TextContent):
+ console.print(block.text)
+ elif isinstance(block, mcp.types.ImageContent):
+ size = len(block.data) * 3 // 4 # rough decoded size
+ console.print(f"[dim][Image: {block.mimeType}, ~{size} bytes][/dim]")
+ elif isinstance(block, mcp.types.AudioContent):
+ size = len(block.data) * 3 // 4
+ console.print(f"[dim][Audio: {block.mimeType}, ~{size} bytes][/dim]")
+ else:
+ console.print(str(block))
+
+
+def _content_block_to_dict(block: mcp.types.ContentBlock) -> dict[str, Any]:
+ """Serialize a single content block to a JSON-safe dict."""
+ if isinstance(block, mcp.types.TextContent):
+ return {"type": "text", "text": block.text}
+ if isinstance(block, mcp.types.ImageContent):
+ return {"type": "image", "mimeType": block.mimeType, "data": block.data}
+ if isinstance(block, mcp.types.AudioContent):
+ return {"type": "audio", "mimeType": block.mimeType, "data": block.data}
+ return {"type": "unknown", "value": str(block)}
+
+
+def _call_result_to_dict(result: CallToolResult) -> dict[str, Any]:
+ """Serialize a ``CallToolResult`` to a JSON-safe dict."""
+
+ content_list = [_content_block_to_dict(block) for block in result.content]
+ out: dict[str, Any] = {"content": content_list, "is_error": result.is_error}
+ if result.structured_content is not None:
+ out["structured_content"] = result.structured_content
+ return out
+
+
+def _tools_to_json(tools: list[mcp.types.Tool]) -> list[dict[str, Any]]:
+ """Serialize a list of tools to JSON-safe dicts."""
+
+ return [
+ {
+ "name": t.name,
+ "description": t.description,
+ "inputSchema": t.inputSchema,
+ **({"outputSchema": t.outputSchema} if t.outputSchema else {}),
+ }
+ for t in tools
+ ]
+
+
+# ---------------------------------------------------------------------------
+# Call handlers (tool, resource, prompt)
+# ---------------------------------------------------------------------------
+
+
+async def _handle_tool_call(
+ client: Client,
+ tool_name: str,
+ arguments: tuple[str, ...],
+ input_json: str | None,
+ json_output: bool,
+) -> None:
+ """Handle a tool call within an open client session."""
+ tools = await client.list_tools()
+ tool_map = {t.name: t for t in tools}
+
+ if tool_name not in tool_map:
+ close_matches = difflib.get_close_matches(
+ tool_name, tool_map.keys(), n=3, cutoff=0.5
+ )
+ msg = f"Tool [cyan]{tool_name}[/cyan] not found."
+ if close_matches:
+ suggestions = ", ".join(f"[cyan]{m}[/cyan]" for m in close_matches)
+ msg += f" Did you mean: {suggestions}?"
+ console.print(f"[bold red]Error:[/bold red] {msg}")
+ sys.exit(1)
+
+ tool = tool_map[tool_name]
+ parsed_args = parse_tool_arguments(arguments, input_json, tool.inputSchema)
+
+ required = set(tool.inputSchema.get("required", []))
+ provided = set(parsed_args.keys())
+ missing = required - provided
+ if missing:
+ missing_str = ", ".join(f"[cyan]{m}[/cyan]" for m in sorted(missing))
+ console.print(
+ f"[bold red]Error:[/bold red] Missing required arguments: {missing_str}"
+ )
+ console.print()
+ sig = format_tool_signature(tool)
+ console.print(f" [dim]{sig}[/dim]")
+ sys.exit(1)
+
+ result = await client.call_tool(tool_name, parsed_args, raise_on_error=False)
+
+ if json_output:
+ console.print_json(json.dumps(_call_result_to_dict(result)))
+ else:
+ _format_call_result_text(result)
+
+ if result.is_error:
+ sys.exit(1)
+
+
+async def _handle_resource(
+ client: Client,
+ uri: str,
+ json_output: bool,
+) -> None:
+ """Handle a resource read within an open client session."""
+ contents = await client.read_resource(uri)
+
+ if json_output:
+ data = []
+ for block in contents:
+ if isinstance(block, mcp.types.TextResourceContents):
+ data.append(
+ {
+ "uri": str(block.uri),
+ "mimeType": block.mimeType,
+ "text": block.text,
+ }
+ )
+ elif isinstance(block, mcp.types.BlobResourceContents):
+ data.append(
+ {
+ "uri": str(block.uri),
+ "mimeType": block.mimeType,
+ "blob": block.blob,
+ }
+ )
+ console.print_json(json.dumps(data))
+ return
+
+ for block in contents:
+ if isinstance(block, mcp.types.TextResourceContents):
+ console.print(block.text)
+ elif isinstance(block, mcp.types.BlobResourceContents):
+ size = len(block.blob) * 3 // 4
+ console.print(f"[dim][Blob: {block.mimeType}, ~{size} bytes][/dim]")
+
+
+async def _handle_prompt(
+ client: Client,
+ prompt_name: str,
+ arguments: tuple[str, ...],
+ input_json: str | None,
+ json_output: bool,
+) -> None:
+ """Handle a prompt get within an open client session."""
+ # Prompt arguments are always string->string, but we reuse
+ # parse_tool_arguments for the key=value / --input-json parsing.
+ # Pass an empty schema so values stay as strings.
+ parsed_args = parse_tool_arguments(arguments, input_json, {"type": "object"})
+
+ prompts = await client.list_prompts()
+ prompt_map = {p.name: p for p in prompts}
+
+ if prompt_name not in prompt_map:
+ close_matches = difflib.get_close_matches(
+ prompt_name, prompt_map.keys(), n=3, cutoff=0.5
+ )
+ msg = f"Prompt [cyan]{prompt_name}[/cyan] not found."
+ if close_matches:
+ suggestions = ", ".join(f"[cyan]{m}[/cyan]" for m in close_matches)
+ msg += f" Did you mean: {suggestions}?"
+ console.print(f"[bold red]Error:[/bold red] {msg}")
+ sys.exit(1)
+
+ result = await client.get_prompt(prompt_name, parsed_args or None)
+
+ if json_output:
+ data: dict[str, Any] = {}
+ if result.description:
+ data["description"] = result.description
+ data["messages"] = [
+ {
+ "role": msg.role,
+ "content": _content_block_to_dict(msg.content),
+ }
+ for msg in result.messages
+ ]
+ console.print_json(json.dumps(data))
+ return
+
+ for msg in result.messages:
+ console.print(f"[bold]{msg.role}:[/bold]")
+ if isinstance(msg.content, mcp.types.TextContent):
+ console.print(f" {msg.content.text}")
+ elif isinstance(msg.content, mcp.types.ImageContent):
+ size = len(msg.content.data) * 3 // 4
+ console.print(
+ f" [dim][Image: {msg.content.mimeType}, ~{size} bytes][/dim]"
+ )
+ else:
+ console.print(f" {msg.content}")
+ console.print()
+
+
+# ---------------------------------------------------------------------------
+# Commands
+# ---------------------------------------------------------------------------
+
+
+async def list_command(
+ server_spec: Annotated[
+ str | None,
+ cyclopts.Parameter(
+ help="Server URL, Python file, MCPConfig JSON, or .js file",
+ ),
+ ] = None,
+ *,
+ command: Annotated[
+ str | None,
+ cyclopts.Parameter(
+ "--command",
+ help="Stdio command to connect to (e.g. 'npx -y @mcp/server')",
+ ),
+ ] = None,
+ transport: Annotated[
+ Literal["http", "sse"] | None,
+ cyclopts.Parameter(
+ name=["--transport", "-t"],
+ help="Force transport type for URL targets (http or sse)",
+ ),
+ ] = None,
+ resources: Annotated[
+ bool,
+ cyclopts.Parameter("--resources", help="Also list resources"),
+ ] = False,
+ prompts: Annotated[
+ bool,
+ cyclopts.Parameter("--prompts", help="Also list prompts"),
+ ] = False,
+ input_schema: Annotated[
+ bool,
+ cyclopts.Parameter("--input-schema", help="Show full input schemas"),
+ ] = False,
+ output_schema: Annotated[
+ bool,
+ cyclopts.Parameter("--output-schema", help="Show full output schemas"),
+ ] = False,
+ json_output: Annotated[
+ bool,
+ cyclopts.Parameter("--json", help="Output as JSON"),
+ ] = False,
+ timeout: Annotated[
+ float | None,
+ cyclopts.Parameter("--timeout", help="Connection timeout in seconds"),
+ ] = None,
+ auth: Annotated[
+ str | None,
+ cyclopts.Parameter(
+ "--auth",
+ help="Auth method: 'oauth', a bearer token string, or 'none' to disable",
+ ),
+ ] = None,
+) -> None:
+ """List tools available on an MCP server.
+
+ Examples:
+ fastmcp list http://localhost:8000/mcp
+ fastmcp list server.py
+ fastmcp list mcp.json --json
+ fastmcp list --command 'npx -y @mcp/server' --resources
+ fastmcp list http://server/mcp --transport sse
+ """
+
+ resolved = resolve_server_spec(server_spec, command=command, transport=transport)
+ client = _build_client(resolved, timeout=timeout, auth=auth)
+
+ try:
+ async with client:
+ tools = await client.list_tools()
+
+ if json_output:
+ data: dict[str, Any] = {"tools": _tools_to_json(tools)}
+ if resources:
+ res = await client.list_resources()
+ data["resources"] = [
+ {
+ "uri": str(r.uri),
+ "name": r.name,
+ "description": r.description,
+ "mimeType": r.mimeType,
+ }
+ for r in res
+ ]
+ if prompts:
+ prm = await client.list_prompts()
+ data["prompts"] = [
+ {
+ "name": p.name,
+ "description": p.description,
+ "arguments": [a.model_dump() for a in (p.arguments or [])],
+ }
+ for p in prm
+ ]
+ console.print_json(json.dumps(data))
+ return
+
+ # Text output
+ if not tools:
+ console.print("[dim]No tools found.[/dim]")
+ else:
+ console.print(f"[bold]Tools ({len(tools)})[/bold]")
+ console.print()
+ for tool in tools:
+ sig = format_tool_signature(tool)
+ console.print(f" [cyan]{sig}[/cyan]")
+ if tool.description:
+ console.print(f" {tool.description}")
+ if input_schema:
+ _print_schema("Input", tool.inputSchema)
+ if output_schema and tool.outputSchema:
+ _print_schema("Output", tool.outputSchema)
+ console.print()
+
+ if resources:
+ res = await client.list_resources()
+ console.print(f"[bold]Resources ({len(res)})[/bold]")
+ console.print()
+ if not res:
+ console.print(" [dim]No resources found.[/dim]")
+ for r in res:
+ console.print(f" [cyan]{r.uri}[/cyan]")
+ desc_parts = [r.name or "", r.description or ""]
+ desc = " — ".join(p for p in desc_parts if p)
+ if desc:
+ console.print(f" {desc}")
+ console.print()
+
+ if prompts:
+ prm = await client.list_prompts()
+ console.print(f"[bold]Prompts ({len(prm)})[/bold]")
+ console.print()
+ if not prm:
+ console.print(" [dim]No prompts found.[/dim]")
+ for p in prm:
+ args_str = ""
+ if p.arguments:
+ parts = [a.name for a in p.arguments]
+ args_str = f"({', '.join(parts)})"
+ console.print(f" [cyan]{p.name}{args_str}[/cyan]")
+ if p.description:
+ console.print(f" {p.description}")
+ console.print()
+
+ except Exception as exc:
+ console.print(f"[bold red]Error:[/bold red] {exc}")
+ sys.exit(1)
+
+
+async def call_command(
+ server_spec: Annotated[
+ str | None,
+ cyclopts.Parameter(
+ help="Server URL, Python file, MCPConfig JSON, or .js file",
+ ),
+ ] = None,
+ target: Annotated[
+ str,
+ cyclopts.Parameter(
+ help="Tool name, resource URI, or prompt name (with --prompt)",
+ ),
+ ] = "",
+ *arguments: str,
+ command: Annotated[
+ str | None,
+ cyclopts.Parameter(
+ "--command",
+ help="Stdio command to connect to (e.g. 'npx -y @mcp/server')",
+ ),
+ ] = None,
+ transport: Annotated[
+ Literal["http", "sse"] | None,
+ cyclopts.Parameter(
+ name=["--transport", "-t"],
+ help="Force transport type for URL targets (http or sse)",
+ ),
+ ] = None,
+ prompt: Annotated[
+ bool,
+ cyclopts.Parameter("--prompt", help="Treat target as a prompt name"),
+ ] = False,
+ input_json: Annotated[
+ str | None,
+ cyclopts.Parameter(
+ "--input-json",
+ help="JSON string of arguments (merged with key=value args)",
+ ),
+ ] = None,
+ json_output: Annotated[
+ bool,
+ cyclopts.Parameter("--json", help="Output raw JSON result"),
+ ] = False,
+ timeout: Annotated[
+ float | None,
+ cyclopts.Parameter("--timeout", help="Connection timeout in seconds"),
+ ] = None,
+ auth: Annotated[
+ str | None,
+ cyclopts.Parameter(
+ "--auth",
+ help="Auth method: 'oauth', a bearer token string, or 'none' to disable",
+ ),
+ ] = None,
+) -> None:
+ """Call a tool, read a resource, or get a prompt on an MCP server.
+
+ By default the target is treated as a tool name. If the target
+ contains ``://`` it is treated as a resource URI. Pass ``--prompt``
+ to treat it as a prompt name.
+
+ Arguments are passed as key=value pairs. Use --input-json for complex
+ or nested arguments.
+
+ Examples:
+ fastmcp call server.py greet name=World
+ fastmcp call server.py resource://docs/readme
+ fastmcp call server.py analyze --prompt data='[1,2,3]'
+ fastmcp call http://server/mcp create --input-json '{"tags": ["a","b"]}'
+ """
+
+ if not target:
+ console.print(
+ "[bold red]Error:[/bold red] Missing target.\n\n"
+ "Usage: fastmcp call [key=value ...]\n\n"
+ " target can be a tool name, a resource URI, or a prompt name (with --prompt).\n\n"
+ "Use [cyan]fastmcp list [/cyan] to see available tools."
+ )
+ sys.exit(1)
+
+ resolved = resolve_server_spec(server_spec, command=command, transport=transport)
+ client = _build_client(resolved, timeout=timeout, auth=auth)
+
+ try:
+ async with client:
+ if prompt:
+ await _handle_prompt(client, target, arguments, input_json, json_output)
+ elif "://" in target:
+ await _handle_resource(client, target, json_output)
+ else:
+ await _handle_tool_call(
+ client, target, arguments, input_json, json_output
+ )
+
+ except Exception as exc:
+ console.print(f"[bold red]Error:[/bold red] {exc}")
+ sys.exit(1)
diff --git a/tests/cli/test_client_commands.py b/tests/cli/test_client_commands.py
new file mode 100644
index 000000000..1add45ba2
--- /dev/null
+++ b/tests/cli/test_client_commands.py
@@ -0,0 +1,557 @@
+"""Tests for fastmcp list and fastmcp call CLI commands."""
+
+import json
+from pathlib import Path
+from typing import Any
+from unittest.mock import patch
+
+import mcp.types
+import pytest
+
+from fastmcp import FastMCP
+from fastmcp.cli import client as client_module
+from fastmcp.cli.client import (
+ Client,
+ _build_client,
+ _build_stdio_from_command,
+ _format_call_result_text,
+ _is_http_target,
+ call_command,
+ coerce_value,
+ format_tool_signature,
+ list_command,
+ parse_tool_arguments,
+ resolve_server_spec,
+)
+from fastmcp.client.client import CallToolResult
+from fastmcp.client.transports.stdio import StdioTransport
+
+# ---------------------------------------------------------------------------
+# coerce_value
+# ---------------------------------------------------------------------------
+
+
+class TestCoerceValue:
+ def test_integer(self):
+ assert coerce_value("42", {"type": "integer"}) == 42
+
+ def test_integer_negative(self):
+ assert coerce_value("-7", {"type": "integer"}) == -7
+
+ def test_integer_invalid(self):
+ with pytest.raises(ValueError, match="Expected integer"):
+ coerce_value("abc", {"type": "integer"})
+
+ def test_number(self):
+ assert coerce_value("3.14", {"type": "number"}) == 3.14
+
+ def test_number_integer_value(self):
+ assert coerce_value("5", {"type": "number"}) == 5.0
+
+ def test_number_invalid(self):
+ with pytest.raises(ValueError, match="Expected number"):
+ coerce_value("xyz", {"type": "number"})
+
+ def test_boolean_true_variants(self):
+ for val in ("true", "True", "TRUE", "1", "yes"):
+ assert coerce_value(val, {"type": "boolean"}) is True
+
+ def test_boolean_false_variants(self):
+ for val in ("false", "False", "FALSE", "0", "no"):
+ assert coerce_value(val, {"type": "boolean"}) is False
+
+ def test_boolean_invalid(self):
+ with pytest.raises(ValueError, match="Expected boolean"):
+ coerce_value("maybe", {"type": "boolean"})
+
+ def test_array(self):
+ assert coerce_value("[1, 2, 3]", {"type": "array"}) == [1, 2, 3]
+
+ def test_array_invalid(self):
+ with pytest.raises(ValueError, match="Expected JSON array"):
+ coerce_value("not-json", {"type": "array"})
+
+ def test_object(self):
+ assert coerce_value('{"a": 1}', {"type": "object"}) == {"a": 1}
+
+ def test_string(self):
+ assert coerce_value("hello", {"type": "string"}) == "hello"
+
+ def test_string_default(self):
+ """Unknown or missing type treats value as string."""
+ assert coerce_value("hello", {}) == "hello"
+
+ def test_string_preserves_numeric_looking_values(self):
+ assert coerce_value("42", {"type": "string"}) == "42"
+
+
+# ---------------------------------------------------------------------------
+# parse_tool_arguments
+# ---------------------------------------------------------------------------
+
+
+class TestParseToolArguments:
+ SCHEMA: dict[str, Any] = {
+ "type": "object",
+ "properties": {
+ "query": {"type": "string"},
+ "limit": {"type": "integer"},
+ "verbose": {"type": "boolean"},
+ },
+ "required": ["query"],
+ }
+
+ def test_basic_key_value(self):
+ result = parse_tool_arguments(("query=hello", "limit=10"), None, self.SCHEMA)
+ assert result == {"query": "hello", "limit": 10}
+
+ def test_input_json_only(self):
+ result = parse_tool_arguments((), '{"query": "hello", "limit": 5}', self.SCHEMA)
+ assert result == {"query": "hello", "limit": 5}
+
+ def test_key_value_overrides_input_json(self):
+ result = parse_tool_arguments(
+ ("limit=20",), '{"query": "hello", "limit": 5}', self.SCHEMA
+ )
+ assert result == {"query": "hello", "limit": 20}
+
+ def test_value_containing_equals(self):
+ result = parse_tool_arguments(("query=a=b=c",), None, self.SCHEMA)
+ assert result == {"query": "a=b=c"}
+
+ def test_invalid_arg_format_exits(self):
+ with pytest.raises(SystemExit):
+ parse_tool_arguments(("noequalssign",), None, self.SCHEMA)
+
+ def test_invalid_input_json_exits(self):
+ with pytest.raises(SystemExit):
+ parse_tool_arguments((), "not-valid-json", self.SCHEMA)
+
+ def test_input_json_non_object_exits(self):
+ with pytest.raises(SystemExit):
+ parse_tool_arguments((), "[1,2,3]", self.SCHEMA)
+
+ def test_single_json_object_as_positional(self):
+ result = parse_tool_arguments(
+ ('{"query": "hello", "limit": 5}',), None, self.SCHEMA
+ )
+ assert result == {"query": "hello", "limit": 5}
+
+ def test_json_positional_ignored_when_input_json_set(self):
+ """When --input-json is already provided, a JSON positional arg is not special."""
+ with pytest.raises(SystemExit):
+ parse_tool_arguments(('{"limit": 99}',), '{"query": "hello"}', self.SCHEMA)
+
+ def test_coercion_error_exits(self):
+ with pytest.raises(SystemExit):
+ parse_tool_arguments(("limit=abc",), None, self.SCHEMA)
+
+
+# ---------------------------------------------------------------------------
+# format_tool_signature
+# ---------------------------------------------------------------------------
+
+
+class TestFormatToolSignature:
+ def _make_tool(
+ self,
+ name: str = "my_tool",
+ properties: dict[str, Any] | None = None,
+ required: list[str] | None = None,
+ output_schema: dict[str, Any] | None = None,
+ description: str | None = None,
+ ) -> mcp.types.Tool:
+ input_schema: dict[str, Any] = {"type": "object"}
+ if properties is not None:
+ input_schema["properties"] = properties
+ if required is not None:
+ input_schema["required"] = required
+ return mcp.types.Tool(
+ name=name,
+ description=description,
+ inputSchema=input_schema,
+ outputSchema=output_schema,
+ )
+
+ def test_no_params(self):
+ tool = self._make_tool()
+ assert format_tool_signature(tool) == "my_tool()"
+
+ def test_required_param(self):
+ tool = self._make_tool(
+ properties={"query": {"type": "string"}},
+ required=["query"],
+ )
+ assert format_tool_signature(tool) == "my_tool(query: str)"
+
+ def test_optional_param_with_default(self):
+ tool = self._make_tool(
+ properties={"limit": {"type": "integer", "default": 10}},
+ )
+ assert format_tool_signature(tool) == "my_tool(limit: int = 10)"
+
+ def test_optional_param_without_default(self):
+ tool = self._make_tool(
+ properties={"limit": {"type": "integer"}},
+ )
+ assert format_tool_signature(tool) == "my_tool(limit: int = ...)"
+
+ def test_mixed_required_and_optional(self):
+ tool = self._make_tool(
+ properties={
+ "query": {"type": "string"},
+ "limit": {"type": "integer", "default": 10},
+ },
+ required=["query"],
+ )
+ sig = format_tool_signature(tool)
+ assert sig == "my_tool(query: str, limit: int = 10)"
+
+ def test_with_output_schema(self):
+ tool = self._make_tool(
+ properties={"q": {"type": "string"}},
+ required=["q"],
+ output_schema={"type": "object"},
+ )
+ assert format_tool_signature(tool) == "my_tool(q: str) -> dict"
+
+ def test_anyof_type(self):
+ tool = self._make_tool(
+ properties={"value": {"anyOf": [{"type": "string"}, {"type": "integer"}]}},
+ required=["value"],
+ )
+ assert format_tool_signature(tool) == "my_tool(value: str | int)"
+
+
+# ---------------------------------------------------------------------------
+# resolve_server_spec
+# ---------------------------------------------------------------------------
+
+
+class TestResolveServerSpec:
+ def test_http_url(self):
+ assert (
+ resolve_server_spec("http://localhost:8000/mcp")
+ == "http://localhost:8000/mcp"
+ )
+
+ def test_https_url(self):
+ assert (
+ resolve_server_spec("https://example.com/mcp") == "https://example.com/mcp"
+ )
+
+ def test_python_file_existing(self, tmp_path: Path):
+ py_file = tmp_path / "server.py"
+ py_file.write_text("# empty")
+ result = resolve_server_spec(str(py_file))
+ assert isinstance(result, StdioTransport)
+ assert result.command == "fastmcp"
+ assert result.args == ["run", str(py_file.resolve()), "--no-banner"]
+
+ def test_json_mcp_config(self, tmp_path: Path):
+ config_file = tmp_path / "mcp.json"
+ config = {"mcpServers": {"test": {"url": "http://localhost:8000"}}}
+ config_file.write_text(json.dumps(config))
+ result = resolve_server_spec(str(config_file))
+ assert isinstance(result, dict)
+ assert "mcpServers" in result
+
+ def test_json_fastmcp_config_exits(self, tmp_path: Path):
+ config_file = tmp_path / "fastmcp.json"
+ config_file.write_text(json.dumps({"source": {"type": "file"}}))
+ with pytest.raises(SystemExit):
+ resolve_server_spec(str(config_file))
+
+ def test_json_not_found_exits(self, tmp_path: Path):
+ with pytest.raises(SystemExit):
+ resolve_server_spec(str(tmp_path / "nonexistent.json"))
+
+ def test_directory_exits(self, tmp_path: Path):
+ """Directories should not be treated as file paths."""
+ with pytest.raises(SystemExit):
+ resolve_server_spec(str(tmp_path))
+
+ def test_unrecognised_exits(self):
+ with pytest.raises(SystemExit):
+ resolve_server_spec("some_random_thing")
+
+ def test_command_returns_stdio_transport(self):
+ result = resolve_server_spec(None, command="npx -y @mcp/server")
+ assert isinstance(result, StdioTransport)
+ assert result.command == "npx"
+ assert result.args == ["-y", "@mcp/server"]
+
+ def test_command_single_word(self):
+ result = resolve_server_spec(None, command="myserver")
+ assert isinstance(result, StdioTransport)
+ assert result.command == "myserver"
+ assert result.args == []
+
+ def test_server_spec_and_command_exits(self):
+ with pytest.raises(SystemExit):
+ resolve_server_spec("http://localhost:8000", command="npx server")
+
+ def test_neither_server_spec_nor_command_exits(self):
+ with pytest.raises(SystemExit):
+ resolve_server_spec(None)
+
+ def test_transport_sse_rewrites_url(self):
+ result = resolve_server_spec("http://localhost:8000/mcp", transport="sse")
+ assert result == "http://localhost:8000/mcp/sse"
+
+ def test_transport_sse_no_duplicate_suffix(self):
+ result = resolve_server_spec("http://localhost:8000/sse", transport="sse")
+ assert result == "http://localhost:8000/sse"
+
+ def test_transport_sse_trailing_slash(self):
+ result = resolve_server_spec("http://localhost:8000/mcp/", transport="sse")
+ assert result == "http://localhost:8000/mcp/sse"
+
+ def test_transport_http_leaves_url_unchanged(self):
+ result = resolve_server_spec("http://localhost:8000/mcp", transport="http")
+ assert result == "http://localhost:8000/mcp"
+
+
+# ---------------------------------------------------------------------------
+# _build_stdio_from_command
+# ---------------------------------------------------------------------------
+
+
+class TestBuildStdioFromCommand:
+ def test_simple_command(self):
+ transport = _build_stdio_from_command("uvx my-server")
+ assert transport.command == "uvx"
+ assert transport.args == ["my-server"]
+
+ def test_quoted_args(self):
+ transport = _build_stdio_from_command("npx -y '@scope/server'")
+ assert transport.command == "npx"
+ assert transport.args == ["-y", "@scope/server"]
+
+ def test_empty_command_exits(self):
+ with pytest.raises(SystemExit):
+ _build_stdio_from_command("")
+
+ def test_invalid_shell_syntax_exits(self):
+ with pytest.raises(SystemExit):
+ _build_stdio_from_command("npx 'unterminated")
+
+
+# ---------------------------------------------------------------------------
+# _is_http_target
+# ---------------------------------------------------------------------------
+
+
+class TestIsHttpTarget:
+ def test_http_url(self):
+ assert _is_http_target("http://localhost:8000") is True
+
+ def test_https_url(self):
+ assert _is_http_target("https://example.com/mcp") is True
+
+ def test_file_path(self):
+ assert _is_http_target("/path/to/server.py") is False
+
+ def test_stdio_transport(self):
+ assert _is_http_target(StdioTransport(command="npx", args=[])) is False
+
+ def test_mcp_config_dict(self):
+ """MCPConfig dicts are not HTTP targets — auth is per-server internally."""
+ assert _is_http_target({"mcpServers": {}}) is False
+
+
+# ---------------------------------------------------------------------------
+# _build_client
+# ---------------------------------------------------------------------------
+
+
+class TestBuildClient:
+ def test_http_target_gets_oauth_by_default(self):
+ client = _build_client("http://localhost:8000/mcp")
+ # OAuth is applied during Client init via _set_auth
+ assert client.transport.auth is not None
+
+ def test_stdio_target_no_auth(self):
+ transport = StdioTransport(command="npx", args=["-y", "@mcp/server"])
+ client = _build_client(transport)
+ # Stdio transports don't support auth — no auth should be set
+ assert not hasattr(client.transport, "auth") or client.transport.auth is None
+
+ def test_explicit_auth_none_disables_oauth(self):
+ client = _build_client("http://localhost:8000/mcp", auth="none")
+ # "none" explicitly disables auth, even for HTTP targets
+ assert client.transport.auth is None
+
+ def test_mcp_config_no_auth(self):
+ """MCPConfig dicts handle auth per-server; no top-level auth applied."""
+ client = _build_client({"mcpServers": {"test": {"url": "http://localhost"}}})
+ # MCPConfigTransport doesn't support _set_auth — no crash means success
+ assert client.transport is not None
+
+
+# ---------------------------------------------------------------------------
+# Integration tests — invoke actual CLI commands via monkeypatched _build_client
+# ---------------------------------------------------------------------------
+
+
+def _build_test_server() -> FastMCP:
+ """Create a minimal FastMCP server for integration tests."""
+ server = FastMCP("TestServer")
+
+ @server.tool
+ def greet(name: str) -> str:
+ """Say hello to someone."""
+ return f"Hello, {name}!"
+
+ @server.tool
+ def add(a: int, b: int) -> int:
+ """Add two numbers."""
+ return a + b
+
+ @server.resource("test://greeting")
+ def greeting_resource() -> str:
+ """A static greeting resource."""
+ return "Hello from resource!"
+
+ @server.prompt
+ def ask(topic: str) -> str:
+ """Ask about a topic."""
+ return f"Tell me about {topic}"
+
+ return server
+
+
+@pytest.fixture()
+def _patch_client():
+ """Patch resolve_server_spec and _build_client so CLI commands use the
+ in-process test server without needing a real transport."""
+ server = _build_test_server()
+
+ def fake_resolve(server_spec: Any, **kwargs: Any) -> str:
+ return "fake"
+
+ def fake_build_client(resolved: Any, **kwargs: Any) -> Client:
+ return Client(server)
+
+ with (
+ patch.object(client_module, "resolve_server_spec", side_effect=fake_resolve),
+ patch.object(client_module, "_build_client", side_effect=fake_build_client),
+ ):
+ yield
+
+
+class TestListCommandCLI:
+ @pytest.mark.usefixtures("_patch_client")
+ async def test_list_tools(self, capsys: pytest.CaptureFixture[str]):
+ await list_command("fake://server")
+ captured = capsys.readouterr()
+ assert "greet" in captured.out
+ assert "add" in captured.out
+
+ @pytest.mark.usefixtures("_patch_client")
+ async def test_list_json(self, capsys: pytest.CaptureFixture[str]):
+ await list_command("fake://server", json_output=True)
+ captured = capsys.readouterr()
+ data = json.loads(captured.out)
+ names = {t["name"] for t in data["tools"]}
+ assert "greet" in names
+ assert "add" in names
+
+ @pytest.mark.usefixtures("_patch_client")
+ async def test_list_resources(self, capsys: pytest.CaptureFixture[str]):
+ await list_command("fake://server", resources=True)
+ captured = capsys.readouterr()
+ assert "test://greeting" in captured.out
+
+ @pytest.mark.usefixtures("_patch_client")
+ async def test_list_prompts(self, capsys: pytest.CaptureFixture[str]):
+ await list_command("fake://server", prompts=True)
+ captured = capsys.readouterr()
+ assert "ask" in captured.out
+
+
+class TestCallCommandCLI:
+ @pytest.mark.usefixtures("_patch_client")
+ async def test_call_tool(self, capsys: pytest.CaptureFixture[str]):
+ await call_command("fake://server", "greet", "name=World")
+ captured = capsys.readouterr()
+ assert "Hello, World!" in captured.out
+
+ @pytest.mark.usefixtures("_patch_client")
+ async def test_call_tool_json(self, capsys: pytest.CaptureFixture[str]):
+ await call_command("fake://server", "greet", "name=World", json_output=True)
+ captured = capsys.readouterr()
+ data = json.loads(captured.out)
+ assert data["is_error"] is False
+
+ @pytest.mark.usefixtures("_patch_client")
+ async def test_call_tool_not_found(self):
+ with pytest.raises(SystemExit):
+ await call_command("fake://server", "nonexistent")
+
+ @pytest.mark.usefixtures("_patch_client")
+ async def test_call_tool_missing_args(self):
+ with pytest.raises(SystemExit):
+ await call_command("fake://server", "greet")
+
+ @pytest.mark.usefixtures("_patch_client")
+ async def test_call_resource_by_uri(self, capsys: pytest.CaptureFixture[str]):
+ await call_command("fake://server", "test://greeting")
+ captured = capsys.readouterr()
+ assert "Hello from resource!" in captured.out
+
+ @pytest.mark.usefixtures("_patch_client")
+ async def test_call_resource_json(self, capsys: pytest.CaptureFixture[str]):
+ await call_command("fake://server", "test://greeting", json_output=True)
+ captured = capsys.readouterr()
+ data = json.loads(captured.out)
+ assert isinstance(data, list)
+ assert data[0]["text"] == "Hello from resource!"
+
+ @pytest.mark.usefixtures("_patch_client")
+ async def test_call_prompt(self, capsys: pytest.CaptureFixture[str]):
+ await call_command("fake://server", "ask", "topic=Python", prompt=True)
+ captured = capsys.readouterr()
+ assert "Python" in captured.out
+
+ @pytest.mark.usefixtures("_patch_client")
+ async def test_call_prompt_json(self, capsys: pytest.CaptureFixture[str]):
+ await call_command(
+ "fake://server", "ask", "topic=Python", prompt=True, json_output=True
+ )
+ captured = capsys.readouterr()
+ data = json.loads(captured.out)
+ assert "messages" in data
+
+ @pytest.mark.usefixtures("_patch_client")
+ async def test_call_prompt_not_found(self):
+ with pytest.raises(SystemExit):
+ await call_command("fake://server", "nonexistent", prompt=True)
+
+ async def test_call_missing_target(self):
+ with pytest.raises(SystemExit):
+ await call_command("fake://server", "")
+
+
+# ---------------------------------------------------------------------------
+# Structured content serialization
+# ---------------------------------------------------------------------------
+
+
+class TestFormatCallResult:
+ def test_structured_content_uses_dict_not_data(
+ self, capsys: pytest.CaptureFixture[str]
+ ):
+ """structured_content (raw dict) is used for display, not data (which may
+ be a non-serializable dataclass)."""
+ result = CallToolResult(
+ content=[mcp.types.TextContent(type="text", text="ok")],
+ structured_content={"key": "value"},
+ meta=None,
+ data=object(), # non-serializable on purpose
+ is_error=False,
+ )
+ # Should not raise — uses structured_content, not data
+ _format_call_result_text(result)
+ captured = capsys.readouterr()
+ assert "value" in captured.out