mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-22 21:44:18 +02:00
Merge pull request #904 from jlowin/feature/fastmcp-inspect-command
This commit is contained in:
Jeremiah Lowin
GitHub
commit
3933f4f96c
4 changed files with 836 additions and 0 deletions
|
|
@ -21,6 +21,7 @@ fastmcp --help
|
|||
| `run` | Run a FastMCP server directly | Uses your current environment; you are responsible for ensuring all dependencies are available |
|
||||
| `dev` | Run a server with the MCP Inspector for testing | Creates an isolated environment; dependencies must be explicitly specified with `--with` and/or `--with-editable` |
|
||||
| `install` | Install a server in the Claude desktop app | Creates an isolated environment; dependencies must be explicitly specified with `--with` and/or `--with-editable` |
|
||||
| `inspect` | Generate a JSON report about a FastMCP server | Uses your current environment; you are responsible for ensuring all dependencies are available |
|
||||
| `version` | Display version information | N/A |
|
||||
|
||||
## Command Details
|
||||
|
|
@ -179,6 +180,29 @@ fastmcp install server.py:my_server
|
|||
fastmcp install server.py:my_server -n "My Analysis Server" --with pandas
|
||||
```
|
||||
|
||||
### `inspect`
|
||||
|
||||
<VersionBadge version="2.9.0" />
|
||||
|
||||
Generate a detailed JSON report about a FastMCP server, including information about its tools, prompts, resources, and capabilities.
|
||||
|
||||
```bash
|
||||
fastmcp inspect server.py
|
||||
```
|
||||
|
||||
The command supports the same server specification format as `run` and `install`:
|
||||
|
||||
```bash
|
||||
# Auto-detect server object
|
||||
fastmcp inspect server.py
|
||||
|
||||
# Specify server object
|
||||
fastmcp inspect server.py:my_server
|
||||
|
||||
# Custom output location
|
||||
fastmcp inspect server.py --output analysis.json
|
||||
```
|
||||
|
||||
### `version`
|
||||
|
||||
Display version information about FastMCP and related components.
|
||||
|
|
|
|||
|
|
@ -1,5 +1,6 @@
|
|||
"""FastMCP CLI tools."""
|
||||
|
||||
import asyncio
|
||||
import importlib.metadata
|
||||
import importlib.util
|
||||
import os
|
||||
|
|
@ -11,6 +12,7 @@ from typing import Annotated
|
|||
|
||||
import dotenv
|
||||
import typer
|
||||
from pydantic import TypeAdapter
|
||||
from rich.console import Console
|
||||
from rich.table import Table
|
||||
from typer import Context, Exit
|
||||
|
|
@ -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 FastMCPInfo, inspect_fastmcp
|
||||
from fastmcp.utilities.logging import get_logger
|
||||
|
||||
logger = get_logger("cli")
|
||||
|
|
@ -435,3 +438,98 @@ 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 inspect_fastmcp(server)
|
||||
|
||||
try:
|
||||
# Try to use existing event loop if available
|
||||
asyncio.get_running_loop()
|
||||
# If there's already a loop running, we need to run in a thread
|
||||
import concurrent.futures
|
||||
|
||||
with concurrent.futures.ThreadPoolExecutor() as executor:
|
||||
future = executor.submit(asyncio.run, get_info())
|
||||
info = future.result()
|
||||
except RuntimeError:
|
||||
# No running loop, safe to use asyncio.run
|
||||
info = asyncio.run(get_info())
|
||||
|
||||
info_json = TypeAdapter(FastMCPInfo).dump_json(info, indent=2)
|
||||
|
||||
# Ensure output directory exists
|
||||
output.parent.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# Write JSON report (always pretty-printed)
|
||||
with output.open("w", encoding="utf-8") as f:
|
||||
f.write(info_json.decode("utf-8"))
|
||||
|
||||
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
326
src/fastmcp/utilities/inspect.py
Normal 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 inspect_fastmcp_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 inspect_fastmcp_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 inspect_fastmcp(mcp: FastMCP[Any] | Any) -> FastMCPInfo:
|
||||
"""Extract information from a FastMCP instance into a dataclass.
|
||||
|
||||
This function automatically detects whether the instance is FastMCP v1.x or v2.x
|
||||
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 inspect_fastmcp_v1(mcp)
|
||||
else:
|
||||
return await inspect_fastmcp_v2(mcp)
|
||||
388
tests/utilities/test_inspect.py
Normal file
388
tests/utilities/test_inspect.py
Normal 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,
|
||||
inspect_fastmcp,
|
||||
inspect_fastmcp_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 inspect_fastmcp(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 inspect_fastmcp(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 inspect_fastmcp(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 inspect_fastmcp(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 inspect_fastmcp(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 inspect_fastmcp(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 inspect_fastmcp(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 inspect_fastmcp_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 inspect_fastmcp_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 inspect_fastmcp_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 inspect_fastmcp_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 inspect_fastmcp(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 inspect_fastmcp(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 inspect_fastmcp(mcp1x)
|
||||
info2x = await inspect_fastmcp(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
|
||||
Loading…
Add table
Add a link
Reference in a new issue