Add fastmcp inspect command with detailed server analysis

- 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 <noreply@anthropic.com>

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)

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)

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)

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