diff --git a/docs/clients/cli.mdx b/docs/clients/cli.mdx new file mode 100644 index 000000000..801c7632c --- /dev/null +++ b/docs/clients/cli.mdx @@ -0,0 +1,161 @@ +--- +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' +``` + +### Name-Based Resolution + +If your MCP servers are already configured in an editor or tool, you can refer to them by name instead of spelling out URLs or file paths. The CLI scans config files from Claude Desktop, Claude Code, Cursor, Gemini CLI, and Goose, and matches the name you provide. + +```bash +fastmcp list weather +fastmcp call weather get_forecast city=London +``` + +You can also use the `source:name` form to target a specific source directly, which is useful when the same server name appears in multiple configs or when you want to be explicit about which config you mean. + +```bash +fastmcp list claude-code:my-server +fastmcp call cursor:weather get_forecast city=London +``` + +The available source names are `claude-desktop`, `claude-code`, `cursor`, `gemini`, `goose`, and `project` (for `./mcp.json`). Run `fastmcp discover` to see what's available. + +## Discovering Configured Servers + +`fastmcp discover` scans your local editor and project configurations for MCP server definitions. It checks Claude Desktop, Claude Code (`~/.claude.json`), Cursor workspace configs (walking up from the current directory), Gemini CLI (`~/.gemini/settings.json`), Goose (`~/.config/goose/config.yaml`), and `mcp.json` in the current directory. + +```bash +fastmcp discover +``` + +The output groups servers by source, showing each server's name and transport. Use `--source` to filter to specific sources, and `--json` for machine-readable output. + +```bash +fastmcp discover --source claude-code +fastmcp discover --source cursor --source gemini --json +``` + +Any server that appears here can be used by name (or `source:name`) with `fastmcp list` and `fastmcp call`, which means you can go from "I have a server configured in Claude Code" to querying it without copying any URLs or paths. + +## 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..2fa032e6e 100644 --- a/docs/development/v3-notes/v3-features.mdx +++ b/docs/development/v3-notes/v3-features.mdx @@ -6,6 +6,55 @@ 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: `fastmcp discover` and name-based resolution + +`fastmcp discover` scans editor configs (Claude Desktop, Claude Code, Cursor, Gemini CLI, Goose) and project-level `mcp.json` files for MCP server definitions. Discovered servers can be referenced by name — or `source:name` for precision — in `fastmcp list` and `fastmcp call`. + +```bash +# See all configured servers +fastmcp discover + +# Use a server by name +fastmcp list weather +fastmcp call weather get_forecast city=London + +# Target a specific source with source:name +fastmcp list claude-code:my-server +fastmcp call cursor:weather get_forecast city=London + +# Filter discovery to specific sources +fastmcp discover --source claude-code --source cursor +``` + +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/pyproject.toml b/pyproject.toml index 3003da059..da38224a8 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -16,6 +16,7 @@ dependencies = [ "cyclopts>=4.0.0", "authlib>=1.6.5", "pydantic[email]>=2.11.7", + "pyyaml>=6.0,<7.0", "pyperclip>=1.9.0", "py-key-value-aio[disk,keyring,memory]>=0.3.0,<0.4.0", "uvicorn>=0.35", diff --git a/skills/fastmcp-client-cli/SKILL.md b/skills/fastmcp-client-cli/SKILL.md new file mode 100644 index 000000000..ae66a5e31 --- /dev/null +++ b/skills/fastmcp-client-cli/SKILL.md @@ -0,0 +1,114 @@ +--- +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'` | +| Discovered name | `weather` or `source:name` | + +Servers configured in editor configs (Claude Desktop, Claude Code, Cursor, Gemini CLI, Goose) or project-level `mcp.json` can be referenced by name. Use `source:name` (e.g. `claude-code:my-server`, `cursor:weather`) to target a specific source. Run `fastmcp discover` to see available names. + +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 +``` + +## Discovering Configured Servers + +```bash +# See all MCP servers in editor/project configs +fastmcp discover + +# Filter by source +fastmcp discover --source claude-code + +# JSON output +fastmcp discover --json +``` + +Scans Claude Desktop, Claude Code, Cursor, Gemini CLI, Goose, and `./mcp.json`. Sources: `claude-desktop`, `claude-code`, `cursor`, `gemini`, `goose`, `project`. + +## Workflow Pattern + +Discover tools first, then call them: + +```bash +# 1. See what servers are configured +fastmcp discover + +# 2. See what tools a server has +fastmcp list weather + +# 3. Call a tool +fastmcp call weather get_forecast city=London +``` + +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..85b9d9ff5 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, discover_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,11 @@ 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") +app.command(discover_command, name="discover") + if __name__ == "__main__": app() diff --git a/src/fastmcp/cli/client.py b/src/fastmcp/cli/client.py new file mode 100644 index 000000000..d0d070d98 --- /dev/null +++ b/src/fastmcp/cli/client.py @@ -0,0 +1,964 @@ +"""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.cli.discovery import DiscoveredServer, discover_servers, resolve_name +from fastmcp.client.client import CallToolResult, Client +from fastmcp.client.elicitation import ElicitResult +from fastmcp.client.transports.base import ClientTransport +from fastmcp.client.transports.http import StreamableHttpTransport +from fastmcp.client.transports.sse import SSETransport +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] | ClientTransport: + """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 — name-based resolution via ``resolve_name``. + + 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. Name-based resolution (bare name or source:name) + try: + return resolve_name(spec) + except ValueError as exc: + console.print(f"[bold red]Error:[/bold red] {exc}") + 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] | ClientTransport) -> 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 isinstance(resolved, (StreamableHttpTransport, SSETransport)) + + +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] | ClientTransport, + *, + 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) + + +async def discover_command( + *, + source: Annotated[ + list[str] | None, + cyclopts.Parameter( + "--source", + help="Only show servers from these sources (e.g. claude-code, cursor, gemini)", + ), + ] = None, + json_output: Annotated[ + bool, + cyclopts.Parameter("--json", help="Output as JSON"), + ] = False, +) -> None: + """Discover MCP servers configured in editor and project configs. + + Scans Claude Desktop, Claude Code, Cursor, Gemini CLI, Goose, and + project-level mcp.json files for MCP server definitions. + + Discovered server names can be used directly with ``fastmcp list`` + and ``fastmcp call`` instead of specifying a URL or file path. + + Examples: + fastmcp discover + fastmcp discover --source claude-code + fastmcp discover --source cursor --source gemini --json + fastmcp list weather + fastmcp call cursor:weather get_forecast city=London + """ + + servers = discover_servers() + + if source: + servers = [s for s in servers if s.source in source] + + if json_output: + data: list[dict[str, Any]] = [ + { + "name": s.name, + "source": s.source, + "qualified_name": s.qualified_name, + "transport_summary": s.transport_summary, + "config_path": str(s.config_path), + } + for s in servers + ] + console.print_json(json.dumps(data)) + return + + if not servers: + console.print("[dim]No MCP servers found.[/dim]") + console.print() + console.print("Searched:") + console.print(" • Claude Desktop config") + console.print(" • ~/.claude.json (Claude Code)") + console.print(" • .cursor/mcp.json (walked up from cwd)") + console.print(" • ~/.gemini/settings.json (Gemini CLI)") + console.print(" • ~/.config/goose/config.yaml (Goose)") + console.print(" • ./mcp.json") + return + + from rich.table import Table + + # Group by source + by_source: dict[str, list[DiscoveredServer]] = {} + for s in servers: + by_source.setdefault(s.source, []).append(s) + + for source_name, group in by_source.items(): + console.print() + console.print(f"[bold]Source:[/bold] {source_name}") + console.print(f"[bold]Config:[/bold] [dim]{group[0].config_path}[/dim]") + console.print() + + table = Table( + show_header=True, + header_style="bold", + show_edge=False, + pad_edge=False, + box=None, + padding=(0, 2), + ) + table.add_column("Server", style="cyan") + table.add_column("Transport", style="dim") + + for s in group: + table.add_row(s.name, s.transport_summary) + + console.print(table) + console.print() diff --git a/src/fastmcp/cli/discovery.py b/src/fastmcp/cli/discovery.py new file mode 100644 index 000000000..5acd42d61 --- /dev/null +++ b/src/fastmcp/cli/discovery.py @@ -0,0 +1,375 @@ +"""Discover MCP servers configured in editor config files. + +Scans filesystem-readable config files from editors like Claude Desktop, +Claude Code, Cursor, Gemini CLI, and Goose, as well as project-level +``mcp.json`` files. Each discovered server can be resolved by name +(or ``source:name``) so the CLI can connect without requiring a URL +or file path. +""" + +import json +import os +import sys +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +import yaml + +from fastmcp.client.transports.base import ClientTransport +from fastmcp.mcp_config import ( + MCPConfig, + MCPServerTypes, + RemoteMCPServer, + StdioMCPServer, +) +from fastmcp.utilities.logging import get_logger + +logger = get_logger("cli.discovery") + + +# --------------------------------------------------------------------------- +# Data model +# --------------------------------------------------------------------------- + + +@dataclass(frozen=True) +class DiscoveredServer: + """A single MCP server found in an editor or project config.""" + + name: str + source: str + config: MCPServerTypes + config_path: Path + + @property + def qualified_name(self) -> str: + """Fully qualified ``source:name`` identifier.""" + return f"{self.source}:{self.name}" + + @property + def transport_summary(self) -> str: + """Human-readable one-liner describing the transport.""" + cfg = self.config + if isinstance(cfg, StdioMCPServer): + parts = [cfg.command, *cfg.args] + return f"stdio: {' '.join(parts)}" + if isinstance(cfg, RemoteMCPServer): + transport = cfg.transport or "http" + return f"{transport}: {cfg.url}" + return str(type(cfg).__name__) + + +# --------------------------------------------------------------------------- +# Scanners — one per config source +# --------------------------------------------------------------------------- + + +def _normalize_server_entry(entry: dict[str, Any]) -> dict[str, Any]: + """Normalize editor-specific server config fields to MCPConfig format. + + Handles two known differences: + - Claude Code uses ``type`` where MCPConfig uses ``transport`` for + remote servers. + - Gemini CLI uses ``httpUrl`` where MCPConfig uses ``url``. + """ + # Gemini: httpUrl → url + if "httpUrl" in entry and "url" not in entry: + entry = {**entry, "url": entry["httpUrl"]} + del entry["httpUrl"] + + # Claude Code / others: type → transport (for url-based entries only) + if "url" in entry and "type" in entry and "transport" not in entry: + transport = entry["type"] + entry = {k: v for k, v in entry.items() if k != "type"} + entry["transport"] = transport + + return entry + + +def _parse_mcp_servers( + servers_dict: dict[str, Any], + *, + source: str, + config_path: Path, +) -> list[DiscoveredServer]: + """Parse an ``mcpServers``-style dict into discovered servers.""" + if not servers_dict: + return [] + + normalized = { + name: _normalize_server_entry(entry) + for name, entry in servers_dict.items() + if isinstance(entry, dict) + } + + try: + config = MCPConfig.from_dict({"mcpServers": normalized}) + except Exception as exc: + logger.warning("Could not parse MCP servers from %s: %s", config_path, exc) + return [] + + return [ + DiscoveredServer( + name=name, source=source, config=server, config_path=config_path + ) + for name, server in config.mcpServers.items() + ] + + +def _parse_mcp_config(path: Path, source: str) -> list[DiscoveredServer]: + """Parse an mcpServers-style JSON file into discovered servers.""" + try: + text = path.read_text() + except OSError as exc: + logger.debug("Could not read %s: %s", path, exc) + return [] + + try: + data: dict[str, Any] = json.loads(text) + except json.JSONDecodeError as exc: + logger.warning("Invalid JSON in %s: %s", path, exc) + return [] + + if not isinstance(data, dict) or "mcpServers" not in data: + return [] + + return _parse_mcp_servers(data["mcpServers"], source=source, config_path=path) + + +def _scan_claude_desktop() -> list[DiscoveredServer]: + """Scan the Claude Desktop config file.""" + if sys.platform == "win32": + config_dir = Path(Path.home(), "AppData", "Roaming", "Claude") + elif sys.platform == "darwin": + config_dir = Path(Path.home(), "Library", "Application Support", "Claude") + elif sys.platform.startswith("linux"): + config_dir = Path( + os.environ.get("XDG_CONFIG_HOME", Path.home() / ".config"), "Claude" + ) + else: + return [] + + path = config_dir / "claude_desktop_config.json" + return _parse_mcp_config(path, "claude-desktop") + + +def _scan_claude_code(start_dir: Path) -> list[DiscoveredServer]: + """Scan ``~/.claude.json`` for global and project-scoped MCP servers.""" + path = Path.home() / ".claude.json" + try: + text = path.read_text() + except OSError: + return [] + + try: + data: dict[str, Any] = json.loads(text) + except json.JSONDecodeError as exc: + logger.warning("Invalid JSON in %s: %s", path, exc) + return [] + + if not isinstance(data, dict): + return [] + + results: list[DiscoveredServer] = [] + + # Global servers + if global_servers := data.get("mcpServers"): + if isinstance(global_servers, dict): + results.extend( + _parse_mcp_servers( + global_servers, source="claude-code", config_path=path + ) + ) + + # Project-scoped servers matching start_dir + resolved_dir = str(start_dir.resolve()) + projects = data.get("projects", {}) + if isinstance(projects, dict): + project_data = projects.get(resolved_dir, {}) + if isinstance(project_data, dict): + if project_servers := project_data.get("mcpServers"): + if isinstance(project_servers, dict): + results.extend( + _parse_mcp_servers( + project_servers, + source="claude-code", + config_path=path, + ) + ) + + return results + + +def _scan_cursor_workspace(start_dir: Path) -> list[DiscoveredServer]: + """Walk up from *start_dir* looking for ``.cursor/mcp.json``.""" + current = start_dir.resolve() + home = Path.home().resolve() + + while True: + candidate = current / ".cursor" / "mcp.json" + if candidate.is_file(): + return _parse_mcp_config(candidate, "cursor") + + parent = current.parent + # Stop at filesystem root or home directory + if parent == current or current == home: + break + current = parent + + return [] + + +def _scan_project_mcp_json(start_dir: Path) -> list[DiscoveredServer]: + """Check for ``mcp.json`` in *start_dir*.""" + candidate = start_dir.resolve() / "mcp.json" + if candidate.is_file(): + return _parse_mcp_config(candidate, "project") + return [] + + +def _scan_gemini(start_dir: Path) -> list[DiscoveredServer]: + """Scan Gemini CLI settings for MCP servers. + + Checks both user-level ``~/.gemini/settings.json`` and project-level + ``.gemini/settings.json``. + """ + results: list[DiscoveredServer] = [] + + # User-level + user_path = Path.home() / ".gemini" / "settings.json" + results.extend(_parse_mcp_config(user_path, "gemini")) + + # Project-level + project_path = start_dir.resolve() / ".gemini" / "settings.json" + if project_path != user_path: + results.extend(_parse_mcp_config(project_path, "gemini")) + + return results + + +def _scan_goose() -> list[DiscoveredServer]: + """Scan Goose config for MCP server extensions. + + Goose uses YAML (``~/.config/goose/config.yaml``) with a different + schema — MCP servers are defined as ``extensions`` with ``type: stdio``. + """ + if sys.platform == "win32": + config_dir = Path( + os.environ.get("APPDATA", Path.home() / "AppData" / "Roaming"), + "Block", + "goose", + "config", + ) + else: + config_dir = Path( + os.environ.get("XDG_CONFIG_HOME", Path.home() / ".config"), + "goose", + ) + + path = config_dir / "config.yaml" + try: + text = path.read_text() + except OSError: + return [] + + try: + data = yaml.safe_load(text) + except yaml.YAMLError as exc: + logger.warning("Invalid YAML in %s: %s", path, exc) + return [] + + if not isinstance(data, dict): + return [] + + extensions = data.get("extensions", {}) + if not isinstance(extensions, dict): + return [] + + # Convert Goose extensions to mcpServers format + servers: dict[str, Any] = {} + for name, ext in extensions.items(): + if not isinstance(ext, dict): + continue + if not ext.get("enabled", True): + continue + ext_type = ext.get("type", "") + if ext_type == "stdio" and "cmd" in ext: + servers[name] = { + "command": ext["cmd"], + "args": ext.get("args", []), + "env": ext.get("envs", {}), + } + elif ext_type == "sse" and "uri" in ext: + servers[name] = {"url": ext["uri"], "transport": "sse"} + + return _parse_mcp_servers(servers, source="goose", config_path=path) + + +# --------------------------------------------------------------------------- +# Public API +# --------------------------------------------------------------------------- + + +def discover_servers(start_dir: Path | None = None) -> list[DiscoveredServer]: + """Run all scanners and return the combined results. + + Duplicate names across sources are preserved — callers can + use :pyattr:`DiscoveredServer.qualified_name` to disambiguate. + """ + cwd = start_dir or Path.cwd() + results: list[DiscoveredServer] = [] + results.extend(_scan_claude_desktop()) + results.extend(_scan_claude_code(cwd)) + results.extend(_scan_cursor_workspace(cwd)) + results.extend(_scan_gemini(cwd)) + results.extend(_scan_goose()) + results.extend(_scan_project_mcp_json(cwd)) + return results + + +def resolve_name(name: str, start_dir: Path | None = None) -> ClientTransport: + """Resolve a server name (or ``source:name``) to a transport. + + Raises :class:`ValueError` when the name is not found or is ambiguous. + """ + servers = discover_servers(start_dir) + + # Qualified form: "cursor:weather" + if ":" in name: + source, server_name = name.split(":", 1) + matches = [s for s in servers if s.source == source and s.name == server_name] + if not matches: + raise ValueError( + f"No server named '{server_name}' found in source '{source}'." + ) + return matches[0].config.to_transport() + + # Bare name: "weather" + matches = [s for s in servers if s.name == name] + + if not matches: + if servers: + available = ", ".join(sorted({s.name for s in servers})) + raise ValueError(f"No server named '{name}' found. Available: {available}") + locations = [ + "Claude Desktop config", + "~/.claude.json (Claude Code)", + ".cursor/mcp.json (walked up from cwd)", + "~/.gemini/settings.json (Gemini CLI)", + "~/.config/goose/config.yaml (Goose)", + "./mcp.json", + ] + raise ValueError( + f"No server named '{name}' found. Searched: {', '.join(locations)}" + ) + + if len(matches) == 1: + return matches[0].config.to_transport() + + # Ambiguous — list qualified alternatives + alternatives = ", ".join(f"'{m.qualified_name}'" for m in matches) + raise ValueError( + f"Ambiguous server name '{name}' — found in multiple sources. " + f"Use a qualified name: {alternatives}" + ) diff --git a/src/fastmcp/prompts/function_prompt.py b/src/fastmcp/prompts/function_prompt.py index 4648c269f..c2ba6f44a 100644 --- a/src/fastmcp/prompts/function_prompt.py +++ b/src/fastmcp/prompts/function_prompt.py @@ -297,13 +297,27 @@ class FunctionPrompt(Prompt): # Convert string arguments to expected types BEFORE validation kwargs = self._convert_string_arguments(kwargs) + # Filter out arguments that aren't in the function signature + # This is important for security: dependencies should not be overridable + # from external callers. self.fn is wrapped by without_injected_parameters, + # so we only accept arguments that are in the wrapped function's signature. + sig = inspect.signature(self.fn) + valid_params = set(sig.parameters.keys()) + kwargs = {k: v for k, v in kwargs.items() if k in valid_params} + + # Use type adapter to validate arguments and handle Field() defaults + # This matches the behavior of tools in function_tool + type_adapter = get_cached_typeadapter(self.fn) + # self.fn is wrapped by without_injected_parameters which handles # dependency resolution internally if inspect.iscoroutinefunction(self.fn): - result = await self.fn(**kwargs) + result = await type_adapter.validate_python(kwargs) else: # Run sync functions in threadpool to avoid blocking the event loop - result = await call_sync_fn_in_threadpool(self.fn, **kwargs) + result = await call_sync_fn_in_threadpool( + type_adapter.validate_python, kwargs + ) # Handle sync wrappers that return awaitables (e.g., partial(async_fn)) if inspect.isawaitable(result): result = await result 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 diff --git a/tests/cli/test_discovery.py b/tests/cli/test_discovery.py new file mode 100644 index 000000000..716694353 --- /dev/null +++ b/tests/cli/test_discovery.py @@ -0,0 +1,668 @@ +"""Tests for MCP server discovery and name-based resolution.""" + +import json +from pathlib import Path +from typing import Any + +import pytest +import yaml + +from fastmcp.cli.client import _is_http_target, resolve_server_spec +from fastmcp.cli.discovery import ( + DiscoveredServer, + _normalize_server_entry, + _parse_mcp_config, + _scan_claude_code, + _scan_claude_desktop, + _scan_cursor_workspace, + _scan_gemini, + _scan_goose, + _scan_project_mcp_json, + discover_servers, + resolve_name, +) +from fastmcp.client.transports.http import StreamableHttpTransport +from fastmcp.client.transports.sse import SSETransport +from fastmcp.client.transports.stdio import StdioTransport +from fastmcp.mcp_config import RemoteMCPServer, StdioMCPServer + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +_STDIO_CONFIG: dict[str, Any] = { + "mcpServers": { + "weather": { + "command": "npx", + "args": ["-y", "@mcp/weather"], + }, + "github": { + "command": "npx", + "args": ["-y", "@mcp/github"], + "env": {"GITHUB_TOKEN": "xxx"}, + }, + } +} + +_REMOTE_CONFIG: dict[str, Any] = { + "mcpServers": { + "api": { + "url": "http://localhost:8000/mcp", + }, + } +} + + +def _write_config(path: Path, data: dict[str, Any]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(data)) + + +# --------------------------------------------------------------------------- +# DiscoveredServer properties +# --------------------------------------------------------------------------- + + +class TestDiscoveredServer: + def test_qualified_name(self): + server = DiscoveredServer( + name="weather", + source="claude-desktop", + config=StdioMCPServer(command="npx", args=["-y", "@mcp/weather"]), + config_path=Path("/fake/config.json"), + ) + assert server.qualified_name == "claude-desktop:weather" + + def test_transport_summary_stdio(self): + server = DiscoveredServer( + name="weather", + source="cursor", + config=StdioMCPServer(command="npx", args=["-y", "@mcp/weather"]), + config_path=Path("/fake/config.json"), + ) + assert server.transport_summary == "stdio: npx -y @mcp/weather" + + def test_transport_summary_remote(self): + server = DiscoveredServer( + name="api", + source="project", + config=RemoteMCPServer(url="http://localhost:8000/mcp"), + config_path=Path("/fake/config.json"), + ) + assert server.transport_summary == "http: http://localhost:8000/mcp" + + def test_transport_summary_remote_sse(self): + server = DiscoveredServer( + name="api", + source="project", + config=RemoteMCPServer(url="http://localhost:8000/sse", transport="sse"), + config_path=Path("/fake/config.json"), + ) + assert server.transport_summary == "sse: http://localhost:8000/sse" + + +# --------------------------------------------------------------------------- +# _parse_mcp_config +# --------------------------------------------------------------------------- + + +class TestParseMcpConfig: + def test_valid_config(self, tmp_path: Path): + path = tmp_path / "config.json" + _write_config(path, _STDIO_CONFIG) + servers = _parse_mcp_config(path, "test-source") + assert len(servers) == 2 + names = {s.name for s in servers} + assert names == {"weather", "github"} + assert all(s.source == "test-source" for s in servers) + assert all(s.config_path == path for s in servers) + + def test_missing_file(self, tmp_path: Path): + path = tmp_path / "nonexistent.json" + servers = _parse_mcp_config(path, "test") + assert servers == [] + + def test_invalid_json(self, tmp_path: Path): + path = tmp_path / "bad.json" + path.write_text("{not json") + servers = _parse_mcp_config(path, "test") + assert servers == [] + + def test_no_mcp_servers_key(self, tmp_path: Path): + path = tmp_path / "config.json" + _write_config(path, {"something": "else"}) + servers = _parse_mcp_config(path, "test") + assert servers == [] + + def test_empty_mcp_servers(self, tmp_path: Path): + path = tmp_path / "config.json" + _write_config(path, {"mcpServers": {}}) + servers = _parse_mcp_config(path, "test") + assert servers == [] + + def test_remote_server(self, tmp_path: Path): + path = tmp_path / "config.json" + _write_config(path, _REMOTE_CONFIG) + servers = _parse_mcp_config(path, "test") + assert len(servers) == 1 + assert isinstance(servers[0].config, RemoteMCPServer) + assert servers[0].config.url == "http://localhost:8000/mcp" + + +# --------------------------------------------------------------------------- +# Scanner: Claude Desktop +# --------------------------------------------------------------------------- + + +class TestScanClaudeDesktop: + def test_finds_config(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch): + config_dir = tmp_path / "Claude" + config_path = config_dir / "claude_desktop_config.json" + _write_config(config_path, _STDIO_CONFIG) + + monkeypatch.setattr("fastmcp.cli.discovery.Path.home", lambda: tmp_path) + # Force darwin for deterministic path + monkeypatch.setattr("fastmcp.cli.discovery.sys.platform", "darwin") + + # We need to override the path construction. On macOS it's + # ~/Library/Application Support/Claude — create that. + mac_dir = tmp_path / "Library" / "Application Support" / "Claude" + mac_path = mac_dir / "claude_desktop_config.json" + _write_config(mac_path, _STDIO_CONFIG) + + servers = _scan_claude_desktop() + assert len(servers) == 2 + assert all(s.source == "claude-desktop" for s in servers) + + def test_missing_config(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch): + monkeypatch.setattr("fastmcp.cli.discovery.Path.home", lambda: tmp_path) + monkeypatch.setattr("fastmcp.cli.discovery.sys.platform", "darwin") + servers = _scan_claude_desktop() + assert servers == [] + + +# --------------------------------------------------------------------------- +# Normalize server entry +# --------------------------------------------------------------------------- + + +class TestNormalizeServerEntry: + def test_remote_type_becomes_transport(self): + entry = {"url": "http://localhost:8000/sse", "type": "sse"} + result = _normalize_server_entry(entry) + assert result["transport"] == "sse" + assert "type" not in result + + def test_remote_with_transport_unchanged(self): + entry = {"url": "http://localhost:8000/mcp", "transport": "http"} + result = _normalize_server_entry(entry) + assert result["transport"] == "http" + + def test_stdio_type_unchanged(self): + """Stdio entries have ``type`` as a proper field — leave it alone.""" + entry = {"command": "npx", "args": [], "type": "stdio"} + result = _normalize_server_entry(entry) + assert result["type"] == "stdio" + + def test_gemini_http_url_becomes_url(self): + entry = {"httpUrl": "https://api.example.com/mcp/"} + result = _normalize_server_entry(entry) + assert result["url"] == "https://api.example.com/mcp/" + assert "httpUrl" not in result + + def test_gemini_http_url_does_not_override_url(self): + entry = {"url": "http://real.com", "httpUrl": "http://other.com"} + result = _normalize_server_entry(entry) + assert result["url"] == "http://real.com" + + +# --------------------------------------------------------------------------- +# Scanner: Claude Code +# --------------------------------------------------------------------------- + + +def _claude_code_config( + *, + global_servers: dict[str, Any] | None = None, + project_path: str | None = None, + project_servers: dict[str, Any] | None = None, +) -> dict[str, Any]: + """Build a minimal ~/.claude.json structure.""" + data: dict[str, Any] = {} + if global_servers is not None: + data["mcpServers"] = global_servers + if project_path and project_servers is not None: + data["projects"] = {project_path: {"mcpServers": project_servers}} + return data + + +class TestScanClaudeCode: + def test_global_servers(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch): + monkeypatch.setattr("fastmcp.cli.discovery.Path.home", lambda: tmp_path) + config_path = tmp_path / ".claude.json" + _write_config( + config_path, + _claude_code_config(global_servers=_STDIO_CONFIG["mcpServers"]), + ) + servers = _scan_claude_code(tmp_path) + assert len(servers) == 2 + assert all(s.source == "claude-code" for s in servers) + + def test_project_servers(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch): + monkeypatch.setattr("fastmcp.cli.discovery.Path.home", lambda: tmp_path) + project_dir = tmp_path / "my-project" + project_dir.mkdir() + config_path = tmp_path / ".claude.json" + _write_config( + config_path, + _claude_code_config( + project_path=str(project_dir), + project_servers={"api": {"url": "http://localhost:8000/mcp"}}, + ), + ) + servers = _scan_claude_code(project_dir) + assert len(servers) == 1 + assert servers[0].name == "api" + + def test_global_and_project_combined( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ): + monkeypatch.setattr("fastmcp.cli.discovery.Path.home", lambda: tmp_path) + project_dir = tmp_path / "proj" + project_dir.mkdir() + config_path = tmp_path / ".claude.json" + _write_config( + config_path, + _claude_code_config( + global_servers={"global-tool": {"command": "echo", "args": ["hi"]}}, + project_path=str(project_dir), + project_servers={"local-tool": {"command": "cat", "args": []}}, + ), + ) + servers = _scan_claude_code(project_dir) + names = {s.name for s in servers} + assert names == {"global-tool", "local-tool"} + + def test_type_normalized_to_transport( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ): + """Claude Code uses ``type: sse`` — verify it becomes ``transport``.""" + monkeypatch.setattr("fastmcp.cli.discovery.Path.home", lambda: tmp_path) + config_path = tmp_path / ".claude.json" + _write_config( + config_path, + _claude_code_config( + global_servers={ + "sse-server": { + "type": "sse", + "url": "http://localhost:8000/sse", + } + } + ), + ) + servers = _scan_claude_code(tmp_path) + assert len(servers) == 1 + assert isinstance(servers[0].config, RemoteMCPServer) + assert servers[0].config.transport == "sse" + + def test_missing_file(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch): + monkeypatch.setattr("fastmcp.cli.discovery.Path.home", lambda: tmp_path) + servers = _scan_claude_code(tmp_path) + assert servers == [] + + def test_no_matching_project(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch): + monkeypatch.setattr("fastmcp.cli.discovery.Path.home", lambda: tmp_path) + config_path = tmp_path / ".claude.json" + _write_config( + config_path, + _claude_code_config( + project_path="/some/other/project", + project_servers={"tool": {"command": "echo", "args": []}}, + ), + ) + servers = _scan_claude_code(tmp_path) + assert servers == [] + + +# --------------------------------------------------------------------------- +# Scanner: Cursor workspace +# --------------------------------------------------------------------------- + + +class TestScanCursorWorkspace: + def test_finds_config_in_cwd(self, tmp_path: Path): + cursor_path = tmp_path / ".cursor" / "mcp.json" + _write_config(cursor_path, _STDIO_CONFIG) + servers = _scan_cursor_workspace(tmp_path) + assert len(servers) == 2 + assert all(s.source == "cursor" for s in servers) + + def test_finds_config_in_parent(self, tmp_path: Path): + cursor_path = tmp_path / ".cursor" / "mcp.json" + _write_config(cursor_path, _STDIO_CONFIG) + child = tmp_path / "src" / "deep" + child.mkdir(parents=True) + servers = _scan_cursor_workspace(child) + assert len(servers) == 2 + + def test_stops_at_home(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch): + # Place config above home — should not be found + monkeypatch.setattr("fastmcp.cli.discovery.Path.home", lambda: tmp_path) + above_home = tmp_path.parent / ".cursor" / "mcp.json" + _write_config(above_home, _STDIO_CONFIG) + child = tmp_path / "project" + child.mkdir() + servers = _scan_cursor_workspace(child) + assert servers == [] + + def test_no_config(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch): + # Confine walk to tmp_path so it doesn't find sibling test dirs + monkeypatch.setattr("fastmcp.cli.discovery.Path.home", lambda: tmp_path) + servers = _scan_cursor_workspace(tmp_path) + assert servers == [] + + +# --------------------------------------------------------------------------- +# Scanner: project mcp.json +# --------------------------------------------------------------------------- + + +class TestScanProjectMcpJson: + def test_finds_config(self, tmp_path: Path): + config_path = tmp_path / "mcp.json" + _write_config(config_path, _STDIO_CONFIG) + servers = _scan_project_mcp_json(tmp_path) + assert len(servers) == 2 + assert all(s.source == "project" for s in servers) + + def test_no_config(self, tmp_path: Path): + servers = _scan_project_mcp_json(tmp_path) + assert servers == [] + + +# --------------------------------------------------------------------------- +# Scanner: Gemini CLI +# --------------------------------------------------------------------------- + + +class TestScanGemini: + def test_user_level_config(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch): + monkeypatch.setattr("fastmcp.cli.discovery.Path.home", lambda: tmp_path) + config_path = tmp_path / ".gemini" / "settings.json" + _write_config(config_path, _STDIO_CONFIG) + servers = _scan_gemini(tmp_path) + assert len(servers) == 2 + assert all(s.source == "gemini" for s in servers) + + def test_project_level_config( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ): + monkeypatch.setattr("fastmcp.cli.discovery.Path.home", lambda: tmp_path) + project_dir = tmp_path / "my-project" + project_dir.mkdir() + config_path = project_dir / ".gemini" / "settings.json" + _write_config(config_path, _STDIO_CONFIG) + servers = _scan_gemini(project_dir) + assert len(servers) == 2 + + def test_http_url_normalized(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch): + """Gemini uses ``httpUrl`` — verify it becomes ``url``.""" + monkeypatch.setattr("fastmcp.cli.discovery.Path.home", lambda: tmp_path) + config_path = tmp_path / ".gemini" / "settings.json" + _write_config( + config_path, + { + "mcpServers": { + "api": {"httpUrl": "https://api.example.com/mcp/"}, + } + }, + ) + servers = _scan_gemini(tmp_path) + assert len(servers) == 1 + assert isinstance(servers[0].config, RemoteMCPServer) + assert servers[0].config.url == "https://api.example.com/mcp/" + + def test_missing_config(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch): + monkeypatch.setattr("fastmcp.cli.discovery.Path.home", lambda: tmp_path) + servers = _scan_gemini(tmp_path) + assert servers == [] + + +# --------------------------------------------------------------------------- +# Scanner: Goose +# --------------------------------------------------------------------------- + +_GOOSE_CONFIG = { + "extensions": { + "developer": { + "enabled": True, + "name": "developer", + "type": "builtin", + }, + "tavily": { + "cmd": "npx", + "args": ["-y", "mcp-tavily-search"], + "enabled": True, + "envs": {"TAVILY_API_KEY": "xxx"}, + "type": "stdio", + }, + "disabled-tool": { + "cmd": "echo", + "args": ["hi"], + "enabled": False, + "type": "stdio", + }, + } +} + + +class TestScanGoose: + def test_finds_stdio_extensions( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ): + monkeypatch.setattr("fastmcp.cli.discovery.Path.home", lambda: tmp_path) + monkeypatch.delenv("XDG_CONFIG_HOME", raising=False) + config_dir = tmp_path / ".config" / "goose" + config_path = config_dir / "config.yaml" + config_path.parent.mkdir(parents=True) + config_path.write_text(yaml.dump(_GOOSE_CONFIG)) + # Force non-windows platform for path logic + monkeypatch.setattr("fastmcp.cli.discovery.sys.platform", "linux") + servers = _scan_goose() + assert len(servers) == 1 + assert servers[0].name == "tavily" + assert servers[0].source == "goose" + assert isinstance(servers[0].config, StdioMCPServer) + assert servers[0].config.command == "npx" + + def test_skips_builtin_and_disabled( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ): + monkeypatch.setattr("fastmcp.cli.discovery.Path.home", lambda: tmp_path) + monkeypatch.delenv("XDG_CONFIG_HOME", raising=False) + config_dir = tmp_path / ".config" / "goose" + config_path = config_dir / "config.yaml" + config_path.parent.mkdir(parents=True) + config_path.write_text(yaml.dump(_GOOSE_CONFIG)) + monkeypatch.setattr("fastmcp.cli.discovery.sys.platform", "linux") + servers = _scan_goose() + names = {s.name for s in servers} + assert "developer" not in names + assert "disabled-tool" not in names + + def test_missing_config(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch): + monkeypatch.setattr("fastmcp.cli.discovery.Path.home", lambda: tmp_path) + monkeypatch.setattr("fastmcp.cli.discovery.sys.platform", "linux") + servers = _scan_goose() + assert servers == [] + + +# --------------------------------------------------------------------------- +# discover_servers +# --------------------------------------------------------------------------- + + +def _suppress_user_scanners(monkeypatch: pytest.MonkeyPatch) -> None: + """Suppress all scanners that read real user config files.""" + monkeypatch.setattr("fastmcp.cli.discovery._scan_claude_desktop", lambda: []) + monkeypatch.setattr("fastmcp.cli.discovery._scan_claude_code", lambda start_dir: []) + monkeypatch.setattr("fastmcp.cli.discovery._scan_gemini", lambda start_dir: []) + monkeypatch.setattr("fastmcp.cli.discovery._scan_goose", lambda: []) + + +class TestDiscoverServers: + def test_combines_sources(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch): + # Set up project mcp.json + project_config = tmp_path / "mcp.json" + _write_config(project_config, _STDIO_CONFIG) + + # Set up cursor config + cursor_config = tmp_path / ".cursor" / "mcp.json" + _write_config(cursor_config, _REMOTE_CONFIG) + + _suppress_user_scanners(monkeypatch) + + servers = discover_servers(start_dir=tmp_path) + sources = {s.source for s in servers} + assert "project" in sources + assert "cursor" in sources + assert len(servers) == 3 # 2 from project + 1 from cursor + + def test_preserves_duplicates( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ): + """Same server name in multiple sources should appear multiple times.""" + project_config = tmp_path / "mcp.json" + _write_config(project_config, _STDIO_CONFIG) + + cursor_config = tmp_path / ".cursor" / "mcp.json" + _write_config(cursor_config, _STDIO_CONFIG) + + _suppress_user_scanners(monkeypatch) + + servers = discover_servers(start_dir=tmp_path) + weather_servers = [s for s in servers if s.name == "weather"] + assert len(weather_servers) == 2 + assert {s.source for s in weather_servers} == {"cursor", "project"} + + +# --------------------------------------------------------------------------- +# resolve_name +# --------------------------------------------------------------------------- + + +class TestResolveName: + @pytest.fixture(autouse=True) + def _isolate_scanners(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch): + """Suppress scanners that read real user configs and confine walks to tmp_path.""" + _suppress_user_scanners(monkeypatch) + monkeypatch.setattr("fastmcp.cli.discovery.Path.home", lambda: tmp_path) + + def test_unique_match(self, tmp_path: Path): + config_path = tmp_path / "mcp.json" + _write_config(config_path, _STDIO_CONFIG) + transport = resolve_name("weather", start_dir=tmp_path) + assert isinstance(transport, StdioTransport) + + def test_qualified_match(self, tmp_path: Path): + config_path = tmp_path / "mcp.json" + _write_config(config_path, _STDIO_CONFIG) + transport = resolve_name("project:weather", start_dir=tmp_path) + assert isinstance(transport, StdioTransport) + + def test_not_found_with_servers(self, tmp_path: Path): + config_path = tmp_path / "mcp.json" + _write_config(config_path, _STDIO_CONFIG) + with pytest.raises(ValueError, match="No server named 'nope'.*Available"): + resolve_name("nope", start_dir=tmp_path) + + def test_not_found_no_servers(self, tmp_path: Path): + with pytest.raises(ValueError, match="No server named 'nope'.*Searched"): + resolve_name("nope", start_dir=tmp_path) + + def test_ambiguous_name(self, tmp_path: Path): + project_config = tmp_path / "mcp.json" + _write_config(project_config, _STDIO_CONFIG) + cursor_config = tmp_path / ".cursor" / "mcp.json" + _write_config(cursor_config, _STDIO_CONFIG) + with pytest.raises(ValueError, match="Ambiguous server name 'weather'"): + resolve_name("weather", start_dir=tmp_path) + + def test_ambiguous_resolved_by_qualified(self, tmp_path: Path): + project_config = tmp_path / "mcp.json" + _write_config(project_config, _STDIO_CONFIG) + cursor_config = tmp_path / ".cursor" / "mcp.json" + _write_config(cursor_config, _STDIO_CONFIG) + transport = resolve_name("cursor:weather", start_dir=tmp_path) + assert isinstance(transport, StdioTransport) + + def test_qualified_not_found(self, tmp_path: Path): + config_path = tmp_path / "mcp.json" + _write_config(config_path, _STDIO_CONFIG) + with pytest.raises( + ValueError, match="No server named 'nope' found in source 'project'" + ): + resolve_name("project:nope", start_dir=tmp_path) + + def test_remote_server_resolves_to_http_transport(self, tmp_path: Path): + config_path = tmp_path / "mcp.json" + _write_config(config_path, _REMOTE_CONFIG) + transport = resolve_name("api", start_dir=tmp_path) + assert isinstance(transport, StreamableHttpTransport) + + +# --------------------------------------------------------------------------- +# Integration: resolve_server_spec falls through to name resolution +# --------------------------------------------------------------------------- + + +class TestResolveServerSpecNameFallback: + def test_bare_name_resolves(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch): + config_path = tmp_path / "mcp.json" + _write_config(config_path, _STDIO_CONFIG) + _suppress_user_scanners(monkeypatch) + monkeypatch.setattr("fastmcp.cli.discovery.Path.home", lambda: tmp_path) + + # Monkeypatch resolve_name in client module to use our tmp_path + original_resolve = resolve_name + + def patched_resolve(name: str, start_dir: Path | None = None) -> Any: + return original_resolve(name, start_dir=tmp_path) + + monkeypatch.setattr("fastmcp.cli.client.resolve_name", patched_resolve) + + result = resolve_server_spec("weather") + assert isinstance(result, StdioTransport) + + def test_url_takes_priority_over_name(self): + """URLs should be resolved before name lookup.""" + result = resolve_server_spec("http://localhost:8000/mcp") + assert result == "http://localhost:8000/mcp" + + +# --------------------------------------------------------------------------- +# Integration: _is_http_target detects transport objects +# --------------------------------------------------------------------------- + + +class TestIsHttpTargetTransports: + def test_streamable_http_transport(self): + transport = StreamableHttpTransport("http://localhost:8000/mcp") + assert _is_http_target(transport) is True + + def test_sse_transport(self): + transport = SSETransport("http://localhost:8000/sse") + assert _is_http_target(transport) is True + + def test_stdio_transport(self): + transport = StdioTransport(command="echo", args=["hello"]) + assert _is_http_target(transport) is False + + def test_string_url(self): + assert _is_http_target("http://localhost:8000") is True + + def test_string_non_url(self): + assert _is_http_target("server.py") is False + + def test_dict_config(self): + assert _is_http_target({"mcpServers": {}}) is False diff --git a/tests/prompts/test_prompt.py b/tests/prompts/test_prompt.py index 30c8336e7..05d7c0f17 100644 --- a/tests/prompts/test_prompt.py +++ b/tests/prompts/test_prompt.py @@ -552,6 +552,88 @@ class TestPromptResult: assert mcp_result.meta == {"key": "value"} +class TestPromptFieldDefaults: + """Test prompts with Field() defaults.""" + + async def test_field_with_default(self): + """Test that Field(default=...) correctly provides default values.""" + + from pydantic import Field + + def prompt_with_defaults( + required: str = Field(description="Required parameter"), + optional: str = Field( + default="default_value", description="Optional parameter" + ), + ) -> str: + return f"required={required}, optional={optional}" + + prompt = Prompt.from_function(prompt_with_defaults) + result = await prompt.render(arguments={"required": "test"}) + assert result.messages == [Message("required=test, optional=default_value")] + + async def test_annotated_field_with_default_in_signature(self): + """Test that Annotated[type, Field(...)] with default in signature works.""" + from typing import Annotated + + from pydantic import Field + + def prompt_with_annotated( + required: Annotated[str, Field(description="Required parameter")], + optional: Annotated[ + str, Field(description="Optional parameter") + ] = "default_value", + ) -> str: + return f"required={required}, optional={optional}" + + prompt = Prompt.from_function(prompt_with_annotated) + result = await prompt.render(arguments={"required": "test"}) + assert result.messages == [Message("required=test, optional=default_value")] + + async def test_multiple_field_defaults(self): + """Test multiple parameters with Field() defaults.""" + from pydantic import Field + + def prompt_with_multiple_defaults( + name: str = Field(description="Name"), + greeting: str = Field(default="Hello", description="Greeting"), + punctuation: str = Field(default="!", description="Punctuation"), + ) -> str: + return f"{greeting}, {name}{punctuation}" + + prompt = Prompt.from_function(prompt_with_multiple_defaults) + + # Test with only required parameter + result1 = await prompt.render(arguments={"name": "World"}) + assert result1.messages == [Message("Hello, World!")] + + # Test overriding one default + result2 = await prompt.render(arguments={"name": "World", "greeting": "Hi"}) + assert result2.messages == [Message("Hi, World!")] + + # Test overriding all defaults + result3 = await prompt.render( + arguments={"name": "World", "greeting": "Greetings", "punctuation": "."} + ) + assert result3.messages == [Message("Greetings, World.")] + + async def test_field_defaults_with_type_conversion(self): + """Test Field() defaults work with type conversion for non-string types.""" + from pydantic import Field + + def prompt_with_typed_defaults( + count: int = Field(description="Count"), + multiplier: int = Field(default=2, description="Multiplier"), + ) -> str: + return f"result={count * multiplier}" + + prompt = Prompt.from_function(prompt_with_typed_defaults) + + # Pass count as string (MCP requirement), should use default for multiplier + result = await prompt.render(arguments={"count": "5"}) + assert result.messages == [Message("result=10")] + + class TestPromptCallableAndConcurrency: """Test prompts with callable objects and concurrent execution.""" diff --git a/tests/resources/test_resource_template.py b/tests/resources/test_resource_template.py index 2cb1d1d4a..ed0b37376 100644 --- a/tests/resources/test_resource_template.py +++ b/tests/resources/test_resource_template.py @@ -1007,3 +1007,83 @@ class TestQueryParameterWithWildcards: assert result["path"] == "src/test/data.txt" assert result["encoding"] == "utf-8" # default assert result["lines"] == 50 # provided + + +class TestResourceTemplateFieldDefaults: + """Test resource templates with Field() defaults.""" + + async def test_field_with_default(self): + """Test that Field(default=...) correctly provides default values in resource templates.""" + from pydantic import Field + + def get_data( + id: str = Field(description="Resource ID"), + format: str = Field(default="json", description="Output format"), + ) -> str: + return f"id={id}, format={format}" + + template = ResourceTemplate.from_function( + fn=get_data, + uri_template="data://{id}{?format}", + name="test", + ) + + # Test with only required parameter + resource = await template.create_resource("data://123", {"id": "123"}) + result = await resource.read() + assert result == "id=123, format=json" + + # Test with override + resource = await template.create_resource( + "data://123?format=xml", {"id": "123", "format": "xml"} + ) + result = await resource.read() + assert result == "id=123, format=xml" + + async def test_multiple_field_defaults(self): + """Test multiple query parameters with Field() defaults.""" + from typing import Any + + from pydantic import Field + + def fetch_data( + resource_id: str = Field(description="Resource ID"), + limit: int = Field(default=10, description="Result limit"), + offset: int = Field(default=0, description="Result offset"), + format: str = Field(default="json", description="Output format"), + ) -> dict[str, Any]: + return { + "resource_id": resource_id, + "limit": limit, + "offset": offset, + "format": format, + } + + template = ResourceTemplate.from_function( + fn=fetch_data, + uri_template="api://{resource_id}{?limit,offset,format}", + name="test", + ) + + # Test with only required parameter - all defaults should apply + resource1 = await template.create_resource( + "api://user123", {"resource_id": "user123"} + ) + result1 = await resource1.read() + assert isinstance(result1, dict) + assert result1["resource_id"] == "user123" + assert result1["limit"] == 10 + assert result1["offset"] == 0 + assert result1["format"] == "json" + + # Test with some overrides + resource2 = await template.create_resource( + "api://user123?limit=50&format=xml", + {"resource_id": "user123", "limit": "50", "format": "xml"}, + ) + result2 = await resource2.read() + assert isinstance(result2, dict) + assert result2["resource_id"] == "user123" + assert result2["limit"] == 50 # overridden + assert result2["offset"] == 0 # default + assert result2["format"] == "xml" # overridden diff --git a/uv.lock b/uv.lock index d70b4e42c..32c27dc99 100644 --- a/uv.lock +++ b/uv.lock @@ -696,6 +696,7 @@ dependencies = [ { name = "pydantic", extra = ["email"] }, { name = "pyperclip" }, { name = "python-dotenv" }, + { name = "pyyaml" }, { name = "rich" }, { name = "uvicorn" }, { name = "watchfiles" }, @@ -763,6 +764,7 @@ requires-dist = [ { name = "pydocket", marker = "extra == 'tasks'", specifier = ">=0.17.2" }, { name = "pyperclip", specifier = ">=1.9.0" }, { name = "python-dotenv", specifier = ">=1.1.0" }, + { name = "pyyaml", specifier = ">=6.0,<7.0" }, { name = "rich", specifier = ">=13.9.4" }, { name = "uvicorn", specifier = ">=0.35" }, { name = "watchfiles", specifier = ">=1.0.0" },