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