ceos/fastmcp
1
0
Fork 0
mirror of https://github.com/PrefectHQ/fastmcp.git synced 2026-08-21 13:04:18 +02:00
Code Issues Projects Releases Packages Wiki Activity

Add fastmcp inspect command with detailed server analysis

Browse source 
- 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>
This commit is contained in:
Jeremiah Lowin 2025-06-21 16:40:54 -04:00
commit 6a3a3077b7
4 changed files with 1175 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

107
src/fastmcp/cli/cli.py
View file

@ -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)

326
src/fastmcp/utilities/inspect.py Normal file
View file

@ -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)

354
tests/cli/test_inspect.py Normal file
View file

@ -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)

388
tests/utilities/test_inspect.py Normal file
View file

@ -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