From 6a3a3077b7d3894148116a9a51155a79cc36e0bd Mon Sep 17 00:00:00 2001 From: Jeremiah Lowin <153965+jlowin@users.noreply.github.com> Date: Sat, 21 Jun 2025 16:40:54 -0400 Subject: [PATCH 1/4] Add fastmcp inspect command with detailed server analysis MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Add comprehensive server inspection utility supporting both FastMCP 1.x and 2.x - Create detailed info dataclasses for tools, prompts, resources, and templates - Implement CLI command with path:object notation and JSON output - Add version reporting (fastmcp_version, mcp_version, server_version) - Include comprehensive unit tests for utilities and CLI 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude --- src/fastmcp/cli/cli.py | 107 +++++++++ src/fastmcp/utilities/inspect.py | 326 ++++++++++++++++++++++++++ tests/cli/test_inspect.py | 354 ++++++++++++++++++++++++++++ tests/utilities/test_inspect.py | 388 +++++++++++++++++++++++++++++++ 4 files changed, 1175 insertions(+) create mode 100644 src/fastmcp/utilities/inspect.py create mode 100644 tests/cli/test_inspect.py create mode 100644 tests/utilities/test_inspect.py diff --git a/src/fastmcp/cli/cli.py b/src/fastmcp/cli/cli.py index f3e326270..57c77da73 100644 --- a/src/fastmcp/cli/cli.py +++ b/src/fastmcp/cli/cli.py @@ -1,7 +1,9 @@ """FastMCP CLI tools.""" +import asyncio import importlib.metadata import importlib.util +import json import os import platform import subprocess @@ -19,6 +21,7 @@ import fastmcp from fastmcp.cli import claude from fastmcp.cli import run as run_module from fastmcp.server.server import FastMCP +from fastmcp.utilities.inspect import get_fastmcp_info from fastmcp.utilities.logging import get_logger logger = get_logger("cli") @@ -435,3 +438,107 @@ def install( else: logger.error(f"Failed to install {name} in Claude app") sys.exit(1) + + +@app.command() +def inspect( + server_spec: str = typer.Argument( + ..., + help="Python file to inspect, optionally with :object suffix", + ), + output: Annotated[ + Path, + typer.Option( + "--output", + "-o", + help="Output file path for the JSON report (default: server-info.json)", + ), + ] = Path("server-info.json"), +) -> None: + """Inspect a FastMCP server and generate a JSON report. + + This command analyzes a FastMCP server (v1.x or v2.x) and generates + a comprehensive JSON report containing information about the server's + name, instructions, version, tools, prompts, resources, templates, + and capabilities. + + Examples: + fastmcp inspect server.py + fastmcp inspect server.py -o report.json + fastmcp inspect server.py:mcp -o analysis.json + fastmcp inspect path/to/server.py:app -o /tmp/server-info.json + """ + + # Parse the server specification + file, server_object = run_module.parse_file_path(server_spec) + + logger.debug( + "Inspecting server", + extra={ + "file": str(file), + "server_object": server_object, + "output": str(output), + }, + ) + + try: + # Import the server + server = run_module.import_server(file, server_object) + + # Get server information + async def get_info(): + return await get_fastmcp_info(server) + + info = asyncio.run(get_info()) + + # Convert to dict for JSON serialization + def convert_dataclass_to_dict(obj): + """Convert dataclass instances to dicts for JSON serialization.""" + if hasattr(obj, "__dataclass_fields__"): + return { + k: convert_dataclass_to_dict(v) for k, v in obj.__dict__.items() + } + elif isinstance(obj, list): + return [convert_dataclass_to_dict(item) for item in obj] + elif isinstance(obj, set): + return list(obj) + elif hasattr(obj, "model_dump"): # Pydantic models + return obj.model_dump() + elif hasattr(obj, "__dict__"): # Other objects with __dict__ + return { + k: convert_dataclass_to_dict(v) for k, v in obj.__dict__.items() + } + else: + return obj + + info_dict = convert_dataclass_to_dict(info) + + # Ensure output directory exists + output.parent.mkdir(parents=True, exist_ok=True) + + # Write JSON report (always pretty-printed) + with output.open("w", encoding="utf-8") as f: + json.dump(info_dict, f, indent=2, ensure_ascii=False) + + logger.info(f"Server inspection complete. Report saved to {output}") + + # Print summary to console + console.print( + f"[bold green]✓[/bold green] Inspected server: [bold]{info.name}[/bold]" + ) + console.print(f" Tools: {len(info.tools)}") + console.print(f" Prompts: {len(info.prompts)}") + console.print(f" Resources: {len(info.resources)}") + console.print(f" Templates: {len(info.templates)}") + console.print(f" Report saved to: [cyan]{output}[/cyan]") + + except Exception as e: + logger.error( + f"Failed to inspect server: {e}", + extra={ + "server_spec": server_spec, + "error": str(e), + }, + ) + console.print(f"[bold red]✗[/bold red] Failed to inspect server: {e}") + sys.exit(1) diff --git a/src/fastmcp/utilities/inspect.py b/src/fastmcp/utilities/inspect.py new file mode 100644 index 000000000..9ddd2c61f --- /dev/null +++ b/src/fastmcp/utilities/inspect.py @@ -0,0 +1,326 @@ +"""Utilities for inspecting FastMCP instances.""" + +from __future__ import annotations + +import importlib.metadata +from dataclasses import dataclass +from typing import Any + +from mcp.server.fastmcp import FastMCP as FastMCP1x + +import fastmcp +from fastmcp.server.server import FastMCP + + +@dataclass +class ToolInfo: + """Information about a tool.""" + + key: str + name: str + description: str | None + input_schema: dict[str, Any] + annotations: dict[str, Any] | None = None + tags: list[str] | None = None + enabled: bool | None = None + + +@dataclass +class PromptInfo: + """Information about a prompt.""" + + key: str + name: str + description: str | None + arguments: list[dict[str, Any]] | None = None + tags: list[str] | None = None + enabled: bool | None = None + + +@dataclass +class ResourceInfo: + """Information about a resource.""" + + key: str + uri: str + name: str | None + description: str | None + mime_type: str | None = None + tags: list[str] | None = None + enabled: bool | None = None + + +@dataclass +class TemplateInfo: + """Information about a resource template.""" + + key: str + uri_template: str + name: str | None + description: str | None + mime_type: str | None = None + tags: list[str] | None = None + enabled: bool | None = None + + +@dataclass +class FastMCPInfo: + """Information extracted from a FastMCP instance.""" + + name: str + instructions: str | None + fastmcp_version: str + mcp_version: str + server_version: str + tools: list[ToolInfo] + prompts: list[PromptInfo] + resources: list[ResourceInfo] + templates: list[TemplateInfo] + capabilities: dict[str, Any] + + +async def get_fastmcp_info_v2(mcp: FastMCP[Any]) -> FastMCPInfo: + """Extract information from a FastMCP v2.x instance. + + Args: + mcp: The FastMCP v2.x instance to inspect + + Returns: + FastMCPInfo dataclass containing the extracted information + """ + # Get all the components using FastMCP2's direct methods + tools_dict = await mcp.get_tools() + prompts_dict = await mcp.get_prompts() + resources_dict = await mcp.get_resources() + templates_dict = await mcp.get_resource_templates() + + # Extract detailed tool information + tool_infos = [] + for key, tool in tools_dict.items(): + # Convert to MCP tool to get input schema + mcp_tool = tool.to_mcp_tool(name=key) + tool_infos.append( + ToolInfo( + key=key, + name=tool.name or key, + description=tool.description, + input_schema=mcp_tool.inputSchema if mcp_tool.inputSchema else {}, + annotations=tool.annotations.model_dump() if tool.annotations else None, + tags=list(tool.tags) if tool.tags else None, + enabled=tool.enabled, + ) + ) + + # Extract detailed prompt information + prompt_infos = [] + for key, prompt in prompts_dict.items(): + prompt_infos.append( + PromptInfo( + key=key, + name=prompt.name or key, + description=prompt.description, + arguments=[arg.model_dump() for arg in prompt.arguments] + if prompt.arguments + else None, + tags=list(prompt.tags) if prompt.tags else None, + enabled=prompt.enabled, + ) + ) + + # Extract detailed resource information + resource_infos = [] + for key, resource in resources_dict.items(): + resource_infos.append( + ResourceInfo( + key=key, + uri=key, # For v2, key is the URI + name=resource.name, + description=resource.description, + mime_type=resource.mime_type, + tags=list(resource.tags) if resource.tags else None, + enabled=resource.enabled, + ) + ) + + # Extract detailed template information + template_infos = [] + for key, template in templates_dict.items(): + template_infos.append( + TemplateInfo( + key=key, + uri_template=key, # For v2, key is the URI template + name=template.name, + description=template.description, + mime_type=template.mime_type, + tags=list(template.tags) if template.tags else None, + enabled=template.enabled, + ) + ) + + # Basic MCP capabilities that FastMCP supports + capabilities = { + "tools": {"listChanged": True}, + "resources": {"subscribe": False, "listChanged": False}, + "prompts": {"listChanged": False}, + "logging": {}, + } + + return FastMCPInfo( + name=mcp.name, + instructions=mcp.instructions, + fastmcp_version=fastmcp.__version__, + mcp_version=importlib.metadata.version("mcp"), + server_version=fastmcp.__version__, # v2.x uses FastMCP version + tools=tool_infos, + prompts=prompt_infos, + resources=resource_infos, + templates=template_infos, + capabilities=capabilities, + ) + + +async def get_fastmcp_info_v1(mcp: Any) -> FastMCPInfo: + """Extract information from a FastMCP v1.x instance using a Client. + + Args: + mcp: The FastMCP v1.x instance to inspect + + Returns: + FastMCPInfo dataclass containing the extracted information + """ + from fastmcp import Client + + # Use a client to interact with the FastMCP1x server + async with Client(mcp) as client: + # Get components via client calls (these return MCP objects) + mcp_tools = await client.list_tools() + mcp_prompts = await client.list_prompts() + mcp_resources = await client.list_resources() + + # Try to get resource templates (FastMCP 1.x does have templates) + try: + mcp_templates = await client.list_resource_templates() + except Exception: + mcp_templates = [] + + # Extract detailed tool information from MCP Tool objects + tool_infos = [] + for mcp_tool in mcp_tools: + # Extract annotations if they exist + annotations = None + if hasattr(mcp_tool, "annotations") and mcp_tool.annotations: + if hasattr(mcp_tool.annotations, "model_dump"): + annotations = mcp_tool.annotations.model_dump() + elif isinstance(mcp_tool.annotations, dict): + annotations = mcp_tool.annotations + else: + annotations = None + + tool_infos.append( + ToolInfo( + key=mcp_tool.name, # For 1.x, key and name are the same + name=mcp_tool.name, + description=mcp_tool.description, + input_schema=mcp_tool.inputSchema if mcp_tool.inputSchema else {}, + annotations=annotations, + tags=None, # 1.x doesn't have tags + enabled=None, # 1.x doesn't have enabled field + ) + ) + + # Extract detailed prompt information from MCP Prompt objects + prompt_infos = [] + for mcp_prompt in mcp_prompts: + # Convert arguments if they exist + arguments = None + if hasattr(mcp_prompt, "arguments") and mcp_prompt.arguments: + arguments = [arg.model_dump() for arg in mcp_prompt.arguments] + + prompt_infos.append( + PromptInfo( + key=mcp_prompt.name, # For 1.x, key and name are the same + name=mcp_prompt.name, + description=mcp_prompt.description, + arguments=arguments, + tags=None, # 1.x doesn't have tags + enabled=None, # 1.x doesn't have enabled field + ) + ) + + # Extract detailed resource information from MCP Resource objects + resource_infos = [] + for mcp_resource in mcp_resources: + resource_infos.append( + ResourceInfo( + key=str(mcp_resource.uri), # For 1.x, key and uri are the same + uri=str(mcp_resource.uri), + name=mcp_resource.name, + description=mcp_resource.description, + mime_type=mcp_resource.mimeType, + tags=None, # 1.x doesn't have tags + enabled=None, # 1.x doesn't have enabled field + ) + ) + + # Extract detailed template information from MCP ResourceTemplate objects + template_infos = [] + for mcp_template in mcp_templates: + template_infos.append( + TemplateInfo( + key=str( + mcp_template.uriTemplate + ), # For 1.x, key and uriTemplate are the same + uri_template=str(mcp_template.uriTemplate), + name=mcp_template.name, + description=mcp_template.description, + mime_type=mcp_template.mimeType, + tags=None, # 1.x doesn't have tags + enabled=None, # 1.x doesn't have enabled field + ) + ) + + # Basic MCP capabilities + capabilities = { + "tools": {"listChanged": True}, + "resources": {"subscribe": False, "listChanged": False}, + "prompts": {"listChanged": False}, + "logging": {}, + } + + return FastMCPInfo( + name=mcp.name, + instructions=getattr(mcp, "instructions", None), + fastmcp_version=fastmcp.__version__, # Report current fastmcp version + mcp_version=importlib.metadata.version("mcp"), + server_version="1.0", # FastMCP 1.x version + tools=tool_infos, + prompts=prompt_infos, + resources=resource_infos, + templates=template_infos, # FastMCP1x does have templates + capabilities=capabilities, + ) + + +def _is_fastmcp_v1(mcp: Any) -> bool: + """Check if the given instance is a FastMCP v1.x instance.""" + + # Check if it's an instance of FastMCP1x and not FastMCP2 + return isinstance(mcp, FastMCP1x) and not isinstance(mcp, FastMCP) + + +async def get_fastmcp_info(mcp: FastMCP[Any] | Any) -> FastMCPInfo: + """Extract information from a FastMCP instance into a dataclass. + + This function automatically detects whether the instance is FastMCP v1.x or v2.x + and uses the appropriate extraction method. + + Args: + mcp: The FastMCP instance to inspect (v1.x or v2.x) + + Returns: + FastMCPInfo dataclass containing the extracted information + """ + if _is_fastmcp_v1(mcp): + return await get_fastmcp_info_v1(mcp) + else: + return await get_fastmcp_info_v2(mcp) diff --git a/tests/cli/test_inspect.py b/tests/cli/test_inspect.py new file mode 100644 index 000000000..fa2f992cb --- /dev/null +++ b/tests/cli/test_inspect.py @@ -0,0 +1,354 @@ +"""Tests for the CLI inspect command.""" + +import json +import tempfile +from pathlib import Path + +from typer.testing import CliRunner + +from fastmcp.cli.cli import app + + +class TestInspectCommand: + """Tests for the fastmcp inspect CLI command.""" + + def setup_method(self): + """Set up test fixtures.""" + self.runner = CliRunner() + + def test_inspect_basic_server(self): + """Test inspecting a basic FastMCP 2.x server.""" + # Create a temporary server file + server_content = ''' +from fastmcp import FastMCP + +mcp = FastMCP("TestServer", instructions="A test server") + +@mcp.tool +def add(a: int, b: int) -> int: + """Add two numbers.""" + return a + b + +@mcp.resource("resource://data") +def get_data() -> str: + """Get test data.""" + return "test data" + +@mcp.prompt +def test_prompt(message: str) -> list: + """Test prompt.""" + return [{"role": "user", "content": message}] +''' + + with tempfile.NamedTemporaryFile(mode="w", suffix=".py", delete=False) as f: + f.write(server_content) + server_file = f.name + + with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f: + output_file = f.name + + try: + # Run the inspect command + result = self.runner.invoke( + app, ["inspect", server_file, "-o", output_file] + ) + + assert result.exit_code == 0 + assert "✓ Inspected server: TestServer" in result.stdout + assert "Tools: 1" in result.stdout + assert "Prompts: 1" in result.stdout + assert "Resources: 1" in result.stdout + + # Check the JSON output + with open(output_file) as f: + data = json.load(f) + + assert data["name"] == "TestServer" + assert data["instructions"] == "A test server" + assert "fastmcp_version" in data + assert "mcp_version" in data + assert "server_version" in data + + # Check tools + assert len(data["tools"]) == 1 + tool = data["tools"][0] + assert tool["key"] == "add" + assert tool["name"] == "add" + assert tool["description"] == "Add two numbers." + assert "input_schema" in tool + assert tool["enabled"] is True + + # Check resources + assert len(data["resources"]) == 1 + resource = data["resources"][0] + assert resource["key"] == "resource://data" + assert resource["uri"] == "resource://data" + assert resource["name"] == "get_data" + + # Check prompts + assert len(data["prompts"]) == 1 + prompt = data["prompts"][0] + assert prompt["key"] == "test_prompt" + assert prompt["name"] == "test_prompt" + assert prompt["description"] == "Test prompt." + + # Check capabilities + assert "capabilities" in data + assert "tools" in data["capabilities"] + + finally: + # Clean up + Path(server_file).unlink(missing_ok=True) + Path(output_file).unlink(missing_ok=True) + + def test_inspect_with_object_spec(self): + """Test inspecting a server with object specification.""" + server_content = ''' +from fastmcp import FastMCP + +server = FastMCP("ObjectSpecServer") + +@server.tool +def multiply(a: int, b: int) -> int: + """Multiply two numbers.""" + return a * b +''' + + with tempfile.NamedTemporaryFile(mode="w", suffix=".py", delete=False) as f: + f.write(server_content) + server_file = f.name + + with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f: + output_file = f.name + + try: + # Run the inspect command with object specification + result = self.runner.invoke( + app, ["inspect", f"{server_file}:server", "-o", output_file] + ) + + assert result.exit_code == 0 + assert "✓ Inspected server: ObjectSpecServer" in result.stdout + + # Check the JSON output + with open(output_file) as f: + data = json.load(f) + + assert data["name"] == "ObjectSpecServer" + assert len(data["tools"]) == 1 + assert data["tools"][0]["name"] == "multiply" + + finally: + # Clean up + Path(server_file).unlink(missing_ok=True) + Path(output_file).unlink(missing_ok=True) + + def test_inspect_default_output(self): + """Test inspecting with default output filename.""" + server_content = ''' +from fastmcp import FastMCP + +mcp = FastMCP("DefaultOutputServer") + +@mcp.tool +def test_tool() -> str: + """Test tool.""" + return "test" +''' + + with tempfile.NamedTemporaryFile(mode="w", suffix=".py", delete=False) as f: + f.write(server_content) + server_file = f.name + + try: + # Run the inspect command without specifying output file + result = self.runner.invoke(app, ["inspect", server_file]) + + assert result.exit_code == 0 + assert "✓ Inspected server: DefaultOutputServer" in result.stdout + assert "Report saved to: server-info.json" in result.stdout + + # Check the default output file exists + default_output = Path("server-info.json") + assert default_output.exists() + + # Check the JSON content + with open(default_output) as f: + data = json.load(f) + + assert data["name"] == "DefaultOutputServer" + + finally: + # Clean up + Path(server_file).unlink(missing_ok=True) + Path("server-info.json").unlink(missing_ok=True) + + def test_inspect_invalid_server_file(self): + """Test inspecting a non-existent server file.""" + result = self.runner.invoke( + app, ["inspect", "nonexistent.py", "-o", "output.json"] + ) + + assert result.exit_code == 1 + # The error happens at the file parsing level, so no stdout output + + def test_inspect_server_with_error(self): + """Test inspecting a server file with syntax errors.""" + server_content = """ +from fastmcp import FastMCP + +mcp = FastMCP("ErrorServer") +# Syntax error below +@mcp.tool +def broken_tool( +""" + + with tempfile.NamedTemporaryFile(mode="w", suffix=".py", delete=False) as f: + f.write(server_content) + server_file = f.name + + try: + result = self.runner.invoke( + app, ["inspect", server_file, "-o", "output.json"] + ) + + assert result.exit_code == 1 + assert "✗ Failed to inspect server:" in result.stdout + + finally: + # Clean up + Path(server_file).unlink(missing_ok=True) + Path("output.json").unlink(missing_ok=True) + + def test_inspect_comprehensive_json_structure(self): + """Test that the JSON output has the correct structure.""" + server_content = ''' +from fastmcp import FastMCP + +mcp = FastMCP("ComprehensiveServer", instructions="Full test server") + +@mcp.tool +def calculate(x: int, y: int) -> int: + """Calculate something.""" + return x + y + +@mcp.resource("resource://static") +def static_resource() -> str: + """Static resource.""" + return "static" + +@mcp.resource("resource://template/{id}") +def template_resource(id: str) -> str: + """Template resource.""" + return f"data-{id}" + +@mcp.prompt +def analysis_prompt(data: str) -> list: + """Analysis prompt.""" + return [{"role": "user", "content": f"Analyze: {data}"}] +''' + + with tempfile.NamedTemporaryFile(mode="w", suffix=".py", delete=False) as f: + f.write(server_content) + server_file = f.name + + with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f: + output_file = f.name + + try: + result = self.runner.invoke( + app, ["inspect", server_file, "-o", output_file] + ) + + assert result.exit_code == 0 + + # Load and validate JSON structure + with open(output_file) as f: + data = json.load(f) + + # Check top-level structure + required_fields = [ + "name", + "instructions", + "fastmcp_version", + "mcp_version", + "server_version", + "tools", + "prompts", + "resources", + "templates", + "capabilities", + ] + for field in required_fields: + assert field in data, f"Missing field: {field}" + + # Check version fields are strings + assert isinstance(data["fastmcp_version"], str) + assert isinstance(data["mcp_version"], str) + assert isinstance(data["server_version"], str) + + # Check that we have the expected components + assert len(data["tools"]) == 1 + assert len(data["resources"]) == 1 + assert len(data["templates"]) == 1 + assert len(data["prompts"]) == 1 + + # Check tool structure + tool = data["tools"][0] + tool_fields = [ + "key", + "name", + "description", + "input_schema", + "annotations", + "tags", + "enabled", + ] + for field in tool_fields: + assert field in tool, f"Missing tool field: {field}" + + # Check resource structure + resource = data["resources"][0] + resource_fields = [ + "key", + "uri", + "name", + "description", + "mime_type", + "tags", + "enabled", + ] + for field in resource_fields: + assert field in resource, f"Missing resource field: {field}" + + # Check template structure + template = data["templates"][0] + template_fields = [ + "key", + "uri_template", + "name", + "description", + "mime_type", + "tags", + "enabled", + ] + for field in template_fields: + assert field in template, f"Missing template field: {field}" + + # Check prompt structure + prompt = data["prompts"][0] + prompt_fields = [ + "key", + "name", + "description", + "arguments", + "tags", + "enabled", + ] + for field in prompt_fields: + assert field in prompt, f"Missing prompt field: {field}" + + finally: + # Clean up + Path(server_file).unlink(missing_ok=True) + Path(output_file).unlink(missing_ok=True) diff --git a/tests/utilities/test_inspect.py b/tests/utilities/test_inspect.py new file mode 100644 index 000000000..0ca03b6c2 --- /dev/null +++ b/tests/utilities/test_inspect.py @@ -0,0 +1,388 @@ +"""Tests for the inspect.py module.""" + +# Import FastMCP1x for testing (always available since mcp is a dependency) +from mcp.server.fastmcp import FastMCP as FastMCP1x + +import fastmcp +from fastmcp import Client, FastMCP +from fastmcp.utilities.inspect import ( + FastMCPInfo, + ToolInfo, + _is_fastmcp_v1, + get_fastmcp_info, + get_fastmcp_info_v1, +) + + +class TestFastMCPInfo: + """Tests for the FastMCPInfo dataclass.""" + + def test_fastmcp_info_creation(self): + """Test that FastMCPInfo can be created with all required fields.""" + tool = ToolInfo( + key="tool1", name="tool1", description="Test tool", input_schema={} + ) + info = FastMCPInfo( + name="TestServer", + instructions="Test instructions", + fastmcp_version="1.0.0", + mcp_version="1.0.0", + server_version="1.0.0", + tools=[tool], + prompts=[], + resources=[], + templates=[], + capabilities={"tools": {"listChanged": True}}, + ) + + assert info.name == "TestServer" + assert info.instructions == "Test instructions" + assert info.fastmcp_version == "1.0.0" + assert info.mcp_version == "1.0.0" + assert info.server_version == "1.0.0" + assert len(info.tools) == 1 + assert info.tools[0].name == "tool1" + assert info.capabilities == {"tools": {"listChanged": True}} + + def test_fastmcp_info_with_none_instructions(self): + """Test that FastMCPInfo works with None instructions.""" + info = FastMCPInfo( + name="TestServer", + instructions=None, + fastmcp_version="1.0.0", + mcp_version="1.0.0", + server_version="1.0.0", + tools=[], + prompts=[], + resources=[], + templates=[], + capabilities={}, + ) + + assert info.instructions is None + + +class TestGetFastMCPInfo: + """Tests for the get_fastmcp_info function.""" + + async def test_empty_server(self): + """Test get_fastmcp_info with an empty server.""" + mcp = FastMCP("EmptyServer", instructions="Empty server for testing") + + info = await get_fastmcp_info(mcp) + + assert info.name == "EmptyServer" + assert info.instructions == "Empty server for testing" + assert info.fastmcp_version == fastmcp.__version__ + assert info.mcp_version is not None + assert info.server_version == fastmcp.__version__ # v2.x uses FastMCP version + assert info.tools == [] + assert info.prompts == [] + assert info.resources == [] + assert info.templates == [] + assert "tools" in info.capabilities + assert "resources" in info.capabilities + assert "prompts" in info.capabilities + assert "logging" in info.capabilities + + async def test_server_with_tools(self): + """Test get_fastmcp_info with a server that has tools.""" + mcp = FastMCP("ToolServer") + + @mcp.tool + def add_numbers(a: int, b: int) -> int: + return a + b + + @mcp.tool + def greet(name: str) -> str: + return f"Hello, {name}!" + + info = await get_fastmcp_info(mcp) + + assert info.name == "ToolServer" + assert len(info.tools) == 2 + tool_names = [tool.name for tool in info.tools] + assert "add_numbers" in tool_names + assert "greet" in tool_names + + async def test_server_with_resources(self): + """Test get_fastmcp_info with a server that has resources.""" + mcp = FastMCP("ResourceServer") + + @mcp.resource("resource://static") + def get_static_data() -> str: + return "Static data" + + @mcp.resource("resource://dynamic/{param}") + def get_dynamic_data(param: str) -> str: + return f"Dynamic data: {param}" + + info = await get_fastmcp_info(mcp) + + assert info.name == "ResourceServer" + assert len(info.resources) == 1 # Static resource + assert len(info.templates) == 1 # Dynamic resource becomes template + resource_uris = [res.uri for res in info.resources] + template_uris = [tmpl.uri_template for tmpl in info.templates] + assert "resource://static" in resource_uris + assert "resource://dynamic/{param}" in template_uris + + async def test_server_with_prompts(self): + """Test get_fastmcp_info with a server that has prompts.""" + mcp = FastMCP("PromptServer") + + @mcp.prompt + def analyze_data(data: str) -> list: + return [{"role": "user", "content": f"Analyze: {data}"}] + + @mcp.prompt("custom_prompt") + def custom_analysis(text: str) -> list: + return [{"role": "user", "content": f"Custom: {text}"}] + + info = await get_fastmcp_info(mcp) + + assert info.name == "PromptServer" + assert len(info.prompts) == 2 + prompt_names = [prompt.name for prompt in info.prompts] + assert "analyze_data" in prompt_names + assert "custom_prompt" in prompt_names + + async def test_comprehensive_server(self): + """Test get_fastmcp_info with a server that has all component types.""" + mcp = FastMCP("ComprehensiveServer", instructions="A server with everything") + + # Add a tool + @mcp.tool + def calculate(x: int, y: int) -> int: + return x * y + + # Add a resource + @mcp.resource("resource://data") + def get_data() -> str: + return "Some data" + + # Add a template + @mcp.resource("resource://item/{id}") + def get_item(id: str) -> str: + return f"Item {id}" + + # Add a prompt + @mcp.prompt + def analyze(content: str) -> list: + return [{"role": "user", "content": content}] + + info = await get_fastmcp_info(mcp) + + assert info.name == "ComprehensiveServer" + assert info.instructions == "A server with everything" + assert info.fastmcp_version == fastmcp.__version__ + + # Check all components are present + assert len(info.tools) == 1 + tool_names = [tool.name for tool in info.tools] + assert "calculate" in tool_names + + assert len(info.resources) == 1 + resource_uris = [res.uri for res in info.resources] + assert "resource://data" in resource_uris + + assert len(info.templates) == 1 + template_uris = [tmpl.uri_template for tmpl in info.templates] + assert "resource://item/{id}" in template_uris + + assert len(info.prompts) == 1 + prompt_names = [prompt.name for prompt in info.prompts] + assert "analyze" in prompt_names + + # Check capabilities + assert "tools" in info.capabilities + assert "resources" in info.capabilities + assert "prompts" in info.capabilities + assert "logging" in info.capabilities + + async def test_server_no_instructions(self): + """Test get_fastmcp_info with a server that has no instructions.""" + mcp = FastMCP("NoInstructionsServer") + + info = await get_fastmcp_info(mcp) + + assert info.name == "NoInstructionsServer" + assert info.instructions is None + + async def test_server_with_client_integration(self): + """Test that the extracted info matches what a client would see.""" + mcp = FastMCP("IntegrationServer") + + @mcp.tool + def test_tool() -> str: + return "test" + + @mcp.resource("resource://test") + def test_resource() -> str: + return "test resource" + + @mcp.prompt + def test_prompt() -> list: + return [{"role": "user", "content": "test"}] + + # Get info using our function + info = await get_fastmcp_info(mcp) + + # Verify using client + async with Client(mcp) as client: + tools = await client.list_tools() + resources = await client.list_resources() + prompts = await client.list_prompts() + + assert len(info.tools) == len(tools) + assert len(info.resources) == len(resources) + assert len(info.prompts) == len(prompts) + + assert info.tools[0].name == tools[0].name + assert info.resources[0].uri == str(resources[0].uri) + assert info.prompts[0].name == prompts[0].name + + +class TestFastMCP1xCompatibility: + """Tests for FastMCP 1.x compatibility.""" + + async def test_fastmcp1x_detection(self): + """Test that FastMCP1x instances are correctly detected.""" + mcp1x = FastMCP1x("Test1x") + mcp2x = FastMCP("Test2x") + + assert _is_fastmcp_v1(mcp1x) is True + assert _is_fastmcp_v1(mcp2x) is False + + async def test_fastmcp1x_empty_server(self): + """Test get_fastmcp_info_v1 with an empty FastMCP1x server.""" + mcp = FastMCP1x("Test1x") + + info = await get_fastmcp_info_v1(mcp) + + assert info.name == "Test1x" + assert info.instructions is None + assert info.fastmcp_version == fastmcp.__version__ + assert info.mcp_version is not None + assert info.server_version == "1.0" # v1.x servers use "1.0" + assert info.tools == [] + assert info.prompts == [] + assert info.resources == [] + assert info.templates == [] # No templates added in this test + assert "tools" in info.capabilities + + async def test_fastmcp1x_with_tools(self): + """Test get_fastmcp_info_v1 with a FastMCP1x server that has tools.""" + mcp = FastMCP1x("Test1x") + + @mcp.tool() + def add_numbers(a: int, b: int) -> int: + return a + b + + @mcp.tool() + def greet(name: str) -> str: + return f"Hello, {name}!" + + info = await get_fastmcp_info_v1(mcp) + + assert info.name == "Test1x" + assert len(info.tools) == 2 + tool_names = [tool.name for tool in info.tools] + assert "add_numbers" in tool_names + assert "greet" in tool_names + + async def test_fastmcp1x_with_resources(self): + """Test get_fastmcp_info_v1 with a FastMCP1x server that has resources.""" + mcp = FastMCP1x("Test1x") + + @mcp.resource("resource://data") + def get_data() -> str: + return "Some data" + + info = await get_fastmcp_info_v1(mcp) + + assert info.name == "Test1x" + assert len(info.resources) == 1 + resource_uris = [res.uri for res in info.resources] + assert "resource://data" in resource_uris + assert len(info.templates) == 0 # No templates added in this test + + async def test_fastmcp1x_with_prompts(self): + """Test get_fastmcp_info_v1 with a FastMCP1x server that has prompts.""" + mcp = FastMCP1x("Test1x") + + @mcp.prompt("analyze") + def analyze_data(data: str) -> list: + return [{"role": "user", "content": f"Analyze: {data}"}] + + info = await get_fastmcp_info_v1(mcp) + + assert info.name == "Test1x" + assert len(info.prompts) == 1 + prompt_names = [prompt.name for prompt in info.prompts] + assert "analyze" in prompt_names + + async def test_dispatcher_with_fastmcp1x(self): + """Test that the main get_fastmcp_info function correctly dispatches to v1.""" + mcp = FastMCP1x("Test1x") + + @mcp.tool() + def test_tool() -> str: + return "test" + + info = await get_fastmcp_info(mcp) + + assert info.name == "Test1x" + assert len(info.tools) == 1 + tool_names = [tool.name for tool in info.tools] + assert "test_tool" in tool_names + assert len(info.templates) == 0 # No templates added in this test + + async def test_dispatcher_with_fastmcp2x(self): + """Test that the main get_fastmcp_info function correctly dispatches to v2.""" + mcp = FastMCP("Test2x") + + @mcp.tool + def test_tool() -> str: + return "test" + + info = await get_fastmcp_info(mcp) + + assert info.name == "Test2x" + assert len(info.tools) == 1 + tool_names = [tool.name for tool in info.tools] + assert "test_tool" in tool_names + + async def test_fastmcp1x_vs_fastmcp2x_comparison(self): + """Test that both versions can be inspected and compared.""" + mcp1x = FastMCP1x("Test1x") + mcp2x = FastMCP("Test2x") + + @mcp1x.tool() + def tool1x() -> str: + return "1x" + + @mcp2x.tool + def tool2x() -> str: + return "2x" + + info1x = await get_fastmcp_info(mcp1x) + info2x = await get_fastmcp_info(mcp2x) + + assert info1x.name == "Test1x" + assert info2x.name == "Test2x" + assert len(info1x.tools) == 1 + assert len(info2x.tools) == 1 + + tool1x_names = [tool.name for tool in info1x.tools] + tool2x_names = [tool.name for tool in info2x.tools] + assert "tool1x" in tool1x_names + assert "tool2x" in tool2x_names + + # Check server versions + assert info1x.server_version == "1.0" + assert info2x.server_version == fastmcp.__version__ + + # No templates added in these tests + assert len(info1x.templates) == 0 + assert len(info2x.templates) == 0 From bc1e185141d9e76fef4a99037c367ca3c974fc49 Mon Sep 17 00:00:00 2001 From: Jeremiah Lowin <153965+jlowin@users.noreply.github.com> Date: Sat, 21 Jun 2025 16:43:40 -0400 Subject: [PATCH 2/4] Use dataclasses.asdict() for better maintainability MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude --- src/fastmcp/cli/cli.py | 23 ++++++++++------------- 1 file changed, 10 insertions(+), 13 deletions(-) diff --git a/src/fastmcp/cli/cli.py b/src/fastmcp/cli/cli.py index 57c77da73..4b31502dc 100644 --- a/src/fastmcp/cli/cli.py +++ b/src/fastmcp/cli/cli.py @@ -492,26 +492,23 @@ def inspect( info = asyncio.run(get_info()) # Convert to dict for JSON serialization - def convert_dataclass_to_dict(obj): - """Convert dataclass instances to dicts for JSON serialization.""" - if hasattr(obj, "__dataclass_fields__"): - return { - k: convert_dataclass_to_dict(v) for k, v in obj.__dict__.items() - } - elif isinstance(obj, list): - return [convert_dataclass_to_dict(item) for item in obj] + from dataclasses import asdict + + def convert_for_json(obj): + """Convert objects for JSON serialization.""" + if isinstance(obj, list): + return [convert_for_json(item) for item in obj] elif isinstance(obj, set): return list(obj) elif hasattr(obj, "model_dump"): # Pydantic models return obj.model_dump() - elif hasattr(obj, "__dict__"): # Other objects with __dict__ - return { - k: convert_dataclass_to_dict(v) for k, v in obj.__dict__.items() - } else: return obj - info_dict = convert_dataclass_to_dict(info) + info_dict = asdict( + info, + dict_factory=lambda fields: {k: convert_for_json(v) for k, v in fields}, + ) # Ensure output directory exists output.parent.mkdir(parents=True, exist_ok=True) From 8506f784538fa304c241a50e17c1f3e9ea41b8b6 Mon Sep 17 00:00:00 2001 From: Jeremiah Lowin <153965+jlowin@users.noreply.github.com> Date: Sat, 21 Jun 2025 20:04:39 -0400 Subject: [PATCH 3/4] Fix event loop conflict in inspect CLI command Co-Authored-By: Claude --- src/fastmcp/cli/cli.py | 38 ++-- src/fastmcp/utilities/inspect.py | 10 +- tests/cli/test_inspect.py | 354 ------------------------------- tests/utilities/test_inspect.py | 34 +-- 4 files changed, 38 insertions(+), 398 deletions(-) delete mode 100644 tests/cli/test_inspect.py diff --git a/src/fastmcp/cli/cli.py b/src/fastmcp/cli/cli.py index 4b31502dc..d3e34524e 100644 --- a/src/fastmcp/cli/cli.py +++ b/src/fastmcp/cli/cli.py @@ -3,7 +3,6 @@ import asyncio import importlib.metadata import importlib.util -import json import os import platform import subprocess @@ -13,6 +12,7 @@ from typing import Annotated import dotenv import typer +from pydantic import TypeAdapter from rich.console import Console from rich.table import Table from typer import Context, Exit @@ -21,7 +21,7 @@ import fastmcp from fastmcp.cli import claude from fastmcp.cli import run as run_module from fastmcp.server.server import FastMCP -from fastmcp.utilities.inspect import get_fastmcp_info +from fastmcp.utilities.inspect import FastMCPInfo, inspect_fastmcp from fastmcp.utilities.logging import get_logger logger = get_logger("cli") @@ -487,35 +487,29 @@ def inspect( # Get server information async def get_info(): - return await get_fastmcp_info(server) + return await inspect_fastmcp(server) - info = asyncio.run(get_info()) + try: + # Try to use existing event loop if available + asyncio.get_running_loop() + # If there's already a loop running, we need to run in a thread + import concurrent.futures - # Convert to dict for JSON serialization - from dataclasses import asdict + with concurrent.futures.ThreadPoolExecutor() as executor: + future = executor.submit(asyncio.run, get_info()) + info = future.result() + except RuntimeError: + # No running loop, safe to use asyncio.run + info = asyncio.run(get_info()) - def convert_for_json(obj): - """Convert objects for JSON serialization.""" - if isinstance(obj, list): - return [convert_for_json(item) for item in obj] - elif isinstance(obj, set): - return list(obj) - elif hasattr(obj, "model_dump"): # Pydantic models - return obj.model_dump() - else: - return obj - - info_dict = asdict( - info, - dict_factory=lambda fields: {k: convert_for_json(v) for k, v in fields}, - ) + info_json = TypeAdapter(FastMCPInfo).dump_json(info, indent=2) # Ensure output directory exists output.parent.mkdir(parents=True, exist_ok=True) # Write JSON report (always pretty-printed) with output.open("w", encoding="utf-8") as f: - json.dump(info_dict, f, indent=2, ensure_ascii=False) + f.write(info_json.decode("utf-8")) logger.info(f"Server inspection complete. Report saved to {output}") diff --git a/src/fastmcp/utilities/inspect.py b/src/fastmcp/utilities/inspect.py index 9ddd2c61f..da73acf72 100644 --- a/src/fastmcp/utilities/inspect.py +++ b/src/fastmcp/utilities/inspect.py @@ -79,7 +79,7 @@ class FastMCPInfo: capabilities: dict[str, Any] -async def get_fastmcp_info_v2(mcp: FastMCP[Any]) -> FastMCPInfo: +async def inspect_fastmcp_v2(mcp: FastMCP[Any]) -> FastMCPInfo: """Extract information from a FastMCP v2.x instance. Args: @@ -179,7 +179,7 @@ async def get_fastmcp_info_v2(mcp: FastMCP[Any]) -> FastMCPInfo: ) -async def get_fastmcp_info_v1(mcp: Any) -> FastMCPInfo: +async def inspect_fastmcp_v1(mcp: Any) -> FastMCPInfo: """Extract information from a FastMCP v1.x instance using a Client. Args: @@ -308,7 +308,7 @@ def _is_fastmcp_v1(mcp: Any) -> bool: return isinstance(mcp, FastMCP1x) and not isinstance(mcp, FastMCP) -async def get_fastmcp_info(mcp: FastMCP[Any] | Any) -> FastMCPInfo: +async def inspect_fastmcp(mcp: FastMCP[Any] | Any) -> FastMCPInfo: """Extract information from a FastMCP instance into a dataclass. This function automatically detects whether the instance is FastMCP v1.x or v2.x @@ -321,6 +321,6 @@ async def get_fastmcp_info(mcp: FastMCP[Any] | Any) -> FastMCPInfo: FastMCPInfo dataclass containing the extracted information """ if _is_fastmcp_v1(mcp): - return await get_fastmcp_info_v1(mcp) + return await inspect_fastmcp_v1(mcp) else: - return await get_fastmcp_info_v2(mcp) + return await inspect_fastmcp_v2(mcp) diff --git a/tests/cli/test_inspect.py b/tests/cli/test_inspect.py deleted file mode 100644 index fa2f992cb..000000000 --- a/tests/cli/test_inspect.py +++ /dev/null @@ -1,354 +0,0 @@ -"""Tests for the CLI inspect command.""" - -import json -import tempfile -from pathlib import Path - -from typer.testing import CliRunner - -from fastmcp.cli.cli import app - - -class TestInspectCommand: - """Tests for the fastmcp inspect CLI command.""" - - def setup_method(self): - """Set up test fixtures.""" - self.runner = CliRunner() - - def test_inspect_basic_server(self): - """Test inspecting a basic FastMCP 2.x server.""" - # Create a temporary server file - server_content = ''' -from fastmcp import FastMCP - -mcp = FastMCP("TestServer", instructions="A test server") - -@mcp.tool -def add(a: int, b: int) -> int: - """Add two numbers.""" - return a + b - -@mcp.resource("resource://data") -def get_data() -> str: - """Get test data.""" - return "test data" - -@mcp.prompt -def test_prompt(message: str) -> list: - """Test prompt.""" - return [{"role": "user", "content": message}] -''' - - with tempfile.NamedTemporaryFile(mode="w", suffix=".py", delete=False) as f: - f.write(server_content) - server_file = f.name - - with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f: - output_file = f.name - - try: - # Run the inspect command - result = self.runner.invoke( - app, ["inspect", server_file, "-o", output_file] - ) - - assert result.exit_code == 0 - assert "✓ Inspected server: TestServer" in result.stdout - assert "Tools: 1" in result.stdout - assert "Prompts: 1" in result.stdout - assert "Resources: 1" in result.stdout - - # Check the JSON output - with open(output_file) as f: - data = json.load(f) - - assert data["name"] == "TestServer" - assert data["instructions"] == "A test server" - assert "fastmcp_version" in data - assert "mcp_version" in data - assert "server_version" in data - - # Check tools - assert len(data["tools"]) == 1 - tool = data["tools"][0] - assert tool["key"] == "add" - assert tool["name"] == "add" - assert tool["description"] == "Add two numbers." - assert "input_schema" in tool - assert tool["enabled"] is True - - # Check resources - assert len(data["resources"]) == 1 - resource = data["resources"][0] - assert resource["key"] == "resource://data" - assert resource["uri"] == "resource://data" - assert resource["name"] == "get_data" - - # Check prompts - assert len(data["prompts"]) == 1 - prompt = data["prompts"][0] - assert prompt["key"] == "test_prompt" - assert prompt["name"] == "test_prompt" - assert prompt["description"] == "Test prompt." - - # Check capabilities - assert "capabilities" in data - assert "tools" in data["capabilities"] - - finally: - # Clean up - Path(server_file).unlink(missing_ok=True) - Path(output_file).unlink(missing_ok=True) - - def test_inspect_with_object_spec(self): - """Test inspecting a server with object specification.""" - server_content = ''' -from fastmcp import FastMCP - -server = FastMCP("ObjectSpecServer") - -@server.tool -def multiply(a: int, b: int) -> int: - """Multiply two numbers.""" - return a * b -''' - - with tempfile.NamedTemporaryFile(mode="w", suffix=".py", delete=False) as f: - f.write(server_content) - server_file = f.name - - with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f: - output_file = f.name - - try: - # Run the inspect command with object specification - result = self.runner.invoke( - app, ["inspect", f"{server_file}:server", "-o", output_file] - ) - - assert result.exit_code == 0 - assert "✓ Inspected server: ObjectSpecServer" in result.stdout - - # Check the JSON output - with open(output_file) as f: - data = json.load(f) - - assert data["name"] == "ObjectSpecServer" - assert len(data["tools"]) == 1 - assert data["tools"][0]["name"] == "multiply" - - finally: - # Clean up - Path(server_file).unlink(missing_ok=True) - Path(output_file).unlink(missing_ok=True) - - def test_inspect_default_output(self): - """Test inspecting with default output filename.""" - server_content = ''' -from fastmcp import FastMCP - -mcp = FastMCP("DefaultOutputServer") - -@mcp.tool -def test_tool() -> str: - """Test tool.""" - return "test" -''' - - with tempfile.NamedTemporaryFile(mode="w", suffix=".py", delete=False) as f: - f.write(server_content) - server_file = f.name - - try: - # Run the inspect command without specifying output file - result = self.runner.invoke(app, ["inspect", server_file]) - - assert result.exit_code == 0 - assert "✓ Inspected server: DefaultOutputServer" in result.stdout - assert "Report saved to: server-info.json" in result.stdout - - # Check the default output file exists - default_output = Path("server-info.json") - assert default_output.exists() - - # Check the JSON content - with open(default_output) as f: - data = json.load(f) - - assert data["name"] == "DefaultOutputServer" - - finally: - # Clean up - Path(server_file).unlink(missing_ok=True) - Path("server-info.json").unlink(missing_ok=True) - - def test_inspect_invalid_server_file(self): - """Test inspecting a non-existent server file.""" - result = self.runner.invoke( - app, ["inspect", "nonexistent.py", "-o", "output.json"] - ) - - assert result.exit_code == 1 - # The error happens at the file parsing level, so no stdout output - - def test_inspect_server_with_error(self): - """Test inspecting a server file with syntax errors.""" - server_content = """ -from fastmcp import FastMCP - -mcp = FastMCP("ErrorServer") -# Syntax error below -@mcp.tool -def broken_tool( -""" - - with tempfile.NamedTemporaryFile(mode="w", suffix=".py", delete=False) as f: - f.write(server_content) - server_file = f.name - - try: - result = self.runner.invoke( - app, ["inspect", server_file, "-o", "output.json"] - ) - - assert result.exit_code == 1 - assert "✗ Failed to inspect server:" in result.stdout - - finally: - # Clean up - Path(server_file).unlink(missing_ok=True) - Path("output.json").unlink(missing_ok=True) - - def test_inspect_comprehensive_json_structure(self): - """Test that the JSON output has the correct structure.""" - server_content = ''' -from fastmcp import FastMCP - -mcp = FastMCP("ComprehensiveServer", instructions="Full test server") - -@mcp.tool -def calculate(x: int, y: int) -> int: - """Calculate something.""" - return x + y - -@mcp.resource("resource://static") -def static_resource() -> str: - """Static resource.""" - return "static" - -@mcp.resource("resource://template/{id}") -def template_resource(id: str) -> str: - """Template resource.""" - return f"data-{id}" - -@mcp.prompt -def analysis_prompt(data: str) -> list: - """Analysis prompt.""" - return [{"role": "user", "content": f"Analyze: {data}"}] -''' - - with tempfile.NamedTemporaryFile(mode="w", suffix=".py", delete=False) as f: - f.write(server_content) - server_file = f.name - - with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f: - output_file = f.name - - try: - result = self.runner.invoke( - app, ["inspect", server_file, "-o", output_file] - ) - - assert result.exit_code == 0 - - # Load and validate JSON structure - with open(output_file) as f: - data = json.load(f) - - # Check top-level structure - required_fields = [ - "name", - "instructions", - "fastmcp_version", - "mcp_version", - "server_version", - "tools", - "prompts", - "resources", - "templates", - "capabilities", - ] - for field in required_fields: - assert field in data, f"Missing field: {field}" - - # Check version fields are strings - assert isinstance(data["fastmcp_version"], str) - assert isinstance(data["mcp_version"], str) - assert isinstance(data["server_version"], str) - - # Check that we have the expected components - assert len(data["tools"]) == 1 - assert len(data["resources"]) == 1 - assert len(data["templates"]) == 1 - assert len(data["prompts"]) == 1 - - # Check tool structure - tool = data["tools"][0] - tool_fields = [ - "key", - "name", - "description", - "input_schema", - "annotations", - "tags", - "enabled", - ] - for field in tool_fields: - assert field in tool, f"Missing tool field: {field}" - - # Check resource structure - resource = data["resources"][0] - resource_fields = [ - "key", - "uri", - "name", - "description", - "mime_type", - "tags", - "enabled", - ] - for field in resource_fields: - assert field in resource, f"Missing resource field: {field}" - - # Check template structure - template = data["templates"][0] - template_fields = [ - "key", - "uri_template", - "name", - "description", - "mime_type", - "tags", - "enabled", - ] - for field in template_fields: - assert field in template, f"Missing template field: {field}" - - # Check prompt structure - prompt = data["prompts"][0] - prompt_fields = [ - "key", - "name", - "description", - "arguments", - "tags", - "enabled", - ] - for field in prompt_fields: - assert field in prompt, f"Missing prompt field: {field}" - - finally: - # Clean up - Path(server_file).unlink(missing_ok=True) - Path(output_file).unlink(missing_ok=True) diff --git a/tests/utilities/test_inspect.py b/tests/utilities/test_inspect.py index 0ca03b6c2..723738242 100644 --- a/tests/utilities/test_inspect.py +++ b/tests/utilities/test_inspect.py @@ -9,8 +9,8 @@ from fastmcp.utilities.inspect import ( FastMCPInfo, ToolInfo, _is_fastmcp_v1, - get_fastmcp_info, - get_fastmcp_info_v1, + inspect_fastmcp, + inspect_fastmcp_v1, ) @@ -69,7 +69,7 @@ class TestGetFastMCPInfo: """Test get_fastmcp_info with an empty server.""" mcp = FastMCP("EmptyServer", instructions="Empty server for testing") - info = await get_fastmcp_info(mcp) + info = await inspect_fastmcp(mcp) assert info.name == "EmptyServer" assert info.instructions == "Empty server for testing" @@ -97,7 +97,7 @@ class TestGetFastMCPInfo: def greet(name: str) -> str: return f"Hello, {name}!" - info = await get_fastmcp_info(mcp) + info = await inspect_fastmcp(mcp) assert info.name == "ToolServer" assert len(info.tools) == 2 @@ -117,7 +117,7 @@ class TestGetFastMCPInfo: def get_dynamic_data(param: str) -> str: return f"Dynamic data: {param}" - info = await get_fastmcp_info(mcp) + info = await inspect_fastmcp(mcp) assert info.name == "ResourceServer" assert len(info.resources) == 1 # Static resource @@ -139,7 +139,7 @@ class TestGetFastMCPInfo: def custom_analysis(text: str) -> list: return [{"role": "user", "content": f"Custom: {text}"}] - info = await get_fastmcp_info(mcp) + info = await inspect_fastmcp(mcp) assert info.name == "PromptServer" assert len(info.prompts) == 2 @@ -171,7 +171,7 @@ class TestGetFastMCPInfo: def analyze(content: str) -> list: return [{"role": "user", "content": content}] - info = await get_fastmcp_info(mcp) + info = await inspect_fastmcp(mcp) assert info.name == "ComprehensiveServer" assert info.instructions == "A server with everything" @@ -204,7 +204,7 @@ class TestGetFastMCPInfo: """Test get_fastmcp_info with a server that has no instructions.""" mcp = FastMCP("NoInstructionsServer") - info = await get_fastmcp_info(mcp) + info = await inspect_fastmcp(mcp) assert info.name == "NoInstructionsServer" assert info.instructions is None @@ -226,7 +226,7 @@ class TestGetFastMCPInfo: return [{"role": "user", "content": "test"}] # Get info using our function - info = await get_fastmcp_info(mcp) + info = await inspect_fastmcp(mcp) # Verify using client async with Client(mcp) as client: @@ -258,7 +258,7 @@ class TestFastMCP1xCompatibility: """Test get_fastmcp_info_v1 with an empty FastMCP1x server.""" mcp = FastMCP1x("Test1x") - info = await get_fastmcp_info_v1(mcp) + info = await inspect_fastmcp_v1(mcp) assert info.name == "Test1x" assert info.instructions is None @@ -283,7 +283,7 @@ class TestFastMCP1xCompatibility: def greet(name: str) -> str: return f"Hello, {name}!" - info = await get_fastmcp_info_v1(mcp) + info = await inspect_fastmcp_v1(mcp) assert info.name == "Test1x" assert len(info.tools) == 2 @@ -299,7 +299,7 @@ class TestFastMCP1xCompatibility: def get_data() -> str: return "Some data" - info = await get_fastmcp_info_v1(mcp) + info = await inspect_fastmcp_v1(mcp) assert info.name == "Test1x" assert len(info.resources) == 1 @@ -315,7 +315,7 @@ class TestFastMCP1xCompatibility: def analyze_data(data: str) -> list: return [{"role": "user", "content": f"Analyze: {data}"}] - info = await get_fastmcp_info_v1(mcp) + info = await inspect_fastmcp_v1(mcp) assert info.name == "Test1x" assert len(info.prompts) == 1 @@ -330,7 +330,7 @@ class TestFastMCP1xCompatibility: def test_tool() -> str: return "test" - info = await get_fastmcp_info(mcp) + info = await inspect_fastmcp(mcp) assert info.name == "Test1x" assert len(info.tools) == 1 @@ -346,7 +346,7 @@ class TestFastMCP1xCompatibility: def test_tool() -> str: return "test" - info = await get_fastmcp_info(mcp) + info = await inspect_fastmcp(mcp) assert info.name == "Test2x" assert len(info.tools) == 1 @@ -366,8 +366,8 @@ class TestFastMCP1xCompatibility: def tool2x() -> str: return "2x" - info1x = await get_fastmcp_info(mcp1x) - info2x = await get_fastmcp_info(mcp2x) + info1x = await inspect_fastmcp(mcp1x) + info2x = await inspect_fastmcp(mcp2x) assert info1x.name == "Test1x" assert info2x.name == "Test2x" From 84d5750f6d241214cf2c5ae56177c5f734f4cae5 Mon Sep 17 00:00:00 2001 From: Jeremiah Lowin <153965+jlowin@users.noreply.github.com> Date: Sat, 21 Jun 2025 20:08:29 -0400 Subject: [PATCH 4/4] Update cli.mdx Co-Authored-By: Claude --- docs/patterns/cli.mdx | 24 ++++++++++++++++++++++++ 1 file changed, 24 insertions(+) diff --git a/docs/patterns/cli.mdx b/docs/patterns/cli.mdx index 19e1ad1ab..84654975b 100644 --- a/docs/patterns/cli.mdx +++ b/docs/patterns/cli.mdx @@ -21,6 +21,7 @@ fastmcp --help | `run` | Run a FastMCP server directly | Uses your current environment; you are responsible for ensuring all dependencies are available | | `dev` | Run a server with the MCP Inspector for testing | Creates an isolated environment; dependencies must be explicitly specified with `--with` and/or `--with-editable` | | `install` | Install a server in the Claude desktop app | Creates an isolated environment; dependencies must be explicitly specified with `--with` and/or `--with-editable` | +| `inspect` | Generate a JSON report about a FastMCP server | Uses your current environment; you are responsible for ensuring all dependencies are available | | `version` | Display version information | N/A | ## Command Details @@ -179,6 +180,29 @@ fastmcp install server.py:my_server fastmcp install server.py:my_server -n "My Analysis Server" --with pandas ``` +### `inspect` + + + +Generate a detailed JSON report about a FastMCP server, including information about its tools, prompts, resources, and capabilities. + +```bash +fastmcp inspect server.py +``` + +The command supports the same server specification format as `run` and `install`: + +```bash +# Auto-detect server object +fastmcp inspect server.py + +# Specify server object +fastmcp inspect server.py:my_server + +# Custom output location +fastmcp inspect server.py --output analysis.json +``` + ### `version` Display version information about FastMCP and related components.